Station Device

This module implements the MCCS station device.

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

An implementation of a station beam Tango device for MCCS.

AcquireDataForCalibration(channel: int) → tuple[list[ska_control_model.ResultCode], list[Optional[str]]][source]

Instruct the SpsStation to start acquiring calibration data from the tiles.

Parameters:: channel – the frequency channel to calibrate for.
Returns:: A tuple containing a return code and a string message indicating status. The message is for information purpose only.

ApplyConfiguration(transaction_id: str) → tuple[list[ska_control_model.ResultCode], list[Optional[str]]][source]

Apply the aggregated channel table to this Station’s SpsStation.

Parameters:: transaction_id – transaction id for the configuration
Returns:: A tuple containing a return code and a string message indicating status. The message is for information purpose only.
Example:

>>> dp = tango.DeviceProxy("low-mccs/station/ci-1")
>>> dp.command_inout("ApplyConfiguration")

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

Set the pointing delay parameters of this Station’s Tiles.

Parameters:: argin – switch time, in ISO formatted time. Default: now
Returns:: A tuple containing a return code and a string message indicating status. The message is for information purpose only.
Example:

>>> dp = tango.DeviceProxy("low-mccs/station/ci-1")
>>> time_string = switch time as ISO formatted time
>>> dp.command_inout("ApplyPointingDelays", time_string)

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

Set the beamformer table entries for a station beam.

Entries are defined as a flattened 2D array, for a maximum of 48 entries Each entry is 7 channels long and corresponds to 8 consecutive frequency channels.

Parameters:: argin – list of channel block description. Elements are: * channel block index: value in range 0:47 for the channel block to set * start_channel - (int) region starting channel, even in range 0 to 510 * beam_index - (int) beam used for this region with range 0 to 47 * subarray_id - (int) Subarray: 0 is reserved for unallocated entry * subarray_logical_channel - (int) logical channel # in the subarray * subarray_beam_id - (int) ID of the subarray beam * substation_id - (int) Substation * aperture_id: ID of the aperture (station*100+substation?)
Returns:: A tuple containing a return code and a string message indicating status. The message is for information purpose only.
Example:

>>> dp = tango.DeviceProxy("low-mccs/station/ci-1")
>>> dp.command_inout("ConfigureChannels", block_table)

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

Configure semi static information like position of antennas.

Parameters:: argin – Semi static information
Returns:: A tuple containing a return code and a string message indicating status. The message is for information purpose only.
Example:

>>> dp = tango.DeviceProxy("low-mccs/station/ci-1")
>>> config = json.dumps({
    "station": {
        "StationId": 1,
        "ref_latitude": 1.0,
        "ref_longitude": 1.0,
        "ref_height": 1.0,
    }
})
>>> dp.command_inout("ConfigureSemiStatic", config)

DeallocateSubarray(subarray_id: int) → tuple[list[ska_control_model.ResultCode], list[Optional[str]]][source]

Deallocates entries relative to a subarray in aggregate tables.

Parameters:: subarray_id – the ID of the subarray to deallocate
Returns:: A tuple containing a return code and a string message indicating status. The message is for information purpose only.

EndScan(subarray_id: int) → tuple[list[ska_control_model.ResultCode], list[Optional[str]]][source]

Stop the current scan associated with the station_beam.

Parameters:: subarray_id – the subarray for which the command applies
Returns:: A tuple containing a return code and a string message indicating status. The message is for information purpose only.

GetPointingDelays(argin: str) → ndarray[source]

Get Pointing Coefficients.

Parameters:: argin – stringified dict of args
Returns:: The pointing delays as pairs of (delay, delay rate) in EEP order.

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

Class for handling the GetPointingDelays() command.

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

Initialise a new instance.

Parameters:

component_manager – the device to which this command belongs.
logger – a logger for this command to use.

do(*args: Any, **kwargs: Any) → ndarray[source]

Implement MccsStation.GetPointingDelaysCommand() command.

Parameters:

args – unspecified positional arguments. This should be empty and is provided for type hinting only
kwargs – unspecified keyword arguments. This should be empty and is provided for type hinting only

Returns:

json encoded string containing list of dictionaries

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

A class for MccsStation’s Init command.

The do() method below is called upon MccsStation’s initialisation.

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

Initialise the MccsStation.

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.

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

Initialise this station’s tiles.

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

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

Set the pointing delay parameters of this Station’s Tiles.

Parameters:: argin – an array containing a beam index followed by antenna delays in antenna EEP order. 1 + 256*2 = 513 elements.
Returns:: A tuple containing a return code and a string message indicating status. The message is for information purpose only.
Example:

>>> dp = tango.DeviceProxy("low-mccs/station/ci-1")
>>> dp.command_inout("LoadPointingDelays", delay_list)

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

Start the scan associated with the station_beam.

Parameters:: argin – Configuration parameters encoded in a json string
Returns:: A tuple containing a return code and a string message indicating status. The message is for information purpose only.

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

