radloggerpy.device.device_manager module

class radloggerpy.device.device_manager.DeviceManager

Bases: object

Factory for device creation and management

The following theory of operation is not finalized and alternative solutions are not only welcome but encouraged:

Each device is run on the threadpool and gets scheduled and descheduled in accordance to the number of concurrent workers. Devices are expected to check for data and subsequently return to sleep upon waking up. The amount of time in between wake-ups should be long enough to give other devices time to retrieve data but short enough to have relevant timing data.

Devices are expected to run in a endless loop, upon returning they will NOT get automatically rescheduled back into the queue of the threadpool. Since all devices inherit the Device super class this class will provide methods to store data. The storage and retrieval methods for data can be assumed to be thread-safe by the device.

The polling rate of DeviceManager to retrieve data from devices depends on the /systems/ used to store data permanently. Some online platforms do not allow to specify timestamps while uploading data. This in turn requires a high polling rate to be able to ensure measurements get uploaded with accurate time information.

_DEVICE_MAP = None

Private Map of device types and corresponding implementations

class _I

Bound to radloggerpy.device.device.Device

alias of TypeVar(‘_I’, bound=Device)

class _U

Bound to radloggerpy.database.objects.device.DeviceObject

alias of TypeVar(‘_U’, bound=DeviceObject)

static _get_device_module(module)

_mng_devices

List of ManagedDevice devices see ManagedDevice

check_devices()

Check the status of the devices and handle failures

TODO(Dantali0n): This method should use the get_state method of devices

instead of relying on the futures to much.

static get_device_class(device_obj: Type[_U]) → Type[_I]

Determines the matching device class for a device object

The device object must specify a concrete implementation

static get_device_implementations()

Return a collection of all device implementations

Access implementations their INTERFACE to determine how they map to radloggerpy.types.device_interfaces.DeviceInterfaces

Returns:

Return type:

static get_device_interfaces()

Return a collection of all device interfaces their abstract classes

Access abstract classes their INTERFACE to determine how they map to radloggerpy.types.device_interfaces.DeviceInterfaces

Returns:

Return type:

static get_device_map()

Return dictionary mapping device types to all concrete classes

The map will only be generated the first time this method is called and is subsequently stored in _DEVICE_MAP.

The dictionary structure follows the following schema:

OrderedDict([(DeviceInterfaces.SERIAL,[devices.ArduinoGeigerPcb])])

Returns:

Ordered dictionary mapping DeviceInterface enums to concrete classes

Return type:

OrderedDict with DeviceTypes as key and lists as values

launch_device(device_obj: Type[_U])

Submit the device and its parameter to the threadpool

Submitted devices are maintained as ManagedDevice instances, this enables to correlate a future to its corresponding device.

class radloggerpy.device.device_manager.ManagedDevice(future: Future, device: Type[_I])

Bases: object

Small data structure to keep track of running devices

class _I

Bound to radloggerpy.device.device.Device

alias of TypeVar(‘_I’, bound=Device)

consecutive_errors: int = 0

device: Device

future: Future