copapy.SerialTarget#

class copapy.SerialTarget(port=None, device=Device(teensy41), optimization='O3', timeout=10.0, connection=None)#

Bases: _TaskTarget

Target on a device running the Zephyr runner, connected over a serial port. The periodic tasks run in real time on the device after run().

Connect to a device.

Parameters:

  • port (str | None) – Serial port (e.g. ‘COM11’ or ‘/dev/ttyACM0’), by default the port is searched by the USB id of the device. Requires pyserial.

  • device (Device) – Device description, e.g. copapy.devices.teensy41

  • optimization (str) – Optimization level of the stencils

  • timeout (float) – Timeout for answers of the runner in seconds

  • connection (Connection | None) – Already opened byte stream to the runner instead of a port

bind(variable, channel)#

Binds a value to an I/O channel of the device. Bindings are used by the next deploy.

A copapy value (input) reads the channel at the release of each task using it, it can not be written by write_value anymore. On an output channel it reads back the level of the output.

A value computed by a task writes the output channel when the task publishes its outputs. The value must be an output of the task.

Parameters:

  • variable (value[Any]) – Value to bind, of the type of the channel (see IoChannel.kind)

  • channel (IoChannel) – Channel of the target, e.g. target.io.gpio(‘led’)

Return type:

None

buffer(variable)#

Returns memory of an array on the target as memoryview.

Only available for the local target.

Return type:

memoryview[Any]

close()#

Closes the serial port. The tasks keep running on the device.

Return type:

None

deploy(*tasks)#

Compiles the tasks for the device and loads them. Previously deployed tasks are removed, the device must be stopped.

Parameters:

tasks (Any) – Tasks or sequences of tasks. All other arguments (values, vectors, tensors, arrays and sequences of them) are the outputs of one additional task without period.

Return type:

None

exec(task=None)#

Runs one job of a deployed task: reading inputs, computing and publishing the outputs. While the tasks are running, only tasks without period can be run.

Parameters:

task (Task | None) – Task to run, can be omitted if only one task is deployed

Return type:

None

read_value(variables)#

Reads the numeric value of a copapy type.

Overloads:

  • variables (value[T]) → T

  • variables (value[int] | value[float] | int | float) → float | int | bool

  • variables (ArrayType[T]) → ArrayType[T]

  • variables (Iterable[T | value[T]]) → list[T]

  • variables (array[T]) → Any

  • variables (Any) → Any

Parameters:

variables (Any) – Variable or multiple variables to read

Returns:

Numeric value or values. Arrays are returned as nested lists, arrays marked with map_to_buffer or map_to_numpy as memoryview or numpy array. Sequences and dicts are returned as lists and dicts of the values of their items.

reset()#

Resets the device, the connection must be opened again afterwards

Return type:

None

run()#

Starts the real-time execution of the periodic tasks, all tasks are released at the start. Inputs can be written and outputs read while the tasks are running.

Return type:

None

sdb: stencil_database#
stats(task=None)#

Statistics of a deployed task. On devices the statistics are reset by run().

Parameters:

task (Task | None) – Task, can be omitted if only one task is deployed

Returns:

  • runs: Number of jobs run

  • overruns: Number of skipped releases because the previous job was not finished

  • exec_time: CPU time of the last job in seconds (without the time the job was preempted by other tasks)

  • exec_time_max: Maximum CPU time of a job in seconds

  • exec_time_mean: Mean CPU time of the jobs in seconds

  • response_time_max: Maximum time from the release to the end of a job in seconds (must be below the period)

  • start_delay_max: Maximum time from the release to the start of a job in seconds

  • degraded: The task currently runs with lower priority than the communication with the host, because the tasks used too much CPU time (load shedding on devices)

  • degradations: Number of times the task was degraded

Return type:

dict[str, Any]

step(until)#

Runs the periodic tasks in simulated time (not in real time) on the device, see LocalTarget.step. The tasks must be stopped.

Return type:

None

stop()#

Stops the real-time execution after the running jobs

Return type:

None

tick: float#
property time: float#

Simulated time in seconds since deploy

write_value(variables, data)#

Write to a copapy value on the target.

Parameters:

  • variables (value[Any] | array[Any] | Iterable[value[Any]]) – Single variable, array or multiple variables to overwrite

  • value – Single value or multiple values to write, (nested) sequences for arrays

Return type:

None