Component subpackage

This module implements infrastructure for component management in SAT.LMC.

class DeviceComponentManager(fqdn, logger, communication_status_changed_callback, component_state_callback, health_changed_callback=None)

An abstract component manager for a Tango device component.

__init__(fqdn, logger, communication_status_changed_callback, component_state_callback, health_changed_callback=None)

Initialise a new instance.

Parameters:

  • fqdn (str) – the FQDN of the device

  • logger (Logger) – the logger to be used by this object.

  • communication_status_changed_callback (Callable[[CommunicationStatus], None]) – callback to be called when the status of the communications channel between the component manager and its component changes

  • component_state_callback (Callable) – callback to be called when the component faults (or stops faulting)

  • health_changed_callback (Optional[Callable[[Optional[HealthState]], None]]) – callback to be called when the health state of the device changes. The value it is called with will normally be a HealthState, but may be None if the admin mode of the device indicates that the device’s health should not be included in upstream health rollup.

property health: HealthState | None

Return the evaluated health state of the device.

This will be either the health state that the device reports, or None if the device is in an admin mode that indicates that its health should not be rolled up.

Returns:

the evaluated health state of the device.

start_communicating()

Establish communication with the component, then start monitoring.

This is a public method that enqueues the work to be done.

Return type:

None

stop_communicating()

Cease monitoring the component, and break off all communication with it.

Return type:

None

class SatComponentManager(logger, communication_status_callback=None, component_state_callback=None, **kwargs)

A component manager for Sat LMC.

Therefore this class accepts two callback arguments: one for when communication with the component changes and one for when the component state changes. In the last case, callback hooks are provided so that the component can indicate the change to this component manager.

__init__(logger, communication_status_callback=None, component_state_callback=None, **kwargs)

Initialise a new instance.

Parameters:

  • logger (Logger) – a logger for this object to use

  • communication_status_callback (Optional[Callable[[CommunicationStatus], None]]) – callback to be called when the state of communications between the component manager and its component changes.

  • component_state_callback (Optional[Callable[[bool], None]]) – callback to be called when the state of the component changes.

  • kwargs (Any) – other keyword args

abort_commands(task_callback=None)

Abort all tasks queued & running.

Parameters:

task_callback (Optional[Callable]) – callback to be called whenever the status of the task changes.

Return type:

tuple[TaskStatus, str]

Returns:

the completed status and no implemented message

property communication_status: CommunicationStatus

Return the communication status of this component manager.

This is implemented as a replacement for the is_communicating property, which should be deprecated.

Returns:

status of the communication channel with the component.

component_state_changed(fault)

Handle notification that the component’s fault status has changed.

This is a callback hook, to be passed to the managed component.

Parameters:

fault (bool) – whether the component has faulted. If False, then this is a notification that the component has recovered from a fault.

Return type:

None

property fault: bool | None

Return whether this component manager is currently experiencing a fault.

Returns:

whether this component manager is currently experiencing a fault.

property is_communicating: bool

Return communication with the component is established.

SatLmc uses the more expressive communication_status for this, but this is still needed as a base classes hook.

Returns:

whether communication with the component is established.

off(task_callback=None)

Turn the component off.

Parameters:

task_callback (Optional[Callable]) – callback to be called when the status of the command changes

Return type:

tuple[TaskStatus, str]

Returns:

the completed status and not implemented message

on(task_callback=None)

Turn the component on.

Parameters:

task_callback (Optional[Callable]) – callback to be called when the status of the command changes

Return type:

tuple[TaskStatus, str]

Returns:

the completed status and not implemented message

reset(task_callback=None)

Reset the component (from fault state).

Parameters:

task_callback (Optional[Callable]) – callback to be called when the status of the command changes

Return type:

tuple[TaskStatus, str]

Returns:

the completed status and not implemented message

standby(task_callback=None)

Put the component into low-power standby mode.

Parameters:

task_callback (Optional[Callable]) – callback to be called when the status of the command changes

Return type:

tuple[TaskStatus, str]

Returns:

the completed status and not implemented message

start_communicating()

Start communicating with the component.

Return type:

None

stop_communicating()

Break off communicating with the component.

Return type:

None

update_communication_status(communication_status)

Handle a change in communication state.

This is a helper method for use by subclasses.

Parameters:

communication_status (CommunicationStatus) – the new communication status of the component manager.

Return type:

None

update_component_state(fault)

Update the component fault status, calling callbacks as required.

This is a helper method for use by subclasses.

Parameters:

fault (Optional[bool]) – whether the component has faulted. If False, then this is a notification that the component has recovered from a fault.

Return type:

None

check_communicating(func)

Return a function that checks component communication before calling a function.

The component manager needs to have established communications with the component, in order for the function to be called.

This function is intended to be used as a decorator:

@check_communicating
def scan(self):
    ...
Parameters:

func (TypeVar(Wrapped, bound= Callable[..., Any])) – the wrapped function

Return type:

TypeVar(Wrapped, bound= Callable[..., Any])

Returns:

the wrapped function