ska_pst.lmc.device_proxy

This module provides a factory for tango.DeviceProxy instances.

This code is based off the SKA TANGO Examples class.

class ska_pst.lmc.device_proxy.ChangeEventSubscription(subscription_id: int, device: PstDeviceProxy, callbacks: List[Callable], attribute_name: str, logger: Logger)[source]

Class to act as a handle for a change event subscription.

Instances of this class can be used to programmatically unsubscribe from a change event, without having to have access to the device or the subscription id.

property callbacks: List[Callable]: Get callbacks for current subscription.

property subscribed: bool: Check if subscription is still subscribed.

unsubscribe() → None[source]

Unsubscribe to the change event.

Use this to method to unsubscribe to listening to a change event of as device. As this is potentially called from a Python thread this will try to run this within a TANGO OmniThread using a background thread.

class ska_pst.lmc.device_proxy.DeviceProxyFactory[source]

Simple factory to create tango.DeviceProxy instances.

This class is an easy attempt to develop the concept developed by MCCS team in the following confluence page: https://confluence.skatelescope.org/display/SE/Running+BDD+tests+in+multiple+harnesses

It is a factory class which provide the ability to create an object of type DeviceProxy. If a proxy had already been created it will reuse that instance.

When testing the static variable _test_context is an instance of the TANGO class MultiDeviceTestContext.

More information on tango testing can be found at the following link: https://pytango.readthedocs.io/en/stable/testing.html

classmethod get_device(fqdn: str, green_mode: tango.GreenMode = tango.GreenMode.Synchronous, logger: Optional[Logger] = None) → PstDeviceProxy[source]

Return a :py:class::PstDeviceProxy.

This will return an existing proxy if already created, else it will create a tango.DeviceProxy and then wrap it as a :py:class::PstDeviceProxy.

Parameters

fqdn – the FQDN of the TANGO device that the proxy is for.
green_mode – the TANGO green mode, the default is GreenMode.Synchronous.
logger – the Python logger to use for the proxy.

class ska_pst.lmc.device_proxy.PstDeviceProxy(fqdn: str, logger: Logger, device: tango.DeviceProxy)[source]

A DeviceProxy wrapper class.

This class is used to wrap device proxies and abstract away from the TANGO API. This class is designed to provide passthrough methods/attributes that would already be available.

At the moment this is a very simple API wrapper but could be built up, like what is done in MCCS that allows the device’s to try to connect and wait for being initialised.

property device: tango.DeviceProxy: Get TANGO Device Proxy object.

is_subscribed_to_events(attribute_name: str) → bool[source]

Check if there is an active event subscription for attribute.

Checks if there is a ChangeEventSubscription for the attribute and if it is actively subscribed.

Parameters: attribute_name – the name of the attribute to check if there is an active event subscription.

subscribe_change_event(attribute_name: str, callback: Callable, stateless: bool = False) → ChangeEventSubscription[source]

Subscribe to change events.

This method is used to subscribe to an attribute changed event on the given proxy: object. This is similar to:

device.subscribe_event(
    attribute_name,
    tango.EventType.CHANGE_EVENT,
    callback,
    stateless=stateless,
)

This method also returns a ChangeEventSubscription which can be used to: later unsubscribe from change events on the device proxy.

Parameters

attribute_name – the name of the attribute on the device proxy to subscribe to.
callback – the callback for TANGO to call when an event has happened.
stateless – whether to use the TANGO stateless event model or not, default is False.

Returns

a ChangeEventSubscription that can be used to later to unsubscribe from.

unsubscribe_change_event(subscription: ChangeEventSubscription) → None[source]

Unsubscribe to change events for a given subscription.

This method is used to unsubscribe to an attribute changed event on the given: proxy object. This is similar to:

device.unsubscribe_event(subscription_id)

Parameters: subscription – the subscription to unsubscribe to.

wait_for_initialised() → None[source]

Wait for the device proxy is up and initalised.

Only after the device is initialised, device.state() != DevState.INIT is the device ready to be used.