Controller subpackage

This subpackage implements MCCS controller functionality.

It includes a controller Tango device and a CLI.

class ControllerComponentManager(*args: Any, **kwargs: Any)[source]

A component manager for an MCCS controller.

This component manager has three jobs:

  • Monitoring of the devices in the MCCS subsystem

  • Powering the MCCS subsystem off and on

  • Allocating resources to subarrays

__init__(subarray_trls: Iterable[str], station_trls: Iterable[str], subarray_beam_trls: Iterable[str], station_beam_trls: Iterable[str], logger: Logger, obs_command_timeout: int, communication_state_callback: Callable[[ska_control_model.CommunicationStatus], None], component_state_callback: Callable[[...], None]) → None[source]

Initialise a new instance.

Parameters:

  • subarray_trls – TRLS of all subarray devices

  • station_trls – TRLS of all station devices

  • subarray_beam_trls – TRLS of all subarray beam devices

  • station_beam_trls – TRLS of all station beam devices

  • obs_command_timeout – the default timeout for obs commands in seconds.

  • logger – the logger to be used by this object.

  • communication_state_callback – callback to be called when the status of the communications channel between the component manager and its component changes

  • component_state_callback – callback to be called when the component state changes

allocate(task_callback: Callable | None = None, *, interface: str | None = None, subarray_id: int, subarray_beams: list[dict]) → tuple[ska_control_model.TaskStatus, str][source]

Allocate and distribute a set of unallocated MCCS resources to a subarray.

The kwargs argument specifies the overall sub-array composition in terms of which stations should be allocated to the specified subarray_beam. It contains: :param interface: the schema version this is running against. :param subarray_id: int, ID of the subarray which requires allocation :param subarray_beams: list of dictionaries, each sepcifying a beam allocation :param task_callback: callback to signal end of command :return: A tuple containing a task status and a unique id

get_health_trl(trl: str) → ska_control_model.HealthState | None[source]

Return the health of a subdevice with given TRL.

Parameters:

trl – TRL of device to return health of

Returns:

health of device given by TRL

Raises:

ValueError – if TRL is for an invalid device

get_healths(device_type: str = 'all') → dict[str, dict[str, Optional[str]]][source]

Return subdevice healths.

Parameters:

device_type – the type of device to return the health state of its instances.

Returns:

dictionary of [device_type: [TRL : HealthState]] for each device of that device type, or all devices if called without an argument.

Raises:

ValueError – if device_type is not a valid device type.

get_resources(subarray_id: int) → str[source]

Return a dictionary of the resources assigned to a given subarray.

Parameters:

subarray_id – The subarray ID of the resources

Returns:

json formatted dictionary

off(task_callback: Callable | None = None) → tuple[ska_control_model.TaskStatus, str][source]

Turn off the MCCS subsystem.

Parameters:

task_callback – Update task state, defaults to None

Returns:

a TaskStatus and message

on(task_callback: Callable | None = None) → tuple[ska_control_model.TaskStatus, str][source]

Turn on the MCCS subsystem.

Parameters:

task_callback – Update task state, defaults to None

Returns:

task status and message

property power_state: ska_control_model.PowerState | None

Return my power state.

Returns:

my power state

release(subarray_id: int) → tuple[ska_control_model.ResultCode, str][source]

Release a subarray’s resources.

Parameters:

subarray_id – ID of the subarray which requires release

Returns:

a result code and message

release_all() → tuple[ska_control_model.ResultCode, str][source]

Release all subarrays resources.

Returns:

a result code and message

restart_subarray(subarray_id: int, task_callback: Callable | None = None) → tuple[ska_control_model.TaskStatus, str][source]

Restart an MCCS subarray.

Parameters:

  • subarray_id – an integer subarray_id.

  • task_callback – Update task state, defaults to None

Returns:

a task status and a message

standby(task_callback: Callable | None = None) → tuple[ska_control_model.TaskStatus, str][source]

Put the MCCS subsystem in standby mode.

Parameters:

task_callback – Update task state, defaults to None

Returns:

task status and message

