alab_management.device_view.device module

alab_management.device_view.device module#

Define the base class of devices.

class BaseDevice(name, description=None, *args, **kwargs)[source]#

Bases: ABC

The abstract class of device.

All the devices should be inherited from this class

name(str)#

Type:: the name of device, which is the unique identifier of this device

description(Optional[str])#

the device type, how to set up and so on.

Type:: description of this kind of device, which can include

args#

Type:: arguments that will be passed to the device class

kwargs#

Type:: keyword arguments that will be passed to the device class

abstract connect()[source]#

Connect to any devices here. This will be called by alabos to make connections to devices at the appropriate time.

This method must be defined even if no device connections are required! Just return in this case.

abstract property description: str#: A short description of the device. This will be stored in the database + displayed in the dashboard. This must be declared in subclasses of BaseDevice!.

dict_in_database(name, default_value=None)[source]#

Create a dict attribute that is stored in the database. Note: nested dicts/lists are not supported!.

Parameters:

name (str) – The name of the attribute
default_value (Optional[dict]) – The default value of the attribute. if None (default), will default to an empty dict.

Return type:

DictInDatabase

Returns:

Class instance to access the attribute. Acts like a normal Dict, but is stored in the database.

abstract disconnect()[source]#

Disconnect from devices here. This will be called by alabos to release connections to devices at the appropriate time.

This method must be defined even if no device connections are required! Just return in this case.

get_message()[source]#

Returns the device message to be displayed on the dashboard.

Note: this method is used instead of python getter/setters because the DeviceWrapper can currently only access methods, not properties.

Return type:: str

abstract is_running()[source]#

Check whether this device is running.

Return type:: bool

list_in_database(name, default_value=None)[source]#

Create a list attribute that is stored in the database. Note: nested dicts/lists are not supported!.

Parameters:

name (str) – The name of the attribute
default_value (Optional[list]) – The default value of the attribute. if None (default), will default to an empty list.

Return type:

ListInDatabase

Returns:

Class instance to access the attribute. Acts like a normal List, but is stored in the database.

request_maintenance(prompt, options)[source]#

Request maintenance input from the user. This will display a prompt to the user and wait for them to select an option. The selected option will be returned.

Parameters:

prompt (str) – the text to display to the user
options (list[Any]) – the options to display to the user. This should be a list of strings.

retrieve_signal(signal_name, within=None)[source]#

Retrieve a signal from the database.

Parameters:

signal_name (str) – device signal name. This should match the signal_name passed to the @log_device_signal decorator
within (Optional[datetime.timedelta], optional) – timedelta defining how far back to pull logs from (relative to current time). Defaults to None.

Returns:

Dict: Dictionary of signal result. Single value vs lists depends on whether within was None or not, respectively. Form is: { “device_name”: “device_name”, “signal_name”: “signal_name”, “value”: “signal_value” or [“signal_value_1”, “signal_value_2”, …]], “timestamp”: “timestamp” or [ “timestamp_1”, “timestamp_2”, …] }

abstract property sample_positions: list[SamplePosition]#

The sample positions describe the position that can hold a sample. The name of sample position will be the unique identifier of this sample position. It does not store any coordinates information about where the position is in the lab. Users need to map the sample positions to real lab coordinates manually.

Note

It doesn’t matter in which device class a sample position is defined. We use name attribute to identify them.

Here is an example of how to define some sample positions

@property
def sample_positions(self):
    return [
        SamplePosition(
            "inside",
            description="The position inside the furnace, where the samples are heated",
            number=8,
        ),
        SamplePosition(
            "furnace_table",
            description="Temporary position to transfer samples",
            number=16,
        ),
    ]

set_message(message)[source]#

Sets the device message to be displayed on the dashboard.

Note: this method is used instead of python getter/setters because the DeviceWrapper can currently only access methods, not properties.

class DeviceSignalEmitter(device)[source]#

Bases: object

This class is responsible for periodically logging device signals to the database. It is intended to be used as a singleton, and should be instantiated once per device.

get_methods_to_log()[source]#

Log the data from all methods decorated with @log_signal to the database.

Collected all the methods that are decorated with @log_signal and return a dictionary of the form:

