Station subpackage

This subpackage implements station functionality for MCCS.

class MccsStation(*args, **kwargs)

An implementation of a station beam Tango device for MCCS.

ApplyConfiguration(transaction_id)

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

Parameters:

transaction_id (str) – transaction id for the configuration

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.

Example:

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

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

Parameters:

argin (str) – switch time, in ISO formatted time. Default: now

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.

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)

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[int]) – 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?)

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.

Example:

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

Configure semi static information like position of antennas.

Parameters:

argin (str) – Semi static information

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.

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)

Deallocates entries relative to a subarray in aggregate tables.

Parameters:

subarray_id (int) – the ID of the subarray to deallocate

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.

EndScan(subarray_id)

Stop the current scan associated with the station_beam.

Parameters:

subarray_id (int) – the subarray for which the command applies

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.

GetPointingDelays(argin)

Get Pointing Coefficients.

Parameters:

argin (str) – stringified dict of args

Return type:

Tuple[ndarray, ndarray]

Returns:

The pointing delays

class InitCommand(*args, **kwargs)

A class for MccsStation’s Init command.

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

do(*args, **kwargs)

Initialise the MccsStation.

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.

Initialise()

Initialise this station’s tiles.

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.

LoadPointingDelays(argin)

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

Parameters:

argin (ndarray) – an array containing a beam index followed by antenna delays

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.

Example:

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

Start the scan associated with the station_beam.

Parameters:

argin (str) – Configuration parameters encoded in a json string

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.

StartAcquisition(argin)

Start data acquisition.

Parameters:

argin (str) – json dictionary with optional keywords:

  • start_time - (ISO UTC time) start time

  • delay - (int) delay start if StartTime is not specified, default 2s

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.

class StartAcquisitionCommand(command_tracker, component_manager, callback=None, logger=None)

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

Initialise a new instance.

Parameters:

  • command_tracker (CommandTracker) – the device’s command tracker

  • component_manager (StationComponentManager) – the device’s component manager

  • callback (Optional[Callable]) – an optional callback to be called when this command starts and finishes.

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

StopTracking(track_id)

Stop tracking an object.

Parameters:

track_id (int) – The ID of the thread you wish to stop tracking.

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.

StopTrackingAll()

Stop all tracking.

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.

TrackObject(argin)

Track an object through the sky.

Parameters:

argin (str) – Configuration parameters encoded in a json string

Return type:

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

Returns:

The id of the tracking thread.

__init__(*args, **kwargs)

Initialise this device object.

Parameters:

  • args (Any) – positional args to the init

  • kwargs (Any) – keyword args to the init

beamTrls()

Return the TRLs of station beams associated with this station.

Return type:

list[str]

Returns:

the TRLs of station beams associated with this station

beamformerTable()

Return the ids of the channels configured for this beam.

Return type:

list[list[int]]

Returns:

channel table

calibrationCoefficients()

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)

Return type:

list[float]

Returns:

the calibration coefficients

calibrationJobId()

Return the calibration job id.

Return type:

int

Returns:

the calibration job id

create_component_manager()

Create and return a component manager for this device.

Return type:

StationComponentManager

Returns:

a component manager for this device.

daqJobId()

Return the DAQ job id.

Return type:

int

Returns:

the DAQ job id

dataDirectory()

Return the data directory.

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

Return type:

str

Returns:

the data directory

delayCentre(value)

Set the delay centre of the station.

Parameters:

value (list[float]) – WGS84 position

Return type:

None

healthModelParams(argin)

Set the params for health transition rules.

Parameters:

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

Return type:

None

health_changed(health)

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 (HealthState) – the new health value

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

isCalibrated()

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

Return type:

bool

Returns:

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

isConfigured()

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

Return type:

bool

Returns:

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

isInitialised()

Return true if all tiles in the SpsStation are initialised.

Return type:

bool

Returns:

true if all tiles in the SpsStation are initialised.

isSynchronised()

Return true if all tiles in the SpsStation are synchronised.

Return type:

bool

Returns:

true if all tiles in the SpsStation are synchronised.

is_On_allowed()

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

Return type:

bool

Returns:

True if the command is allowed

numberOfChannels()

Return the total number of channels in the beamformer.

Return type:

int

Returns:

the total number of channels

outsideTemperature()

Return the OutsideTemperature.

Return type:

Optional[float]

Returns:

the OutsideTemperature.

refHeight()

Return the refHeight attribute.

Return type:

float

Returns:

the ellipsoidal height of the station reference position

refLatitude()

Return the refLatitude attribute.

Return type:

float

Returns:

the WGS84 Latitude of the station reference position

refLongitude()

Return the refLongitude attribute.

Return type:

float

Returns:

the WGS84 Longitude of the station reference position

transientBufferTrl()

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

Return type:

str

Returns:

the TRL of the TANGO device that managers the transient buffer

class StationComponentManager(station_id, ref_latitude, ref_longitude, ref_height, field_station_trl, antenna_trls, station_calibrator_trl, sps_station_trl, logger, max_workers, communication_state_callback, component_state_callback)

A component manager for a station.

