device_manager Package¶

High level overview¶

Conctete subclasses of IDeviceManager() represent a silf.backend.commons.device._device.Device to the ExperimentManager(), it has following responsibilities:

Manage modes of the Device
Store and return metadata that enables construction of experiment GUI
Aggregate and modify resuts from the Device to and return modivied results to the user.
Optionally separate Device inside its own subprocess.

Mode management¶

DeviceManager is aware of current mode as it gets it as a parameter for IDeviceManager.power_up() method.

Moreover methods that return metadata, also get mode as a parameter, in this case it means: “Get metadata for appripriate mode”.

Metadata management¶

Each DeviceManager should describe input fields it needs for particular mode. These input fields are returned from IDeviceManager.get_input_fields(). Output fielfs are retuned from IDeviceManager.get_output_fields().

Default DeviceManager implementations allow to define metadata declaratively: see: DefaultDeviceManager and SingleModeDeviceManager.

Result aggregation¶

All DeviceManager concrete implementation derive from ResultMixin that provides sensible way to aggregate results.

Basic device manager classes¶

class silf.backend.commons.device_manager._device_manager.IDeviceManager(experiment_name, config)¶

Bases: object

Instances of this class serve as proxy and translator between main experiment Experiment and Device.

TODO: add link do the experiment

It translated unfiltered input from user to input recognizable to device, it also contains metadata needed by the experiment.

applicable_modes¶

List of modes this device can be in. It is used purely for validation purposes, if we try to power_up() this device in a mode that is outside of applicable_modes exception should be raised.

If value returned is None it means that this device is mode agnostic,

Returns:	Modes applicable for this device
Return type:	list of strings or None.

apply_settings(settings)¶

Applies settings to the device.

Parameters:	settings (dict) – settings to be applied. Can contain settings that are not interesting for this device, these should be filtered out. It contains raw input from json parser, values might need to be converted to python objects.
Raises:	error.SILFProtocolError – if there are any errors in settings, or if there is any error during setting the equipement up.

check_settings(mode, settings)¶

Checks settings for a particular mode. This method is implemented, using native validation of class:.ControlSuite, using get_input_fields().

Parameters:	mode (str) – Mode for which settings are checked. settings (`SettingSuite`) – Settings to be checked.
Raises:	SILFProtocolError – When settings are invalid

config_section = None¶: Section that customizes this device in config file

current_mode¶: Current mode this device is in. :return: Current mode :rtype: str

current_result_map¶

Current results, that is: all results produces in current series. This method returns a plain dictionary.

Returns:	Current results
Return type:	immutable `dict`

current_results¶

Current results, that is: all results produces in current series. This returns class that is serializable and ready to be sent. For developer friendly version use current_result_map().

This property is resetted during stop() method.

Returns:	Current results
Return type:	`ResultSuite`

current_settings¶

Currenty applied settings.

This property is resetted during power_down() method.

Returns:	Currently applied settings. Mapping from setting name to setting value.
Return type:	`dict`

device_name = 'default'¶: Name of the device. Should be unique in the experiment.

device_state¶: State of device managed by this manager.

get_input_fields(mode)¶: Returns _control suite for mode :param str mode: mode for which we get controls :rtype: ControlSuite

get_output_fields(mode)¶

Returns output fields for mode :param str mode: mode for which we get output fields

Returns:	Returns output fields for mode
Return type:	`OutputFieldSuite`

has_result(name)¶

Returns:	Returns true if this instance has result named name in it’s current results. Result is considered valid if it’s value resolves to `True` (not None, if collection is must be not empty, etc).
Return type:	bool

pop_results()¶

Return new results that could be sent to user. This method mutates the state of this instance, immidiately after a call to pop_results() next call will return empty ResultSuite.

Returns:	Recently acquired results
Return type:	`ResultSuite`

power_down(suspend=False)¶

Powers down the devie, if suspend is false it also cleans up the remote device kills and remote proces.

Parameters:	suspend (bool) – If `False` will kill the remote process.

power_up(mode)¶

Powers up the device and sets it’s mode (if applicable).

Parameters:	mode (str) – Mode to initialize device in.

query()¶

This method will be called periodically it should perform any neccesary communication with the device to refresh state of this instane.

It should exit in in less than 50ms.

running¶

If this device is running it means:

That start() was called
That calls to perform_diagnostics(), power_up(), apply_settings(), start() should fail.
Results will be updated periodically.

If this device is not running it means:

Results are None

Returns:	Whether this device is running.
Return type:	`bool`

