Core objects

Parameters

class concert.base.Parameter(fget=None, fset=None, data=None, check=None, help=None)

A parameter with getter and setter.

Parameters are similar to normal Python properties and can additionally trigger state checks. If fget or fset is not given, you must implement the accessor functions named _set_name and _get_name:

from concert.base import Parameter, State

class SomeClass(object):

    state = State(default='standby')

    def actual(self):
        return 'moving'

    param = Parameter(check=check(source='standby',
                                            target=['standby', 'moving'],
                                            check=actual))

    def _set_param(self, value):
        pass

    def _get_param(self):
        pass

When a Parameter is attached to a class, you can modify it by accessing its associated ParameterValue with a dictionary access:

obj = SomeClass()
print(obj['param'])

fget is a callable that is called when reading the parameter. fset is called when the parameter is written to.

data is passed to the state check function.

check is a check() that changes states when a value is written to the parameter.

help is a string describing the parameter in more detail.

class concert.base.ParameterValue(instance, parameter)

Value object of a Parameter.

get(*args, **kwargs)

Get concrete value of this object.

If wait_on is not None, it must be a future on which this method joins.

lock(permanent=False)

Lock parameter for writing. If permament is True the parameter cannot be unlocked anymore.

locked

Return True if the parameter is locked for writing.

restore()

Restore the last value saved with ParameterValue.stash().

If the parameter can only be read or no value has been saved, this operation does nothing.

set(*args, **kwargs)

Set concrete value on the object.

If wait_on is not None, it must be a future on which this method joins.

stash(*args, **kwargs)

Save the current value internally on a growing stack.

If the parameter is writable the current value is saved on a stack and to be later retrieved with ParameterValue.restore().

unlock()

Unlock parameter for writing.

wait(value, sleep_time=<Quantity(0.1, 'second')>, timeout=None)

Wait until the parameter value is value. sleep_time is the time to sleep between consecutive checks. timeout specifies the maximum waiting time.

writable

Return True if the parameter is writable.

class concert.base.Quantity(unit, fget=None, fset=None, lower=None, upper=None, data=None, check=None, help=None)

Bases: concert.base.Parameter

A Parameter associated with a unit.

fget, fset, data, check and help are identical to the Parameter constructor arguments.

unit is a Pint quantity. lower and upper denote soft limits between the Quantity values can lie.

class concert.base.QuantityValue(instance, quantity)

Bases: concert.base.ParameterValue

lock_limits(permanent=False)

Lock limits, if permanent is True the limits cannot be unlocker anymore.

unlock_limits()

Unlock limits.

wait(value, eps=None, sleep_time=<Quantity(0.1, 'second')>, timeout=None)

Wait until the parameter value is value. eps is the allowed discrepancy between the actual value and value. sleep_time is the time to sleep between consecutive checks. timeout specifies the maximum waiting time.

Collection of parameters

class concert.base.Parameterizable

Collection of parameters.

For each class of type Parameterizable, Parameter can be set as class attributes

class Device(Parameterizable):

    def get_something(self):
        return 'something'

    something = Parameter(get_something)

There is a simple Parameter and a parameter which models a physical quantity Quantity.

A Parameterizable is iterable and returns its parameters of type ParameterValue or its subclasses

for param in device:
    print("name={}".format(param.name))

To access a single name parameter object, you can use the [] operator:

param = device['position']
print param.is_readable()

If the parameter name does not exist, a ParameterError is raised.

Each parameter value is accessible as a property. If a device has a position it can be read and written with:

param.position = 0 * q.mm
print param.position
install_parameters(params)

Install parameters at run-time.

params is a dictionary mapping parameter names to Parameter objects.

lock(permanent=False)

Lock all the parameters for writing. If permanent is True, the parameters cannot be unlocked anymore.

restore(*args, **kwargs)

Restore all parameters saved with Parameterizable.stash().

stash(*args, **kwargs)

Save all writable parameters that can be restored with Parameterizable.restore().

The values are stored on a stacked, hence subsequent saved states can be restored one by one.

unlock()

Unlock all the parameters for writing.

State machine

class concert.base.State(default=None, fget=None, fset=None, data=None, check=None, help=None)

Finite state machine.

Use this on a class, to keep some sort of known state. In order to enforce restrictions, you would decorate methods on the class with check():

class SomeObject(object):

    state = State(default='standby')

    @check(source='*', target='moving')
    def move(self):
        pass

In case your device doesn’t provide information on its state you can use the transition() to store the state in an instance of your device:

@transition(immediate='moving', target='standby')
def _set_some_param(self, param_value):
    # when the method starts device state is set to *immediate*
    # long operation goes here
    pass
    # the state is set to *target* in the end

Accessing the state variable will return the current state value, i.e.:

obj = SomeObject()
assert obj.state == 'standby'

The state cannot be set explicitly by:

obj.state = 'some_state'

but the object needs to provide methods which transition out of states, the same holds for transitioning out of error states. If the _get_state() method is implemented in the device it is always used to get the state, otherwise the state is stored in software.

concert.base.check(source='*', target=None)

Decorates a method for checking the device state.

source denotes the source state that must be present at the time of invoking the decorated method. target is the state that the state object will be after successful completion of the method or a list of possible target states.

