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

class ControlCommand:

A control command for a module.

ControlCommand(**kwds)

Helper for @overload to raise when called.

command: str

Custom command string (used when kind is CUSTOM).

duration: int

Optional duration associated with this command, in milliseconds.

kind: ControlCommandKind

The ControlCommandKind of this command.

class ControlCommandKind:

The type of a control command sent to controllable modules.

Members:

UNKNOWN

START : Start an operation.

PAUSE : Pause an operation; can be resumed with START.

STOP : Stop an operation.

STEP : Advance operation by one step.

CUSTOM : Custom command.

ControlCommandKind(value: <class 'SupportsInt'>)

CUSTOM: ClassVar[ControlCommandKind]

PAUSE: ClassVar[ControlCommandKind]

START: ClassVar[ControlCommandKind]

STEP: ClassVar[ControlCommandKind]

STOP: ClassVar[ControlCommandKind]

UNKNOWN: ClassVar[ControlCommandKind]

value: int

class DataType:

Identifies the data type of a port / stream.

Members:

ControlCommand : Module control command.

TableRow : A row of tabular data.

Frame : A video frame.

LineCommand : Outbound command to a hardware signal line.

LineReading : Timestamped reading from a hardware signal line.

SignalBlockI32 : A block of 32-bit integer samples.

SignalBlockU16 : A block of 16-bit unsigned integer samples.

SignalBlockF32 : A block of 32-bit float samples.

DataType(value: <class 'SupportsInt'>)

ControlCommand: ClassVar[DataType]

Frame: ClassVar[DataType]

LineCommand: ClassVar[DataType]

LineReading: ClassVar[DataType]

SignalBlockF32: ClassVar[DataType]

SignalBlockI32: ClassVar[DataType]

SignalBlockU16: ClassVar[DataType]

TableRow: ClassVar[DataType]

value: int

class EdlDataset:

An EDL dataset that holds data files for a modality.

def add_aux_file( self, file_name: str, key: str = '', summary: str = '', scan: bool = False) -> pathlib._local.Path:

Returns the path where the data should be written (pathlib.Path).

def path_for_basename(self, basename: str) -> pathlib._local.Path:

Return the path for a data file with the given basename (pathlib.Path).

The file is not registered. Use a scan pattern to include it in the manifest.

def set_attribute(self, key: str, value: Any) -> None:

Set an attribute on this dataset.

def set_data_file( self, file_name: str, summary: str = '', scan: bool = False) -> pathlib._local.Path:

Returns the path where the data should be written (pathlib.Path).

Parameters

scan: If True, treat file_name as a glob pattern scanned at save time.

is_empty: bool

True if no data file has been registered yet.

Name of this dataset (str).

path: pathlib._local.Path

Filesystem path of this dataset's directory (pathlib.Path).

class EdlGroup:

An EDL group that can contain datasets and sub-groups.

def create_dataset(self, name: str) -> EdlDataset:

Create a dataset with the given name (MUST_CREATE semantics).

def create_group(self, name: str) -> EdlGroup:

Create (or open) a sub-group with the given name.

def set_attribute(self, key: str, value: Any) -> None:

Set an attribute on this group.

Name of this group (str).

path: pathlib._local.Path

Filesystem path of this group (pathlib.Path).

class Frame:

A video frame.

index: int

Number of the frame.

mat: numpy.ndarray

Frame image data as a NumPy array (OpenCV Mat).

time: datetime.timedelta

Time when the frame was recorded, as a duration.

time_usec: int

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

class HwInputLine:

Handle for a hardware input line. Use this to configure an input on the downstream device; subsequent readings arrive as LineReading messages on the consuming module's input port, matched by line_id.

HwInputLine( port: OutputPort, line_id: <class 'SupportsInt'>, pullup: bool = False)

Construct an input-line handle.

Parameters

port: The OutputPort carrying LineCommand messages.
line_id: Hardware line / channel / pin number.
pullup: True to enable the input pull-up resistor (default: False).

def send_mode(self) -> LineCommand:

Emit the SET_MODE command that configures the line as an input.

Call this at the start of every run.

Returns

The LineCommand that was submitted.

line_id: int

The hardware line ID this handle refers to.

class HwOutputLine:

Handle for a hardware output line (Firmata pin, DAQ channel, TTL line, ...).

Construction captures the port and line identity but emits no command. Call send_mode() at the start of every run to configure the line on the downstream device.

