Component Manager Base Classes

FhsComponentManagerBase Class

class ska_mid_cbf_fhs_common.base_classes.device.fhs_component_manager_base.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, ABC

Base 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:: bool

state

The current DevState of the device.

Type:: DevState

get_device_health_state() → HealthState[source]

Get the health state of the device.

Returns:: HealthState The health state of the device.

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.

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.

log_critical(msg: str, transaction_id: str | None = None) → None[source]

log_error(msg: str, transaction_id: str | None = None) → None[source]

log_warning(msg: str, transaction_id: str | None = None) → None[source]

log_info(msg: str, transaction_id: str | None = None) → None[source]

log_debug(msg: str, transaction_id: str | None = None) → None[source]

FhsControllerComponentManagerBase Class

class ska_mid_cbf_fhs_common.base_classes.device.controller.fhs_controller_component_manager_base.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, FhsComponentManagerBase

Extension 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.

abstract property config_schema: dict[str, Any]: The ConfigureScan input JSON schema for this controller.

abstract property config_dataclass: type[FhsControllerBaseConfig]: The ConfigureScan input dataclass for this controller.

device

Reference to the controller device instance.

Type:: FhsControllerBaseDevice

ip_block_list: list[str]

List of IP block IDs managed by this controller.

Type:: list[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]

simulation_mode

Whether the controller is deployed in simulation mode.

Type:: SimulationMode

emulation_mode

Whether the controller is deployed in emulation mode.

Type:: bool

health_monitor

The health state monitor for this controller.

Type:: FhsHealthMonitor

obs_state

The current observation state of this controller.

Type:: ObsState

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.

Parameters:

command_name (str) – name of command for result message
task_callback (Callable) – command tracker update_command_info callback
task_abort_event (Event) – task executor abort event

Returns:

True if abort event is set, otherwise False

Return type:

bool

start_communicating() → None[source]: Establish communication with the component, then start monitoring.

stop_communicating() → None[source]: Close communication with the component, then stop monitoring.

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:: bool

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:: bool

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:: bool

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]

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]

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 none
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]

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 none
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]

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 none
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]

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]

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]]

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]

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]

FhsLowLevelComponentManagerBase Class

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