Python Module API

Python Module API

Syntalos provides a Python API to easily build new modules. Python modules do not run within the Syntalos process, and instead communicate with the main application via an interface provided by the syntalos_mlink Python module. This API can be used from either the Python Script module, or by standalone modules that are written in Python entirely.

The Python interface is documented below.

syntalos_mlink

Syntalos Python Module Interface

def list_types_synonyms() -> Any:

list_types_synonyms() -> object

class SyntalosPyError(builtins.Exception):

Common base class for all non-exit exceptions.

class Frame(pybind11_builtins.pybind11_object):

A video frame.

Frame()

__init__(self: syntalos_mlink.Frame) -> None

index: int

Number of the frame.

time: datetime.timedelta

Time when the frame was recorded.

mat: numpy.ndarray

Frame image data.

time_usec: int

Time when the frame was recorded, as integer in µs.

class ControlCommandKind(pybind11_builtins.pybind11_object):

Members:

UNKNOWN

START

PAUSE

STOP

STEP

CUSTOM

ControlCommandKind(value: <class 'SupportsInt'>)

__init__(self: syntalos_mlink.ControlCommandKind, value: typing.SupportsInt) -> None

name: str

name(self: object, /) -> str

value: int
class ControlCommand(pybind11_builtins.pybind11_object):
ControlCommand(**kwds)

__init__(args, *kwargs) Overloaded function.

  1. __init__(self: syntalos_mlink.ControlCommand) -> None

  2. __init__(self: syntalos_mlink.ControlCommand, arg0: syntalos_mlink.ControlCommandKind) -> None

duration: int
command: str
class FirmataCommandKind(pybind11_builtins.pybind11_object):

Members:

UNKNOWN

NEW_DIG_PIN

NEW_ANA_PIN

IO_MODE

WRITE_ANALOG

WRITE_DIGITAL

WRITE_DIGITAL_PULSE

SYSEX

FirmataCommandKind(value: <class 'SupportsInt'>)

__init__(self: syntalos_mlink.FirmataCommandKind, value: typing.SupportsInt) -> None

name: str

name(self: object, /) -> str

value: int
WRITE_DIGITAL_PULSE: ClassVar[FirmataCommandKind] = <FirmataCommandKind.WRITE_DIGITAL_PULSE: 6>
class FirmataControl(pybind11_builtins.pybind11_object):
FirmataControl(**kwds)

__init__(args, *kwargs) Overloaded function.

  1. __init__(self: syntalos_mlink.FirmataControl) -> None

  2. __init__(self: syntalos_mlink.FirmataControl, arg0: syntalos_mlink.FirmataCommandKind) -> None

pin_id: int
pin_name: str
is_output: bool
is_pullup: bool
value: int
class FirmataData(pybind11_builtins.pybind11_object):
FirmataData()

__init__(self: syntalos_mlink.FirmataData) -> None

pin_id: int
pin_name: str
value: int
is_digital: bool
time: datetime.timedelta

Time when the data was acquired.

time_usec: int

Time when the data was acquired, as integer in µs.

class IntSignalBlock(pybind11_builtins.pybind11_object):

A block of timestamped integer signal data.

IntSignalBlock()

__init__(self: syntalos_mlink.IntSignalBlock) -> None

timestamps: Annotated[numpy.ndarray[tuple[Any, ...], numpy.dtype[numpy.uint64]], '[m, 1]']

Timestamps of the data blocks.

data: Annotated[numpy.ndarray[tuple[Any, ...], numpy.dtype[numpy.int32]], '[m, n]']

The data matrix.

length: int
rows: int
cols: int
class FloatSignalBlock(pybind11_builtins.pybind11_object):

A block of timestamped float signal data.

FloatSignalBlock()

__init__(self: syntalos_mlink.FloatSignalBlock) -> None

timestamps: Annotated[numpy.ndarray[tuple[Any, ...], numpy.dtype[numpy.uint64]], '[m, 1]']