HwOutputLine( port: OutputPort, line_id: <class 'SupportsInt'>, analog: bool = False)

Construct an output-line handle.

Parameters

port: The OutputPort carrying LineCommand messages.
line_id: Hardware line / channel / pin number.
analog: True if this is an analog DAC output, False for digital (default).

def pulse( self, duration_usec: <class 'SupportsInt'> = 2500) -> LineCommand:

Emit a digital pulse on the line (digital lines only).

Parameters

duration_msec: Pulse duration in microseconds (default: 2500).

Raises

SyntalosPyError: If this is an analog line.

Returns

The LineCommand that was submitted.

def pulse_msec( self, duration_msec: <class 'SupportsInt'> = 50) -> LineCommand:

Emit a digital pulse on the line (digital lines only).

Parameters

duration_msec: Pulse duration in milliseconds (default: 50).

Raises

SyntalosPyError: If this is an analog line.

Returns

The LineCommand that was submitted.

def send_mode(self) -> LineCommand:

Emit the SET_MODE command that configures the line as an output.

Call this at the start of every run.

Returns

The LineCommand that was submitted.

def set_analog_value(self, value: <class 'SupportsInt'>) -> LineCommand:

Write an analog DAC value to the line (analog lines only).

Parameters

value: DAC code (interpretation is device-specific).

Raises

SyntalosPyError: If this is a digital line.

Returns

The LineCommand that was submitted.

def set_value(self, value: bool) -> LineCommand:

Write a digital value to the line (digital lines only).

Parameters

value: True / 1 for HIGH, False / 0 for LOW.

Raises

SyntalosPyError: If this is an analog line.

Returns

The LineCommand that was submitted.

line_id: int

The hardware line ID this handle refers to.

class InputPort:

A module input port.

Obtain an instance via get_input_port().

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

Limit the number of items delivered to on_data per second.

Parameters

items_per_sec: Maximum items per second; 0 disables throttling.

metadata: dict[str, typing.Any]

Read-only dict[str, object] of metadata provided by the upstream module for this port.

Values are native Python types: int, float, str, bool, None, MetaSize, or list / dict for nested structures.

The unique port ID string.

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

Callback invoked with each incoming data item.

Assign a callable that accepts a single argument of the port's data type (e.g. Frame, TableRow). Set to None to remove the callback.

Type: Callable[[object], None] | None.

class LineCommand:

Command issued to a hardware signal line.

LineCommand(**kwds)

Helper for @overload to raise when called.

duration_usec: int

Pulse duration for *_PULSE kinds, in microseconds.

extra: bytes

Optional device-specific payload (used by DEVICE_SPECIFIC).

flags: LineModeFlags

Mode flags (used by LineCommandKind.SET_MODE).

kind: LineCommandKind

The LineCommandKind to execute.

line_id: int

Hardware line / channel / pin number.

value: int

Digital: 0/1; analog: DAC code.

class LineCommandKind:

Type of operation requested on a hardware signal line.

Members:

UNKNOWN

SET_MODE : Configure a line's direction / mode (uses flags).

WRITE_DIGITAL : Set a digital line to a value (True/False).

WRITE_ANALOG : Set an analog line to a value.

WRITE_DIGITAL_PULSE : Pulse a digital line for duration.

WRITE_ANALOG_PULSE : Drive an analog line for duration.

DEVICE_SPECIFIC : Device-defined payload carried in extra.

LineCommandKind(value: <class 'SupportsInt'>)

DEVICE_SPECIFIC: ClassVar[LineCommandKind]

SET_MODE: ClassVar[LineCommandKind]

UNKNOWN: ClassVar[LineCommandKind]

WRITE_ANALOG: ClassVar[LineCommandKind]

WRITE_ANALOG_PULSE: ClassVar[LineCommandKind]

WRITE_DIGITAL: ClassVar[LineCommandKind]

WRITE_DIGITAL_PULSE: ClassVar[LineCommandKind]

value: int

class LineModeFlags:

Composable mode flags for LineCommandKind.SET_MODE.

Members:

NONE

IS_INPUT : Line is an input.

IS_OUTPUT : Line is an output.

PULL_UP : Input pull-up resistor enabled (inputs only).

LineModeFlags(value: <class 'SupportsInt'>)

IS_INPUT: ClassVar[LineModeFlags]

