Device Base Classes

FhsBaseDevice Class

class ska_mid_cbf_fhs_common.base_classes.device.fhs_base_device.FhsBaseDevice(*args: Any, **kwargs: Any)

Bases: SKABaseDevice

Base class containing basic functionality for all FHS-level devices.

Initialise object.

device_id

ID of the device.

Type:

device_property(dtype=int)

device_version_num

Version of the device.

Type:

device_property(dtype=str)

device_gitlab_hash

Gitlab hash of the device.

Type:

device_property(dtype=str)

simulation_mode

1 if the device is deployed in simulation mode, 0 otherwise.

Type:

device_property(dtype=int)

emulation_mode

1 if the device is deployed in emulation mode, 0 otherwise.

Type:

device_property(dtype=int)

bitstream_id

ID of the bitstream used for this device.

Type:

device_property(dtype=str)

bitstream_version

Version of the bitstream used for this device.

Type:

device_property(dtype=str)

communicationState

Read-only Tango attribute specifying the communication state of the device.

Returns:

The communication state of the device.

Return type:

CommunicationStatus

simulationMode: attribute_from_signal

Simulation mode attribute of the device.

Some devices may implement both modes, while others will have simulators that set simulationMode to True while the real devices always set simulationMode to False.

init_command_objects(commands_and_methods: list[tuple] | None = None) → None

Set up all long-running command objects.

Parameters:

commands_and_methods (list[tuple] | None, optional) – The list of commands to set up for this device, each a tuple of the form (“TangoCommandName”, “python_method_name”). Default is None (no commands will be set up).

init_fast_command_objects(commands_and_classes: list[tuple]) → None

Set up all fast command objects.

Parameters:

commands_and_classes (list[tuple]) – The list of commands to set up for this device, each a tuple of the form (“TangoCommandName”, PythonFastCommandClass).

InitCommand = None

init_device() → None

Initialize the Tango device after startup.

set_local_change_events() → None

Set attribute change events specific to the implementing class.

get_dev_state() → DevState

Get the DevState of the device.

Returns:

The DevState of the device.

Return type:

DevState

On: Callable[[BaseInterface], DevVarLongStringArrayType] | None = None

Off: Callable[[BaseInterface], DevVarLongStringArrayType] | None = None

Standby: Callable[[BaseInterface], DevVarLongStringArrayType] | None = None

Reset: Callable[[BaseInterface], DevVarLongStringArrayType] | None = None

is_On_allowed = None

Abort() → tuple[list[ResultCode], list[str]]

Abort any executing long running command(s) and empty the queue.

Abort is itself a long running command that is executed immediately.

Subclasses should override execute_Abort() to change the behaviour of this command.

Returns:

A tuple containing ResultCode.STARTED and the unique ID of the command

AbortCommands() → tuple[list[ResultCode], list[str]]

Empty out long running commands in queue.

Returns:

A tuple containing a return code and a string message indicating status. The message is for information purpose only.

CheckLongRunningCommandStatus(argin: str) → str

Check the status of a long running command by ID.

Parameters:

argin – the command id

Returns:

command status

DebugDevice() → int

Enable remote debugging of this device.

To modify behaviour for this command, modify the do() method of the command class: DebugDeviceCommand.

Returns:

the port the debugger is listening on

GetVersionInfo() → list[str]

Return the version information of the device.

To modify behaviour for this command, modify the do() method of the command class.

Returns:

The result code and the command unique ID

GroupDefinitions

Device property.

Each string in the list is a JSON serialised dict defining the group_name, devices and subgroups in the group. A Tango Group object is created for each item in the list, according to the hierarchy defined. This provides easy access to the managed devices in bulk, or individually.

The general format of the list is as follows, with optional devices and subgroups keys:

[ {"group_name": "<name>",
   "devices": ["<dev name>", ...]},
  {"group_name": "<name>",
   "devices": ["<dev name>", "<dev name>", ...],
   "subgroups" : [{<nested group>},
                    {<nested group>}, ...]},
  ...
  ]

For example, a hierarchy of racks, servers and switches:

[ {"group_name": "servers",
   "devices": ["elt/server/1", "elt/server/2",
                 "elt/server/3", "elt/server/4"]},
  {"group_name": "switches",
   "devices": ["elt/switch/A", "elt/switch/B"]},
  {"group_name": "pdus",
   "devices": ["elt/pdu/rackA", "elt/pdu/rackB"]},
  {"group_name": "racks",
   "subgroups": [
        {"group_name": "rackA",
         "devices": ["elt/server/1", "elt/server/2",
                       "elt/switch/A", "elt/pdu/rackA"]},
        {"group_name": "rackB",
         "devices": ["elt/server/3", "elt/server/4",
                       "elt/switch/B", "elt/pdu/rackB"],
         "subgroups": []}
   ]} ]

LoggingLevelDefault

Device property.

Default logging level at device startup. See LoggingLevel

LoggingTargetsDefault

Device property.

Default logging targets at device startup. See the project readme for details.

SkaLevel

Device property.

Indication of importance of the device in the SKA hierarchy to support drill-down navigation: 1..6, with 1 highest.

TangoClassClass

alias of FhsBaseDeviceClass

TangoClassName = 'FhsBaseDevice'

adminMode: attribute_from_signal

Admin mode attribute of the device.

buildState

Read the Build State of the device.

Returns:

the build state of the device

commandedState: attribute_from_signal

Attribute for the last commanded Operating State of the device.

This should be set by subclasses of this interface by writing to _commanded_state.

controlMode: attribute_from_signal

Control mode attribute of the device.

The control mode of the device is either REMOTE or LOCAL. Tango Device accepts only from a ‘local’ client and ignores commands and queries received from TM or any other ‘remote’ clients. The Local clients has to release LOCAL control before REMOTE clients can take control again.

healthState

Health state attribute of the device.

This should be set by subclasses of this interface by writing to _health_state.

is_Off_allowed = None

loggingLevel

Read the logging level of the device.

Initialises to LoggingLevelDefault on startup. See LoggingLevel

Returns:

Logging level of the device.

loggingTargets

Read the additional logging targets of the device.

Note that this excludes the handlers provided by the ska_ser_logging library defaults - initialises to LoggingTargetsDefault on startup.

Returns:

Logging level of the device.

lrcFinished

Expose a signal as a Tango attribute.

lrcProtocolVersions

Return supported protocol versions.

Returns:

A tuple containing the lower and upper bounds of supported long running command protocol versions.

testMode: attribute_from_signal

Test mode attribute of the device.

Either no test mode or an indication of the test mode.

versionId

Read the Version Id of the device.

Returns:

the version id of the device

is_Standby_allowed = None

is_Reset_allowed = None

execute_On() → tuple[list[ResultCode], list[str]]

Execute the standard On command.

This method must be overridden by a subclass to enable the Tango command, otherwise the command will not be part of the device interface. The submit_lrc_task() decorator can be used to make Reset a long running command.

If implemented, this command should turn the device on and result in the operational state transitioning to ON or ALARM. The command should not return until the transition has occurred.

Returns:

A tuple containing a return code and a string message indicating status or a command ID. The message is for information purpose only.

execute_Off() → tuple[list[ResultCode], list[str]]

Execute the standard Off command.

This method must be overridden by a subclass to enable the Tango command, otherwise the command will not be part of the device interface. The submit_lrc_task() decorator can be used to make Reset a long running command.

If implemented, this command should turn the device off and result in the operational state transitioning to OFF. The command should not return until the transition has occurred.

Returns:

A tuple containing a return code and a string message indicating status or a command ID. The message is for information purpose only.

execute_Standby() → tuple[list[ResultCode], list[str]]

Execute the standard Standby command.

This method must be overridden by a subclass to enable the Tango command, otherwise the command will not be part of the device interface. The submit_lrc_task() decorator can be used to make Reset a long running command.

If implemented, this command should put the device in standby mode and result in the operational state transitioning to STANDBY. The command should not return until the transition has occurred.

Returns:

A tuple containing a return code and a string message indicating status or a command ID. The message is for information purpose only.