Timestamps of the data blocks.

data: Annotated[numpy.ndarray[tuple[Any, ...], numpy.dtype[numpy.float64]], '[m, n]']

The data matrix.

length: int
rows: int
cols: int
class InputPort(pybind11_builtins.pybind11_object):

Representation of a module input port.

on_data: Callable[[typing.Any], None]

Set function to be called when new data arrives.

metadata: dict[str, typing.Any]

Obtain the metadata associated with this input port.

def set_throttle_items_per_sec(self, items_per_sec: <class 'SupportsInt'>) -> None:

Limit the amount of input received to a set amount of elements per second.

name: str
class OutputPort(pybind11_builtins.pybind11_object):

Representation of a module output port.

def submit(self, arg0: Any) -> None:

Submit the given entity to the output port for transfer to its destination(s).

name: str
def set_metadata_value(self, arg0: str, arg1: Any) -> None:

Set (immutable) metadata value for this port.

def set_metadata_value_size(self, arg0: str, arg1: Any) -> None:

Set (immutable) metadata value for a 2D size type for this port.

def firmata_register_digital_pin( self, pin_id: <class 'SupportsInt'>, name: str, is_output: bool, is_pullup: bool = False) -> FirmataControl:

Convenience function to create a command to register a named digital pin and immediately submit it on this port. The registered pin can later be referred to by its name.

def firmata_submit_digital_value(self, name: str, value: bool) -> FirmataControl:

Convenience function to write a digital value to a named pin.

def firmata_submit_digital_pulse( self, name: str, duration_msec: <class 'SupportsInt'> = 50) -> FirmataControl:

Convenience function to emit a digital pulse on a named pin.

def println(text: str) -> None:

Print text to stdout.

def raise_error(message: str) -> None:

Emit an error message string, immediately terminating the current action and (if applicable) the experiment.

def time_since_start_msec() -> int:

Get time since experiment started in milliseconds.

def time_since_start_usec() -> int:

Get time since experiment started in microseconds.

def wait(msec: <class 'SupportsInt'>) -> None:

Wait (roughly) for the given amount of milliseconds without blocking communication with the master process.

def wait_sec(sec: <class 'SupportsInt'>) -> None:

Wait (roughly) for the given amount of seconds without blocking communication with the master process.

def await_data(timeout_usec: <class 'SupportsInt'> = -1) -> None:

Wait for new data to arrive and call selected callbacks. Also keep communication with the Syntalos master process.

def is_running() -> bool:

Return True if the experiment is still running, False if we are supposed to shut down.

def schedule_delayed_call( delay_msec: <class 'SupportsInt'>, callable_fn: Callable[[], None]) -> None:

Schedule call to a callable to be processed after a set amount of milliseconds.

def get_input_port(id: str) -> Any:

Get reference to input port with the give ID.

def get_output_port(id: str) -> Any:

Get reference to output port with the give ID.

def call_on_show_settings(callable_fn: Callable[[bytearray], None]) -> None:

Call the given function when the module's settings dialog should be shown. The callback receives settings as bytes.

def call_on_show_display(callable_fn: Callable[[], None]) -> None:

Call the given function when the module's display window should be shown.

def save_settings(settings_data: bytearray) -> None:

Send module settings data as bytes to Syntalos for safekeeping.

def new_firmatactl_with_id_name( kind: FirmataCommandKind, pin_id: <class 'SupportsInt'>, name: str) -> FirmataControl:

Create new Firmata control command with a given pin ID and registered name.

def new_firmatactl_with_id( kind: FirmataCommandKind, pin_id: <class 'SupportsInt'>) -> FirmataControl:

Create new Firmata control command with a given pin ID.

def new_firmatactl_with_name( kind: FirmataCommandKind, name: str) -> FirmataControl:

Create new Firmata control command with a given pin name (the name needs to be registered previously).