Controller subpackage

This subpackage implements MCCS controller functionality.

It includes a controller Tango device and a CLI.

class ControllerComponentManager(subarray_trls, station_trls, subarray_beam_trls, station_beam_trls, logger, max_workers, communication_state_callback, component_state_callback)

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, station_trls, subarray_beam_trls, station_beam_trls, logger, max_workers, communication_state_callback, component_state_callback)

Initialise a new instance.

Parameters:

  • subarray_trls (Iterable[str]) – TRLS of all subarray devices

  • station_trls (Iterable[str]) – TRLS of all station devices

  • subarray_beam_trls (Iterable[str]) – TRLS of all subarray beam devices

  • station_beam_trls (Iterable[str]) – TRLS of all station beam devices

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

  • max_workers (int) – nos. of worker threads

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

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

allocate(task_callback=None, *, interface=None, subarray_id, subarray_beams)

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: :type interface: Optional[str] :param interface: the schema version this is running against. :type subarray_id: int :param subarray_id: int, ID of the subarray which requires allocation :type subarray_beams: list[dict] :param subarray_beams: list of dictionaries, each sepcifying a beam allocation :type task_callback: Optional[Callable] :param task_callback: callback to signal end of command :rtype: tuple[TaskStatus, str] :return: A tuple containing a task status and a unique id

get_health_trl(trl)

Return the health of a subdevice with given TRL.

Parameters:

trl (str) – TRL of device to return health of

Return type:

Optional[HealthState]

Returns:

health of device given by TRL

Raises:

ValueError – if TRL is for an invalid device

get_healths(device_type='all')

Return subdevice healths.

Parameters:

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

Return type:

dict[str, dict[str, Optional[str]]]

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)

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

Parameters:

subarray_id (int) – The subarray ID of the resources

Return type:

str

Returns:

json formatted dictionary

off(task_callback=None)

Turn off the MCCS subsystem.

Parameters:

task_callback (Optional[Callable]) – Update task state, defaults to None

Return type:

tuple[TaskStatus, str]

Returns:

a TaskStatus and message

on(task_callback=None)

Turn on the MCCS subsystem.

Parameters:

task_callback (Optional[Callable]) – Update task state, defaults to None

Return type:

tuple[TaskStatus, str]

Returns:

task status and message

property power_state: ska_control_model.PowerState | None

Return my power state.

Returns:

my power state

release(subarray_id)

Release a subarray’s resources.

Parameters:

subarray_id (int) – ID of the subarray which requires release

Return type:

tuple[ResultCode, str]

Returns:

a result code and message

release_all()

Release all subarrays resources.

Return type:

tuple[ResultCode, str]

Returns:

a result code and message

restart_subarray(subarray_id, task_callback=None)

Restart an MCCS subarray.

Parameters:

  • subarray_id (int) – an integer subarray_id.

  • task_callback (Optional[Callable]) – Update task state, defaults to None

Return type:

tuple[TaskStatus, str]

Returns:

a task status and a message

standby(task_callback=None)

Put the MCCS subsystem in standby mode.

Parameters:

task_callback (Optional[Callable]) – Update task state, defaults to None

Return type:

tuple[TaskStatus, str]

Returns:

task status and message

start_communicating()

Establish communication with the station components.

Return type:

None

stop_communicating()

Break off communication with the station components.

Return type:

None

class ControllerHealthModel(component_state_callback, station_trls, subarray_beam_trls, station_beam_trls, thresholds)

A health model for a controller.

__init__(component_state_callback, station_trls, subarray_beam_trls, station_beam_trls, thresholds)

Initialise a new instance.

Parameters:

  • component_state_callback (Callable[..., None]) – callback to be called whenever there is a change to this this health model’s evaluated health state.

  • station_trls (Sequence[str]) – the TRLs of this controller’s stations

  • subarray_beam_trls (Sequence[str]) – the TRLs of this controller’s subarray beams

  • station_beam_trls (Sequence[str]) – the TRLs of this controller’s station beams

  • thresholds (dict[str, float]) – The thresholds for number of sub devices healthy

evaluate_health()

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.

Return type:

HealthState

Returns:

an overall health of the controller

station_beam_health_changed(station_beam_trl, station_beam_health)

Handle a change in station beam health.

Parameters:

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

  • station_beam_health (Optional[HealthState]) – 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.

Return type:

None

station_health_changed(station_trl, station_health)

Handle a change in station health.

Parameters:

  • station_trl (str) – the TRL of the station whose health has changed

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

Return type:

None

subarray_beam_health_changed(subarray_beam_trl, subarray_beam_health)

Handle a change in subarray beam health.

Parameters:

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

  • subarray_beam_health (Optional[HealthState]) – 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.

Return type:

None

class ControllerResourceManager(subarrays, subarray_beams, station_beams, channel_blocks)

A resource manager for the controller component manager.

__init__(subarrays, subarray_beams, station_beams, channel_blocks)

Initialise a new instance.

Parameters:

  • subarrays (Iterable[str]) – all subarrays to be managed by this resource manager

  • subarray_beams (Iterable[str]) – all subarray beams to be managed by this resource manager

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

  • channel_blocks (Iterable[int]) – all channel blocks to be managed by this resource manager