start_communicating() → None[source]

Establish communication with the station components.

stop_communicating() → None[source]

Break off communication with the station components.

class ControllerHealthModel(*args: Any, **kwargs: Any)[source]

A health model for a controller.

__init__(component_state_callback: Callable[[...], None], station_trls: Sequence[str], subarray_beam_trls: Sequence[str], station_beam_trls: Sequence[str], thresholds: dict[str, float]) → None[source]

Initialise a new instance.

Parameters:

  • component_state_callback – callback to be called whenever there is a change to this this health model’s evaluated health state.

  • station_trls – the TRLs of this controller’s stations

  • subarray_beam_trls – the TRLs of this controller’s subarray beams

  • station_beam_trls – the TRLs of this controller’s station beams

  • thresholds – The thresholds for number of sub devices healthy

evaluate_health() → tuple[ska_control_model.HealthState, str][source]

Compute overall health of the controller.

The overall health is based on the fault and communication status of the controller overall, together with the health of the stations, subarray beams and station beams.

This implementation simply sets the health of the controller to the health of its least healthy component.

Returns:

an overall health of the controller

station_beam_health_changed(station_beam_trl: str, station_beam_health: ska_control_model.HealthState | None) → None[source]

Handle a change in station beam health.

Parameters:

  • station_beam_trl – the TRL of the station beam whose health has changed

  • station_beam_health – the health state of the specified station beam, or None if the station beam’s admin mode indicates that its health should not be rolled up.

station_health_changed(station_trl: str, station_health: ska_control_model.HealthState | None) → None[source]

Handle a change in station health.

Parameters:

  • station_trl – the TRL of the station whose health has changed

  • station_health – the health state of the specified station, or None if the station’s admin mode indicates that its health should not be rolled up.

subarray_beam_health_changed(subarray_beam_trl: str, subarray_beam_health: ska_control_model.HealthState | None) → None[source]

Handle a change in subarray beam health.

Parameters:

  • subarray_beam_trl – the TRL of the subarray beam whose health has changed

  • subarray_beam_health – the health state of the specified subarray beam, or None if the subarray beam’s admin mode indicates that its health should not be rolled up.

class ControllerResourceManager(subarrays: Iterable[str], subarray_beams: Iterable[str], station_beams: Iterable[str], channel_blocks: Iterable[int])[source]

A resource manager for the controller component manager.

__init__(subarrays: Iterable[str], subarray_beams: Iterable[str], station_beams: Iterable[str], channel_blocks: Iterable[int]) → None[source]

Initialise a new instance.

Parameters:

  • subarrays – all subarrays to be managed by this resource manager

  • subarray_beams – all subarray beams to be managed by this resource manager

  • station_beams – all station beams to be managed by this resource manager (as a resource pool)

  • channel_blocks – all channel blocks to be managed by this resource manager

allocate(subarray: str, **resources: Iterable[Hashable]) → None[source]

Allocate resources to a subarray.

Parameters:

  • subarray – the subarray to which resources are to be allocated

  • resources –

    the resources to allocate. Each keyword specifies a resource type, with the value a list of the resources of that type to be allocated. For example:

    controller_resource_manager.allocate(
    "low-mccs/subarray/01",
    stations=[
        ["low-mccs/station/001", "low-mccs/station/002"]
    ],
    station_beams=[["low-mccs/beam/01", "low-mccs/beam/02"]]
    channel_blocks=[2, 3],
)

deallocate(**resources: Iterable[Hashable]) → None[source]

