Abstract High-level KATCP Client API (resource)

A high-level abstract interface to KATCP clients, sensors and requests.

class katcp.resource.KATCPDummyRequest(name, description, is_active=<function <lambda>>)

Bases: katcp.resource.KATCPRequest

Dummy counterpart to KATCPRequest that always returns a successful reply

Methods

KATCPDummyRequest.is_active() True if resource for this request is active
KATCPDummyRequest.issue_request(*args, **kwargs)
class katcp.resource.KATCPReply

Bases: katcp.resource._KATCPReplyTuple

Container for return messages of KATCP request (reply and informs).

This is based on a named tuple with ‘reply’ and ‘informs’ fields so that the KATCPReply object can still be unpacked into a normal tuple.

Parameters:

reply : katcp.Message object

Reply message returned by katcp request

informs : list of katcp.Message objects

List of inform messages returned by KATCP request

Attributes

KATCPReply.messages List of all messages returned by KATCP request, reply first.
KATCPReply.reply Alias for field number 0
KATCPReply.informs Alias for field number 1
The instance evaluates to nonzero (i.e. truthy) if the request succeeded.  

Methods

KATCPReply.count(...)
KATCPReply.index((value, [start, ...) Raises ValueError if the value is not present.
messages

List of all messages returned by KATCP request, reply first.

succeeded

True if request succeeded (i.e. first reply argument is ‘ok’).

class katcp.resource.KATCPRequest(name, description, is_active=<function <lambda>>)

Bases: object

Abstract Base class to serve as the definition of the KATCPRequest API.

Wrapper around a specific KATCP request to a given KATCP device. Each available KATCP request for a particular device has an associated KATCPRequest object in the object hierarchy. This wrapper is mainly for interactive convenience. It provides the KATCP request help string as a docstring and pretty-prints the result of the request.

Methods

KATCPRequest.is_active() True if resource for this request is active
KATCPRequest.issue_request(*args, **kwargs) Signature as for __call__
description

Description of KATCP request as obtained from the ?help request.

is_active()

True if resource for this request is active

issue_request(*args, **kwargs)

Signature as for __call__

Do the request immediately without checking active state.

name

Name of the KATCP request.

class katcp.resource.KATCPResource

Bases: object

Base class to serve as the definition of the KATCPResource API.

A class C implementing the KATCPResource API should register itself using KATCPResource.register(C) or subclass KATCPResource directly. A complication involved with subclassing is that all the abstract properties must be implemented as properties; normal instance attributes cannot be used.

Attributes

Apart from the abstract properties described below  
TODO Describe how hierarchies are implemented. Also all other descriptions  
here so that the sphinx doc can be autogenerated from here.  

Methods

KATCPResource.is_active()
KATCPResource.list_sensors([filter, ...]) List sensors available on this resource matching certain criteria.
KATCPResource.set_active(active)
KATCPResource.set_sampling_strategies(*args, ...) Set a sampling strategy for all sensors that match the specified filter.
KATCPResource.set_sampling_strategy(*args, ...) Set a sampling strategy for a specific sensor.
KATCPResource.wait(*args, **kwargs) Wait for a sensor in this resource to satisfy a condition.
address

Address of the underlying client/device.

Type: tuple(host, port) or None, with host a string and port an integer.

If this KATCPResource is not associated with a specific KATCP device (e.g. it is only a top-level container for a hierarchy of KATCP resources), the address should be None.

children

AttrDict of subordinate KATCPResource objects keyed by their names.

description

Description of this KATCP resource.

is_connected

Indicate whether the underlying client/device is connected or not.

list_sensors(filter='', strategy=False, status='', use_python_identifiers=True, tuple=False, refresh=False)

List sensors available on this resource matching certain criteria.

Parameters:

filter : string, optional

Filter each returned sensor’s name against this regexp if specified. To ease the dichotomy between Python identifier names and actual sensor names, the default is to search on Python identifier names rather than KATCP sensor names, unless use_python_identifiers below is set to False. Note that the sensors of subordinate KATCPResource instances may have inconsistent names and Python identifiers, better to always search on Python identifiers in this case.

strategy : {False, True}, optional

Only list sensors with a set strategy if True

status : string, optional

Filter each returned sensor’s status against this regexp if given

use_python_identifiers : {True, False}, optional

Match on python identfiers even the the KATCP name is available.

tuple : {True, False}, optional, Default: False

Return backwards compatible tuple instead of SensorResultTuples

refresh : {True, False}, optional, Default: False

If set the sensor values will be refreshed with get_value before returning the results.

Returns:

sensors : list of SensorResultTuples, or list of tuples

List of matching sensors presented as named tuples. The object field is the KATCPSensor object associated with the sensor. Note that the name of the object may not match name if it originates from a subordinate device.

name

Name of this KATCP resource.

parent

Parent KATCPResource object of this subordinate resource, or None.

req

Attribute root/container for all KATCP request wrappers.

Each KATCP request that is exposed on a KATCP device should have a corresponding KATCPRequest object so that calling

resource.req.request_name(arg1, arg2, ...)

sends a ‘?request-name arg1 arg2 ...’ message to the KATCP device and waits for the associated inform-reply and reply messages.

For a KATCPResource object that exposes a hierarchical device it can choose to include lower-level request handlers here such that resource.req.dev_request() maps to resource.dev.req.request().

sensor

Attribute root/container for all KATCP sensor wrappers.

Each KATCP sensor that is exposed on a KATCP device should have a corresponding KATCPSensor object so that

resource.sensor.sensor_name

corresponds to a sensor named e.g. ‘sensor-name’, where the object or attribute name is an escaped/Pythonised version of the original sensor name (see escape_name() for the escape mechanism). Hopefully the device is not crazy enough to have multiple sensors that map to the same Python identifier.

A KATCPResource object that exposes a hierarchical device can choose to include lower-level sensors here such that resource.sensor.dev_sensorname maps to resource.dev.sensor.sensorname.

set_sampling_strategies(*args, **kwargs)

Set a sampling strategy for all sensors that match the specified filter.

Parameters:

filter : string

The regular expression filter to use to select the sensors to which to apply the specified strategy. Use “” to match all sensors. Is matched using list_sensors().

strategy_and_params : seq of str or str

As tuple contains (<strat_name>, [<strat_parm1>, ...]) where the strategy names and parameters are as defined by the KATCP spec. As str contains the same elements in space-separated form.

**list_sensor_args : keyword arguments

Passed to the list_sensors() call as kwargs

Returns:

sensors_strategies : tornado Future

resolves with a dict with the Python identifier names of the sensors as keys and the value a tuple:

(success, info) with

sucess : bool

True if setting succeeded for this sensor, else False

info : tuple

normalised sensor strategy and parameters as tuple if success == True else, sys.exc_info() tuple for the error that occured.

set_sampling_strategy(*args, **kwargs)

Set a sampling strategy for a specific sensor.

Parameters:

sensor_name : string

The specific sensor.

strategy_and_params : seq of str or str

As tuple contains (<strat_name>, [<strat_parm1>, ...]) where the strategy names and parameters are as defined by the KATCP spec. As str contains the same elements in space-separated form.

Returns:

sensors_strategies : tornado Future

resolves with a dict with the Python identifier names of the sensors as keys and the value a tuple:

(success, info) with

sucess : bool

True if setting succeeded for this sensor, else False

info : tuple

normalised sensor strategy and parameters as tuple if success == True else, sys.exc_info() tuple for the error that occured.

wait(*args, **kwargs)

Wait for a sensor in this resource to satisfy a condition.

Parameters:

sensor_name : string

The name of the sensor to check

condition_or_value : obj or callable, or seq of objs or callables

If obj, sensor.value is compared with obj. If callable, condition_or_value(reading) is called, and must return True if its condition is satisfied. Since the reading is passed in, the value, status, timestamp or received_timestamp attributes can all be used in the check.

timeout : float or None

The timeout in seconds (None means wait forever)

Returns:

This command returns a tornado Future that resolves with True when the :

sensor value satisfies the condition, or False if the condition is :

still not satisfied after a given timeout period. :

Raises:

:class:`KATCPSensorError` :

If the sensor does not have a strategy set, or if the named sensor is not present

exception katcp.resource.KATCPResourceError

Bases: exceptions.Exception

Error raised for resource-related errors

exception katcp.resource.KATCPResourceInactive

Bases: katcp.resource.KATCPResourceError

Raised when a request is made to an inactive resource

class katcp.resource.KATCPSensor(sensor_description, sensor_manager)

Bases: object

Wrapper around a specific KATCP sensor on a given KATCP device.

Each available KATCP sensor for a particular device has an associated KATCPSensor object in the object hierarchy. This wrapper is mainly for interactive convenience. It provides the KATCP request help string as a docstring and registers listeners. Subclasses need to call the base class version of __init__().

Methods

KATCPSensor.call_listeners(reading)
KATCPSensor.clear_listeners() Clear any registered listeners to updates from this sensor.
KATCPSensor.get_reading(*args, **kwargs) Get a fresh sensor reading from the KATCP resource
KATCPSensor.get_status(*args, **kwargs) Get a fresh sensor status from the KATCP resource
KATCPSensor.get_value(*args, **kwargs) Get a fresh sensor value from the KATCP resource
KATCPSensor.is_listener(listener)
KATCPSensor.parse_value(s_value) Parse a value from a string.
KATCPSensor.register_listener(listener[, ...]) Add a callback function that is called when sensor value is updated.
KATCPSensor.set(timestamp, status, value) Set sensor with a given received value, matches katcp.Sensor.set()
KATCPSensor.set_formatted(raw_timestamp, ...) Set sensor using KATCP string formatted inputs
KATCPSensor.set_sampling_strategy(strategy) Set current sampling strategy for sensor
KATCPSensor.set_strategy(strategy[, params]) Set current sampling strategy for sensor.
KATCPSensor.set_value(value[, status, timestamp]) Set sensor value with optinal specification of status and timestamp
KATCPSensor.unregister_listener(listener) Remove a listener callback added with register_listener().
KATCPSensor.wait(condition_or_value[, timeout]) Wait for the sensor to satisfy a condition.
clear_listeners()

Clear any registered listeners to updates from this sensor.

get_reading(*args, **kwargs)

Get a fresh sensor reading from the KATCP resource

Returns:reply : tornado Future resolving with KATCPSensorReading object
get_status(*args, **kwargs)

Get a fresh sensor status from the KATCP resource

Returns:reply : tornado Future resolving with KATCPSensorReading object
get_value(*args, **kwargs)

Get a fresh sensor value from the KATCP resource

Returns:reply : tornado Future resolving with KATCPSensorReading object
name

Name of this KATCPSensor

normalised_name

Normalised name of this KATCPSensor that can be used as a python identifier

parent_name

Name of the parent of this KATCPSensor

parse_value(s_value)

Parse a value from a string.

Parameters:

s_value : str

A string value to attempt to convert to a value for the sensor.

Returns:

value : object

A value of a type appropriate to the sensor.

reading

Most recently received sensor reading as KATCPSensorReading instance

register_listener(listener, reading=False)

Add a callback function that is called when sensor value is updated. The callback footprint is received_timestamp, timestamp, status, value.

Parameters:

listener : function

Callback signature: if reading listener(katcp_sensor, reading) where

katcp_sensor is this KATCPSensor instance reading is an instance of KATCPSensorReading

Callback signature: default, if not reading

listener(received_timestamp, timestamp, status, value)

sampling_strategy

Current sampling strategy

set(timestamp, status, value)

Set sensor with a given received value, matches katcp.Sensor.set()

set_formatted(raw_timestamp, raw_status, raw_value, major)

Set sensor using KATCP string formatted inputs

Mirrors katcp.Sensor.set_formatted().

This implementation is empty. Will, during instantiation, be overridden by the set_formatted() method of a katcp.Sensor object.

set_sampling_strategy(strategy)

Set current sampling strategy for sensor

Parameters:

strategy : seq of str or str

As tuple contains (<strat_name>, [<strat_parm1>, ...]) where the strategy names and parameters are as defined by the KATCP spec. As str contains the same elements in space-separated form.

Returns:

done : tornado Future that resolves when done or raises KATCPSensorError

set_strategy(strategy, params=None)

Set current sampling strategy for sensor. Add this footprint for backwards compatibility.

Parameters:

strategy : seq of str or str

As tuple contains (<strat_name>, [<strat_parm1>, ...]) where the strategy names and parameters are as defined by the KATCP spec. As str contains the same elements in space-separated form.

params : seq of str or str

(<strat_name>, [<strat_parm1>, ...])

Returns:

done : tornado Future that resolves when done or raises KATCPSensorError

set_value(value, status=1, timestamp=None)

Set sensor value with optinal specification of status and timestamp

unregister_listener(listener)

Remove a listener callback added with register_listener().

Parameters:

listener : function

Reference to the callback function that should be removed

wait(condition_or_value, timeout=None)

Wait for the sensor to satisfy a condition.

Parameters:

condition_or_value : obj or callable, or seq of objs or callables

If obj, sensor.value is compared with obj. If callable, condition_or_value(reading) is called, and must return True if its condition is satisfied. Since the reading is passed in, the value, status, timestamp or received_timestamp attributes can all be used in the check. TODO: Sequences of conditions (use SensorTransitionWaiter thingum?)

timeout : float or None

The timeout in seconds (None means wait forever)

Returns:

This command returns a tornado Future that resolves with True when the :

sensor value satisfies the condition. It will never resolve with False; :

if a timeout is given a TimeoutError happens instead. :

Raises:

:class:`KATCPSensorError` :

If the sensor does not have a strategy set

:class:`tornado.gen.TimeoutError` :

If the sensor condition still fails after a stated timeout period

exception katcp.resource.KATCPSensorError

Bases: katcp.resource.KATCPResourceError

Raised if a problem occured dealing with as KATCPSensor operation

class katcp.resource.KATCPSensorReading

Bases: katcp.resource.KATCPSensorReading

Sensor reading as a (received_timestamp, timestamp, istatus, value) tuple.

Attributes

KATCPSensorReading.received_timestamp Alias for field number 0
KATCPSensorReading.timestamp Alias for field number 1
KATCPSensorReading.istatus Alias for field number 2
KATCPSensorReading.value Alias for field number 3

Methods

KATCPSensorReading.count(...)
KATCPSensorReading.index((value, [start, ...) Raises ValueError if the value is not present.
status

Returns the string representation of sensor status, eg ‘nominal’

class katcp.resource.KATCPSensorsManager

Bases: object

Sensor management class used by KATCPSensor. Abstracts communications details.

This class should arrange:

  1. A mechanism for setting sensor strategies
  2. A mechanism for polling a sensor value
  3. Keeping track of- and reapplying sensor strategies after reconnect, etc.
  4. Providing local time. This is doing to avoid direct calls to time.time, allowing accelerated time testing / simulation / dry-running

Methods

KATCPSensorsManager.get_sampling_strategy(...) Get the current sampling strategy for the named sensor
KATCPSensorsManager.poll_sensor(sensor_name) Poll sensor and arrange for sensor object to be updated
KATCPSensorsManager.reapply_sampling_strategies() Reapply all sensor strategies using cached values
KATCPSensorsManager.set_sampling_strategy(...) Set the sampling strategy for the named sensor
KATCPSensorsManager.time() Returns the current time (in seconds since UTC epoch)
get_sampling_strategy(sensor_name)

Get the current sampling strategy for the named sensor

Parameters:

sensor_name : str

Name of the sensor

Returns:

strategy : tornado Future that resolves with tuple of str

contains (<strat_name>, [<strat_parm1>, ...]) where the strategy names and parameters are as defined by the KATCP spec

poll_sensor(sensor_name)

Poll sensor and arrange for sensor object to be updated

reapply_sampling_strategies()

Reapply all sensor strategies using cached values

Would typically be called when a connection is re-established. Should not raise errors when resetting stratgies for sensors that no longer exist on the KATCP resource.

set_sampling_strategy(sensor_name, strategy_and_parms)

Set the sampling strategy for the named sensor

Parameters:

sensor_name : str

Name of the sensor

strategy : seq of str or str

As tuple contains (<strat_name>, [<strat_parm1>, ...]) where the strategy names and parameters are as defined by the KATCP spec. As str contains the same elements in space-separated form.

Returns:

done : tornado Future that resolves when done or raises KATCPSensorError

Notes

It is recommended that implementations use normalize_strategy_parameters() to process the strategy_and_parms parameter, since it will deal with both string and list versions and makes sure that numbers are represented as strings in a consistent format.

This method should arrange for the strategy to be set on the underlying network device or whatever other implementation is used. This strategy should also be automatically re-set if the device is reconnected, etc. If a strategy is set for a non-existing sensor, it should still cache the strategy and ensure that is applied whenever said sensor comes into existance. This allows an applications to pre-set strategies for sensors before synced / connected to a device.

time()

Returns the current time (in seconds since UTC epoch)

class katcp.resource.SensorResultTuple

Bases: katcp.resource.SensorResultTuple

Per-sensor result of list_sensors() method

Attributes

SensorResultTuple.object Alias for field number 0
SensorResultTuple.name Alias for field number 1
SensorResultTuple.python_identifier Alias for field number 2
SensorResultTuple.description Alias for field number 3
SensorResultTuple.type Alias for field number 4
SensorResultTuple.units Alias for field number 5
SensorResultTuple.reading Alias for field number 6

Methods

SensorResultTuple.count(...)
SensorResultTuple.index((value, [start, ...) Raises ValueError if the value is not present.
katcp.resource.escape_name(name)

Escape sensor and request names to be valid Python identifiers.

katcp.resource.normalize_strategy_parameters(params)

Normalize strategy parameters to be a list of strings.

Parameters:

params : (space-delimited) string or sequence of strings/numbers Parameters

expected by SampleStrategy object, in various forms, where the first parameter is the name of the strategy.

Returns:

params : tuple of strings

Strategy parameters as a list of strings