ska_mid_cbf_fhs_common Module
Common library for FHS devices, services and related classes.
- class ska_mid_cbf_fhs_common.BaseSimulatorApi(ip_block_name: str, logger: Logger)[source]
Bases:
FhsBaseApiInterface
- class ska_mid_cbf_fhs_common.FhsBaseDevice(*args: Any, **kwargs: Any)[source]
Bases:
SKABaseDeviceBase class containing basic functionality for all FHS-level devices.
Initialise object.
- 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,devicesandsubgroupsin 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
devicesandsubgroupskeys:[ {"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": []} ]} ]
- InitCommand = None
- 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.
- 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
- 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
- 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)
- 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_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
ONorALARM. 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.
- 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.
- get_dev_state() DevState[source]
Get the DevState of the device.
- Returns:
The DevState of the device.
- Return type:
DevState
- healthState
Health state attribute of the device.
This should be set by subclasses of this interface by writing to
_health_state.
- init_command_objects(commands_and_methods: list[tuple] | None = None) None[source]
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[source]
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).
- is_Off_allowed = None
- is_On_allowed = None
- is_Reset_allowed = None
- is_Standby_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.
- classmethod run(args=None, **kwargs) Any | None[source]
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
- set_local_change_events() None[source]
Set attribute change events specific to the implementing class.
- 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
- admin_mode_model: ska_control_model.AdminModeModel
State model used to ensure that admin mode transitions are allowed.
- op_state_model: ska_control_model.OpStateModel
State model used to ensure that operational state transitions are allowed.
- class ska_mid_cbf_fhs_common.FhsFastCommand(component_manager: FhsComponentManagerT, logger: Logger | None = None, validator: ArgumentValidator | None = None)[source]
Bases:
FastCommand,Generic[FhsComponentManagerT]TODO: Document this class properly
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.
- class ska_mid_cbf_fhs_common.FhsComponentManagerBase(*args: Any, attr_change_callback: Callable[[str, Any], None] | None = None, attr_archive_callback: Callable[[str, Any], None] | None = None, health_state_callback: Callable[[HealthState], None] | None = None, logger: Logger, **kwargs: Any)[source]
Bases:
TaskExecutorComponentManager,ABCBase component manager class for all FHS devices.
- Parameters:
*args (
Any) – Any arbitrary positional arguments to pass to the superclass constructor.attr_change_callback (
Callable[[str, Any], None] | None, optional) – Callback that is called when an attribute change event occurs. Default is None.attr_archive_callback (
Callable[[str, Any], None] | None, optional) – Callback that is called when an attribute archive event occurs. Default is None.health_state_callback (
Callable[[HealthState], None] | None, optional) – Callback that is called when a HealthState change occurs. Default is None.logger (
logging.Logger) – A logger instance to be used by this component manager.**kwargs (
Any) – Any arbitrary keyword arguments to pass to the superclass init method.
- property faulty: bool | None
Whether this component manager is currently experiencing a fault.
- Type:
- get_device_health_state() HealthState[source]
Get the health state of the device.
- Returns:
HealthStateThe health state of the device.
- set_fault_and_failed() None[source]
Set the component state to faulty and update its health state to FAILED. This is to be called when an exception occurs in the component manager.
- start_communicating() None[source]
Establish communication with the component, then start monitoring.
- stop_communicating() None[source]
Cease monitoring the component, and break off all communication with it.
- update_device_health_state(health_state: HealthState) None[source]
Handle a health state change. This is a helper method for use by subclasses.
- Parameters:
health_state (
HealthState) – The new health state of the component manager.
- state
The current DevState of the device.
- Type:
DevState
- class ska_mid_cbf_fhs_common.FhsLowLevelBaseDevice(*args: Any, **kwargs: Any)[source]
Bases:
FhsBaseDeviceFHS Base low level device class. This class should not be used unless absolutely necessary, and otherwise it will likely be deprecated/removed in a future release.
Initialise object.
- 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
- class ConfigureCommand(component_manager: FhsComponentManagerT, logger: Logger | None = None, validator: ArgumentValidator | None = None)[source]
Bases:
FhsFastCommandInitialise 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(argin: str) tuple[ResultCode, str][source]
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
- 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
- class DeconfigureCommand(component_manager: FhsComponentManagerT, logger: Logger | None = None, validator: ArgumentValidator | None = None)[source]
Bases:
FhsFastCommandInitialise 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][source]
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 GetStatusCommand(component_manager: FhsComponentManagerT, logger: Logger | None = None, validator: ArgumentValidator | None = None)[source]
Bases:
FhsFastCommandInitialise 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(clear: bool = False) tuple[ResultCode, str][source]
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
- 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,devicesandsubgroupsin 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
devicesandsubgroupskeys:[ {"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.
- class RecoverCommand(component_manager: FhsComponentManagerT, logger: Logger | None = None, validator: ArgumentValidator | None = None)[source]
Bases:
FhsFastCommandInitialise 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][source]
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
- 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
FhsLowLevelBaseDeviceClass
- TangoClassName = 'FhsLowLevelBaseDevice'
- TestCmd() tuple[list[ResultCode], list[str]][source]
‘TestCmd’ TANGO command
No input parameter (DevVoid)
- Returns:
(not documented)
- Return type:
DevVarLongStringArray
- 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_path
TANGO property
- 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
- 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
- configure(config: str) tuple[list[ResultCode], list[str]][source]
‘configure’ TANGO command
Configuration json.
- Returns:
(not documented)
- Return type:
DevVarLongStringArray
- 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.
- deconfigure() tuple[list[ResultCode], list[str]][source]
‘deconfigure’ TANGO command
No input parameter (DevVoid)
- Returns:
(not documented)
- Return type:
DevVarLongStringArray
- 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)
- emulator_base_url
TANGO property
- emulator_id
TANGO property
- emulator_ip_block_id
TANGO property
- firmware_ip_block_id
TANGO property
- getstatus(clear: bool) tuple[list[ResultCode], list[str]][source]
‘getstatus’ TANGO command
- Parameters:
clear (DevBoolean) – (not documented)
- Returns:
(not documented)
- Return type:
DevVarLongStringArray
- healthState
Health state attribute of the device.
This should be set by subclasses of this interface by writing to
_health_state.
- health_monitor_poll_interval
TANGO property
- 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.
- recover() tuple[list[ResultCode], list[str]][source]
‘recover’ TANGO command
No input parameter (DevVoid)
- Returns:
(not documented)
- Return type:
DevVarLongStringArray
- 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)
- start() tuple[list[ResultCode], list[str]][source]
‘start’ TANGO command
No input parameter (DevVoid)
- Returns:
(not documented)
- Return type:
DevVarLongStringArray
- stop() tuple[list[ResultCode], list[str]][source]
‘stop’ TANGO command
No input parameter (DevVoid)
- Returns:
(not documented)
- Return type:
DevVarLongStringArray
- 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
- class ska_mid_cbf_fhs_common.FhsLowLevelComponentManagerBase(*args: Any, device: FhsLowLevelBaseDevice, simulator_api: FhsBaseApiInterface, **kwargs: Any)[source]
Bases:
FhsComponentManagerBaseLow level component manager base class. This class should not be used unless absolutely necessary, and otherwise it will likely be deprecated/removed in a future release.
- Parameters:
*args (
Any) – Any arbitrary positional arguments to pass to the superclass constructor.attr_change_callback (
Callable[[str, Any], None] | None, optional) – Callback that is called when an attribute change event occurs. Default is None.attr_archive_callback (
Callable[[str, Any], None] | None, optional) – Callback that is called when an attribute archive event occurs. Default is None.health_state_callback (
Callable[[HealthState], None] | None, optional) – Callback that is called when a HealthState change occurs. Default is None.logger (
logging.Logger) – A logger instance to be used by this component manager.**kwargs (
Any) – Any arbitrary keyword arguments to pass to the superclass init method.
- class ska_mid_cbf_fhs_common.FhsControllerBaseConfig[source]
Bases:
DataClassJsonMixinBase class for FHS controller ConfigureScan dataclasses.
- class ska_mid_cbf_fhs_common.FhsControllerBaseDevice(*args: Any, **kwargs: Any)[source]
Bases:
FhsBaseDevice,SKAObsDevice,Generic[FhsControllerComponentManagerT]Extension of the FHS device base class containing additional functionality for all FHS controller devices.
Initialise object.
- 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
- ClearLoggingLevelOverrides() tuple[list[ResultCode], list[str]][source]
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 ClearLoggingLevelOverridesCommand(component_manager: FhsComponentManagerT, logger: Logger | None = None, validator: ArgumentValidator | None = None)[source]
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][source]
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
- ConfigureScan(config: str) tuple[list[ResultCode], list[str]][source]
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]]
- 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
- EndScan(transaction_id: str | None = None) tuple[list[ResultCode], list[str]][source]
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]]
- GetStatus(ip_blocks: list[str] | None = None) tuple[list[ResultCode], list[str]][source]
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]]
- class GetStatusCommand(component_manager: FhsComponentManagerT, logger: Logger | None = None, validator: ArgumentValidator | None = None)[source]
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][source]
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
- 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
- GoToIdle(transaction_id: str | None = None) tuple[list[ResultCode], list[str]][source]
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]]
- GroupDefinitions
Device property.
Each string in the list is a JSON serialised dict defining the
group_name,devicesandsubgroupsin 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
devicesandsubgroupskeys:[ {"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.
- ObsReset(transaction_id: str | None = None) tuple[list[ResultCode], list[str]][source]
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]]
- Scan(scan_schema: str) tuple[list[ResultCode], list[str]][source]
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]]
- 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'
- UpdateGlobalLoggingLevel(logging_level: str) tuple[list[ResultCode], list[str]][source]
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]]
- class UpdateGlobalLoggingLevelCommand(component_manager: FhsComponentManagerT, logger: Logger | None = None, validator: ArgumentValidator | None = None)[source]
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][source]
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
- UpdateIPBlockLoggingLevels(input_json: str) tuple[list[ResultCode], list[str]][source]
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]]
- class UpdateIPBlockLoggingLevelsCommand(component_manager: FhsComponentManagerT, logger: Logger | None = None, validator: ArgumentValidator | None = None)[source]
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][source]
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
- 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_path
File path to the bitstream used by the 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
- abstract property component_manager_class: type[FhsControllerComponentManagerT]
The component manager class associated with this controller device.
- Type:
type[FhsControllerComponentManagerT]
- 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.
- create_component_manager() FhsControllerComponentManagerT[source]
Instantiate the component manager for this device.
- Returns:
The instantiated component manager.
- Return type:
FhsControllerComponentManagerT
- 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)
- emulator_base_url
Base URL of the emulator used by the device.
- Type:
device_property(dtype=str)
- emulator_id
ID of the emulator used by the device.
- Type:
device_property(dtype=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]]
- healthState
Health state attribute of the device.
This should be set by subclasses of this interface by writing to
_health_state.
- 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:
- 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]
- ip_blocks
Base64-encoded string containing the deploy-time properties of all IP blocks managed by the controller.
- Type:
device_property(dtype=str)
- loggingLevel
Read the logging level of the device.
Initialises to LoggingLevelDefault on startup. See
LoggingLevel- Returns:
Logging level of the device.
- loggingLevelOverrides
Read-only Tango attribute specifying any overridden IP block logging levels.
- Returns:
JSON string containing any overridden IP block logging levels.
- Return type:
- 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.
- logging_level
Logging level of the device.
- Type:
device_property(dtype=str)
- 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]]
- 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
- admin_mode_model: ska_control_model.AdminModeModel
State model used to ensure that admin mode transitions are allowed.
- op_state_model: ska_control_model.OpStateModel
State model used to ensure that operational state transitions are allowed.
- class ska_mid_cbf_fhs_common.FhsControllerComponentManagerBase(*args: Any, device: FhsBaseDevice, logger: Logger, simulation_mode: SimulationMode = SimulationMode.FALSE, attr_change_callback: Callable[[str, Any], None] | None = None, attr_archive_callback: Callable[[str, Any], None] | None = None, health_state_callback: Callable[[HealthState], None] | None = None, obs_state_action_callback: Callable[[str], None] | None = None, obs_command_running_callback: Callable[[str, bool], None] | None = None, emulation_mode: bool = False, create_log_file: bool = True, **kwargs: Any)[source]
Bases:
ObsDeviceComponentManager,FhsComponentManagerBaseExtension of the FHS observing device component manager base class, containing additional functionality for all FHS controller component managers.
- Parameters:
*args (
Any) – Any arbitrary positional arguments to pass to the superclass constructor.device (
FhsBaseDevice) – A reference to the controller device instance.logger (
logging.Logger) – A logger instance to be used by this component manager.simulation_mode (
SimulationMode, optional) – Whether the controller is deployed in simulation mode (SimulationMode.TRUE) or not (SimulationMode.FALSE). Default is SimulationMode.FALSE.attr_change_callback (
Callable[[str, Any], None] | None, optional) – Callback that is called when an attribute change event occurs. Default is None.attr_archive_callback (
Callable[[str, Any], None] | None, optional) – Callback that is called when an attribute archive event occurs. Default is None.health_state_callback (
Callable[[HealthState], None] | None, optional) – Callback that is called when a HealthState change occurs. Default is None.obs_state_action_callback (
Callable[[str, bool], None] | None, optional) – Callback that is called when the controller’s observation state changes. Default is None.obs_command_running_callback (
Callable[[str, bool], None] | None, optional) – Callback that is called when an observing command starts running. Default is None.emulation_mode (
bool, optional) – Whether the controller is deployed in emulation mode or not. Default is False.create_log_file (
bool, optional) – Whether or not to create a log file for this controller. Default is True.**kwargs (
Any) – Any arbitrary keyword arguments to pass to the superclass init method.
- abort_commands(task_callback: TaskCallbackType | None = None) tuple[TaskStatus, str][source]
Run a task to abort all commands and stop any started IP blocks.
- Parameters:
task_callback (
Optional[Callable], optional) – A callback to pass to the superclass to be run when the task status changes. Default is None.- Returns:
The status of the task and an informative message string.
- Return type:
tuple[TaskStatus, str]
- clear_logging_level_overrides() tuple[ResultCode, str][source]
Clear all IP block logging level overrides. This is the implementation for the ClearLoggingLevelOverrides command.
- Returns:
The Tango result code, and an informative message string.
- Return type:
tuple[ResultCode, str]
- abstract property config_dataclass: type[FhsControllerBaseConfig]
The ConfigureScan input dataclass for this controller.
- abstract property config_schema: dict[str, Any]
The ConfigureScan input JSON schema for this controller.
- configure_scan(argin: str, task_callback: Callable | None = None) tuple[TaskStatus, str][source]
Submit the task to start running the ConfigureScan command implementation.
- Parameters:
argin (
str) – The configuration JSON string from the command’s input argument.task_callback (
Optional[Callable], optional) – A callback to run when the task status changes. Default is None.
- Returns:
The status of the task and an informative message string.
- Return type:
tuple[TaskStatus, str]
- end_scan(argin: str | None = None, task_callback: Callable | None = None) tuple[TaskStatus, str][source]
Submit the task to start running the EndScan command implementation.
- Parameters:
argin (
str) – The Transaction id from the command’s input argument, can be nonetask_callback (
Optional[Callable], optional) – A callback to run when the task status changes. Default is None.
- Returns:
The status of the task and an informative message string.
- Return type:
tuple[TaskStatus, str]
- get_status(ip_blocks: list[str]) tuple[ResultCode, dict[str, dict]][source]
Fetch the status(es) from the specified IP block(s). This is the implementation for the GetStatus command.
- Parameters:
ip_blocks (
list[str]) – The list of IP blocks for which to fetch the status.- Returns:
The Tango result code, and a dictionary containing all of the requested IP block statuses.
- Return type:
tuple[ResultCode, dict[str, dict]]
- go_to_idle(argin: str | None = None, task_callback: Callable | None = None) tuple[TaskStatus, str][source]
Submit the task to start running the GoToIdle command implementation.
- Parameters:
argin (
str) – The Transaction id from the command’s input argument, can be nonetask_callback (
Optional[Callable], optional) – A callback to run when the task status changes. Default is None.
- Returns:
The status of the task and an informative message string.
- Return type:
tuple[TaskStatus, str]
- is_allowed(error_msg: str, obs_states: list[ObsState]) bool[source]
Determine whether the current ObsState is in the provided list of states, and log a warning message if not.
- Returns:
True if the current ObsState is in the list provided, False otherwise.
- Return type:
- is_go_to_idle_allowed() bool[source]
Determine whether the GoToIdle command is allowed from the current ObsState.
- Returns:
True if the GoToIdle command is allowed, False otherwise.
- Return type:
- is_obs_reset_allowed() bool[source]
Determine whether the ObsReset command is allowed from the current ObsState.
- Returns:
True if the ObsReset command is allowed, False otherwise.
- Return type:
- obs_reset(argin: str | None = None, task_callback: Callable | None = None) tuple[TaskStatus, str][source]
Submit the task to start running the ObsReset command implementation.
- Parameters:
argin (
str) – The Transaction id from the command’s input argument, can be nonetask_callback (
Optional[Callable], optional) – A callback to run when the task status changes. Default is None.
- Returns:
The status of the task and an informative message string.
- Return type:
tuple[TaskStatus, str]
- scan(argin: str, task_callback: Callable | None = None) tuple[TaskStatus, str][source]
Submit the task to start running the Scan command implementation.
- Parameters:
argin (
str) – The scan schema JSON string from the command’s input argument.task_callback (
Optional[Callable], optional) – A callback to run when the task status changes. Default is None.
- Returns:
The status of the task and an informative message string.
- Return type:
tuple[TaskStatus, str]
- start_communicating() None[source]
Establish communication with the component, then start monitoring.
- task_abort_event_is_set(command_name: str, task_callback: Callable, task_abort_event: Event) bool[source]
Helper method for checking task abort event during command thread.
- update_global_logging_level(logging_level: str) tuple[ResultCode, str][source]
Update the controller’s global logging level. This is the implementation for the UpdateGlobalLoggingLevel command.
- Parameters:
logging_level (
str) – The new logging level.- Returns:
The Tango result code, and an informative message string.
- Return type:
tuple[ResultCode, str]
- update_ip_block_logging_levels(input_json: str) tuple[ResultCode, str][source]
Assign a specified logging level override to all specified IP block(s). This is the implementation for the UpdateIPBlockLoggingLevels command.
- Parameters:
input_json (
str) – A JSON string in the format: {“ip_blocks”: [“SomeIPBlockIDOne”, “SomeIPBlockIDTwo”], “level”: “DEBUG”}- Returns:
The Tango result code, and an informative message string.
- Return type:
tuple[ResultCode, str]
- ip_block_aliases: dict[str, list[str]]
Mapping of IP block aliases to their base IDs.
- Type:
dict[str, list[str]]
- ip_block_manager_map: dict[str, BaseIPBlockManager]
Mapping of IP block IDs and aliases to their respective manager classes.
- Type:
dict[str, BaseIPBlockManager]
- device
Reference to the controller device instance.
- Type:
- simulation_mode
Whether the controller is deployed in simulation mode.
- Type:
SimulationMode
- health_monitor
The health state monitor for this controller.
- Type:
- obs_state
The current observation state of this controller.
- Type:
ObsState
- class ska_mid_cbf_fhs_common.BaseIPBlockManager(ip_block_id: str, controlling_device_name: str, bitstream_path: str, bitstream_id: str, bitstream_version: str, firmware_ip_block_id: str, simulation_mode: SimulationMode = SimulationMode.TRUE, emulation_mode: bool = False, emulator_ip_block_id: str | None = None, emulator_id: str | None = None, emulator_base_url: str | None = None, logging_level: str = 'INFO', create_log_file: bool = True, **additional_properties: Any)[source]
Bases:
ABC,Generic[IPBlockConfigT,IPBlockStatusT,FhsBaseApiInterfaceT]Base class for IP Block Manager classes. Provides basic API method wrappers and sets up IP block-level logging.
- Parameters:
ip_block_id (
str) – The ID of the IP block being managed.controlling_device_name (
str) – The name of the Tango device controlling this IP block manager.bitstream_path (
str) – The path to the bitstream being used for this IP block.bitstream_id (
str) – The ID of the bitstream being used for this IP block.bitstream_version (
str) – The version of the bitstream being used for this IP block.firmware_ip_block_id (
str) – The firmware ID of the IP block being managed.simulation_mode (
SimulationMode, optional) – Whether the controller is deployed in simulation mode (SimulationMode.TRUE) or not (SimulationMode.FALSE). Default is SimulationMode.FALSE.emulation_mode (
bool, optional) – Whether the controller is deployed in emulation mode or not. Default is False.emulator_ip_block_id (
str) – The emulator ID of the IP block being managed.emulator_id (
str) – The ID of the bitstream emulator this manager will communicate with in emulation mode.emulator_base_url (
str) – The base URL of the bitstream emulator this manager will communicate with in emulation mode.logging_level (
str) – The logging level to initialize this IP block manager with.create_log_file (
bool, optional) – Whether or not to create a log file for this IP block manager. Default is True.**additional_properties (
Any) – Any additional properties for a particular IP block subclass.
- abstract property config_dataclass: type[IPBlockConfigT]
The configuration dataclass for this IP block.
- configure(config: IPBlockConfigT | None = None) int[source]
Configure the IP block.
- Parameters:
config (
IPBlockConfigT | None, optional) – A configuration to apply, or None (no configuration). Default is None.
- deconfigure(config: IPBlockConfigT | None = None) int[source]
Deconfigure the IP block.
- Parameters:
config (
IPBlockConfigT | None, optional) – A config object to reference during deconfiguration, or None. Default is None.
- property emulator_api_class: type[EmulatorApi]
The emulator API base class for this IP block. Should only be overridden if additional API methods are required on top of the base API.
- Type:
type[EmulatorApi]
- property firmware_api_class: type[FirmwareApi]
The firmware API base class for this IP block. Should only be overridden if additional API methods are required on top of the base API.
- Type:
type[FirmwareApi]
- abstract property simulator_api_class: type[BaseSimulatorApi]
The simulator API class for this IP block.
- Type:
type[BaseSimulatorApi]
- status(clear: bool = False) IPBlockStatusT | None[source]
Get the current status from the IP block.
- Parameters:
clear (
bool, optional) – Whether or not to clear counters in firmware (or the emulator). Default is False.- Returns:
The status retrieved from the IP block if successful, or None if retrieval was unsuccessful.
- Return type:
IPBlockStatusT | None
- class ska_mid_cbf_fhs_common.BaseMonitoringIPBlockManager(ip_block_id: str, controlling_device_name: str, bitstream_path: str, bitstream_id: str, bitstream_version: str, firmware_ip_block_id: str, simulation_mode: ~ska_control_model.simulation_mode.SimulationMode = SimulationMode.TRUE, emulation_mode: bool = False, emulator_ip_block_id: str | None = None, emulator_id: str | None = None, emulator_base_url: str | None = None, logging_level: str = 'INFO', health_monitor_poll_interval: float = 30.0, update_health_state_callback: ~typing.Callable = <function BaseMonitoringIPBlockManager.<lambda>>, create_log_file: bool = True, **additional_properties: ~typing.Any)[source]
Bases:
BaseIPBlockManager[IPBlockConfigT,IPBlockStatusT,FhsBaseApiInterfaceT],Generic[IPBlockConfigT,IPBlockStatusT,FhsBaseApiInterfaceT]Base class for health monitoring IP Block Manager classes.
- Parameters:
ip_block_id (
str) – The ID of the IP block being managed.controlling_device_name (
str) – The name of the Tango device controlling this IP block manager.bitstream_path (
str) – The path to the bitstream being used for this IP block.bitstream_id (
str) – The ID of the bitstream being used for this IP block.bitstream_version (
str) – The version of the bitstream being used for this IP block.firmware_ip_block_id (
str) – The firmware ID of the IP block being managed.simulation_mode (
SimulationMode, optional) – Whether the controller is deployed in simulation mode (SimulationMode.TRUE) or not (SimulationMode.FALSE). Default is SimulationMode.FALSE.emulation_mode (
bool, optional) – Whether the controller is deployed in emulation mode or not. Default is False.emulator_ip_block_id (
str) – The emulator ID of the IP block being managed.emulator_id (
str) – The ID of the bitstream emulator this manager will communicate with in emulation mode.emulator_base_url (
str) – The base URL of the bitstream emulator this manager will communicate with in emulation mode.logging_level (
str) – The logging level to initialize this IP block manager with.health_monitor_poll_interval (
float) – The interval, in seconds, at which to run health state polling.update_health_state_callback (
Callable) – Callback to run whenever the IP block’s health state changes.create_log_file (
bool, optional) – Whether or not to create a log file for this IP block manager. Default is True.**additional_properties (
Any) – Any additional properties for a particular IP block subclass.
- abstractmethod get_status_healthstates(status: IPBlockStatusT) dict[str, HealthState][source]
Determine the health states of various parts of the status retrieved from the IP block. This runs once per polling cycle.
- Parameters:
status (
IPBlockStatusT) – The status returned by the IP block’s API.- Returns:
A dictionary mapping status items to their determined HealthStates, which will be used to determine the overall health state of the IP block.
- Return type:
dict[str, HealthState]
- logger: Logger
The logger used by this IP block manager.
- Type:
Logger
- health_monitor
The health state monitor for this IP block.
- Type:
- class ska_mid_cbf_fhs_common.WidebandPowerMeterManager(ip_block_id: str, controlling_device_name: str, bitstream_path: str, bitstream_id: str, bitstream_version: str, firmware_ip_block_id: str, simulation_mode: SimulationMode = SimulationMode.TRUE, emulation_mode: bool = False, emulator_ip_block_id: str | None = None, emulator_id: str | None = None, emulator_base_url: str | None = None, logging_level: str = 'INFO', create_log_file: bool = True, **additional_properties: Any)[source]
Bases:
BaseIPBlockManager[WidebandPowerMeterConfig,WidebandPowerMeterStatus,FhsBaseApiInterface]Wideband Power Meter IP block manager.
- Parameters:
ip_block_id (
str) – The ID of the IP block being managed.controlling_device_name (
str) – The name of the Tango device controlling this IP block manager.bitstream_path (
str) – The path to the bitstream being used for this IP block.bitstream_id (
str) – The ID of the bitstream being used for this IP block.bitstream_version (
str) – The version of the bitstream being used for this IP block.firmware_ip_block_id (
str) – The firmware ID of the IP block being managed.simulation_mode (
SimulationMode, optional) – Whether the controller is deployed in simulation mode (SimulationMode.TRUE) or not (SimulationMode.FALSE). Default is SimulationMode.FALSE.emulation_mode (
bool, optional) – Whether the controller is deployed in emulation mode or not. Default is False.emulator_ip_block_id (
str) – The emulator ID of the IP block being managed.emulator_id (
str) – The ID of the bitstream emulator this manager will communicate with in emulation mode.emulator_base_url (
str) – The base URL of the bitstream emulator this manager will communicate with in emulation mode.logging_level (
str) – The logging level to initialize this IP block manager with.create_log_file (
bool, optional) – Whether or not to create a log file for this IP block manager. Default is True.**additional_properties (
Any) – Any additional properties for a particular IP block subclass.
- property config_dataclass: type[WidebandPowerMeterConfig]
The configuration dataclass for the Wideband Power Meter.
- Type:
type[WidebandPowerMeterConfig]
- deconfigure(config: WidebandPowerMeterConfig | None = None) int[source]
Deconfigure the Wideband Power Meter
- property simulator_api_class: type[WidebandPowerMeterSimulator]
The simulator API class for the Wideband Power Meter.
- Type:
type[WidebandPowerMeterSimulator]
- property status_dataclass: type[WidebandPowerMeterStatus]
The status dataclass for the Wideband Power Meter.
- Type:
type[WidebandPowerMeterStatus]
- class ska_mid_cbf_fhs_common.WidebandPowerMeterConfig(averaging_time: float, flagging: numpy.uint8, transaction_id: str | None = None)[source]
Bases:
DataClassJsonMixin- flagging: uint8
- class ska_mid_cbf_fhs_common.WidebandPowerMeterStatus(timestamp: float, avg_power_pol_x: float, avg_power_pol_y: float, nd_on_power_pol_x: float, nd_on_power_pol_y: float, nd_off_power_pol_x: float, nd_off_power_pol_y: float, nd_trn_power_pol_x: float, nd_trn_power_pol_y: float, flag: bool, overflow: numpy.uint8)[source]
Bases:
DataClassJsonMixin- overflow: uint8
- class ska_mid_cbf_fhs_common.WidebandPowerMeterSimulator(ip_block_name: str, logger: Logger)[source]
Bases:
BaseSimulatorApi
- class ska_mid_cbf_fhs_common.FSPMode(*values)[source]
Bases:
IntEnum- UNKNOWN = -1
- CORR = 0
- PST = 1
- PSS = 2
- class ska_mid_cbf_fhs_common.FtileEthernetManager(ip_block_id: str, controlling_device_name: str, bitstream_path: str, bitstream_id: str, bitstream_version: str, firmware_ip_block_id: str, simulation_mode: ~ska_control_model.simulation_mode.SimulationMode = SimulationMode.TRUE, emulation_mode: bool = False, emulator_ip_block_id: str | None = None, emulator_id: str | None = None, emulator_base_url: str | None = None, logging_level: str = 'INFO', health_monitor_poll_interval: float = 30.0, update_health_state_callback: ~typing.Callable = <function BaseMonitoringIPBlockManager.<lambda>>, create_log_file: bool = True, **additional_properties: ~typing.Any)[source]
Bases:
BaseMonitoringIPBlockManager[FtileEthernetConfig,FtileEthernetStatus,FhsBaseApiInterface]Ftile Ethernet IP block manager.
- Parameters:
ip_block_id (
str) – The ID of the IP block being managed.controlling_device_name (
str) – The name of the Tango device controlling this IP block manager.bitstream_path (
str) – The path to the bitstream being used for this IP block.bitstream_id (
str) – The ID of the bitstream being used for this IP block.bitstream_version (
str) – The version of the bitstream being used for this IP block.firmware_ip_block_id (
str) – The firmware ID of the IP block being managed.simulation_mode (
SimulationMode, optional) – Whether the controller is deployed in simulation mode (SimulationMode.TRUE) or not (SimulationMode.FALSE). Default is SimulationMode.FALSE.emulation_mode (
bool, optional) – Whether the controller is deployed in emulation mode or not. Default is False.emulator_ip_block_id (
str) – The emulator ID of the IP block being managed.emulator_id (
str) – The ID of the bitstream emulator this manager will communicate with in emulation mode.emulator_base_url (
str) – The base URL of the bitstream emulator this manager will communicate with in emulation mode.logging_level (
str) – The logging level to initialize this IP block manager with.health_monitor_poll_interval (
float) – The interval, in seconds, at which to run health state polling.update_health_state_callback (
Callable) – Callback to run whenever the IP block’s health state changes.create_log_file (
bool, optional) – Whether or not to create a log file for this IP block manager. Default is True.**additional_properties (
Any) – Any additional properties for a particular IP block subclass.
- property config_dataclass: type[FtileEthernetConfig]
The configuration dataclass for the F-tile Ethernet block.
- Type:
type[FtileEthernetConfig]
- get_status_healthstates(status: FtileEthernetStatus) dict[str, HealthState][source]
Determine the health states of various parts of the status retrieved from the IP block. This runs once per polling cycle.
- Parameters:
status (
IPBlockStatusT) – The status returned by the IP block’s API.- Returns:
A dictionary mapping status items to their determined HealthStates, which will be used to determine the overall health state of the IP block.
- Return type:
dict[str, HealthState]
- property simulator_api_class: type[FtileEthernetSimulator]
The simulator API class for the F-tile Ethernet block.
- Type:
type[FtileEthernetSimulator]
- property status_dataclass: type[FtileEthernetStatus]
The status dataclass for the F-tile Ethernet block.
- Type:
type[FtileEthernetStatus]
- class ska_mid_cbf_fhs_common.FtileEthernetConfig(loopback: bool = False, unidirectional: bool = False)[source]
Bases:
DataClassJsonMixin
- class ska_mid_cbf_fhs_common.FtileEthernetStatus(rx_mac: ska_mid_cbf_fhs_common.ip_blocks.ftile_ethernet.ftile_ethernet_manager.FtileEthernetMacStats = <factory>, tx_mac: ska_mid_cbf_fhs_common.ip_blocks.ftile_ethernet.ftile_ethernet_manager.FtileEthernetMacStats = <factory>, rx_fec: ska_mid_cbf_fhs_common.ip_blocks.ftile_ethernet.ftile_ethernet_manager.FtileEthernetFecStats = <factory>, tx_pll_locked: numpy.uint8 = 0, tx_lanes_stable: bool = False, rx_cdr_locked: numpy.uint8 = 0, rx_deskew: bool = False, rx_deskew_change: bool = False, rx_pcs_ready: bool = False, tx_ready: bool = False, rx_ready: bool = False, ready: bool = False)[source]
Bases:
DataClassJsonMixin- rx_cdr_locked: uint8 = 0
- tx_pll_locked: uint8 = 0
- rx_mac: FtileEthernetMacStats
- tx_mac: FtileEthernetMacStats
- rx_fec: FtileEthernetFecStats
- class ska_mid_cbf_fhs_common.FtileEthernetMacStats(fragments: numpy.uint32 = 0, jabbers: numpy.uint32 = 0, fcs_err: numpy.uint32 = 0, pause_err: numpy.uint32 = 0, multicast: ska_mid_cbf_fhs_common.ip_blocks.ftile_ethernet.ftile_ethernet_manager.FtileEthernetCastStats = <factory>, broadcast: ska_mid_cbf_fhs_common.ip_blocks.ftile_ethernet.ftile_ethernet_manager.FtileEthernetCastStats = <factory>, unicast: ska_mid_cbf_fhs_common.ip_blocks.ftile_ethernet.ftile_ethernet_manager.FtileEthernetCastStats = <factory>, packet_count_64b: numpy.uint64 = 0, packet_count_65to127b: numpy.uint64 = 0, packet_count_128to255b: numpy.uint64 = 0, packet_count_256to511b: numpy.uint64 = 0, packet_count_512to1023b: numpy.uint64 = 0, packet_count_1024to1518: numpy.uint64 = 0, packet_count_1519tomaxb: numpy.uint64 = 0, packet_count_oversize: numpy.uint32 = 0, pause: numpy.uint64 = 0, runt: numpy.uint32 = 0, frame_start: numpy.uint64 = 0, length_error: numpy.uint32 = 0, pfc_error: numpy.uint32 = 0, pfc: numpy.uint32 = 0, malformed: numpy.uint32 = 0, dropped: numpy.uint32 = 0, bad_length_type: numpy.uint32 = 0, payload_bytes: numpy.uint64 = 0, total_bytes: numpy.uint64 = 0)[source]
Bases:
DataClassJsonMixin- bad_length_type: uint32 = 0
- dropped: uint32 = 0
- fcs_err: uint32 = 0
- fragments: uint32 = 0
- frame_start: uint64 = 0
- jabbers: uint32 = 0
- length_error: uint32 = 0
- malformed: uint32 = 0
- packet_count_1024to1518: uint64 = 0
- packet_count_128to255b: uint64 = 0
- packet_count_1519tomaxb: uint64 = 0
- packet_count_256to511b: uint64 = 0
- packet_count_512to1023b: uint64 = 0
- packet_count_64b: uint64 = 0
- packet_count_65to127b: uint64 = 0
- packet_count_oversize: uint32 = 0
- pause: uint64 = 0
- pause_err: uint32 = 0
- payload_bytes: uint64 = 0
- pfc: uint32 = 0
- pfc_error: uint32 = 0
- runt: uint32 = 0
- total_bytes: uint64 = 0
- multicast: FtileEthernetCastStats
- broadcast: FtileEthernetCastStats
- unicast: FtileEthernetCastStats
- class ska_mid_cbf_fhs_common.FtileEthernetFecStats(corrected_codewords: list[numpy.uint64] = <factory>, uncorrected_codewords: list[numpy.uint64] = <factory>)[source]
Bases:
DataClassJsonMixin
- class ska_mid_cbf_fhs_common.FtileEthernetCastStats(data_err: numpy.uint32 = 0, ctrl_err: numpy.uint32 = 0, data_ok: numpy.uint64 = 0, ctrl_ok: numpy.uint64 = 0)[source]
Bases:
DataClassJsonMixin- ctrl_err: uint32 = 0
- ctrl_ok: uint64 = 0
- data_err: uint32 = 0
- data_ok: uint64 = 0
- class ska_mid_cbf_fhs_common.FtileEthernetSimulator(ip_block_name: str, logger: Logger)[source]
Bases:
BaseSimulatorApi
- ska_mid_cbf_fhs_common.find(func: Callable[[T], bool], iterable: Iterable[T]) T | None[source]
Find the first element in the iterable for which the specified function returns True.
- Parameters:
func (
Callable[[T], bool]) – The function to run against each element in the iterable.iterable (
Iterable[T]) – The iterable to check.
- Returns:
The first element in the iterable for which the specified function returns True, or None if no matching element was found.
- Return type:
T | None
- ska_mid_cbf_fhs_common.find_index(func: Callable[[T], bool], iterable: Iterable[T]) int | None[source]
Find the index of the first element in the iterable for which the specified function returns True.
- Parameters:
func (
Callable[[T], bool]) – The function to run against each element in the iterable.iterable (
Iterable[T]) – The iterable to check.
- Returns:
The index of the first element in the iterable for which the specified function returns True, or None if no matching element was found.
- Return type:
- ska_mid_cbf_fhs_common.calculate_gain_multiplier(p_max: int | float | MPFloat, measured_power: int | float | MPFloat, headroom: int | float | MPFloat) MPFloat[source]
Calculate the gain multiplier required to apply to the signal to achieve the optimal output power adhering to the provided headroom.
- Parameters:
p_max (
FloatLike) – Maximum signal power, in decibels (dB).measured_power (
FloatLike) – Average power measurement (is this in volts?). Must be within the interval (0, 1].headroom (
FloatLike) – Requested RFI headroom, in decibels (dB).
- Returns:
The gain multiplier to apply to the signal.
- Return type:
- ska_mid_cbf_fhs_common.calculate_expected_power(p_max: int | float | MPFloat, headroom: int | float | MPFloat) MPFloat[source]
Calculate the expected optimal output power adhering to the provided headroom.
- Parameters:
p_max (
FloatLike) – Maximum signal power, in decibels (dB).headroom (
FloatLike) – Requested RFI headroom, in decibels (dB).
- Returns:
The expected optimal output power.
- Return type:
- ska_mid_cbf_fhs_common.calculate_output_power(measured_power: int | float | MPFloat, multiplier: int | float | MPFloat) MPFloat[source]
Calculate the power achieved after applying the specified gain multiplier.
- Parameters:
measured_power (
FloatLike) – Average power measurement (is this in volts?). Must be within the interval (0, 1].multiplier (
FloatLike) – Gain multiplier to apply to the measured power.
- Returns:
The calculated output power.
- Return type:
- class ska_mid_cbf_fhs_common.MPFloat(val: int | float | MPFloat, **kwargs)[source]
Bases:
mpfMultiprecision or arbitrary precision float. Wraps the mpmath.mpf multiprecision float class such that creating an MPFloat from a Python float does not lose precision.
A new mpf can be created from a Python float, an int, a or a decimal string representing a number in floating-point format.
- context
- class ska_mid_cbf_fhs_common.NonBlockingFunction(fn: Callable[[P], R], *args, **kwargs)[source]
Bases:
Generic[P,R]Class that wraps a given function to run it in a thread and return a result.
- static await_all(*fns: NonBlockingFunction) list[Any][source]
Wait for all of the specified non-blocking functions to complete and return a list of all of their results.
- ska_mid_cbf_fhs_common.non_blocking(fn: Callable[[P], R])[source]
Decorator which turns a function into a non-blocking function.
- ska_mid_cbf_fhs_common.ska_time() float[source]
Get the current time relative to the SKA epoch (1999-12-31 23:59:28 UTC), in seconds.
- class ska_mid_cbf_fhs_common.EmulatorApi(bitstream_path: str, emulator_ip_block_id: str, emulator_id: str, emulator_base_url: str, logger: Logger)[source]
Bases:
FhsBaseApiInterface
- class ska_mid_cbf_fhs_common.FirmwareApi(bitstream_path: str, firmware_ip_block_id: str, logger: Logger)[source]
Bases:
FhsBaseApiInterface
- class ska_mid_cbf_fhs_common.FhsHealthMonitor(logger: Logger, get_device_health_state: Callable, update_health_state_callback: Callable, check_registers_callback: Callable | None = None, api: FhsBaseApiInterface | None = None, status_func_name: str = 'status', poll_interval=1.0)[source]
Bases:
object
- class ska_mid_cbf_fhs_common.RegisterPollingThread(logger: Logger, api: FhsBaseApiInterface, status_func_name: str, check_registers_callback: Callable, add_health_state: Callable, merge_health_states: Callable, poll_interval=1.0)[source]
Bases:
PollingThread- Parameters:
poll_interval (
float, optional) – The time interval, in seconds, between each poll. Default is 1.0.immediate_first_poll (
bool, optional) – If True, execute the first poll immediately when the polling thread is started. Otherwise, wait poll_interval seconds before executing the first poll. Default is True.poll_body (
Callable[[None], bool] | None, optional) – A function to run in each polling cycle. If the function returns True on any given call, polling will stop immediately afterward. This can be useful to poll only until a certain condition is met. Any other return value will be ignored and polling will continue as normal. If this parameter is set to None, the default _poll_body method will be used, which does nothing unless overridden by a subclass. Default is None.poll_body_args (
tuple[Any], optional) – A tuple of positional arguments to pass to the poll body function every time it is run. Default is an empty tuple.poll_body_kwargs (
dict[str, Any] | None, optional) – A dictionary of keyword arguments to pass to the poll body function every time it is run. Default is an empty dictionary.poll_exc_body (
Callable[[None], Any] | None, optional) – A function to run when/if an exception is caught during execution of the polling thread. If set to None, the default _poll_exc_body method will be used, which does nothing unless overridden by a subclass. Default is None.poll_exc_body_args (
tuple[Any], optional) – A tuple of positional arguments to pass to the poll exception function if it is run. Default is an empty tuple.poll_exc_body_kwargs (
dict[str, Any] | None, optional) – A dictionary of keyword arguments to pass to the poll exception function if it is run. Default is an empty dictionary.
- class ska_mid_cbf_fhs_common.PollingThread(poll_interval: float = 1.0, immediate_first_poll: bool = True, poll_body: Callable[[None], Any] | None = None, poll_body_args: tuple[Any] = (), poll_body_kwargs: dict[str, Any] | None = None, poll_exc_body: Callable[[None], Any] | None = None, poll_exc_body_args: tuple[Any] = (), poll_exc_body_kwargs: dict[str, Any] | None = None)[source]
Bases:
ThreadGeneric polling thread class, which wraps a standard Python thread. The _poll_body and _poll_exc_body methods may be overridden by subclasses to provide specific polling functionality.
- Parameters:
poll_interval (
float, optional) – The time interval, in seconds, between each poll. Default is 1.0.immediate_first_poll (
bool, optional) – If True, execute the first poll immediately when the polling thread is started. Otherwise, wait poll_interval seconds before executing the first poll. Default is True.poll_body (
Callable[[None], bool] | None, optional) – A function to run in each polling cycle. If the function returns True on any given call, polling will stop immediately afterward. This can be useful to poll only until a certain condition is met. Any other return value will be ignored and polling will continue as normal. If this parameter is set to None, the default _poll_body method will be used, which does nothing unless overridden by a subclass. Default is None.poll_body_args (
tuple[Any], optional) – A tuple of positional arguments to pass to the poll body function every time it is run. Default is an empty tuple.poll_body_kwargs (
dict[str, Any] | None, optional) – A dictionary of keyword arguments to pass to the poll body function every time it is run. Default is an empty dictionary.poll_exc_body (
Callable[[None], Any] | None, optional) – A function to run when/if an exception is caught during execution of the polling thread. If set to None, the default _poll_exc_body method will be used, which does nothing unless overridden by a subclass. Default is None.poll_exc_body_args (
tuple[Any], optional) – A tuple of positional arguments to pass to the poll exception function if it is run. Default is an empty tuple.poll_exc_body_kwargs (
dict[str, Any] | None, optional) – A dictionary of keyword arguments to pass to the poll exception function if it is run. Default is an empty dictionary.
- run()[source]
Method representing the thread’s activity.
You may override this method in a subclass. The standard run() method invokes the callable object passed to the object’s constructor as the target argument, if any, with sequential and keyword arguments taken from the args and kwargs arguments, respectively.
- class ska_mid_cbf_fhs_common.FhsObsStateMachine(callback: Callable | None = None, **extra_kwargs: Any)[source]
Bases:
LockedMachineThe observation state machine used by a generic FHS base class sub-element ObsDevice.
NOTE: entirely taken from ska-mid-cbd-mcs which in turn was taken from ska-csp-lmc-base CspSubElementObsStateMachine, to decouple FHS devices from their ska-tango-base version dependency
Initialise the model.
- Parameters:
callback (
Optional[Callable], optional) – A callback to be called when the state changes. Default is None.**extra_kwargs (
Any) – Any additional keyword arguments to pass to the superclass initialiser (useful for graphing)
- ABORT_COMPLETED = 'abort_completed'
- ABORT_INVOKED = 'abort_invoked'
- COMPONENT_FAULT = 'component_fault'
- CONFIGURE_COMPLETED = 'configure_completed'
- CONFIGURE_INVOKED = 'configure_invoked'
- DECONFIGURE_COMPLETED = 'deconfigure_completed'
- DECONFIGURE_INVOKED = 'deconfigure_invoked'
- GO_TO_IDLE = 'go_to_idle'
- OBSRESET_COMPLETED = 'obsreset_completed'
- OBSRESET_INVOKED = 'obsreset_invoked'
- RECOVER_COMPLETED = 'recover_completed'
- RECOVER_INVOKED = 'recover_invoked'
- START_COMPLETED = 'start_completed'
- START_INVOKED = 'start_invoked'
- STOP_COMPLETED = 'stop_completed'
- STOP_INVOKED = 'stop_invoked'
- class ska_mid_cbf_fhs_common.FhsObsStateModel(logger: ~logging.Logger, callback: ~typing.Callable[[~ska_control_model.obs_state.ObsState], None] | None = None, state_machine_factory: ~typing.Callable[[...], ~transitions.extensions.locking.LockedMachine] = <class 'ska_mid_cbf_fhs_common.state_model.fhs_obs_state.FhsObsStateMachine'>)[source]
Bases:
objectImplements the observation state model for subarray.
The model supports all of the states of the
ObsStateenum:EMPTY: the subarray is unresourced
RESOURCING: the subarray is performing a resourcing operation
IDLE: the subarray is resourced but unconfigured
CONFIGURING: the subarray is performing a configuring operation
READY: the subarray is resourced and configured
SCANNING: the subarray is scanning
ABORTING: the subarray is aborting
ABORTED: the subarray has aborted
RESETTING: the subarray is resetting from an ABORTED or FAULT state back to IDLE
RESTARTING: the subarray is restarting from an ABORTED or FAULT state back to EMPTY
FAULT: the subarray has encountered a observation fault.
Initialise the model.
- Parameters:
logger (
logging.Logger) – the logger to be used by this state model.callback (
Callable[[ObsState], None] | None, optional) – A callback to be called when a transition causes a change to device obs_state. Default is None.state_machine_factory (
Callable[..., Machine], optional) – a callable that returns a state machine for this model to use. Default is FhsObsStateMachine.
- is_action_allowed(action: str, raise_if_disallowed: bool = False) bool[source]
Return whether a given action is allowed in the current state.
- Parameters:
- Returns:
whether the action is allowed in the current state
- Return type:
- Raises:
StateModelError – The action is unknown to the state machine
- class ska_mid_cbf_fhs_common.ConfigurableThreadedTestTangoContextManager(**multi_device_context_kwargs)[source]
Bases:
ThreadedTestTangoContextManagerInitialise a new instance.
- class ska_mid_cbf_fhs_common.DeviceTestUtils[source]
Bases:
objectHelper class containing useful static methods for FHS device testing.
- static assert_lrc_completed(device_under_test: DeviceProxy, event_tracer: TangoEventTracer, timeout: float = 30.0, command_name: str | None = None, passing_result_codes: Iterable[ResultCode] | None = None, command_id: str | None = None, message: str | None = None, instant_fail_on_non_passing_result: bool = True, custom_non_passing_results: Iterable[ResultCode] | None = None) None[source]
Assert that a long-running command has completed with one of a set of provided result codes.
- Parameters:
device_under_test (
DeviceProxy) – The device being tested.event_tracer (
TangoEventTracer) – The event tracer for the device being tested. Note that this tracer must be set up to listen for longRunningCommandResult change events.timeout (
float, optional) – A timeout in seconds to wait for the command to complete. Default is 30.0.command_name (
str | None, optional) – Optionally, a command name to match against. Default is None, in which case any LRC completion will cause the assertion to pass.passing_result_codes (
Iterable[ResultCode] | None, optional) – A set of result codes which allow the assertion to pass. Default is [ResultCode.OK].command_id (
str | None, optional) – Optionally, a specific command ID to match against. Overrides command_name if both are provided. Default is None, in which case any LRC completion will cause the assertion to pass.message (
str | None, optional) – Optionally, a specific success/error message to match against. Default is None, in which case the returned message will be ignored.instant_fail_on_non_passing_result (
bool, optional) – If True, an LRC result (for the specified command, if provided) with a result code not matching any of the passing_result_codes, and/or a message not matching the provided message, will cause the assertion to instantly fail. If the custom_non_passing_results parameter is provided, that set of codes will be matched against instead. Default is True.custom_non_passing_results (
Iterable[ResultCode] | None, optional) – A specific set of result codes which cause the assertion to instantly fail. Ignored if the instant_fail_on_non_passing_result parameter is False. Default is None.
- static lrc_early_stop_matcher(result_codes: Iterable[ResultCode], command_name: str | None = None, message: str | None = None, command_id: str | None = None, inverted: bool = False) Callable[[ReceivedEvent], bool][source]
Return an “early stop” matcher function, which returns True if a received event is a long-running command result, matches a given set of result codes, and optionally matches a command name.
Intended for use in the with_early_stop function of a SKA Tango assertion. See: https://developer.skao.int/projects/ska-tango-testing/en/0.8.0/guide/integration/advanced_use_cases.html
- Parameters:
result_codes (
Iterable[ResultCode]) – The result codes to match against.command_name (
str | None, optional) – An optional command name to match against. Default is None.message (
str | None, optional) – An optional specific success/error message to match against. Default is None.command_id (
str | None, optional) – An optional specific command ID to match against. Overrides command_name if provided. Default is None.inverted (
bool, optional) – If True, check for any result code and/or message EXCEPT the one(s) specified. Default is False.
- Returns:
The created early stop matcher function.
- Return type:
Callable[[ReceivedEvent], bool]
- static lrc_result_matcher(result_codes: Iterable[ResultCode], command_name: str | None = None, message: str | None = None, command_id: str | None = None, inverted: bool = False) Callable[[ReceivedEvent], bool][source]
Return a custom matcher function, which returns True if a received long-running command result matches a given set of result codes and optionally a command name.
Intended for use in the custom_matcher parameter of a SKA Tango assertion. See: https://developer.skao.int/projects/ska-tango-testing/en/0.8.0/guide/integration/advanced_use_cases.html
- Parameters:
result_codes (
Iterable[ResultCode]) – The result codes to match against.command_name (
str | None, optional) – An optional command name to match against. Default is None.message (
str | None, optional) – An optional specific success/error message to match against. Default is None.command_id (
str | None, optional) – An optional specific command ID to match against. Overrides command_name if provided. Default is None.inverted (
bool, optional) – If True, check for any result code and/or message EXCEPT the one(s) specified. Default is False.
- Returns:
The created custom matcher function.
- Return type:
Callable[[ReceivedEvent], bool]
- static polling_test_setup(device_under_test: DeviceProxy, event_tracer: TangoEventTracer, config_json_file: str | None = None, event_timeout: float = 30.0)[source]
Perform initial setup for a polling-based low-level device test, including calling Configure() (if desired) and Start().
- Parameters:
device_under_test (
DeviceProxy) – The device under test.event_tracer (
TangoEventTracer) – The event tracer being used for the device under test.config_json_file (
str | None, optional) – Optional path to a config file. If provided, its contents will be passed to Configure(), otherwise Configure() will be skipped. Default is None.event_timeout (
float, optional) – The event timeout in seconds to use when awaiting the Start() command. Default is 30.0.
- class ska_mid_cbf_fhs_common.SimModeCMBase(*args: Any, attr_change_callback: Callable[[str, Any], None] | None = None, attr_archive_callback: Callable[[str, Any], None] | None = None, **kwargs: Any)[source]
Bases:
TaskExecutorComponentManagerA base class for FHS device simulation mode component managers. FHS device instances can make use component managers that inherit from this class and FhsSimMode-derived mixins to simulate their own behaviour.
SimModeCMBase-derived simulation mode component managers should initialize to default behaviour and supply an interface matching the one expected of the genuine component manager that is being replaced; property getter methods are used to return attribute_overrides values instead of expected attributes, and functools.partial is used to override LRC submit_task methods with sim_command, which submits a generic LRC task that pulls from command_overrides values to generate expected behaviour.
Initialize default attribute and command overrides.
Requires only logger, communication_state_callback and component_state_callback arguments for BaseComponentManager initialization.
- get_attribute_override(attr_name)[source]
Return specified attribute override from queue
- Parameters:
attribute – specified attribute to return the value of
- get_command_override(command)[source]
Return specified command override from queue
- Parameters:
command – specified command to return the value of
- sim_command(*args: Any, task_callback: Callable, command_name: str, **kwargs: Any) tuple[TaskStatus, str][source]
Submit simulated command operation method to task executor queue. Along with the submitted _sim_command thread, provides the ability to modify LRC behaviour asynchronously to the device, thus approaching a truer, more tunable simulation of LRCs.
- Parameters:
task_callback (
Callable) – Callback function to update task statuscommand_name (
str) – name of the command to be simulated
- Returns:
A tuple containing a return code and a string message indicating status. The message is for information purpose only.
- Return type:
tuple[TaskStatus, str]
- start_communicating(*args, **kwargs) None[source]
Simulate basic start_communicating behaviour. Updates communication state and power state.
- class ska_mid_cbf_fhs_common.SimModeObsCMBase(*args: Any, **kwargs: Any)[source]
Bases:
SimModeCMBaseA base class for FHS observing device simulation mode component managers.
Initialize default obs device attribute and command overrides.
- abort_commands(task_callback: TaskCallbackType | None = None) tuple[TaskStatus, str][source]
Abort all tasks queued & running.
- Parameters:
task_callback (
TaskCallbackType | None, optional) – callback to be called whenever the statusNone. (of the task changes. Default is)
- class ska_mid_cbf_fhs_common.FhsSimMode(self: tango._tango.Device_6Impl, klass: tango._tango.DeviceClass, name: str, description: str = 'A Tango device', state: tango._tango.DevState = <DevState.UNKNOWN: 13>, status: str = 'Not initialised')[source]
Bases:
DeviceA base simulation mode mixin class for Mid.CBF FHS devices.
Adds the simOverrides attribute which allows for overriding device behaviour when in simulation mode when used alongside SimModeCMBase-derived component managers.
- TangoClassClass
alias of
FhsSimModeClass
- TangoClassName = 'FhsSimMode'
- simOverrides
Attribute value overrides (JSON dict)
- class ska_mid_cbf_fhs_common.FhsObsSimMode(self: tango._tango.Device_6Impl, klass: tango._tango.DeviceClass, name: str, description: str = 'A Tango device', state: tango._tango.DevState = <DevState.UNKNOWN: 13>, status: str = 'Not initialised')[source]
Bases:
FhsSimModeA base simulation mode mixin class for Mid.CBF FHS observing devices.
- TangoClassClass
alias of
FhsObsSimModeClass
- TangoClassName = 'FhsObsSimMode'
- simOverrides
Attribute value overrides (JSON dict)