concert.base.transition(immediate=None, target=None)

Change software state of a device to immediate. After the function execution finishes change the state to target.

Devices

class concert.devices.base.Device

Bases: concert.base.Parameterizable

A Device provides locked access to a real-world device.

It implements the context protocol to provide locking:

with device:
    # device is locked
    device.parameter = 1 * q.m
    ...

# device is unlocked again

Asynchronous execution

exception concert.async.KillException

Exception that may be thrown during the execution of an async() decorated function. The function may run cleanup code.

concert.async.async()

A decorator for functions which are executed asynchronously.

concert.async.threaded()

Threaded execution of a function func.

class concert.async.Dispatcher

Core dispatcher

send(sender, message)

Send message from sender.

subscribe(sender, message, handler)

Subscribe to a message sent by sender.

When message is sent by sender, handler is called with sender as the only argument.

unsubscribe(sender, message, handler)

Remove handler from the subscribers to (sender, message).

concert.async.resolve(result)

Return a list of tuples (x, y, ...) from a process that returns a list of futures each returning a single tuple (x, y, ...).

concert.async.wait(futures)

Wait for the list of futures to finish and raise exceptions if happened.

Exceptions

class concert.base.UnitError

Raised when an operation is passed value with an incompatible unit.

class concert.base.LimitError

Raised when an operation is passed a value that exceeds a limit.

class concert.base.ParameterError(parameter)

Raised when a parameter is accessed that does not exists.

class concert.base.AccessorNotImplementedError

Raised when a setter or getter is not implemented.

class concert.base.ReadAccessError(parameter)

Raised when user tries to change a parameter that cannot be written.

class concert.base.WriteAccessError(parameter)

Raised when user tries to read a parameter that cannot be read.

class concert.base.StateError(error_state, msg=None)

Raised in state check functions of devices.

Configuration

concert.config.DISABLE_ASYNC

Disable asynchronous execution by returning a dummy future which is not executed synchronusly.

concert.config.DISABLE_GEVENT

Turn of gevent support and fall back to ThreadPoolExecutor approach.

Sessions

concert.session.utils.code_of(func)

Show implementation of func.

concert.session.utils.ddoc()

Render device documentation.

concert.session.utils.dstate()

Render device state in a table.

concert.session.utils.get_default_table(field_names, widths=None)

Return a prettytable styled for use in the shell. field_names is a list of table header strings.

concert.session.utils.pdoc(hide_blacklisted=True)

Render process documentation.

Networking

Networking package facilitates all network connections, e.g. sockets and Tango.

Socket Connections

class concert.networking.base.SocketConnection(host, port, return_sequence='n')

A two-way socket connection. return_sequence is a string appended after every command indicating the end of it, the default value is a newline (n).

execute(data)

Execute command and wait for response (thread safe).

recv()

Read data from the socket. The result is first stripped from the trailing return sequence characters and then returned.

send(data)

Send data to the peer. The return sequence characters are appended to the data before it is sent.

class concert.networking.aerotech.Connection(host, port)

Aerotech socket connection.

recv()

Return properly interpreted answer from the controller.

TANGO

Tango devices are interfaced by PyTango, one can obtain the DeviceProxy by the get_tango_device() function.

concert.networking.base.get_tango_device(uri, peer=None)

Get a Tango device by specifying its uri. If peer is given change the tango_host specifying which database to connect to. Format is host:port as a string.

Helpers

class concert.helpers.Bunch(values)

Encapsulate a list or dictionary to provide attribute-like access.

Common use cases look like this:

d = {'foo': 123, 'bar': 'baz'}
b = Bunch(d)
print(b.foo)
>>> 123

l = ['foo', 'bar']
b = Bunch(l)
print(b.foo)
>>> 'foo'
class concert.helpers.Command(name, opts)

Command class for the CLI script

Command objects are loaded at run-time and injected into Concert’s command parser.

name denotes the name of the sub-command parser, e.g. “mv” for the MoveCommand. opts must be an argparse-compatible dictionary of command options.

run(*args, **kwargs)

Run the command

exception concert.helpers.WaitError

Raised on busy waiting timeouts

concert.helpers.busy_wait(condition, sleep_time=<Quantity(0.1, 'second')>, timeout=None)

Busy wait until a callable condition returns True. sleep_time is the time to sleep between consecutive checks of condition. If timeout is given and the condition doesn’t return True within the time specified by it a WaitingError is raised.

class concert.helpers.expects(*args, **kwargs)

Decorator which determines expected arguments for the function and also check correctness of given arguments. If input arguments differ from expected ones, exception TypeError will be raised.

For numeric arguments use Numeric class with 2 parameters: dimension of the array and units (optional). E.g. “Numeric (1)” means function expects one number or “Numeric (2, q.mm)” means function expects expression like [4,5]*q.mm

Common use case looks like this:

@expects (Camera, LinearMotor, pixelsize = Numeric(2, q.mm)) def foo(camera, motor, pixelsize = None):

pass
concert.helpers.memoize(func)

Memoize the result of func.

Remember the result of func depending on its arguments. Note, that this requires that the function is free from any side effects, e.g. returns the same value given the same arguments.