alab_management.device_view.device_view module

alab_management.device_view.device_view module#

Wrapper over the devices collection.

exception DeviceConnectionError[source]#

Bases: Exception

Generic error signifying that connection to a device has failed.

class DevicePauseStatus(value)[source]#

Bases: Enum

Pause status of the Device. This is used to pause the device outside of the typical Task queue (like by an operator for maintenance or refilling consumables).

PAUSED = 3#
RELEASED = 1#
REQUESTED = 2#
class DeviceTaskStatus(value)[source]#

Bases: Enum

The Task status of devices. Used by TaskManager to decide whether a Device is available for to execute a Task.

ERROR = 4#
IDLE = 2#
OCCUPIED = 3#
UNKNOWN = 1#
class DeviceView(connect_to_devices=False)[source]#

Bases: object

Device view provides API to get/set the status of a device as well as request ownership of one device.

add_devices_to_db()[source]#

Insert device definitions to db, which includes devices’ name, descriptions, parameters, type (class name).

When one device’s name has already appeared in the database, a NameError will be raised. Device name is a unique identifier for a device

execute_command(device_name, method, *args, **kwargs)[source]#

Call a callable function (method) with *args and **kwargs on device_name.

get_all()[source]#

Get all the devices in the database, used for dashboard.

Return type:

list[dict[str, Any]]

get_all_attributes(device_name)[source]#

Returns the device attributes.

Parameters:

device_name (str) – name of the device to get the attributes for

Return type:

dict[str, Any]

Returns:

dict: device attributes

get_attribute(device_name, attribute)[source]#

Gets a device attribute. Attributes are used to store device-specific values in the database.

Parameters:

  • device_name (str) – name of the device to get the attribute for

  • attribute (str) – attribute to be retrieved

Return type:

Any

Returns:

Any: attribute value

get_available_devices(device_str, type_or_name, task_id=None)[source]#

Given device type, it will return all the device with this type.

If only_idle set to True, only the idle devices will be returned (or ones have the same task id)

Parameters:

  • device_str (str) – the type of device

  • type_or_name (str) – “type” or “name” to specify whether searching for a type of device by Type(BaseDevice), or for a specific device by name

  • task_id (Optional[ObjectId]) – the id of task that requests this device

Return type:

list[dict[str, str | bool]]

Returns:

[{“name”: device_name, “need_release”: bool}] The entry need_release indicates whether a device needs to be released when __exit__ method is called in the DevicesLock.

get_device(device_name)[source]#

Get device by device name, if not found, raises ValueError.

Return type:

dict[str, Any]

get_devices_by_task(task_id)[source]#

Get devices given a task id (regardless of its status!).

Return type:

list[BaseDevice]

get_message(device_name)[source]#

Gets the current device message. Message is used to communicate device state with the user dashboard.

Parameters:

device_name (str) – name of the device to set the message for

Return type:

str

get_samples_on_device(device_name)[source]#

Get all the samples on a device.

This function will directly call the SampleView.get_samples_on_device method for backward compatibility. :rtype: dict[str, list[ObjectId]]

Note

(in version 1.0.1) This function has been moved to SampleView class.

get_status(device_name)[source]#

Get device status by device name, if not found, raise ValueError.

Return type:

DeviceTaskStatus

occupy_device(device, task_id)[source]#

Occupy a device with given task id.

pause_device(device_name)[source]#

Request pause for a specific device.

query_property(device_name, prop)[source]#

Query the property value of a device (with device_name) with prop.

If there is no such device with name device_name, a ValueError shall be raised. If there is no such property with name prop, a AttributeError shall be raised.

release_device(device_name)[source]#

Release a device.

device: name of device to be released

request_devices(task_id, device_names_str=None, device_types_str=None)[source]#

Request a list of device, this function will return the name of devices if all the requested device is ready.

Note

There should be no duplicated devices in the device_type, or a ValueError shall be raised

Parameters:

  • task_id (ObjectId) – the id of task that requests these devices

  • device_names_str (Optional[Collection[str]]) – the requested device names. If None, no device name is requested

  • device_types_str (Optional[Collection[str]]) – the requested device types. If None, no device type is requested

Return type:

dict[str, dict[str, str | bool]] | None

Returns:

{“device_type_name”: {“name”: device_name, “need_release”: need_release (bool)}} or None

set_all_attributes(device_name, attributes)[source]#

Sets the device attributes.

Parameters:

  • device_name (str) – name of the device to set the attributes for

  • attributes (dict) – attributes to be set

set_attribute(device_name, attribute, value)[source]#

Sets a device attribute. Attributes are used to store device-specific values in the database.

Parameters:

  • device_name (str) – name of the device to set the attribute for

  • attribute (str) – attribute to be set

  • value (Any) – attribute value

set_message(device_name, message)[source]#

Sets the device message. Message is used to communicate device state with the user dashboard.

Parameters:

  • device_name (str) – name of the device to set the message for

  • message (str) – message to be set

sync_device_status()[source]#

Sync the device status (usually when the system is set up).

Some devices may still be running, so it is not usable now. We will set the status to OCCUPIED

unpause_device(device_name)[source]#

Unpause a device.

Contents