IS_OUTPUT: ClassVar[LineModeFlags]

NONE: ClassVar[LineModeFlags]

PULL_UP: ClassVar[LineModeFlags]

value: int

class LineReading:

Timestamped scalar reading from a hardware signal line.

line_id: int

Hardware line / channel / pin number.

time: datetime.timedelta

Time when the reading was acquired, as a duration.

time_usec: int

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

value: int

Sampled value.

class MetaSize:

Two-dimensional size value used in stream metadata.

MetaSize(**kwds)

Helper for @overload to raise when called.

height: int

Height in pixels (or other integer units).

width: int

Width in pixels (or other integer units).

class OutputPort:

A module output port.

Obtain an instance via get_output_port().

def set_metadata_value(self, key: str, value: Any) -> None:

Set a metadata entry for this port.

Metadata must be set before the run starts (i.e. in prepare()); it is immutable once acquisition begins.

Parameters

key: Metadata key string.
value: Metadata value. Accepted types: int, float, str, MetaSize, or list of int/str values.

Raises

SyntalosPyError: If the value type is not supported.

def set_metadata_value_size(self, key: str, value: Any) -> None:

Set a 2-D size metadata entry for this port (e.g. 'size' for video streams).

Parameters

key: Metadata key string.
value: Either a MetaSize object or a sequence of exactly two integers [width, height].

Raises

SyntalosPyError: If value does not have exactly two elements.

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

Send a data item to all modules connected to this port.

Parameters

data: Data item matching this port's type (e.g. Frame, TableRow).

Raises

SyntalosPyError: If the item type does not match the port's declared data type.

The unique port ID string.

class SecondaryClockSynchronizer:

Synchronizer for an external steady monotonic clock.

Obtain an instance via SyntalosLink.init_clock_synchronizer().

def process_timestamp( self, master_timestamp_usec: <class 'SupportsInt'>, secondary_acq_timestamp_usec: <class 'SupportsInt'>) -> int:

Process a master/secondary timestamp pair.

Returns the (possibly adjusted) master timestamp, in microseconds.

def set_calibration_points_count(self, count: <class 'SupportsInt'>) -> None:

Set the number of calibration points required to compute the initial offset.

def set_expected_clock_frequency_hz(self, frequency: <class 'SupportsFloat'>) -> None:

Auto-tune tolerance and required calibration points based on the expected DAQ frequency.

def set_strategies(self, strategies: <class 'SupportsInt'>) -> None:

Set the bitset of TimeSyncStrategy values the synchronizer may apply.

def set_tolerance_usec(self, tolerance_usec: <class 'SupportsInt'>) -> None:

Set the tolerance for offset deviations, in microseconds.

def start(self) -> bool:

Initialize and begin synchronization. Returns True on success.

def stop(self) -> None:

Finalize and close the time-sync file (if any).

clock_correction_offset_usec: int

The current dynamic clock-correction offset, in microseconds.

Negative values indicate the secondary clock is running too slow, positive values indicate it is running too fast.

expected_offset_to_master_usec: int

The expected offset from the secondary clock to the master clock, in microseconds.

is_calibrated: bool

Whether the synchronizer has completed its calibration phase.

class SignalBlockF32:

A block of timestamped 32-bit float signal data.

cols: int

Number of columns (channels).

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

2-D data matrix: rows = samples, columns = channels.

length: int

Number of samples (rows) in this block.

rows: int

Number of rows (samples).

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

1-D array of sample timestamps in µs.

class SignalBlockI32:

A block of timestamped 32-bit integer signal data.

cols: int

Number of columns (channels).

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

2-D data matrix: rows = samples, columns = channels.

length: int

Number of samples (rows) in this block.

rows: int

Number of rows (samples).

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

1-D array of sample timestamps in µs.

class SignalBlockU16:

A block of timestamped 16-bit unsigned integer signal data.

cols: int

Number of columns (channels).

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

2-D data matrix: rows = samples, columns = channels.

length: int

Number of samples (rows) in this block.

rows: int

Number of rows (samples).

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

1-D array of sample timestamps in µs.

class SyntalosLink:

Manages the connection to Syntalos.

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

Single-shot data poll for use inside a custom event loop.

Parameters

timeout_usec: Maximum time to block in µs; -1 waits indefinitely.

def await_data_forever(self, event_fn: Callable[[], None] = None) -> None:

Signal initialization complete and run the module event loop.