__init__(station_id, ref_latitude, ref_longitude, ref_height, field_station_trl, antenna_trls, station_calibrator_trl, sps_station_trl, logger, max_workers, communication_state_callback, component_state_callback)

Initialise a new instance.

Parameters:

  • station_id (int) – the id of this station

  • ref_latitude (float) – reference latitude of the station.

  • ref_longitude (float) – reference longitude of the station.

  • ref_height (float) – reference ellipsoidal height of the station.

  • field_station_trl (str) – TRL of the Tango device that manages this station’s FieldStation

  • antenna_trls (Sequence[str]) – TRLs of the Tango devices and manage this station’s antennas

  • station_calibrator_trl (str) – TRL of the Tango devices and manage this station’s station calibrator

  • sps_station_trl (str) – TRL of the Tango devices and manage this station’s Spshw station

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

  • max_workers (int) – the maximum worker threads for the slow commands associated with this component manager.

  • 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

apply_configuration(transaction_id, task_callback=None)

Apply the beamformer and calibration configuration to SpsStation.

Parameters:

  • transaction_id (str) – the transaction id for the configuration

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

Return type:

tuple[TaskStatus, str]

Returns:

a result code and response string

apply_pointing_delays(load_time, task_callback=None)

Submit the apply_pointing method.

This method returns immediately after it submitted self._apply_pointing for execution.

Parameters:

  • load_time (str) – time at which to load the pointing delay

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

Return type:

tuple[TaskStatus, str]

Returns:

a task status and response message

property beamformer_table: list[list[int]]

Return the channel table reformatted as would be needed by ConfigureChannels.

Returns:

reformatted channel table

configure_channels(channel_blocks)

Configure channels for a station beam in the channel table.

Parameters:

channel_blocks (list[int]) – List of channel table entries

Return type:

ResultCode

Returns:

a result code and response string

configure_semi_static(task_callback=None, *, interface=None, station_config, field_station_config, antenna_config)

Submit the configure method.

TODO Check if this is required anymore This method returns immediately after it submitted self._configure_semi_static for execution.

Parameters:

  • interface (Optional[str]) – the schema version this is running against.

  • station_config (dict) – Configuration specification for the station device.

  • field_station_config (Optional[dict]) – Configuration specification for the field station device.

  • antenna_config (Optional[dict]) – Configuration specification for the antenna deviced.

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

Return type:

tuple[TaskStatus, str]

Returns:

a result code and response string

deallocate_subarray(subarray_id)

Clear channels for a station beam in the channel table.

Parameters:

subarray_id (int) – subarray_id to clear

Return type:

ResultCode

Returns:

a result code and response string

end_scan(subarray_id, task_callback=None, task_abort_event=None)

Submit the EndScan slow task.

This method returns immediately after it is submitted for execution.

Parameters:

  • subarray_id (int) – The subarray for which the command applies

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

  • task_abort_event (Optional[Event]) – Check for abort, defaults to None

Return type:

tuple[TaskStatus, str]

Returns:

Task status and response message

get_pointing_delays(task_callback=None, *, interface=None, pointing_type, values, time_step=10.0, reference_time=None)

Get the pointing delays for this station.

Parameters:

  • interface (Optional[str]) – the schema version this is running against.

  • pointing_type (str) – the type of pointing requested

  • values (dict) – the pointing values, either in alt_az or ra_dec

  • reference_time (Optional[str]) – time in which coordinates are equal, in ISO8601 formatted astropy.Time time

  • time_step (float) – How long between each time step in seconds

  • task_callback (Optional[Callable]) – callback to signal end of command

Return type:

tuple[ndarray, ndarray]

Returns:

list of pointing delays

initialise(task_callback=None)

Submit the initialise method.

This method returns immediately after it submitted self._initialise for execution.

Parameters:

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

Return type:

tuple[TaskStatus, str]

Returns:

a task staus and response message

property is_configured: bool

Return whether this station component manager is configured.

Returns:

whether this station component manager is configured.

load_pointing_delays(delays, task_callback=None)

Load the pointing delays for this station.

Parameters:
Return type:

tuple[TaskStatus, str]

Returns:

a result code and list of pointing delays

property number_of_channels: int

Return the total number of channels in the beamformer.

Returns:

the total numebr of channels

off(task_callback=None)

Submit the _off method.

This method returns immediately after it submitted self._off for execution.

Parameters:

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

Return type:

tuple[TaskStatus, str]

Returns:

a result code and response message

on(task_callback=None)

Submit the _on method.

This method returns immediately after it submitted self._on for execution.

Parameters:

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

Return type:

tuple[TaskStatus, str]

Returns:

a task staus and response message

property power_state: ska_control_model.PowerState | None

Return my power state.

Returns:

my power state

property power_state_lock: RLock

Return the power state lock of this component manager.

Returns:

the power state lock of this component manager.

property ref_height: float

Return whether this stations height.

Returns:

this stations height.

property ref_latitude: float

Return whether this stations latitude.

Returns:

this stations latitude.

property ref_longitude: float

Return whether this stations longitude.

Returns:

this stations longitude.