start(callback=None)¶

Starts data acquisition for this device.

Parameters:	callback – Optional callable to be called when devie is started.

state¶

State of this manager, may different from the device_state

Returns:	State of this manager
Return type:	`str`

stop()¶: Stops data acquisition for the device.

tearDown()¶: Cleans up this instance and releases all attached resources.

tear_down()¶: Cleans up this instance and releases all attached resources.

class silf.backend.commons.device_manager._device_manager.DefaultDeviceManager(*args, **kwargs)¶

Bases: silf.backend.commons.device_manager._device_manager.DeclarativeDeviceMixin, silf.backend.commons.device_manager._device_manager.DeviceWrapperMixin

To create instance of this class you need to specify following attributes:

>>> from silf.backend.commons_test.device.mock_voltimeter import MockVoltimeter
>>> class TestDeviceManager(DefaultDeviceManager):
...     DEVICE_CONSTRUCTOR = MockVoltimeter
...     DEVICE_ID = "voltimeter1"
...     CONTROL_MAP = {
...         "mode1" : ControlSuite(NumberControl("foo", "Foo Control")),
...         "mode2" : ControlSuite(NumberControl("foo", "Foo Control"))
...     }
...     OUTPUT_FIELD_MAP = {
...         "mode2" : OutputFieldSuite(),
...         "mode1" : OutputFieldSuite(bar = OutputField("foo", "bar"))
...     }
>>> mngr = TestDeviceManager("FooExperiment", None)
>>> sorted(mngr.applicable_modes)
['mode1', 'mode2']
>>> mngr.get_output_fields('mode1')
<OutputFieldSuite bar=<Result class=foo name=['bar'] type=array metadata={} settings={}> >
>>> mngr.get_input_fields('mode2')
<ControlSuite <Control name=foo type=number live=False label='Foo Control'>>

class silf.backend.commons.device_manager._device_manager.SingleModeDeviceManager(*args, **kwargs)¶

Bases: silf.backend.commons.device_manager._device_manager.SingleModeDeviceMixin, silf.backend.commons.device_manager._device_manager.DeviceWrapperMixin

To create instance of this class you need to specify following attributes:

>>> from silf.backend.commons_test.device.mock_voltimeter import MockVoltimeter
>>> class TestDeviceManager(SingleModeDeviceManager):
...     DEVICE_CONSTRUCTOR = MockVoltimeter
...     DEVICE_ID = "voltimeter1"
...     CONTROLS = ControlSuite(NumberControl("foo", "Foo Control"))
...     OUTPUT_FIELDS = OutputFieldSuite(bar = OutputField("foo", "bar"))
>>> mngr = TestDeviceManager("FooExperiment", None)
>>> mngr.applicable_modes
>>> mngr.get_output_fields('any-mode')
<OutputFieldSuite bar=<Result class=foo name=['bar'] type=array metadata={} settings={}> >
>>> mngr.get_input_fields('any-mode')
<ControlSuite <Control name=foo type=number live=False label='Foo Control'>>

class silf.backend.commons.device_manager._device_manager.ResultMixin(*args, **kwargs)¶

Bases: silf.backend.commons.device_manager._device_manager.IDeviceManager

This mixin handles getting results from the experiment,

Each entry in RESULT_CREATORS define single result that will be sent to the client. So class in example will send only two results: foo and bar, even if device itself will prowide other results.

Contents of particular result fields (foo and bar) are totally dependent on result creators.

Additionaly you can override ResultMixin._convert_result() that allows for for more extennsive costumistaion.

>>> from silf.backend.commons.device_manager import *
>>> from silf.backend.commons.device_manager._result_creator import *
>>> from silf.backend.commons.util.abc_utils import patch_abc

>>> p = patch_abc(ResultMixin)

>>> class TestResultMixin(ResultMixin):
...     RESULT_CREATORS = [
...         AppendResultCreator("foo"),
...         OverrideResultsCreator("bar")
...     ]
...
...     RESULTS = []
...
...     def _pop_results_internal(self):
...         try:
...             return self.RESULTS
...         finally:
...             self.RESULTS = []

>>> test = TestResultMixin(None, None)

>>> test.RESULTS = [{"foo": [1, 2, 3], "bar": 1.15}, {"foo": [4, 5], "bar" : 3.12}]

>>> test.current_results
<ResultSuite bar=<Result pragma=replace value=3.12 > foo=<Result pragma=append value=[1, 2, 3, 4, 5] > >