allocate(subarray, **resources)

Allocate resources to a subarray.

Parameters:

  • subarray (str) – the subarray to which resources are to be allocated

  • resources (Iterable[Hashable]) –

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

Return type:

None

deallocate(**resources)

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

Parameters:

resources (Iterable[Hashable]) –

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

Return type:

None

deallocate_from(subarray)

Deallocate all resources from a subarray.

Parameters:

subarray (str) – the subarray to which resources are to be allocated

Return type:

None

get_allocated(subarray)

Return the resources allocated to a given subarray.

Parameters:

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

Return type:

Mapping[str, Iterable[str]]

Returns:

the resources allocated to the subarray.

set_health(resource_type, resource, is_healthy)

Set the health of a resource.

Parameters:

  • resource_type (str) – the type of resource whose health is to be set

  • resource (Hashable) – the resource whose health is to be set

  • is_healthy (bool) – whether the resource is healthy or not

Return type:

None

set_ready(subarray, is_ready)

Set the health of a resource.

Parameters:

  • subarray (str) – the subarray to be set as ready

  • is_ready (bool) – whether the subarray is ready or not

Return type:

None

class MccsController(*args, **kwargs)

An implementation of a controller Tango device for MCCS.

Allocate(argin)

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 (str) – JSON-formatted string containing an integer subarray ID, and resources to be allocated to that subarray

Return type:

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

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(command_tracker, component_manager, callback=None, logger=None)

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, component_manager, callback=None, logger=None)

Initialise a new AllocateCommand instance.

Parameters:

  • command_tracker (CommandTracker) – the command tracker used to track this command

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

  • callback (Optional[Callable]) – the callback to be called at command completion

  • logger (Optional[Logger]) – a logger for this command to use.

GetAssignedResources(subarray_id)

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

Parameters:

subarray_id (int) – The subarray ID of the resources

Return type:

str

Returns:

json formatted dictionary

GetHealthTrl(argin)

Return health of device given by TRL.

Parameters:

argin (str) – TRL of device to return health of.

Return type:

Optional[str]

Returns:

health of device given by TRL.

class InitCommand(*args, **kwargs)

A class for MccsController’s Init command.

The do() method below is

called during MccsController’s initialisation.

do(*args, **kwargs)

Initialise the attributes and properties of the MccsController.

Parameters:

  • args (Any) – positional args to the component manager method

  • kwargs (Any) – keyword args to the component manager method

Return type:

tuple[ResultCode, str]

Returns:

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

Release(argin)

Release resources from an MCCS Sub-Array.

Parameters:

argin (str) – JSON-formatted string

Return type:

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

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()

Release all resources from an MCCS Array.

Return type:

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

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(component_manager, logger=None)

Class for handling the ReleaseAll() command.

schemas/MccsController_Release_2_0.json

__init__(component_manager, logger=None)

Initialise a new ReleaseAllCommand instance.

Parameters:
do(*args, **kwargs)

Implement MccsController.Release() command functionality.

Parameters:

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

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

Return type:

tuple[ResultCode, str]

Returns:

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

class ReleaseCommand(component_manager, logger=None)

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, logger=None)

Initialise a new ReleaseCommand instance.

Parameters:
do(*args, **kwargs)

Implement MccsController.Release() command functionality.

Parameters:

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

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

Return type:

tuple[ResultCode, str]

Returns:

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

RestartSubarray(argin)

Restart an MCCS subarray.

Parameters:

argin (int) – an integer subarray_id.

Return type:

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

Returns:

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

StandbyFull()

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.

Return type:

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

Returns:

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

StandbyLow()

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.

Return type:

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

Returns:

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

__init__(*args, **kwargs)

Initialise this device object.

Parameters:

  • args (Any) – positional args to the init

  • kwargs (Any) – keyword args to the init

buildState()

Read the Build State of the device.

Return type:

str

Returns:

the build state of the device

create_component_manager()

Create and return a component manager for this device.

Return type:

ControllerComponentManager

Returns:

a component manager for this device.

healthModelParams(argin)

Set the params for health transition rules.

Parameters:

argin (str) – JSON-string of dictionary of health states

Return type:

None

init_command_objects()

Set up the handler objects for Commands.

Return type:

None

init_device()

Initialise the device.

This is overridden here to change the Tango serialisation model.

Return type:

None

stationBeamHealths()

Read the health of station beams controlled by the device.

Return type:

str

Returns:

health of station beams in a json format

stationHealths()

Read the health of stations controlled by the device.

Return type:

str

Returns:

health of stations in a json format

subDeviceHealths()

Read the health of all subdevices controlled by the device.

Return type:

str

Returns:

health of subdevices in a json tree format

subarrayBeamHealths()

Read the health of subarray beams controlled by the device.

Return type:

str

Returns:

health of subarray beams in a json format

subarrayHealths()

Read the health of subarrays controlled by the device.

Return type:

str

Returns:

health of subarrays in a json format

versionId()

Read the Version Id of the device.

Return type:

str

Returns:

the version id of the device