tango_utils module

Generic utility functions for working with Tango devices.

aiv_utils.tango_utils.format_trls(devs)

Returns a compact representation of device names by factoring out a common prefix.

>>> format_trls(["low-mccs/tile/s8-1-tpm15", "low-mccs/tile/s8-1-tpm16"])
'low-mccs/tile/s8-1-(tpm15|tpm16)'

>>> format_trls(["low-mccs/subarray/01", "low-sdp/subarray/01"])
'low-(mccs|sdp)/subarray/01'

>>> format_trls(["low-mccs/station/s8-1", "low-mccs/spsstation/s8-1"])
'low-mccs/(spsstation|station)/s8-1'

>>> format_trls(["domain/family-a/1", "domain/family-a/2", "domain/family-b/1"])
'domain/family-a/1, domain/family-a/2, domain/family-b/1'

>>> format_trls([])
''
aiv_utils.tango_utils.fq_dev_name(dev)

Given a DeviceProxy, return a fully-qualified device name.

Return type:

str

class aiv_utils.tango_utils.group(name)

Like a tango.Group, but iterable, and sets source and timeout.

aiv_utils.tango_utils.member(dev)

Return the member part of the device TRL.

For example, if the TRL is “low-mccs/tile/s8-1-tpm01”, then the member part is “s8-1-tpm01”.

Return type:

str

Returns:

the member part of the device TRL.

aiv_utils.tango_utils.prop(dev, prop_name, astype=None)

Return a list of all values for a device property, optionally coerced to astype.

Return type:

str

aiv_utils.tango_utils.raise_for_command_result(result, ok_codes=frozenset({ResultCode.OK, ResultCode.STARTED, ResultCode.QUEUED}))

Takes a command result, and raises an exception unless it indicates success.

aiv_utils.tango_utils.read_all(devs, attr)

Uses GreenMode.Futures to read the same attr on all devs.

Return type:

list

aiv_utils.tango_utils.restart_devices(devs, force_restartserver=False)

Restarts devs without affecting other devices.

If dev is the only device in its device server, restarts the device server using DServer.RestartServer(). Otherwise, restarts just this device via Init().

aiv_utils.tango_utils.single_prop(dev, prop_name)

Return the first value of a device property.

Properties are always lists but many properties only expect a single value. This convenience method makes accessing those less verbose.

Return type:

str

aiv_utils.tango_utils.tango_attr_value(dev, attr, default=<object object>)

Like getattr(dev, attr, default), but using item lookup rather than attribute access.

aiv_utils.tango_utils.wait_for(devs, attr, desired_value, failed_value=None, timeout=60.0, quiet=False, drop_initial=False)

Block until attr of each of devs has a certain value or matches a predicate.

Parameters:

  • devs (Union[DeviceProxy, Iterable[DeviceProxy]]) – a DeviceProxy or a list of DeviceProxy

  • attr (str) – the name of the attribute to wait for

  • desired_value – an attribute value to wait for, or a custom predicate which may accept either one argument (the value) or two (the value, and the device)

  • failed_value – the value of the attribute which if reached to add the device in the failed list and not wait for any more updates on it. When every other device has reached either the desired value or this value, raise an appropriate exception. Also accepts a custom predicate.

  • timeout (float) – maximum time to wait for each device to meet the condition

  • quiet (bool) – don’t log anything

Return type:

None

Returns:

None

aiv_utils.tango_utils.wait_lrc(dev, cmd, *args, timeout=15, quiet=True)

Waits for a long-running command to complete.

Parameters:

  • dev (DeviceProxy) – a DeviceProxy

  • cmd (str) – the name of the long-running command to execute

  • args – arguments to execute the command with

  • timeout – maximum time to wait for the command to reach status

  • quiet – don’t log anything

Returns:

None