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.

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

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

Allocate(subarray_id: int, subarray_beams: list[dict] | None = None, interface: str | None = None) → ska_tango_base.type_hints.TaskFunctionType[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:

subarray_id – integer subarray ID
subarray_beams – list of subarray beam allocation dicts
interface – schema interface version 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.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,
            ],
        }
        )
    )

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.

Release(subarray_id: int, interface: str | None = None) → ska_tango_base.type_hints.TaskFunctionType[source]

Release resources from an MCCS Sub-Array.

Parameters:

subarray_id – integer ID of the subarray to release
interface – schema interface version 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
            }
        )
    )

ReleaseAll() → ska_tango_base.type_hints.TaskFunctionType[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()

RestartSubarray(argin: int) → ska_tango_base.type_hints.TaskFunctionType[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[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[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.

TriggerAdcEqualisation(station_args: list[dict]) → ska_tango_base.type_hints.TaskFunctionType[source]

Trigger ADC equalisation.

Parameters:: station_args – list of per-station dicts
Returns:: A tuple containing a return code and a string message indicating status. The message is for information purposes only.
Example:

>>> proxy = tango.DeviceProxy("ska-low-mccs/control/control")
>>> proxy.TriggerAdcEqualisation(
        json.dumps(
                {
                    "station_args":
                        [
                            {
                                station_id: 1,
                                target_adc: 17,
                                bias: 1,
                            },
                            {
                                station_id: 2,
                                target_adc: 1,
                            },
                        ]
                }
            )
    )
>>> # For all stations:
>>> proxy.TriggerAdcEqualisation(
        json.dumps("")
    )

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

execute_Off() → ska_tango_base.type_hints.TaskFunctionType[source]

Put the Controller OFF.

Returns:: A tuple containing a return code and a string message indicating status.

execute_On() → ska_tango_base.type_hints.TaskFunctionType[source]

Put the Controller ON.

Returns:: A tuple containing a return code and a string message indicating status.

execute_Standby() → ska_tango_base.type_hints.TaskFunctionType[source]

Put the Controller into STANDBY.

Returns:: A tuple containing a return code and a string message indicating status.

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.

healthThresholds(argin: str) → None[source]

Set the params for health transition rules.

Default health thresholds:

“devices”: (f2f, d2f, d2d),

tuple(int, int, int): Number of devices failed before health failed,
Number of devices degraded before health failed, Number of devices degraded before health degraded

Valid devices are: “subarrays”, “stations”, “subarraybeams”, “stationbeams”.

Parameters:: argin – JSON-string of dictionary of health thresholds

init_device() → None[source]: Initialise the device.

is_Off_allowed(request_type: LRCReqType | None = ska_tango_base.long_running_commands.LRCReqType.ENQUEUE_REQ) → bool[source]

Return whether the OFF command is allowed.

Parameters:: request_type – The request type
Returns:: True if OFF command is allowed
Raises:: CmdNotAllowedError – if there are no devices to turn off

is_On_allowed(request_type: LRCReqType | None = ska_tango_base.long_running_commands.LRCReqType.ENQUEUE_REQ) → bool[source]

Return whether the On command is allowed.

Parameters:: request_type – The request type
Returns:: False if command is not allowed by base class
Raises:: CmdNotAllowedError – if there are no subservient devices

is_Release_allowed(request_type: LRCReqType | None = ska_tango_base.long_running_commands.LRCReqType.ENQUEUE_REQ) → bool[source]

Return whether the Release command is allowed.

Parameters:: request_type – The request type
Returns:: False if command is not allowed by base class
Raises:: CmdNotAllowedError – if there are no subservient devices

is_RestartSubarray_allowed(request_type: LRCReqType | None = ska_tango_base.long_running_commands.LRCReqType.ENQUEUE_REQ) → bool[source]

Return whether the Release command is allowed.

Parameters:: request_type – The request type
Returns:: False if command is not allowed by base class
Raises:: CmdNotAllowedError – if there are no subservient devices

is_StandbyFull_allowed(request_type: LRCReqType | None = ska_tango_base.long_running_commands.LRCReqType.ENQUEUE_REQ) → bool[source]

Return whether the StandbyFull command is allowed.

Parameters:: request_type – The request type
Returns:: False if command is not allowed by base class
Raises:: CmdNotAllowedError – if there are no subservient devices

is_StandbyLow_allowed(request_type: LRCReqType | None = ska_tango_base.long_running_commands.LRCReqType.ENQUEUE_REQ) → bool[source]

Return whether the StandbyLow command is allowed.

Parameters:: request_type – The request type
Returns:: False if command is not allowed by base class
Raises:: CmdNotAllowedError – if there are no subservient devices

is_Standby_allowed(request_type: LRCReqType | None = ska_tango_base.long_running_commands.LRCReqType.ENQUEUE_REQ) → bool[source]

Return whether the Standby command is allowed.

Parameters:: request_type – The request type
Returns:: False if command is not allowed by base class
Raises:: CmdNotAllowedError – if there are no subservient devices

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

resourceSummary() → str[source]

Return a detailed summary of all resources managed by the controller.

This attribute provides comprehensive information about the allocation and availability status of all resources in the system, structured as a JSON object.

The summary includes: 1) Total resources present in the system (inventory of all managed resources) 2) Resources currently in use, including which subarray they’re allocated to 3) Available resources that are not currently allocated 4) Detailed breakdown of resources by station/location

Resources tracked include: - Channel blocks: Frequency channels allocated to stations for processing - Station beams: Beamforming resources that process signals from tiles - Subarray beams: Higher-level beams that combine data from station beams

This attribute is useful for: - Monitoring system capacity and utilization - Diagnosing resource allocation issues - Planning observations based on available resources - Verifying the success of allocation/release operations

Returns:: A JSON-formatted string containing detailed resource information organized by resource type with allocation statistics

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

useNewHealthModel(argin: bool) → None[source]

Set a flag indicating whether this controller is using the new health model.

Parameters:: argin – a flag indicating whether this controller is currently using the new health model.

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