Controller Device

This module contains the SKA Low MCCS Controller device prototype.

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

main(*args: str, **kwargs: str) → int[source]

Entry point for module.

Parameters:

args – positional arguments
kwargs – named arguments

Returns:

exit code