>>> test.current_results
<ResultSuite bar=<Result pragma=replace value=3.12 > foo=<Result pragma=append value=[1, 2, 3, 4, 5] > >

>>> test.pop_results()
<ResultSuite bar=<Result pragma=replace value=3.12 > foo=<Result pragma=append value=[1, 2, 3, 4, 5] > >

>>> test.pop_results()
<ResultSuite bar=<Result pragma=replace value=3.12 > >

>>> test.clear_current_results()

>>> test.pop_results()
<ResultSuite >

>>> test.current_results
<ResultSuite >

Clean up: >>> p.stop()

RESULT_CREATORS = []¶: A list that maps str (name of the result recieved from the device) to instance of :class:’result_appender.ResultAggregator`.

_convert_result(dict)¶: Enables to translate between device and experiment results.

_pop_results_internal()¶: Returns raw results from the device. :return: list of dicts.

clear_current_results()¶: Clears current results (that is after this call it is guaranteed that both current_results and pop_results will be empty. :return:

current_results¶

Return type:	`misc.ResultSuite`

pop_results()¶

Returns:

class silf.backend.commons.device_manager._device_manager.SingleModeDeviceMixin(*args, **kwargs)¶

Bases: silf.backend.commons.device_manager._device_manager.IDeviceManager

Mixin for single current_mode device.

get_input_fields(mode)¶: Returns _control suite for current_mode :param str mode: current_mode for which we get controls (Ignored) :return: :rtype: misc.ControlSuite

get_output_fields(mode)¶: Returns output fields for current_mode :param str mode: current_mode for which we get output fields (Ignored) :return: :rtype: _control.OutputFieldSuite

class silf.backend.commons.device_manager._device_manager.DeclarativeDeviceMixin(*args, **kwargs)¶

Bases: silf.backend.commons.device_manager._device_manager.IDeviceManager

>>> from silf.backend.commons.api import *
>>> from silf.backend.commons.util.abc_utils import patch_abc

>>> foo_mode = ControlSuite(
...   NumberControl("foo", "Insert foo", max_value=10),
...   NumberControl("bar", "Insert bar")
... )

>>> bar_mode = ControlSuite(
...   NumberControl("foo", "Insert foo", max_value=10),
...   NumberControl("bar", "Insert bar")
... )

>>> output_fields = _misc.OutputFieldSuite(
...     bar=_misc.OutputField("foo", "bar")
... )

Enable instantiating DeclarativeDeviceMixin (it is an ABC so you shouldn’t be able to do this normallu >>> p = patch_abc(DeclarativeDeviceMixin)

>>> class DeclarativeDeviceMixinTest(DeclarativeDeviceMixin):
...     CONTROL_MAP = {'foo' : foo_mode, 'bar' : bar_mode}
...     OUTPUT_FIELD_MAP = {
...         'foo' : output_fields,
...         'bar' : output_fields,
...     }

>>> manager = DeclarativeDeviceMixinTest(None, None)

Basic operatoons

>>> manager.get_input_fields('foo')
<ControlSuite <Control name=bar type=number live=False label='Insert bar'> <Control name=foo type=number live=False label='Insert foo'>>

>>> manager.get_input_fields('bar')
<ControlSuite <Control name=bar type=number live=False label='Insert bar'> <Control name=foo type=number live=False label='Insert foo'>>

>>> manager.get_input_fields("baz") 
Traceback (most recent call last):
KeyError: 'baz'

>>> manager.get_output_fields("foo")
<OutputFieldSuite bar=<Result class=foo name=['bar'] type=array metadata={} settings={}> >

Clean up: >>> p.stop()

get_input_fields(mode)¶: Returns _control suite for current_mode :param str mode: current_mode for which we get controls :return: :rtype: misc.ControlSuite

get_output_fields(mode)¶: Returns output fields for current_mode :param str mode: current_mode for which we get output fields :return: :rtype: _control.OutputFieldSuite

class silf.backend.commons.device_manager._device_manager.DeviceWrapperMixin(experiment_name, config)¶

Bases: silf.backend.commons.device_manager._device_manager.ResultMixin

Device manager that wraps a device.

AUTOMATICALLY_POOL_RESULTS = True¶: Boolean value. If true device will automatically pool results on each iteration.

BIND_STATE_TO_DEVICE_STATE = True¶: If true state of this device will be synchronized with state of underlying device. State is updated on every call to query().

DEVICE_CONSTRUCTOR = None¶: Type of the device, or callable that takes single argument: the device id.

DEVICE_ID = ''¶: Id of the device

USE_WORKER = True¶: If this is set to true this instance will spawn device in dofferent process, and all methods will be executed asynchroneusly. If set to False methods will be executed in current thread.

Warning

This is sometimes usefull when debugging stuff, but may break stuff in weird ways. Never use it in production.

_apply_settings_to_device(converted_settings)¶: Applies settings to the device. :param dict converted_settings: Result of settings conversion. :return:

_construct_device()¶: Protected method that is used to construct the internal device. :return: WorkerWrapper

_convert_settings_to_device_format(converted_settings)¶: Converts settings from format readable to user, to format readable to device. :param dict converted_settings: Parsed, validated, settings from the user. :return: Converted settings :rtype: dict

static _exception_translator()¶

Wrapper that catches all DeviceException and raises them again as SilfProtocolError.

>>> with exception_translator():
...     raise DeviceException("Bar")
Traceback (most recent call last):
silf.backend.commons.api.exceptions.SILFProtocolError: Bar

>>> with exception_translator():
...     raise TypeError("Foo")
Traceback (most recent call last):
TypeError: Foo

Rtype None:

_load_settings_without_applying_to_device(settings, live)¶: Loads settings (updates self.current_settings) :param SetingsSuite settings: :return:

apply_settings(settings)¶: See IDeviceManager.apply_settings().

perform_diagnostics(diagnostics_level='short')¶: See IDeviceManager.perform_diagnostics().

power_down(suspend=False, wait_time=10)¶

Powers down the device, and optionally kills it.

See IDeviceManager.power_down().

power_up(mode)¶

Creates and powers up the device.

See IDeviceManager.power_up().

running¶: See IDeviceManager.running().

start(callback=None)¶: See IDeviceManager.start().

stop()¶: See IDeviceManager.stop().

tear_down(wait_time=2)¶

Kills the device if device is present.

See IDeviceManager.tearDown().

Contains classes that cope with merging results taken from the device prior to sending them to the client.

class silf.backend.commons.device_manager._result_creator.ResultCreator(result_name, pragma='append', *, read_from=None)¶

Bases: object

Instances of this class get results from the device and produce appropriate Result instances.

aggregate_results(results)¶: Adds results to results cache. :param dict results: dictionary containing many results values.

clear()¶: Clears the results cache.

has_results¶

pop_results() → silf.backend.commons.api.stanza_content._misc.Result¶

Returns current set of results and possibly removes them from results cache.

Returns:	Experiment results or RESULTS_UNSET if there are no new results to return.
Return type:	`Result` or `None`

result_name = None¶: Name of interesting results

class silf.backend.commons.device_manager._result_creator.AppendResultCreator(result_name, pragma='append', **kwargs)¶

Bases: silf.backend.commons.device_manager._result_creator.ResultCreator

Class that aggregates results from source, and empties cache on each call to pop_results().

It is useful if device provides series of points, each of these points is unique, and should not be resent to client. >>> agg = AppendResultCreator(“foo”) >>> agg.has_results False >>> agg.pop_results() <Result pragma=append value=[] > >>> agg.aggregate_results({“foo” : [1, 2], “bar”: [3, 4]}) >>> agg.pop_results() <Result pragma=append value=[1, 2] > >>> agg.pop_results() <Result pragma=append value=[] > >>> agg.aggregate_results({“foo” : [1, 2], “bar”: [3, 4]}) >>> agg.aggregate_results({“foo” : [2, 3], “bar”: [3, 4]}) >>> agg.has_results True >>> agg.pop_results() <Result pragma=append value=[1, 2, 2, 3] > >>> agg.pop_results() <Result pragma=append value=[] > >>> agg.has_results False

aggregate_results(results)¶

clear_results(results_to_clear=0)¶

has_results¶

pop_results()¶

class silf.backend.commons.device_manager._result_creator.OverrideResultsCreator(result_name, pragma='replace', *, read_from=None)¶

Bases: silf.backend.commons.device_manager._result_creator.ResultCreator

Class that always returns last result. Useful for returning device state to user, and when single result compromises whole result set.

>>> agg = OverrideResultsCreator("foo")
>>> agg.has_results
False
>>> agg.pop_results()
<Result pragma=replace value=[] >
>>> agg.aggregate_results({"foo": [1, 2, 3, 4, 5, 6], "bar": "ignored"})
>>> agg.has_results
True
>>> agg.pop_results()
<Result pragma=replace value=[1, 2, 3, 4, 5, 6] >
>>> agg.pop_results()
<Result pragma=replace value=[1, 2, 3, 4, 5, 6] >
>>> agg.aggregate_results({"foo":  [7, 8, 9, 10, 11, 12], "bar": "ignored"})
>>> agg.pop_results()
<Result pragma=replace value=[7, 8, 9, 10, 11, 12] >
>>> agg.aggregate_results({"bar": "ignored"})
>>> agg.pop_results()
<Result pragma=replace value=[7, 8, 9, 10, 11, 12] >
>>> agg = OverrideResultsCreator("bar", read_from="foo")
>>> agg.has_results
False
>>> agg.pop_results()
<Result pragma=replace value=[] >
>>> agg.aggregate_results({"foo": [1]})
>>> agg.pop_results()
<Result pragma=replace value=[1] >

aggregate_results(results)¶

clear()¶

has_results¶

pop_results()¶

class silf.backend.commons.device_manager._result_creator.XYChartResultCreator(result_name, pragma='replace', *, read_from=())¶

Bases: silf.backend.commons.device_manager._result_creator.ResultCreator

Create this result creator

>>> creator = XYChartResultCreator(result_name="baz", read_from=['foo', 'bar'])

At the beginning it is empty

>>> creator.pop_results()
<Result pragma=replace value=[] >
>>> creator.has_results
False

Now let’s append some results

>>> creator.aggregate_results({"foo": [1, 2, 3], "bar": [-1, -2, -3]})
>>> creator.has_results
True
>>> creator.pop_results()
<Result pragma=replace value=[[1, -1], [2, -2], [3, -3]] >

Results are stored

>>> creator.has_results
True
>>> creator.pop_results()
<Result pragma=replace value=[[1, -1], [2, -2], [3, -3]] >

Results are cleared after call to clear:

>>> creator.clear()
>>> creator.has_results
False
>>> creator.pop_results()
<Result pragma=replace value=[] >

After appending only one coordinate we don’t produce results

>>> creator.aggregate_results({"foo": [1, 2, 3]})
>>> creator.has_results
False
>>> creator.pop_results()
<Result pragma=replace value=[] >

After adding second coordinate we produce them:

>>> creator.aggregate_results({"bar": [1]})
>>> creator.has_results
True
>>> creator.pop_results()
<Result pragma=replace value=[[1, 1]] >

>>> creator = XYChartResultCreator(result_name="baz", read_from=['foo', 'bar'])
>>> creator.aggregate_results({"foo": [1, 1, 1], "bar": [-1, -2, -3]})
>>> creator.pop_results()
<Result pragma=replace value=[[1, -3]] >

aggregate_results(results)¶

clear()¶

has_results¶

pop_results()¶

class silf.backend.commons.device_manager._result_creator.ReturnLastElementResultCreator(result_name, pragma='append', *, read_from=None)¶

Bases: silf.backend.commons.device_manager._result_creator.ResultCreator

>>> agg = ReturnLastElementResultCreator("foo")
>>> agg.has_results()
False
>>> agg.pop_results()
<Result pragma=append value=None >
>>> agg.aggregate_results({'foo' : [1, 2, 3]})
>>> agg.pop_results()
<Result pragma=append value=3 >
>>> agg.pop_results()
<Result pragma=append value=3 >

aggregate_results(results)¶

clear()¶

has_results()¶

pop_results()¶

Series manager¶

This package contains contains device managers for one use-case, we want student to capture series of points, and device is able to capture only single point at a time.

class silf.backend.commons.device_manager._series_manager.ISeriesManager(experiment_name, config)¶

Bases: silf.backend.commons.device_manager._series_manager.AbstractSeriesManager

apply_settings(settings)¶

get_number_of_points_in_series(settings)¶

Calculates number of points in series

Parameters:	settings (dict) – Mapping from setting name to setting value.
Returns:	Number of points in series.
Return type:	`int`

is_series_finished¶

class silf.backend.commons.device_manager._series_manager.SingleModeSeriesManager(experiment_name, config)¶: Bases: silf.backend.commons.device_manager._series_manager.ISeriesManager, silf.backend.commons.device_manager._device_manager.SingleModeDeviceMixin

class silf.backend.commons.device_manager._series_manager.MultiModeSeriesManager(experiment_name, config)¶: Bases: silf.backend.commons.device_manager._series_manager.ISeriesManager, silf.backend.commons.device_manager._device_manager.DeclarativeDeviceMixin

exception silf.backend.commons.device_manager._series_manager.OnFinalSeriesPoint¶: Bases: Exception