Writing devices

Get the code

Concert is developed using Git on the popular GitHub platform. To clone the repository call:

$ git clone https://github.com/ufo-kit/concert

To get started you are encouraged to install the development dependencies via pip:

$ cd concert
$ pip install -r requirements.txt

After that you can simply install the development source with

$ make install

Run the tests

The core of Concert is tested using Python’s standard library unittest module and nose. To run all tests, you can call nose directly in the root directory or run make with the check argument

$ make check

Some tests take a lot of time to complete and are marked with the @slow decorator. To skip them during regular development cycles, you can run

$ make check-fast

You are highly encouraged to add new tests when you are adding a new feature to the core or fixing a known bug.

Basic concepts

The core abstraction of Concert is a Parameter. A parameter has at least a name but most likely also associated setter and getter callables. Moreover, a parameter can have units and limiters associated with it.

The modules related to device creation are found here

concert/
|-- base.py
`-- devices
    |-- base.py
    |-- cameras
    |   |-- base.py
    |   `-- ...
    |-- __init__.py
    |-- motors
    |   |-- base.py
    |   `-- ...
    `-- storagerings
        |-- base.py
        `-- ...

Adding a new device

To add a new device to an existing device class (such as motor, pump, monochromator etc.), a new module has to be added to the corresponding device class package. Inside the new module, the concrete device class must then import the base class, inherit from it and implement all abstract method stubs.

Let’s assume we want to add a new motor called FancyMotor. We first create a new module called fancy.py in the concert/devices/motors directory package. In the fancy.py module, we first import the base class

from concert.devices.motors.base import LinearMotor

Our motor will be a linear one, let’s sub-class LinearMotor:

class FancyMotor(LinearMotor):
    """This is a docstring that can be looked up at run-time by the `ddoc`
    tool."""

In order to install all required parameters, we have to call the base constructor. Now, all that’s left to do, is implementing the abstract methods that would raise a AccessorNotImplementedError:

def _get_position(self):
    # the returned value must have units compatible with units set in
    # the Quantity this getter implements
    return self.position

def _set_position(self, position):
    # position is guaranteed to be in the units set by the respective
    # Quantity
    self.position = position

We guarantee that in setters which implement a Quantity, like the _set_position() above, obtain the value in the exact same units as they were specified in the respective Quantity they implement. E.g. if the above _set_position() implemented a quantity with units set in kilometers, the position of the _set_position() will also be in kilometers. On the other hand the getters do not need to return the exact same quantity but the value must be compatible, so the above _get_position() could return millimeters and the user would get the value in kilometers, as defined in the respective Quantity.

Creating a device class

Defining a new device class involves adding a new package to the concert/devices directory and adding a new base.py class that inherits from Device and defines necessary Parameter and Quantity objects.

In this exercise, we will add a new pump device class. From an abstract point of view, a pump is characterized and manipulated in terms of the volumetric flow rate, e.g. how many cubic millimeters per second of a medium is desired.

First, we create a new base.py into the new concert/devices/pumps directory and import everything that we need:

import quantities as q
from concert.base import Quantity
from concert.devices.base import Device

The Device handles the nitty-gritty details of messaging and parameter handling, so our base pump device must inherit from it. Furthermore, we have to specify which kind of parameters we want to expose and how we get the values for the parameters (by tying them to getter and setter callables):

class Pump(Device):

    flow_rate = Quantity(q.m**3 / q.s,
                         lower=0 * q.m**3 / q.s, upper=1 * q.m**3 / q.s,
                         help="Flow rate of the pump")

    def __init__(self):
        super(Pump, self).__init__()

The flow_rate parameter can only receive values from zero to one cubic meter per second.

We didn’t specify explicit fget and fset functions, which is why implicit setters and getters called _set_flow_rate and _get_flow_rate are installed. The real devices then need to implement these. You can however, also specify explicit setters and getters in order to hook into the get and set process:

class Pump(Device):

    def __init__(self):
        super(Pump, self).__init__()

    def _intercept_get_flow_rate(self):
        return self._get_flow_rate() * 10

    flow_rate = Parameter(unit=q.m**3 / q.s,
                          fget=_intercept_get_flow_rate)

Be aware, that in this case you have to list the parameter after the functions that you want to refer to.

In case you want to specify the name of the accessor function yourself and rely on implementation by subclasses, you have to raise an AccessorNotImplementedError:

class Pump(Device):

    ...

    def _set_flow_rate(self):
        raise AccessorNotImplementedError

State machine

A formally defined finite state machine is necessary to ensure and reason about correct behaviour. Concert provides an implicitly defined, decorator-based state machine. All you need to do is declare a State object on the base device class and apply the transition() decorator on each method that changes the state of a device:

from concert.fsm import State, transition

class Motor(Device):

    state = State(default='open')

    ...

    @transition(source='standby', target='moving')
    def start_moving(self):
        ...

If the source state is valid on such a device, start_moving will run and eventually change the state to moving. In case of two-step functions, an immediate state can be set that is valid throughout the body of the function:

@transition(source='standby', target='standby', immediate='moving')
def move(self):
    ...

Besides single state strings you can also add arrays of strings and a catch-all * state that matches all states.

In some cases it might be necessary to reach more than one target state. For this, you can pass a list of possible target state and must provide a check function that returns the current state. It is called after the decorated function was called:

@transition(source='here', target=['this', 'that'], check=func)
def do_something(self):
    ...

There is no explicit error handling implemented but it can be easily modeled by adding error states and reset functions that transition out of them.

Parameters

In case changing a parameter value causes a state transition, add a transition() to the Parameter object:

class Motor(Device):

    state = State(default='standby')

    velocity = Parameter(unit=q.m / q.s,
                         transition(source='*', target='moving'))