Skip to content

Python API

deimos

Controller

The control program that communicates with hardware peripherals, runs calculations, and dispatches data.

Source code in deimos/deimos.pyi 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
class Controller:
    """
    The control program that communicates with hardware peripherals,
    runs calculations, and dispatches data.
    """

    def __init__(self, op_name: str, op_dir: str, rate_hz: float) -> None:
        """Build a new controller.

        `rate_hz` will be rounded to the nearest nanosecond when converted
        to the sample period.

        This constructor does not run the controller or attach any peripherals.
        """
        ...
    def run(self) -> str:
        """Run the control program."""
        ...
    def run_nonblocking(
        self,
        latest_value_cutoff_freq: float | None = None,
        wait_for_ready: bool = True,
    ) -> RunHandle:
        """Run the control program on a separate thread and return a handle
        for coordination.

        Args:
            latest_value_cutoff_freq: Optional second-order Butterworth low-pass filter
                                      cutoff frequency to apply to latest-value data.
                                      If the selected frequency is outside the viable
                                      range for the filter, the cutoff frequency will
                                      be clamped to the viable bounds and a warning
                                      will be emitted.
            wait_for_ready: Block until the controller has completed its first cycle.
        """
        ...
    def scan(self, timeout_ms: int = 10) -> list[PeripheralLike]:
        """Scan the local network (and any other attached sockets)
        for available peripherals."""
        ...
    def available_inputs(self) -> list[str]:
        """List peripheral inputs that can be written manually."""
        ...
    def graphviz_dot(self) -> str:
        """Render the current calc expression graph as Graphviz DOT text."""
        ...
    def add_peripheral(self, name: str, p: PeripheralLike) -> None:
        """Register a peripheral with the control program"""
        ...
    def attach_hootl_driver(
        self,
        peripheral_name: str,
        transport: HootlTransport,
        end_epoch_ns: int | None = None,
    ) -> HootlRunHandle:
        """Wrap an existing peripheral with a hootl wrapper and start its driver."""
        ...
    def add_calc(self, name: str, calc: CalcLike) -> None:
        """Add a calc to the expression graph that runs on every cycle"""
        ...
    def add_dataframe_dispatcher(
        self,
        name: str,
        max_size_megabytes: int,
        overflow_behavior: Overflow,
    ) -> dispatcher.DataFrameHandle:
        """Add an in-memory dataframe dispatcher and return its shared handle."""
        ...
    def add_dispatcher(self, name: str, dispatcher: DispatcherLike) -> None:
        """Add a dispatcher via a JSON-serializable dispatcher instance."""
        ...
    def dispatcher_names(self) -> list[str]: ...
    def remove_dispatcher(self, name: str) -> bool: ...
    def add_socket(self, name: str, socket: SocketLike) -> None:
        """Add a socket via a JSON-serializable socket instance."""
        ...
    def remove_socket(self, name: str) -> bool:
        """Remove a socket by name."""
        ...
    def set_peripheral_input_source(self, input_field: str, source_field: str) -> None:
        """Connect an entry in the calc graph to a
        command to be sent to the peripheral."""
        ...
    def clear_calcs(self) -> None:
        """Remove all calcs."""
        ...
    def clear_peripherals(self) -> None:
        """Remove all peripherals."""
        ...
    def clear_dispatchers(self) -> None:
        """Remove all dispatchers."""
        ...
    def clear_sockets(self) -> None:
        """Remove all sockets."""
        ...

    @property
    def op_name(self) -> str:
        """
        The name of the operation.
        Used to set database table names, set log and data file names, etc.
        """
        ...
    @op_name.setter
    def op_name(self, v: str) -> None: ...
    @property
    def op_dir(self) -> str:
        """
        The directory where this operation's logs and other data will be placed,
        and where calcs with linked configuration (like a SequenceMachine) can
        find their linked files or folders by relative path.
        """
        ...
    @op_dir.setter
    def op_dir(self, v: str) -> None: ...
    @property
    def dt_ns(self) -> int:
        """[ns] control program cycle period."""
        ...
    @dt_ns.setter
    def dt_ns(self, v: int) -> None: ...
    @property
    def rate_hz(self) -> float:
        """[Hz] control program cycle rate."""
        ...
    @rate_hz.setter
    def rate_hz(self, v: float) -> None: ...
    @property
    def peripheral_loss_of_contact_limit(self) -> int:
        """Number of missed packets from the controller that indicates disconnection."""
        ...
    @peripheral_loss_of_contact_limit.setter
    def peripheral_loss_of_contact_limit(self, v: int) -> None: ...
    @property
    def controller_loss_of_contact_limit(self) -> int:
        """Number of missed packets from a peripheral that indicates disconnection."""
        ...
    @controller_loss_of_contact_limit.setter
    def controller_loss_of_contact_limit(self, v: int) -> None: ...
    @property
    def termination_criteria(self) -> Termination | None:
        """Criteria for exiting the control program."""
        ...
    @termination_criteria.setter
    def termination_criteria(self, v: Termination | None) -> None: ...
    @property
    def loss_of_contact_policy(self) -> LossOfContactPolicy:
        """
        The response of the control program when a peripheral disconnects during run.
        """
        ...

    @loss_of_contact_policy.setter
    def loss_of_contact_policy(self, v: LossOfContactPolicy) -> None: ...
    @property
    def loop_method(self) -> LoopMethod:
        """
        The loop waiting method for the controller.

        Busywaiting is performant, but inefficient;
        relying on the operating system for scheduling is efficient, but not performant.
        """
        ...
    @loop_method.setter
    def loop_method(self, v: LoopMethod) -> None: ...
    @property
    def enable_manual_inputs(self) -> bool:
        """Whether manual input overrides should be applied during the control loop."""
        ...
    @enable_manual_inputs.setter
    def enable_manual_inputs(self, v: bool) -> None: ...