execute_Reset() → tuple[list[ResultCode], list[str]]

Execute the standard Reset command.

This method must be overridden by a subclass to enable the Tango command, otherwise the command will not be part of the device interface. The submit_lrc_task() decorator can be used to make Reset a long running command.

For a device that directly monitors and controls hardware, this command should put that hardware into a known state, for example by clearing buffers and loading default values into registers, or if necessary even by power-cycling and re-initialising the hardware.

Logical devices should generally implement this command to perform a sensible reset of that logical device. For example, aborting any current activities and clearing internal state.

Reset() generally should not change the power state of the device or its hardware:

• If invoking Reset() from STANDBY state, the device would usually be expected to remain in STANDBY.

• If invoking Reset() from ON state, the device would usually be expected to remain in ON.

• If invoking Reset() from FAULT state, the device would usually be expected to transition to ON or remain in FAULT, depending on whether the reset was successful in clearing then fault.

Reset() generally should not propagate to subservient devices. For example, a subsystem controller device should implement :py:meth`!Reset()` to reset the subsystem as a whole, but that probably should not result in all of the subsystem’s hardware being power-cycled.

Returns:

A tuple containing a return code and a string message indicating status. The message is for information purpose only.

classmethod run(args=None, **kwargs) → Any | None

Run the class as a Tango device server.

Parameters:

• *args – Arbitrary positional arguments to be passed to the Tango run_server method

• **kwargs – Arbitrary keyword arguments to be passed to the Tango run_server method

FhsControllerBaseDevice Class

class ska_mid_cbf_fhs_common.base_classes.device.controller.fhs_controller_base_device.FhsControllerBaseDevice(*args: Any, **kwargs: Any)

Bases: FhsBaseDevice, SKAObsDevice, Generic[FhsControllerComponentManagerT]

Extension of the FHS device base class containing additional functionality for all FHS controller devices.

Initialise object.

abstract property component_manager_class: type[FhsControllerComponentManagerT]

The component manager class associated with this controller device.

Type:

type[FhsControllerComponentManagerT]

property long_running_commands: list[tuple[str, str]]

A list of long-running commands defined on this controller, each a tuple of the form (“TangoCommandName”, “python_method_name”).

Type:

list[tuple[str, str]]

property fast_commands: list[tuple[str, FhsFastCommand]]

A list of fast commands defined on this controller, each a tuple of the form (“TangoCommandName”, PythonFastCommandClass).

Type:

list[tuple[str, FhsFastCommand]]

ip_blocks

Base64-encoded string containing the deploy-time properties of all IP blocks managed by the controller.

Type:

device_property(dtype=str)

logging_level

Logging level of the device.

Type:

device_property(dtype=str)

bitstream_path

File path to the bitstream used by the device.

Type:

device_property(dtype=str)

emulator_id

ID of the emulator used by the device.

Type:

device_property(dtype=str)

emulator_base_url

Base URL of the emulator used by the device.

Type:

device_property(dtype=str)

ipBlockIDs

Read-only Tango attribute specifying the list of IP block IDs the controller can interact with.

Returns:

The list of IP block IDs the controller can interact with.

Return type:

list[str]

ipBlockAliases

Read-only Tango attribute specifying all possible aliases for each IP block (which can be used e.g. in command arguments).

Returns:

The list of aliases for all IP blocks.

Return type:

str

loggingLevelOverrides

Read-only Tango attribute specifying any overridden IP block logging levels.

Returns:

JSON string containing any overridden IP block logging levels.

Return type:

str

ConfigureScan(config: str) → tuple[list[ResultCode], list[str]]

Tango command to apply a received scan configuration in preparation for a scan.

Parameters:

config (str) – The configuration JSON string.

Returns:

The Tango result code and a string message indicating status. The message is for information purpose only.

Return type:

tuple[list[ResultCode], list[str]]

Scan(scan_schema: str) → tuple[list[ResultCode], list[str]]

Tango command to start a scan operation.

Parameters:

scan_schema (str) – The scan schema JSON string.

Returns:

The Tango result code and a string message indicating status. The message is for information purpose only.

Return type:

tuple[list[ResultCode], list[str]]

EndScan(transaction_id: str | None = None) → tuple[list[ResultCode], list[str]]

Tango command to end a scan operation.

Parameters:

transaction_id (str) – The Transaction id, can be none

Returns:

The Tango result code and a string message indicating status. The message is for information purpose only.

Return type:

tuple[list[ResultCode], list[str]]

GoToIdle(transaction_id: str | None = None) → tuple[list[ResultCode], list[str]]

Tango command to return the device to an IDLE ObsState.

Parameters:

transaction_id (str) – The Transaction id, can be none

Returns:

The Tango result code and a string message indicating status. The message is for information purpose only.

Return type:

tuple[list[ResultCode], list[str]]

ObsReset(transaction_id: str | None = None) → tuple[list[ResultCode], list[str]]

Tango command to reset the observing state.

Parameters:

transaction_id (str) – The Transaction id, can be none

Returns:

The Tango result code and a string message indicating status. The message is for information purpose only.

Return type:

tuple[list[ResultCode], list[str]]

GetStatus(ip_blocks: list[str] | None = None) → tuple[list[ResultCode], list[str]]

Tango command to fetch the status(es) of any specified IP block(s).

Parameters:

ip_blocks (list[str] | None, optional) – The list of IP block names/IDs to fetch the status for.

Returns:

The Tango result code and a JSON string containing all requested statuses.

Return type:

tuple[list[ResultCode], list[str]]

UpdateGlobalLoggingLevel(logging_level: str) → tuple[list[ResultCode], list[str]]

Tango command to update the global logging level, impacting the controller and any IP blocks without logging level overrides.

Parameters:

logging_level (str) – The new logging level.

Returns:

The Tango result code and a string message indicating status. The message is for information purpose only.

Return type:

tuple[list[ResultCode], list[str]]

UpdateIPBlockLoggingLevels(input_json: str) → tuple[list[ResultCode], list[str]]

Tango command to assign a specified logging level override to all specified IP block(s).

Parameters:

input_json (str) – A JSON string in the format: {“ip_blocks”: [“SomeIPBlockIDOne”, “SomeIPBlockIDTwo”], “level”: “DEBUG”}

Returns:

The Tango result code and a string message indicating status. The message is for information purpose only.

Return type:

tuple[list[ResultCode], list[str]]

ClearLoggingLevelOverrides() → tuple[list[ResultCode], list[str]]

Tango command to clear all IP block logging level overrides, such that all IP blocks will use the global controller logging level.

Returns:

The Tango result code and a string message indicating status. The message is for information purpose only.

Return type:

tuple[list[ResultCode], list[str]]

class GetStatusCommand(component_manager: FhsComponentManagerT, logger: Logger | None = None, validator: ArgumentValidator | None = None)

Bases: FhsFastCommand[FhsControllerComponentManagerT]

FastCommand subclass for the GetStatus command.

Initialise a new BaseCommand instance.

Parameters:

• logger – the logger to be used by this Command. If not provided, then a default module logger will be used.

• validator – an optional validator to use to parse, validate and/or unpack command arguments.

do(ip_blocks: list[str] | None = None) → tuple[ResultCode, str]

Perform the user-specified functionality of the command.

This class provides stub functionality; subclasses should subclass this method with their command functionality.

Parameters:

• args – positional args to the component manager method

• kwargs – keyword args to the component manager method

Raises:

NotImplementedError – method does not exist

class UpdateGlobalLoggingLevelCommand(component_manager: FhsComponentManagerT, logger: Logger | None = None, validator: ArgumentValidator | None = None)

Bases: FhsFastCommand[FhsControllerComponentManagerT]

FastCommand subclass for the UpdateGlobalLoggingLevel command.

Initialise a new BaseCommand instance.

Parameters:

• logger – the logger to be used by this Command. If not provided, then a default module logger will be used.

• validator – an optional validator to use to parse, validate and/or unpack command arguments.

do(logging_level: str) → tuple[ResultCode, str]

