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 | None]][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(argin: str) → tuple[list[ska_control_model.ResultCode], list[str | None]][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[str | None]][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
            }
        )
    )

ReleaseAll() → tuple[list[ska_control_model.ResultCode], list[str | None]][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) → tuple[list[ska_control_model.ResultCode], list[str | None]][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 | None]][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 | None]][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(argin: str) → tuple[list[ska_control_model.ResultCode], list[str | None]][source]

Trigger ADC equalistation.

Parameters:: argin – JSON-formatted string
Returns:: A tuple containing a return code and a string message indicating status. The message is for information purpose only.
Raises:
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.

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_command_objects() → None[source]: Set up the handler objects for Commands.

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

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