controller_loss_of_contact_limit: int property writable

Number of missed packets from a peripheral that indicates disconnection.

dt_ns: int property writable

[ns] control program cycle period.

enable_manual_inputs: bool property writable

Whether manual input overrides should be applied during the control loop.

loop_method: LoopMethod property writable

The loop waiting method for the controller.

Busywaiting is performant, but inefficient; relying on the operating system for scheduling is efficient, but not performant.

loss_of_contact_policy: LossOfContactPolicy property writable

The response of the control program when a peripheral disconnects during run.

op_dir: str property writable

The directory where this operation's logs and other data will be placed, and where calcs with linked configuration (like a SequenceMachine) can find their linked files or folders by relative path.

op_name: str property writable

The name of the operation. Used to set database table names, set log and data file names, etc.

peripheral_loss_of_contact_limit: int property writable

Number of missed packets from the controller that indicates disconnection.

rate_hz: float property writable

[Hz] control program cycle rate.

termination_criteria: Termination | None property writable

Criteria for exiting the control program.

__init__(op_name, op_dir, rate_hz)

Build a new controller.

rate_hz will be rounded to the nearest nanosecond when converted to the sample period.

This constructor does not run the controller or attach any peripherals.

Source code in deimos/deimos.pyi 
145
146
147
148
149
150
151
152
153
def __init__(self, op_name: str, op_dir: str, rate_hz: float) -> None:
    """Build a new controller.

    `rate_hz` will be rounded to the nearest nanosecond when converted
    to the sample period.

    This constructor does not run the controller or attach any peripherals.
    """
    ...

add_calc(name, calc)

Add a calc to the expression graph that runs on every cycle

Source code in deimos/deimos.pyi 
196
197
198
def add_calc(self, name: str, calc: CalcLike) -> None:
    """Add a calc to the expression graph that runs on every cycle"""
    ...

add_dataframe_dispatcher(name, max_size_megabytes, overflow_behavior)

Add an in-memory dataframe dispatcher and return its shared handle.

Source code in deimos/deimos.pyi 
199
200
201
202
203
204
205
206
def add_dataframe_dispatcher(
    self,
    name: str,
    max_size_megabytes: int,
    overflow_behavior: Overflow,
) -> dispatcher.DataFrameHandle:
    """Add an in-memory dataframe dispatcher and return its shared handle."""
    ...

add_dispatcher(name, dispatcher)

Add a dispatcher via a JSON-serializable dispatcher instance.

Source code in deimos/deimos.pyi 
207
208
209
def add_dispatcher(self, name: str, dispatcher: DispatcherLike) -> None:
    """Add a dispatcher via a JSON-serializable dispatcher instance."""
    ...

add_peripheral(name, p)

Register a peripheral with the control program

Source code in deimos/deimos.pyi 
185
186
187
def add_peripheral(self, name: str, p: PeripheralLike) -> None:
    """Register a peripheral with the control program"""
    ...

add_socket(name, socket)

Add a socket via a JSON-serializable socket instance.

Source code in deimos/deimos.pyi 
212
213
214
def add_socket(self, name: str, socket: SocketLike) -> None:
    """Add a socket via a JSON-serializable socket instance."""
    ...

attach_hootl_driver(peripheral_name, transport, end_epoch_ns=None)

Wrap an existing peripheral with a hootl wrapper and start its driver.