Signals IDLE to Syntalos then blocks until a shutdown is requested.

Parameters

event_fn: Optional zero-argument callable, invoked regularly to pump an external event loop.

def create_dataset_in_group( self, group: EdlGroup, name: str) -> EdlDataset:

Create a new dataset inside the given group with MUST_CREATE semantics.

Parameters

group: Parent EdlGroup.
name: Dataset name.

Returns

The new EdlDataset.

def create_default_dataset(self, preferred_name: str = '') -> EdlDataset:

Create the default dataset for this module under the assigned EDL root group.

Uses MUST_CREATE semantics: fails if a dataset with the same name already exists. Must be called inside on_prepare.

Parameters

preferred_name: Dataset name; defaults to the module's group name.

Returns

The new EdlDataset.

def create_storage_group(self, name: str) -> EdlGroup:

Create (or open) a sub-group under the module's assigned EDL root group.

Parameters

name: Group name.

Returns

The EdlGroup.

def init_clock_synchronizer( self, expected_frequency_hz: <class 'SupportsFloat'> = 0.0, id: str = '') -> SecondaryClockSynchronizer:

Create a synchronizer for an external steady monotonic clock.

Must be called while the module is preparing or running (typically inside on_prepare).

Parameters

expected_frequency_hz: Optional expected acquisition frequency in Hz; if greater than zero it auto-tunes the calibration point count and tolerance.
id: Optional synchronizer id; defaults to the module instance id.

Returns

A new SecondaryClockSynchronizer.

def register_input_port( self, id: str, title: str, data_type: DataType) -> InputPort:

Declare a new input port for this module.

Must be called at module level so that Syntalos can discover the port topology and restore project connections before the first run is prepared.

Parameters

id: Unique port identifier used in get_input_port().
title: Human-readable port label shown in the flow-graph editor.
data_type: DataType value, e.g. DataType.Frame, DataType.TableRow.

Returns

An InputPort handle.

def register_output_port( self, id: str, title: str, data_type: DataType) -> OutputPort:

Declare a new output port for this module.

Must be called at module level (top-level script code, not inside a function) so that Syntalos can discover the port topology and restore project connections before the first run is prepared.

Parameters

id: Unique port identifier used in get_output_port().
title: Human-readable port label shown in the flow-graph editor.
data_type: DataType value, e.g. DataType.Frame, DataType.TableRow.

Returns

An OutputPort handle.

def signal_idle(self) -> None:

Signal IDLE (initialization complete) without entering the built-in event loop.

Call this before starting a custom loop with await_data().

allow_async_start: bool

Whether start() is run in parallel with other modules for this module.

On startup, Syntalos can send START to all external modules in parallel,not waiting for them to finish individual START actions. This is the async-start method, and it results in much better aligned start times of modules.However, some modules may need to modify stream metadata in START and must ensure it is receivedbefore their streams are started. Those modules may want to disable async-start for themselves, so they run exclusively and the enginewaits for them to complete START actions.

is_running: bool

Check whether the experiment is still active.

Returns

True while the run is in progress, False once a stop has been requested.

on_load_settings: Callable[[bytes | bytearray, os.PathLike | str | bytes], bool]

The callback receives the loaded settings payload first (bytes / bytearray), followed by the base project directory as pathlib.Path. The callback must return True to signal that the settings were loaded successfully, or False / use raise_error to signal a loading failure. If not registered, incoming settings are ignored.

Type: Callable[[bytes | bytearray, pathlib.Path], bool] | None.

on_prepare: Callable[[], bool]

The callback must return True to signal that preparation succeeded, or False / raise an exception to abort. If not registered the module automatically signals readiness.

Type: Callable[[], bool] | None.

on_save_settings: Callable[[os.PathLike | str | bytes], bytes]

The callback receives the base project directory as a pathlib.Path and must return a bytes-like payload (bytes or bytearray). If not registered, no module settings are saved.

Type: Callable[[pathlib.Path], bytes | bytearray] | None.

on_show_display: Callable[[], None]

Type: Callable[[], None] | None.

on_show_settings: Callable[[], None]

Type: Callable[[], None] | None.

on_start: Callable[[], None]

Type: Callable[[], None] | None.

on_stop: Callable[[], None]

Type: Callable[[], None] | None.

root_group: EdlGroup

The local EDL root group (EdlGroup). Available after prepare().