{
    <method_name>: {
        "interval": <interval_seconds>,
        "signal_name": <signal_name>
    }
}

log_method_to_db(method_name, signal_name)[source]#

Logs a method to the database. This is called by the worker thread.

Parameters:

method_name (str) – the name of the method to call on the device
signal_name (str) – the name of the signal to log to the database

Exceptions:: Any exceptions raised by the method call will be caught and raised directly.

retrieve_signal(signal_name, within=None)[source]#

Retrieve a signal from the database.

Parameters:

signal_name (str) – device signal name. This should match the signal_name passed to the @log_device_signal decorator
within (Optional[datetime.timedelta]) – timedelta defining how far back to pull logs from (relative to current time). Defaults to None.

Returns:

Dict: Dictionary of signal result. Single value vs lists depends on whether within: was None or not, respectively. Form is:

{: “device_name”: “device_name”, “signal_name”: “signal_name”, “value”: “signal_value” or [“signal_value_1”, “signal_value_2”, …]], “timestamp”: “timestamp” or [“timestamp_1”, “timestamp_2”, …]

}

start()[source]#: Start the logging worker thread. This will start logging all methods decorated with @log_signal to the database.

stop()[source]#: Stop the logging worker thread. This will stop logging all.

add_device(device)[source]#: Register a device instance. It is stored in a global dictionary.

get_all_devices()[source]#

Get all the device names in the device registry. This is a shallow copy of the registry.

Return type:

dict[str, BaseDevice]

Returns:

A dictionary of all the devices in the registry. The keys are the device names,: and the values are the device instances.

log_signal(signal_name, interval_seconds)[source]#

This is a decorator for methods within a BaseDevice. Methods decorated with this will be called at the specified interval and the result will be logged to the database under the signal_name provided. The intended use is to track process variables (like a furnace temperature, a pressure sensor, etc.) whenever the device is connected to alabos.

Parameters:

signal_name (str) – Name to attribute to this signal
interval_seconds (int) – Interval at which to log this signal to the database.

mock(return_constant=<function _UNSPECIFIED>, object_type=<function _UNSPECIFIED>)[source]#

A decorator used for mocking functions during simulation.

Parameters:

return_constant (Any, optional) – The constant value to be returned by the mocked function. It can be a value (str, int, float, bool), list of values, or a dictionary specifying return values for keys. Default is None.
object_type (Union[List[Any], Any], optional) – The type or list of types to mock if the function returns an object. Default is None.

Returns:

Decorator function used to mock other functions during simulation.

Raises:

ValueError – If both return_constant and object_type are specified.:
ValueError – If return_constant is not of types: str, int, float, bool, list, or dict.:
ValueError – If object_type is specified and not a list or a class type.:

Note

The decorator mocks the function during simulation based on specified constant values or object types.

Examples

Mocking a function with a constant return value:

@mock(return_constant=42)
def get_data() -> int:
    ...
    a = some_integer
    return a

Mocking a function that returns multiple values in a dictionary:

@mock(return_constant={"twotheta": [0.1, 0.2, 0.3], "counts": [100, 200, 300]})
def run_simulation() -> dict:
    ...
    a = {twotheta: [0.1, 0.2, 0.3]}
    b = {counts: [100, 200, 300]}
    return {**a, **b}

Mocking a function that returns a specific object type:

@mock(object_type=str)
def create_mock_string() -> str:
    return "Mocked String"

Mocking a function that returns a single object type:

from alab_control.ohaus_scale import OhausScale as ScaleDriver

@mock(object_type=ScaleDriver)
def get_driver(self):
    self.driver = ScaleDriver(ip=self.ip_address, timeout=self.TIMEOUT)
    self.driver.set_unit_to_mg()
    return self.driver

Mocking a function that returns a list of object types:

from alab_control.furnace_2416 import FurnaceController
from alab_control.door_controller import DoorController

@mock(object_type=[FurnaceController, DoorController])
def get_driver(self):
    self.driver = FurnaceController(port=self.com_port)
    return self.driver, self.door_controller

alab_management.device_view.device module

Contents

alab_management.device_view.device module#