Source code in deimos/deimos.pyi 
188
189
190
191
192
193
194
195
def attach_hootl_driver(
    self,
    peripheral_name: str,
    transport: HootlTransport,
    end_epoch_ns: int | None = None,
) -> HootlRunHandle:
    """Wrap an existing peripheral with a hootl wrapper and start its driver."""
    ...

available_inputs()

List peripheral inputs that can be written manually.

Source code in deimos/deimos.pyi 
179
180
181
def available_inputs(self) -> list[str]:
    """List peripheral inputs that can be written manually."""
    ...

clear_calcs()

Remove all calcs.

Source code in deimos/deimos.pyi 
222
223
224
def clear_calcs(self) -> None:
    """Remove all calcs."""
    ...

clear_dispatchers()

Remove all dispatchers.

Source code in deimos/deimos.pyi 
228
229
230
def clear_dispatchers(self) -> None:
    """Remove all dispatchers."""
    ...

clear_peripherals()

Remove all peripherals.

Source code in deimos/deimos.pyi 
225
226
227
def clear_peripherals(self) -> None:
    """Remove all peripherals."""
    ...

clear_sockets()

Remove all sockets.

Source code in deimos/deimos.pyi 
231
232
233
def clear_sockets(self) -> None:
    """Remove all sockets."""
    ...

graphviz_dot()

Render the current calc expression graph as Graphviz DOT text.

Source code in deimos/deimos.pyi 
182
183
184
def graphviz_dot(self) -> str:
    """Render the current calc expression graph as Graphviz DOT text."""
    ...

remove_socket(name)

Remove a socket by name.

Source code in deimos/deimos.pyi 
215
216
217
def remove_socket(self, name: str) -> bool:
    """Remove a socket by name."""
    ...

run()

Run the control program.

Source code in deimos/deimos.pyi 
154
155
156
def run(self) -> str:
    """Run the control program."""
    ...

run_nonblocking(latest_value_cutoff_freq=None, wait_for_ready=True)

Run the control program on a separate thread and return a handle for coordination.

Parameters:

Name Type Description Default
latest_value_cutoff_freq float | None

Optional second-order Butterworth low-pass filter cutoff frequency to apply to latest-value data. If the selected frequency is outside the viable range for the filter, the cutoff frequency will be clamped to the viable bounds and a warning will be emitted.

 None
wait_for_ready bool

Block until the controller has completed its first cycle.

 True
Source code in deimos/deimos.pyi 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
def run_nonblocking(
    self,
    latest_value_cutoff_freq: float | None = None,
    wait_for_ready: bool = True,
) -> RunHandle:
    """Run the control program on a separate thread and return a handle
    for coordination.

    Args:
        latest_value_cutoff_freq: Optional second-order Butterworth low-pass filter
                                  cutoff frequency to apply to latest-value data.
                                  If the selected frequency is outside the viable
                                  range for the filter, the cutoff frequency will
                                  be clamped to the viable bounds and a warning
                                  will be emitted.
        wait_for_ready: Block until the controller has completed its first cycle.
    """
    ...

scan(timeout_ms=10)

Scan the local network (and any other attached sockets) for available peripherals.

Source code in deimos/deimos.pyi 
175
176
177
178
def scan(self, timeout_ms: int = 10) -> list[PeripheralLike]:
    """Scan the local network (and any other attached sockets)
    for available peripherals."""
    ...

set_peripheral_input_source(input_field, source_field)

Connect an entry in the calc graph to a command to be sent to the peripheral.

Source code in deimos/deimos.pyi 
218
219
220
221
def set_peripheral_input_source(self, input_field: str, source_field: str) -> None:
    """Connect an entry in the calc graph to a
    command to be sent to the peripheral."""
    ...

LoopMethod

Source code in deimos/deimos.pyi 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
class LoopMethod:
    Performant: ClassVar[Self]
    """
    Use 100% of a CPU to protect timing.
    This increases maximum usable control frequency.
    """

    Efficient: ClassVar[Self]
    """
    Use operating system scheduling to reduce CPU usage
    at the expense of degraded cycle performance.
    Typically viable up to about 50Hz control rate.
    """

    @staticmethod
    def performant() -> Performant:
        """Use 100% of a CPU to protect timing."""
        ...
    @staticmethod
    def efficient() -> Efficient:
        """Use operating system scheduling for lower CPU usage."""
        ...

Efficient: Self class-attribute

Use operating system scheduling to reduce CPU usage at the expense of degraded cycle performance. Typically viable up to about 50Hz control rate.

Performant: Self class-attribute

Use 100% of a CPU to protect timing. This increases maximum usable control frequency.

efficient() staticmethod

Use operating system scheduling for lower CPU usage.

