tomato.driverinterface_1_0 package

tomato.driverinterface_1_0.in_devmap(func)

pydantic model tomato.driverinterface_1_0.Attr

Bases: BaseModel

A Pydantic BaseModel used to describe device attributes.

Show JSON schema

{
   "title": "Attr",
   "description": "A Pydantic :class:`BaseModel` used to describe device attributes.",
   "type": "object",
   "properties": {
      "type": {
         "title": "Type"
      },
      "rw": {
         "default": false,
         "title": "Rw",
         "type": "boolean"
      },
      "status": {
         "default": false,
         "title": "Status",
         "type": "boolean"
      },
      "units": {
         "default": null,
         "title": "Units",
         "type": "string"
      }
   },
   "required": [
      "type"
   ]
}

field type: T [Required]

field rw: bool = False

field status: bool = False

field units: str = None

class tomato.driverinterface_1_0.ModelInterface(settings=None)

Bases: object

An abstract base class specifying the driver interface.

Individual driver modules should expose a DriverInterface which inherits from this abstract class. Only the methods of this class should be used to interact with drivers and their devices.

version: str = '1.0'

class DeviceManager(driver, key, **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.

key: tuple: The key 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: dict[str, list]: Container for cached data on this component.

running: bool

datalock: RLock: Lock object for thread-safe data manipulation.

run(): Helper function for starting the self.thread.

task_runner()

Target function for the self.thread.

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 re-primed for future Tasks at the end of this function.

prepare_task(task: Task, **kwargs: dict): Given a Task, prepare this component for execution by settin all Attrs as specified in the task.technique_params dictionary.

abstract do_task(task: Task, **kwargs: dict)

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 values of all Attrs, so that the component status is consistent with the cached data.

stop_task(**kwargs: dict): Stops the currently running task.

abstract set_attr(attr: str, val: Any, **kwargs: dict): Sets the specified Attr to val.

abstract get_attr(attr: str, **kwargs: dict) → Any: Reads the value of the specified Attr.

get_data(**kwargs: dict) → dict[str, list]: Returns the cached self.data before clearing the cache.

abstract attrs() → dict[str, Attr]: Returns a dict of all available Attrs.

abstract capabilities() → set: Returns a set of all supported techniques.

status(**kwargs) → dict: Compiles a status report from Attrs marked as status=True.

reset(**kwargs) → None: Resets the component to an initial status.

CreateDeviceManager(key, **kwargs): A factory function which is used to pass this instance of the ModelInterface to the new DeviceManager instance.

devmap: dict[tuple, DeviceManager]: Map of registered devices, the tuple keys are component = (address, channel)

settings: dict[str, Any]: A settings map to contain driver-specific settings such as dllpath for BioLogic

dev_register(address: str, channel: str, **kwargs: dict) → Reply

Register a new device component in this driver.

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

The returned Reply should contain the capabilities of the registered component in the data slot.

dev_teardown(key: tuple, **kwargs: dict) → Reply

Emergency stop function.

Should set the device component into a documented, safe state. The function is to be only called in case of critical errors, or when the component is being removed, not as part of normal operation (i.e. it is not intended as a clean-up after task completion).

dev_reset(key: tuple, **kwargs: dict) → Reply

Component reset function.

Should set the device component into a documented, safe state. This function is executed at the end of every job.

dev_set_attr(attr: str, val: Any, key: tuple, **kwargs: dict) → Reply

Set value of the Attr of the specified device component.

Pass-through to the DeviceManager.set_attr() function. No type or read-write validation performed here!

dev_get_attr(attr: str, key: tuple, **kwargs: dict) → Reply

Get value of the Attr from the specified device component.

Pass-through to the DeviceManager.get_attr() function. Units are not returned; those can be queried for all Attrs using self.attrs().

dev_status(key: tuple, **kwargs: dict) → Reply

Get the status report from the specified device component.

Iterates over all Attrs on the component that have status=True and returns their values in the Reply.data as a dict.

task_start(key: tuple, task: Task, **kwargs) → Reply

Submit a Task onto the specified device component.

Pushes the supplied Task into the Queue of the component, then starts the worker thread. Checks that the Task is among the capabilities of this component.

task_status(key: tuple, **kwargs: dict) → Reply

Returns the task readiness status of the specified device component.

The running entry in the data slot of the Reply indicates whether a Task is running. The can_submit entry indicates whether another Task can be queued onto the device component already.

task_stop(key: tuple, **kwargs) → Reply

Stops a running task and returns any collected data.

Pass-through to DriverManager.stop_task() and task_data().

task_data(key: tuple, **kwargs) → Reply

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

Pass-through for DeviceManager.get_data(), with the caveat that the dict[list] which is returned from the component is here converted to a Dataset and annotated using units from attrs().

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

status() → Reply: Returns the driver status. Currently that is the names of the components in the devmap.

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 dev_reset(). This is not a pass-through to dev_teardown().

capabilities(key: tuple, **kwargs) → Reply

Returns the capabilities of the device component.

Pass-through to DriverManager.capabilities().

attrs(key: tuple, **kwargs: dict) → Reply

Query available Attrs on the specified device component.

Pass-through to the DeviceManager.attrs() function.