run_uuid: bytes

The run's collection UUID as 16 raw bytes (UUIDv7). Available after prepare().

class SyntalosPyError(builtins.Exception):

Common base class for all non-exit exceptions.

class TimeSyncStrategy:

Strategies a synchronizer may apply to keep a secondary timeline aligned with the master clock.

Values are bit flags and may be combined with |.

Members:

NONE : Do nothing.

SHIFT_TIMESTAMPS_FWD : Move timestamps forward to match the master clock.

SHIFT_TIMESTAMPS_BWD : Move timestamps backward to match the master clock.

ADJUST_CLOCK : Adjust the secondary clock to match the master clock without changing timestamps.

WRITE_TSYNCFILE : Write a time-sync file for offline offset correction in postprocessing.

TimeSyncStrategy(value: <class 'SupportsInt'>)

ADJUST_CLOCK: ClassVar[TimeSyncStrategy]

NONE: ClassVar[TimeSyncStrategy]

SHIFT_TIMESTAMPS_BWD: ClassVar[TimeSyncStrategy]

SHIFT_TIMESTAMPS_FWD: ClassVar[TimeSyncStrategy]

WRITE_TSYNCFILE: ClassVar[TimeSyncStrategy]

value: int

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

Wait for incoming data and dispatch it to registered on_data callbacks.

Also services the IPC channel to the Syntalos process. Call this regularly inside a run() loop to keep the module responsive.

Parameters

timeout_usec: Maximum time to block in microseconds. Pass -1 (default) to wait until the module is no longer in RUNNING state.

def get_input_port(id: str) -> InputPort | None:

Retrieve a reference to an input port by its ID.

Parameters

id: The port ID (assigned via SyntalosLink.register_input_port() or via the PyScript GUI editor).

Returns

An InputPort handle, or None if no port with that ID exists.

def get_output_port(id: str) -> OutputPort | None:

Retrieve a reference to an output port by its ID.

Parameters

id: The port ID (assigned via SyntalosLink.register_output_port() or via the PyScript GUI editor).

Returns

An OutputPort handle, or None if no port with that ID exists.

def init_link(*, rename_process: bool = False) -> SyntalosLink:

Initialize a connection with Syntalos.

This function must be called only once at program startup, before invoking any other methods on this module.

Parameters

rename_process: Rename the current process to match the module identifier.:return: The active SyntalosLink registry object.

def is_running() -> bool:

Check whether the experiment is still active.

Returns

True while the run is in progress, False once a stop has been requested.

def list_types_synonyms() -> list:

def new_line_command( kind: LineCommandKind, line_id: <class 'SupportsInt'>, value: <class 'SupportsInt'> = 0) -> LineCommand:

Create a LineCommand for the given line.

Parameters

kind: The LineCommandKind to execute.
line_id: Hardware line / channel / pin number.
value: Optional numeric value (digital: 0/1; analog: DAC code).

Returns

A new LineCommand instance.

def print_types_synonyms() -> None:

def println(text: str) -> None:

Print a line of text to stdout.

Parameters

text: The text to print.

def raise_error(message: str) -> None:

Raise a module error, immediately stopping the current run.

Parameters

message: Human-readable error description.

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

Schedule a callable to be invoked after a delay.

The call is executed on the module's event loop, so it is safe to interact with ports and other module state from the callback.

Signature: schedule_delayed_call(delay_msec: int, callable_fn: Callable[[], None]) -> None.

Parameters

delay_msec: Delay before the call is made, in milliseconds. Must be >= 0.
callable_fn: Zero-argument callable to invoke.

Raises

SyntalosPyError: If delay_msec is negative.

def time_since_start_msec() -> int:

Return the time elapsed since the experiment started.

Returns

Elapsed time in milliseconds.

def time_since_start_usec() -> int:

Return the time elapsed since the experiment started.

Returns

Elapsed time in microseconds.

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

Sleep for approximately the given number of milliseconds.

Unlike a plain time.sleep(), this keeps the communication channel to Syntalos alive so that control messages (e.g. stop requests) are still handled.

Parameters

msec: Duration to wait in milliseconds.

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

Sleep for approximately the given number of seconds.

Unlike a plain time.sleep(), this keeps the communication channel to Syntalos alive so that control messages (e.g. stop requests) are still handled.

Parameters

sec: Duration to wait in seconds.

Verifying time delays Network Control Protocol