Perform the user-specified functionality of the command.

This class provides stub functionality; subclasses should subclass this method with their command functionality.

Parameters:

• args – positional args to the component manager method

• kwargs – keyword args to the component manager method

Raises:

NotImplementedError – method does not exist

class UpdateIPBlockLoggingLevelsCommand(component_manager: FhsComponentManagerT, logger: Logger | None = None, validator: ArgumentValidator | None = None)

Bases: FhsFastCommand[FhsControllerComponentManagerT]

FastCommand subclass for the UpdateIPBlockLoggingLevels command.

Initialise a new BaseCommand instance.

Parameters:

• logger – the logger to be used by this Command. If not provided, then a default module logger will be used.

• validator – an optional validator to use to parse, validate and/or unpack command arguments.

do(input_json: str) → tuple[ResultCode, str]

Perform the user-specified functionality of the command.

This class provides stub functionality; subclasses should subclass this method with their command functionality.

Parameters:

• args – positional args to the component manager method

• kwargs – keyword args to the component manager method

Raises:

NotImplementedError – method does not exist

class ClearLoggingLevelOverridesCommand(component_manager: FhsComponentManagerT, logger: Logger | None = None, validator: ArgumentValidator | None = None)

Bases: FhsFastCommand[FhsControllerComponentManagerT]

FastCommand subclass for the ClearLoggingLevelOverrides command.

Initialise a new BaseCommand instance.

Parameters:

• logger – the logger to be used by this Command. If not provided, then a default module logger will be used.

• validator – an optional validator to use to parse, validate and/or unpack command arguments.

do() → tuple[ResultCode, str]

Perform the user-specified functionality of the command.

This class provides stub functionality; subclasses should subclass this method with their command functionality.

Parameters:

• args – positional args to the component manager method

• kwargs – keyword args to the component manager method

Raises:

NotImplementedError – method does not exist

init_device() → None

Initialize the Tango device after startup.

create_component_manager() → FhsControllerComponentManagerT

Instantiate the component manager for this device.

Returns:

The instantiated component manager.

Return type:

FhsControllerComponentManagerT

Abort() → tuple[list[ResultCode], list[str]]

Abort any executing long running command(s) and empty the queue.

Abort is itself a long running command that is executed immediately.

Subclasses should override execute_Abort() to change the behaviour of this command.

Returns:

A tuple containing ResultCode.STARTED and the unique ID of the command

AbortCommands() → tuple[list[ResultCode], list[str]]

Empty out long running commands in queue.

Returns:

A tuple containing a return code and a string message indicating status. The message is for information purpose only.

CheckLongRunningCommandStatus(argin: str) → str

Check the status of a long running command by ID.

Parameters:

argin – the command id

Returns:

command status

DebugDevice() → int

Enable remote debugging of this device.

To modify behaviour for this command, modify the do() method of the command class: DebugDeviceCommand.

Returns:

the port the debugger is listening on

GetVersionInfo() → list[str]

Return the version information of the device.

To modify behaviour for this command, modify the do() method of the command class.

Returns:

The result code and the command unique ID

GroupDefinitions

Device property.

Each string in the list is a JSON serialised dict defining the group_name, devices and subgroups in the group. A Tango Group object is created for each item in the list, according to the hierarchy defined. This provides easy access to the managed devices in bulk, or individually.

The general format of the list is as follows, with optional devices and subgroups keys:

[ {"group_name": "<name>",
   "devices": ["<dev name>", ...]},
  {"group_name": "<name>",
   "devices": ["<dev name>", "<dev name>", ...],
   "subgroups" : [{<nested group>},
                    {<nested group>}, ...]},
  ...
  ]

For example, a hierarchy of racks, servers and switches:

[ {"group_name": "servers",
   "devices": ["elt/server/1", "elt/server/2",
                 "elt/server/3", "elt/server/4"]},
  {"group_name": "switches",
   "devices": ["elt/switch/A", "elt/switch/B"]},
  {"group_name": "pdus",
   "devices": ["elt/pdu/rackA", "elt/pdu/rackB"]},
  {"group_name": "racks",
   "subgroups": [
        {"group_name": "rackA",
         "devices": ["elt/server/1", "elt/server/2",
                       "elt/switch/A", "elt/pdu/rackA"]},
        {"group_name": "rackB",
         "devices": ["elt/server/3", "elt/server/4",
                       "elt/switch/B", "elt/pdu/rackB"],
         "subgroups": []}
   ]} ]

LoggingLevelDefault

Device property.

Default logging level at device startup. See LoggingLevel

LoggingTargetsDefault

Device property.

Default logging targets at device startup. See the project readme for details.

SkaLevel

Device property.

Indication of importance of the device in the SKA hierarchy to support drill-down navigation: 1..6, with 1 highest.

TangoClassClass

alias of FhsControllerBaseDeviceClass

TangoClassName = 'FhsControllerBaseDevice'

adminMode: attribute_from_signal

Admin mode attribute of the device.

bitstream_id

ID of the bitstream used for this device.

Type:

device_property(dtype=str)

bitstream_version

Version of the bitstream used for this device.

Type:

device_property(dtype=str)

buildState

Read the Build State of the device.

Returns:

the build state of the device

commandedObsState

Attribute for the last commanded Observation State of the device.

This should be set by subclasses of this interface by writing to _commanded_obs_state.

commandedState: attribute_from_signal

Attribute for the last commanded Operating State of the device.

This should be set by subclasses of this interface by writing to _commanded_state.

communicationState

Read-only Tango attribute specifying the communication state of the device.

Returns:

The communication state of the device.

Return type:

CommunicationStatus

configurationDelayExpected

Configuration delay expected attribute of the device.

This should be set by subclasses of this interface by writing to _config_delay_expected.

configurationProgress

Configuration progress attribute of the device.

This should be set by subclasses of this interface by writing to _config_progress.

controlMode: attribute_from_signal

Control mode attribute of the device.

The control mode of the device is either REMOTE or LOCAL. Tango Device accepts only from a ‘local’ client and ignores commands and queries received from TM or any other ‘remote’ clients. The Local clients has to release LOCAL control before REMOTE clients can take control again.

device_gitlab_hash

Gitlab hash of the device.

Type:

device_property(dtype=str)

device_id

ID of the device.

Type:

device_property(dtype=int)

device_version_num

Version of the device.

Type:

device_property(dtype=str)

emulation_mode

1 if the device is deployed in emulation mode, 0 otherwise.

Type:

device_property(dtype=int)

healthState

Health state attribute of the device.

This should be set by subclasses of this interface by writing to _health_state.

init_command_objects() → None

Set up all command objects.

loggingLevel

Read the logging level of the device.

Initialises to LoggingLevelDefault on startup. See LoggingLevel

Returns:

Logging level of the device.

loggingTargets

Read the additional logging targets of the device.

Note that this excludes the handlers provided by the ska_ser_logging library defaults - initialises to LoggingTargetsDefault on startup.

Returns:

Logging level of the device.

lrcFinished

Expose a signal as a Tango attribute.

lrcProtocolVersions

Return supported protocol versions.

Returns:

A tuple containing the lower and upper bounds of supported long running command protocol versions.

obsMode

Observation mode attribute of the device.

This should be set by subclasses of this interface by writing to _obs_mode.

obsState

Observation state attribute of the device.

This should be set by subclasses of this interface by writing to _obs_state.

simulationMode: attribute_from_signal

Simulation mode attribute of the device.

Some devices may implement both modes, while others will have simulators that set simulationMode to True while the real devices always set simulationMode to False.

simulation_mode

1 if the device is deployed in simulation mode, 0 otherwise.

Type:

device_property(dtype=int)

testMode: attribute_from_signal

Test mode attribute of the device.

Either no test mode or an indication of the test mode.

versionId

Read the Version Id of the device.

Returns:

the version id of the device

reset_obs_state()

Set the ObsState of the device back to IDLE.

FhsLowLevelBaseDevice Class

Currently not in use and should be considered “tentatively deprecated” unless absolutely needed again.