tomato.driverinterface_3_0 package

Code author: Peter Kraus

pydantic model tomato.driverinterface_3_0.Attr

Bases: BaseModel

A BaseModel used to describe device attributes.

Show JSON schema

{
   "title": "Attr",
   "type": "object",
   "properties": {
      "type": {
         "default": null,
         "title": "Type"
      },
      "rw": {
         "default": false,
         "title": "Rw",
         "type": "boolean"
      },
      "status": {
         "default": false,
         "title": "Status",
         "type": "boolean"
      },
      "units": {
         "anyOf": [
            {
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Units"
      },
      "maximum": {
         "anyOf": [
            {
               "type": "number"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Maximum"
      },
      "minimum": {
         "anyOf": [
            {
               "type": "number"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Minimum"
      },
      "options": {
         "anyOf": [
            {
               "items": {},
               "type": "array",
               "uniqueItems": true
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Options"
      }
   }
}

Config:

• arbitrary_types_allowed: bool = True

field type: type [Required]

Data type of the attribute

field rw: bool = False

Is the attribute read-write?

field status: bool = False

Should the attribute be included in component status?

field units: str | None = None

Default units for the attribute, optional.

field maximum: float | Quantity | None = None

Maximum value for the attribute, optional.

Constraints:

• union_mode = left_to_right

field minimum: float | Quantity | None = None

Minimum value for the attribute, optional.

Constraints:

• union_mode = left_to_right

field options: set | None = None

Allowed set of values for the attribute, optional.

pydantic model tomato.driverinterface_3_0.Status

Bases: BaseModel

A BaseModel used to describe component status.

Show JSON schema

{
   "title": "Status",
   "description": "A :class:`~pydantic.BaseModel` used to describe component status.",
   "type": "object",
   "properties": {
      "connected": {
         "title": "Connected",
         "type": "boolean"
      },
      "state": {
         "anyOf": [
            {
               "enum": [
                  "idle",
                  "meas",
                  "task",
                  "stop"
               ],
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "title": "State"
      },
      "can_submit": {
         "title": "Can Submit",
         "type": "boolean"
      },
      "attrs": {
         "additionalProperties": true,
         "title": "Attrs",
         "type": "object"
      }
   },
   "required": [
      "connected",
      "state",
      "can_submit"
   ]
}

field connected: bool [Required]

Indicates whether component is communicating correctly.

field state: Literal['idle', 'meas', 'task', 'stop'] | None [Required]

Indicates device state:

• idle when component is connected and idle,

• meas when component is doing an idle measurement,

• task when component has a running task,

• stop when component is being torn down or reset,

• None when component is not connected.

field can_submit: bool [Required]

Indicates whether a Task can be sent to component queue.

field attrs: dict[str, Any] [Optional]

Container for any attrs that are returned as part of a status.

class tomato.driverinterface_3_0.ModelInterface(settings: dict[str, Any] | None = None)

Bases: object

An abstract base class specifying the driver interface.

Individual driver modules should expose a DriverInterface as a top-level import, which inherits from this abstract class. Only the methods of this class are used to interact with drivers and their components.

All methods of this class should return Reply objects (except the ComponentFactory() function). However, for better readability, a decorator function to_reply() is provided, so that the types of the return values can be explicitly defined here.

version: str = '3.0'

Version of the DriverInterface.

idle_measurement_interval: int | None = None

The interval (in seconds) after which self.cmp_measure() will be executed, when idle.

property name: str

Property that should return the name of this driver.

devmap: dict[str, ModelComponent]

Map of registered device components, the keys are set from the tomato.models.Component.name.

constants: dict[str, Any]

A map that should be populated with driver-specific run-time constants.

settings: dict[str, Any]

A settings map to contain driver-specific settings such as dllpath for BioLogic

retries: dict[str, int]

Map of components which failed to register, with number of retries as values.

ComponentFactory(name, **kwargs)

A factory function which is used to pass this instance of the ModelInterface to the new ModelDevice instance.

cmp_register(name: str, address: str | None, channel: str | None, **kwargs: dict) → tuple[bool, str, str | None]

Register a new device component in this driver.

Creates a ModelDevice representing a device component, storing it in the self.devmap using the provided address and channel.

Returns the name of the registered component as the Reply.data.

cmp_stop(name: str, **kwargs: dict) → tuple[bool, str, None]

Component stop function, passthrough to ModelComponent.stop().

Should set the device component into a documented, safe state.

cmp_quit(name: str, **kwargs: dict) → tuple[bool, str, None]

Component quit function, passthrough to ModelComponent.stop() and ModelComponent.quit().

Should set the device component into a documented, safe state, then release the component from tomato.

The function is called when the driver is exiting normally.

cmp_reset(name: str, **kwargs: dict) → tuple[bool, str, None]

Component reset function, passthrough to ModelComponent.stop() and ModelComponent.reset().

Should set the device component into a safe state and make it ready to accept new Tasks if possible.

The function is called on completion of each Payload.

cmp_set_attr(attr: str, val: str | int | float | Quantity, name: str, **kwargs: dict) → tuple[bool, str, str | int | float | Quantity]

Set value of the Attr of the specified device component.

Pass-through to the ModelDevice.set_attr() function. No type or read-write validation performed here! Returns the validated or coerced value as the Reply.data.

cmp_get_attr(attr: str, name: str, **kwargs: dict) → tuple[bool, str, str | int | float | Quantity]

Get value of the Attr from the specified device component.

Pass-through to the ModelDevice.get_attr() function. No type coercion is done here. Returns the value as the Reply.data.

cmp_status(name: str, **kwargs: dict) → tuple[bool, str, Status]

Get the status report from the specified device component.

Returns a flag in Reply.data['running'] indicating whether the component is running.

Passthrough to ModelDevice.status(). Returns the dict of attribute values marked as status=True.

cmp_capabilities(name: str, **kwargs) → tuple[bool, str, set]

Returns the capabilities of the device component.

Pass-through to ModelDevice.capabilities(). Returns the set of capabilities in Reply.data.

cmp_attrs(name: str, **kwargs: dict) → tuple[bool, str, dict]

Query available Attrs on the specified device component.

Pass-through to the ModelDevice.attrs() function. Returns the dict of attributes as the Reply.data.

cmp_constants(name: str, **kwargs: dict) → tuple[bool, str, dict]

Query constants on the specified device component and this driver.

Returns the dict of constants as the Reply.data.

cmp_last_data(name: str, **kwargs: dict) → tuple[bool, str, None | Dataset]

Fetch the last stored data on the component.

Passthrough to ModelDevice.get_last_data(). The data in the form of a xarray.Dataset is returned as the Reply.data.

cmp_measure(name: str, **kwargs: dict) → tuple[bool, str, None]

Do a single measurement on the component according to its current configuration.

Fails if the component already has a running task / measurement.

task_start(name: str, task: Task, **kwargs) → tuple[bool, str, set | Task]

Submit a Task onto the specified device component.

Pushes the supplied Task into the Queue of the component, then starts the worker thread (if not already started). Checks that the Task is among the capabilities of this component.

task_status(name: str, **kwargs: dict) → tuple[bool, str, dict]

task_stop(name: str, **kwargs) → tuple[bool, str, Dataset | None]

Stops a running task and returns any collected data.

Pass-through to ModelComponent.stop_task() and ModelInterface.task_data().

If there is any cached data, it is returned as a xarray.Dataset in the Reply.data and the cache is cleared.

task_data(name: str, **kwargs) → tuple[bool, str, Dataset | None]

Return cached task data on the device component and clean the cache.

Pass-through for ModelDevice.get_data(), which should return a xarray.Dataset that is fully annotated.

This function gets called by the job thread every device.pollrate, it therefore incurs some IPC cost.

task_validate(name: str, task: Task, **kwargs) → tuple[bool, str, None]

Validate the provided Task for submission on the component identified by key.

status() → Reply

Returns the driver status. Currently that is the names of the components in the devmap.

quit() → Reply

Driver quit function.

Stops tasks and quits every registered component. Passthrough to ModelInterface.task_stop() and ModelInterface.cmp_quit().

Any driver-specific commands (such as releasing serial port etc.) should be performed ehre.

Called when driver process is exiting.

reset() → Reply

Resets the driver.

Called when the driver process is quitting. Instructs all remaining tasks to stop. Warns when devices linger. Passes through to cmp_reset(). This is not a pass-through to cmp_teardown().

class tomato.driverinterface_3_0.ModelComponent(driver, name, **kwargs)

Bases: object

An abstract base class specifying a manager for an individual component.

This class should handle determining attributes and capabilities of the component, the reading/writing of those attributes, processing of tasks, and caching and returning of task data.

driver: ModelInterface

The parent DriverInterface instance.

name: str

The name in self.driver.devmap referring to this object.

task_list: Queue

A Queue used to pass Tasks to the worker Thread.

thread: Thread

The worker Thread.

data: Dataset | None

Container for cached data on this component.

last_data: Dataset | None

Container for last datapoint on this component.

state: str | None = None

A str holding the component state.

running_task: Task | None = None

datalock: RLock

Lock object for thread-safe data manipulation.

constants: dict[str, Any]

Constant metadata of this component.

task_runner() → None

Target function for the self.thread when handling Tasks.

This function waits for a Task passed using self.task_list, then handles setting all Attrs using the prepare_task() function, and finally handles the main loop of the task, periodically running the do_task() function (using task.sampling_interval) until the maximum task duration (i.e. task.max_duration) is exceeded.

The self.thread is reset to None.

prepare_task(task: Task, **kwargs: dict) → None

Given a Task, prepare this component for execution by setting all Attrs as specified in the task.task_params dictionary.

do_task(task: Task, t_start: float, t_now: float, t_prev: float, **kwargs: dict) → None

Periodically called task execution function.

This function is responsible for updating self.data with new data, i.e. performing the measurement. It should also update the value of self.last_data, so that the component status is consistent with the cached data.

abstract do_measure(**kwargs: dict) → None

One shot execution worker function.

This function is performs a measurement using the current configuration of self.attrs, and stores the result in self.last_data.

stop_task(**kwargs: dict) → None

Stops the currently running task.

abstract set_attr(attr: str, val: str | int | float | Quantity, **kwargs: dict) → str | int | float | Quantity

Sets the specified Attr to val.

This function should handle any data type coercion and validation using e.g. Attr.maximum and Attr.minimum.

Returns the coerced value corresponding to val.

abstract get_attr(attr: str, **kwargs: dict) → str | int | float | Quantity

Reads the value of the specified Attr.

get_data(**kwargs: dict) → Dataset | None

Returns the cached self.data as a xarray.Dataset before clearing the cache.

get_last_data(**kwargs: dict) → Dataset | None

Returns the last_data object as a xarray.Dataset.

abstract attrs(**kwargs) → dict[str, Attr]

Returns a dict of all available Attrs.

abstract capabilities(**kwargs) → set

Returns a set of all supported techniques.

abstract status(**kwargs) → Status

Function indicating component status.

The implementation of this function in the driver module should perform checks whether the components is still reachable (Status.connected) and what state is the component in (Status.state).

The function should also compile a status report using Attrs marked as status=True and return it as Status.attrs.

stop(**kwargs) → None

Stops any activity on this component.

This function should set the component to a safe state.

By default a pass-through to ModelComponent.stop_task().

abstract quit(**kwargs) → None

Quits the component.

This function makes the component ready to quit. When accessed via the ModelInterface.cmp_quit(), it is always called after ModelComponent.stop(), therefore all Tasks on the device can be assumed to be stopped.

reset(**kwargs) → None

Resets the component to an initial status.

This function makes the component ready to accept new Task. When accessed via the ModelInterface.cmp_reset(), it is always called after ModelComponent.stop(), therefore all Tasks on the device can be assumed to be stopped.

Submodules

tomato.driverinterface_3_0.decorators.in_devmap(func)

tomato.driverinterface_3_0.decorators.to_reply(func)

Helper decorator for coercing tuples into Reply.

tomato.driverinterface_3_0.decorators.log_errors(func)

Helper decorator for logging all kinds of errors.

This decorator should be only used on functions in the API of the ModelInterface, as the caught exceptions will cause the driver process to exit.

tomato.driverinterface_3_0.decorators.coerce_val(func)

Decorator for coercing val into the correct format based on Attr data.

This decorator should be applied to the ModelDriver.set_attr() function, in order to check whether the supplied value is allowed (not None, in options, between minimum and maximum) as well as coercing it to the right type and unit.