scan(task_callback=None, *, interface=None, subarray_id, scan_id, start_time=None, duration=0.0)

Submit the Scan slow task.

This method returns immediately after it is submitted for execution.

Parameters:

  • interface (Optional[str]) – the schema version this is running against.

  • subarray_id (int) – The subarray for whic the command applies

  • scan_id (int) – The ID for this scan

  • start_time (Optional[str]) – UTC time for begin of scan, None for immediate start

  • duration (Optional[float]) – Scan duration in seconds. 0.0 or omitted means forever

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

Return type:

tuple[TaskStatus, str]

Returns:

Task status and response message

property scan_ids: list[int]

Return the current scan IDs for each subarray.

Returns:

list of scan IDs starting from subarray 1, 0 = subarray not scanning

setup_pointing_helper()

Set up the pointing helper.

Return type:

None

start_acquisition(task_callback=None, *, start_time=None, delay=2)

Submit the start acquisition method.

This method returns immediately after it submitted self._start_acquisition for execution.

Parameters:

  • start_time (Optional[str]) – the time at which to start data acquisition, defaults to None

  • delay (Optional[int]) – delay start, defaults to 2

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

Return type:

tuple[TaskStatus, str]

Returns:

a task staus and response message

start_communicating()

Establish communication with the station components.

Return type:

None

stop_communicating()

Break off communication with the station components.

Return type:

None

stop_tracking(track_id, task_callback=None)

Stop a tracking thread.

Parameters:

  • track_id (int) – the id of the tracking you wish to stop.

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

Return type:

tuple[TaskStatus, str]

Returns:

a result code and list of pointing delays

stop_tracking_all(task_callback=None)

Stop all tracking threads.

Parameters:

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

Return type:

tuple[TaskStatus, str]

Returns:

a result code and list of pointing delays

property tileprogrammingstate: tuple[str]

Return the tileprogrammingstate of the SpsStation.

Returns:

the tileprogrammingstate of the SpsStation.

track_object(task_callback=None, task_abort_event=None, *, interface=None, pointing_type, values, scan_time, reference_time=None, station_beam_number=0, time_step=1.0)

Submit the track_object slow task.

This method returns immediately after it is submitted for execution.

Parameters:

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

  • task_abort_event (Optional[Event]) – Check for abort, defaults to None

  • interface (Optional[str]) – the schema version this is running against.

  • pointing_type (str) – the type of pointing requested

  • values (dict) – Coordinates for object to be tracked

  • scan_time (float) – Time to scan object in seconds

  • reference_time (Optional[str]) – time in which coordinates are equal, in ISO8601 formatted astropy.Time time

  • station_beam_number (Optional[int]) – The station beam number to be used

  • time_step (Optional[float]) – How long between each time step in seconds

Return type:

tuple[TaskStatus, str]

Returns:

A return code and a unique command ID.

update_communication_state(communication_state)

Update the status of communication with the component.

Overridden here to fire the “is configured” callback whenever communication is freshly established

Parameters:

communication_state (CommunicationStatus) – the status of communication with the component

Return type:

None

class StationHealthModel(field_station_trl, sps_station_trl, antenna_trls, component_state_callback, thresholds=None)

A health model for a station.

__init__(field_station_trl, sps_station_trl, antenna_trls, component_state_callback, thresholds=None)

Initialise a new instance.

Parameters:

  • field_station_trl (str) – the TRL of this station’s FieldStation

  • sps_station_trl (str) – the TRL of this MccsStation’s SpsStation.

  • antenna_trls (Sequence[str]) – the TRLs of this station’s antennas

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

  • thresholds (Optional[dict[str, float]]) – the threshold parameters for the health rules

antenna_health_changed(antenna_trl, antenna_health)

Handle a change in antenna health.

Parameters:

  • antenna_trl (str) – the TRL of the antenna whose health has changed

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

Return type:

None

evaluate_health()

Compute overall health of the station.

The overall health is based on the fault and communication status of the station overall, together with the health of the FieldStation, antennas and SpsStation that it manages.

This implementation simply sets the health of the station to the health of its least healthy component.

Return type:

HealthState

Returns:

an overall health of the station

field_station_health_changed(field_station_trl=None, field_station_health=None)

Handle a change in FieldStation health.

Parameters:

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

  • field_station_trl (Optional[str]) – the TRL of the FieldStation.

Return type:

None

sps_station_health_changed(sps_station_trl, sps_station_health)

Handle a change in SpsStation health.

Parameters:

  • sps_station_trl (str) – the TRL of the SpsStation

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

Return type:

None

class StationObsStateModel(logger, obs_state_changed_callback)

An observation state model for a station.

__init__(logger, obs_state_changed_callback)

Initialise a new instance.

Parameters:

  • logger (Logger) – a logger for this model to use

  • obs_state_changed_callback (Callable[[ObsState], None]) – callback to be called when there is a change to this model’s evaluated observation state.

is_configured_changed(is_configured)

Handle a change in whether the station is configured.

Parameters:

is_configured (bool) – whether the station is configured

Return type:

None

update_obs_state()

Update the observation state, ensuring that the callback is called.

Return type:

None