Start data acquisition.

Parameters:: argin – json dictionary with optional keywords:

start_time - (ISO UTC time) start time
delay - (int) delay start if StartTime is not specified, default 2s

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

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

Class for handling the StartAcquisition() command.

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

{
    "$schema": "https://json-schema.org/draft/2020-12/schema",
    "$id": "https://skao.int/Mccs_StartAcquisition.json",
    "title": "StartAcquisition schema",
    "description": "Schema for StartAcquisition command",
    "type": "object",
    "properties": {
        "start_time": {
            "description": "",
            "type": "string",
            "format": "time"
        },
        "delay": {
            "description": "An acquisition start delay",
            "type": "integer"
        }
    }
}

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

Initialise a new instance.

Parameters:

command_tracker – the device’s command tracker
component_manager – the device’s component manager
callback – an optional callback to be called when this command starts and finishes.
logger – a logger for this command to log with.

StopTracking(track_id: int) → tuple[list[ska_control_model.ResultCode], list[Optional[str]]][source]

Stop tracking an object.

Parameters:: track_id – The ID of the thread you wish to stop tracking.
Returns:: A tuple containing a return code and a string message indicating status. The message is for information purpose only.

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

Stop all tracking.

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

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

Track an object through the sky.

Parameters:: argin – Configuration parameters encoded in a json string
Returns:: The id of the tracking thread.

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

Initialise this device object.

Parameters:

args – positional args to the init
kwargs – keyword args to the init

beamTrls() → list[str][source]

Return the TRLs of station beams associated with this station.

Returns:: the TRLs of station beams associated with this station

beamformerTable() → list[list[int]][source]

Return the ids of the channels configured for this beam.

Returns:: channel table

calibrationCoefficients() → list[float][source]

Return the calibration coefficients for the station.

Todo:: How big should this array be? 4 complex values (Jones matrix) per channel. This station can have up to 16 tiles of up to 16 antennas, so that is 8 x 16 x 16 = 2048 coefficients per channel. But how many channels? 384 channels, 786432 elements per station (402M for SKA Low)
Returns:: the calibration coefficients

calibrationJobId() → int[source]

Return the calibration job id.

Returns:: the calibration job id

create_component_manager() → StationComponentManager[source]

Create and return a component manager for this device.

Returns:: a component manager for this device.

daqJobId() → int[source]

Return the DAQ job id.

Returns:: the DAQ job id

dataDirectory() → str[source]

Return the data directory.

(the parent directory for all files generated by this station)

Returns:: the data directory

dataReceivedResult() → tuple[str, str] | None[source]

Read the result of the receiving of data.

Returns:: A tuple containing the data mode of transmission and a json string with any additional data about the data such as the file name.

delayCentre(value: list[float]) → None[source]

Set the delay centre of the station.

Parameters:: value – WGS84 position

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.

health_changed(health: ska_control_model.HealthState) → None[source]

Handle change in this device’s health state.

This is a callback hook, called whenever the HealthModel’s evaluated health state changes. It is responsible for updating the tango side of things i.e. making sure the attribute is up to date, and events are pushed.

Parameters:: health – the new health value

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.

isCalibrated() → bool[source]

Return a flag indicating whether this station is currently calibrated or not.

Returns:: a flag indicating whether this station is currently calibrated or not.

isConfigured() → bool[source]

Return a flag indicating whether this station is currently configured or not.

Returns:: a flag indicating whether this station is currently configured or not.

isInitialised() → bool[source]

Return true if all tiles in the SpsStation are initialised.

Returns:: true if all tiles in the SpsStation are initialised.

isSynchronised() → bool[source]

Return true if all tiles in the SpsStation are synchronised.

Returns:: true if all tiles in the SpsStation are synchronised.

is_On_allowed() → bool[source]

Check if command Off is allowed in the current device state.

Returns:: True if the command is allowed

lastPointingDelays() → list[float][source]

Return last pointing delays applied to the tiles.

Values are initialised to 0.0 if they haven’t been set. These values are in antenna EEP order.

Returns:: last pointing delays applied to the tiles.

numberOfChannels() → int[source]

Return the total number of channels in the beamformer.

Returns:: the total number of channels

outsideTemperature() → float | None[source]

Return the OutsideTemperature.

Returns:: the OutsideTemperature.

refHeight() → float[source]

Return the refHeight attribute.

Returns:: the ellipsoidal height of the station reference position

refLatitude() → float[source]

Return the refLatitude attribute.

Returns:: the WGS84 Latitude of the station reference position

refLongitude() → float[source]

Return the refLongitude attribute.

Returns:: the WGS84 Longitude of the station reference position

transientBufferTrl() → str[source]

Return the TRL of the TANGO device that managers the transient buffer.

Returns:: the TRL of the TANGO device that managers the transient buffer

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

Entry point for module.

Parameters:

args – positional arguments
kwargs – named arguments

Returns:

exit code