Source code in deimos/deimos.pyi 
58
59
60
61
@staticmethod
def efficient() -> Efficient:
    """Use operating system scheduling for lower CPU usage."""
    ...

performant() staticmethod

Use 100% of a CPU to protect timing.

Source code in deimos/deimos.pyi 
54
55
56
57
@staticmethod
def performant() -> Performant:
    """Use 100% of a CPU to protect timing."""
    ...

LossOfContactPolicy

Source code in deimos/deimos.pyi 
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
class LossOfContactPolicy:
    Terminate: ClassVar[Self]
    """Terminate the control program."""

    Reconnect: ClassVar[Self]
    """Attempt to reconnect to the peripheral"""

    @staticmethod
    def terminate() -> Terminate:
        """Construct a policy that terminates the control program."""
        ...
    @staticmethod
    def reconnect_s(timeout_s: float) -> Reconnect:
        """Construct a reconnect policy with a timeout in seconds."""
        ...
    @staticmethod
    def reconnect_indefinite() -> Reconnect:
        """Construct a reconnect policy with no timeout."""
        ...

Reconnect: Self class-attribute

Attempt to reconnect to the peripheral

Terminate: Self class-attribute

Terminate the control program.

reconnect_indefinite() staticmethod

Construct a reconnect policy with no timeout.

Source code in deimos/deimos.pyi 
94
95
96
97
@staticmethod
def reconnect_indefinite() -> Reconnect:
    """Construct a reconnect policy with no timeout."""
    ...

reconnect_s(timeout_s) staticmethod

Construct a reconnect policy with a timeout in seconds.

Source code in deimos/deimos.pyi 
90
91
92
93
@staticmethod
def reconnect_s(timeout_s: float) -> Reconnect:
    """Construct a reconnect policy with a timeout in seconds."""
    ...

terminate() staticmethod

Construct a policy that terminates the control program.

Source code in deimos/deimos.pyi 
86
87
88
89
@staticmethod
def terminate() -> Terminate:
    """Construct a policy that terminates the control program."""
    ...

Overflow

Source code in deimos/deimos.pyi 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
class Overflow:
    Wrap: ClassVar[Self]
    """Overwrite oldest data first."""

    NewFile: ClassVar[Self]
    """Create a new shard."""

    Error: ClassVar[Self]
    """Emit an error and shut down."""

    @staticmethod
    def wrap() -> Wrap:
        """Wrap back to the beginning of the file."""
        ...
    @staticmethod
    def new_file() -> NewFile:
        """Create a new file on overflow."""
        ...
    @staticmethod
    def error() -> Error:
        """Error on overflow."""
        ...

Error: Self class-attribute

Emit an error and shut down.

NewFile: Self class-attribute

Create a new shard.

Wrap: Self class-attribute

Overwrite oldest data first.

error() staticmethod

Error on overflow.

Source code in deimos/deimos.pyi 
35
36
37
38
@staticmethod
def error() -> Error:
    """Error on overflow."""
    ...

new_file() staticmethod

Create a new file on overflow.

Source code in deimos/deimos.pyi 
31
32
33
34
@staticmethod
def new_file() -> NewFile:
    """Create a new file on overflow."""
    ...

wrap() staticmethod

Wrap back to the beginning of the file.

Source code in deimos/deimos.pyi 
27
28
29
30
@staticmethod
def wrap() -> Wrap:
    """Wrap back to the beginning of the file."""
    ...

Termination

Source code in deimos/deimos.pyi 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
class Termination:
    Timeout: ClassVar[Self]
    """End the control program after some duration from the start of the first cycle."""

    Scheduled: ClassVar[Self]
    """End the control program at a specific UTC system time."""

    @staticmethod
    def timeout_s(s: float) -> Timeout:
        """End after `s` seconds from the start of the first cycle."""
        ...
    @staticmethod
    def scheduled_epoch_ns(ns: int) -> Scheduled:
        """End at a specified absolute system time in UTC nanoseconds."""
        ...

Scheduled: Self class-attribute

End the control program at a specific UTC system time.

Timeout: Self class-attribute

End the control program after some duration from the start of the first cycle.

scheduled_epoch_ns(ns) staticmethod

End at a specified absolute system time in UTC nanoseconds.

Source code in deimos/deimos.pyi 
74
75
76
77
@staticmethod
def scheduled_epoch_ns(ns: int) -> Scheduled:
    """End at a specified absolute system time in UTC nanoseconds."""
    ...

timeout_s(s) staticmethod

End after s seconds from the start of the first cycle.

Source code in deimos/deimos.pyi 
70
71
72
73
@staticmethod
def timeout_s(s: float) -> Timeout:
    """End after `s` seconds from the start of the first cycle."""
    ...