Deallocate resources (regardless of what subarray they are allocated to.

Parameters:

resources –

the resources to deallocate. Each keyword specifies a resource type, with the value a list of the resources of that type to be deallocated. For example:

controller_resource_manager.deallocate(
    stations=[
        "low-mccs/station/001", "low-mccs/station/002"
    ],
    channel_blocks=[2, 3],
)

deallocate_from(subarray: str) → None[source]

Deallocate all resources from a subarray.

Parameters:

subarray – the subarray to which resources are to be allocated

get_allocated(subarray: str) → Mapping[str, Iterable[str]][source]

Return the resources allocated to a given subarray.

Parameters:

subarray – TRL of the subarray for which the allocated resources are to be returned

Returns:

the resources allocated to the subarray.

set_ready(subarray: str, is_ready: bool) → None[source]

Set the health of a resource.

Parameters:

  • subarray – the subarray to be set as ready

  • is_ready – whether the subarray is ready or not

class MccsController(*args: Any, **kwargs: Any)[source]

An implementation of a controller Tango device for MCCS.

Allocate(argin: str) → tuple[list[ska_control_model.ResultCode], list[Optional[str]]][source]

Allocate a set of unallocated MCCS resources to a sub-array.

The JSON argument specifies the overall sub-array composition in terms of which stations should be allocated to the specified Sub-Array.

Parameters:

argin – JSON-formatted string containing an integer subarray ID, and resources to be allocated to that subarray

Returns:

A tuple containing a return code, a string message indicating status and message UID. The string message is for information purposes only, but the message UID is for message management use.

Example:

>>> proxy = tango.DeviceProxy("ska-low-mccs/control/control")
>>> proxy.Allocate(
        json.dumps(
        {
            "interface":
                "https://schema.skao.int/ska-low-mccs-controller-allocate/3.0"
            "subarray_id": 1,
            "subarray_beams": [
                {
                    "subarray_beam_id": 3,
                    "apertures": [
                        {"station_id": 1, "aperture_id": "1.1" },
                        {"station_id": 2, "aperture_id": "2.2" },
                        {"station_id": 2, "aperture_id": "2.3" },
                        {"station_id": 3, "aperture_id": "3.1" },
                        {"station_id": 4, "aperture_id": "4.1" },
                    ],
                }
                "number_of_channels": 32,
            ],
        }
        )
    )
class AllocateCommand(*args: Any, **kwargs: Any)[source]

Class for handling the Allocate() command.

This command takes as input a JSON string that conforms to the following schema:

schemas/MccsController_Allocate_3_0.json

__init__(command_tracker: ska_tango_base.base.CommandTracker, component_manager: ControllerComponentManager, callback: Callable | None = None, logger: Logger | None = None) → None[source]

Initialise a new AllocateCommand instance.

Parameters:

  • command_tracker – the command tracker used to track this command

  • component_manager – the component manager to which this command belongs.

  • callback – the callback to be called at command completion

  • logger – a logger for this command to use.

GetAssignedResources(subarray_id: int) → str[source]

Return a dictionary of the resources assigned to a given subarray.

Parameters:

subarray_id – The subarray ID of the resources

Returns:

json formatted dictionary

GetHealthTrl(argin: str) → str | None[source]

Return health of device given by TRL.

Parameters:

argin – TRL of device to return health of.

Returns:

health of device given by TRL.

class InitCommand(*args: Any, **kwargs: Any)[source]

A class for MccsController’s Init command.

The do() method below is

called during MccsController’s initialisation.

do(*args: Any, **kwargs: Any) → tuple[ska_control_model.ResultCode, str][source]

Initialise the attributes and properties of the MccsController.

Parameters:

  • args – positional args to the component manager method

  • kwargs – keyword args to the component manager method

Returns:

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

Release(argin: str) → tuple[list[ska_control_model.ResultCode], list[Optional[str]]][source]

Release resources from an MCCS Sub-Array.

Parameters:

argin – JSON-formatted string

Returns:

A tuple containing a return code, a string message indicating status and message UID. The string message is for information purposes only, but the message UID is for message management use.

Example:

>>> proxy = tango.DeviceProxy("ska-low-mccs/control/control")
>>> proxy.Release(
        json.dumps(
            {
            "subarray_id": 1,
            "release_all": true
            }
        )
    )
ReleaseAll() → tuple[list[ska_control_model.ResultCode], list[Optional[str]]][source]

Release all resources from an MCCS Array.

Returns:

A tuple containing a return code, a string message indicating status and message UID. The string message is for information purposes only, but the message UID is for message management use.

Example:

>>> proxy = tango.DeviceProxy("ska-low-mccs/control/control")
>>> proxy.ReleaseAll()
class ReleaseAllCommand(*args: Any, **kwargs: Any)[source]

Class for handling the ReleaseAll() command.

schemas/MccsController_Release_2_0.json

__init__(component_manager: ControllerComponentManager, logger: Logger | None = None) → None[source]

Initialise a new ReleaseAllCommand instance.

Parameters:

  • component_manager – the device to which this command belongs.

  • logger – a logger for this command to use.

do(*args: Any, **kwargs: Any) → tuple[ska_control_model.ResultCode, str][source]

Implement MccsController.Release() command functionality.

Parameters:

  • args – Positional arguments. This should be empty and is provided for type hinting purposes only.

  • kwargs – keyword arguments unpacked from the JSON argument to the command.

Returns:

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

class ReleaseCommand(*args: Any, **kwargs: Any)[source]

Class for handling the Release() command.

This command takes as input a JSON string that conforms to the following schema:

schemas/MccsController_Release_2_0.json

__init__(component_manager: ControllerComponentManager, logger: Logger | None = None) → None[source]

Initialise a new ReleaseCommand instance.

Parameters:

  • component_manager – the device to which this command belongs.

  • logger – a logger for this command to use.

do(*args: Any, **kwargs: Any) → tuple[ska_control_model.ResultCode, str][source]

Implement MccsController.Release() command functionality.

Parameters:

  • args – Positional arguments. This should be empty and is provided for type hinting purposes only.

  • kwargs – keyword arguments unpacked from the JSON argument to the command.

Returns:

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

RestartSubarray(argin: int) → tuple[list[ska_control_model.ResultCode], list[Optional[str]]][source]

Restart an MCCS subarray.

Parameters:

argin – an integer subarray_id.

Returns:

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

StandbyFull() → tuple[list[ska_control_model.ResultCode], list[Optional[str]]][source]

Put MCCS into standby mode.

Some elements of SKA Mid have both low and full standby modes, but SKA Low has no such elements. We just need a Standby command, not separate StandbyLow and StandbyFull.

Returns:

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

StandbyLow() → tuple[list[ska_control_model.ResultCode], list[Optional[str]]][source]

Put MCCS into standby mode.

Some elements of SKA Mid have both low and full standby modes, but SKA Low has no such elements. We just need a Standby command, not separate StandbyLow and StandbyFull.

Returns:

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

__init__(*args: Any, **kwargs: Any) → None[source]

Initialise this device object.

Parameters:

  • args – positional args to the init

  • kwargs – keyword args to the init

buildState() → str[source]

Read the Build State of the device.

Returns:

the build state of the device

create_component_manager() → ControllerComponentManager[source]

Create and return a component manager for this device.

Returns:

a component manager for this device.

healthModelParams(argin: str) → None[source]

Set the params for health transition rules.

Parameters:

argin – JSON-string of dictionary of health states

healthReport() → str[source]

Get the health report.

Returns:

the health report.

init_command_objects() → None[source]

Set up the handler objects for Commands.

init_device() → None[source]

Initialise the device.

This is overridden here to change the Tango serialisation model.

missedEvents() → int[source]

Get the amount of missed change events.

Some commands rely on change events from sub-devices, sometime we miss these events, this attribute keeps track of how many we know we have missed.

Returns:

the amount of missed change events

stationBeamHealths() → str[source]

Read the health of station beams controlled by the device.

Returns:

health of station beams in a json format

stationHealths() → str[source]

Read the health of stations controlled by the device.

Returns:

health of stations in a json format

subDeviceHealths() → str[source]

Read the health of all subdevices controlled by the device.

Returns:

health of subdevices in a json tree format

subarrayBeamHealths() → str[source]

Read the health of subarray beams controlled by the device.

Returns:

health of subarray beams in a json format

subarrayHealths() → str[source]

Read the health of subarrays controlled by the device.

Returns:

health of subarrays in a json format

versionId() → str[source]

Read the Version Id of the device.

Returns:

the version id of the device