Station subpackage

This subpackage implements station functionality for MCCS.

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

An implementation of a station Tango device for MCCS.

AcquireDataForCalibration(first_channel: int, last_channel: int, daq_mode: str = 'TCC') → ska_tango_base.type_hints.DevVarLongStringArrayType[source]

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

Parameters:

  • first_channel – The first channel to acquire data for

  • last_channel – The last channel to acquire data for

  • daq_mode – Which correlator to start, default TCC.

Returns:

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

ApplyCalibration(load_time: str) → ska_tango_base.type_hints.TaskFunctionType[source]

Load the calibration coefficients at the specified time delay.

Parameters:

load_time – load time, in ISO formatted time. If not set, defaults to calibrationLoadDelay seconds in the future.

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("ApplyCalibration", "")
ApplyConfiguration(**kwargs: Any) → ska_tango_base.type_hints.DevVarLongStringArrayType[source]

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

Parameters:

kwargs – containing optional transaction_id, calibration_id, subarray_id.

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(load_time: str) → ska_tango_base.type_hints.TaskFunctionType[source]

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

Parameters:

load_time – 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)
BeamformerRunningForSubarray(subarray_id: int) → bool[source]

Get Beamformer Running status for a specific subarray.

Parameters:

subarray_id – subarray_id, in the range 1 to 16

Returns:

True if the beamformer is running

ConfigureChannels(argin: list[int]) → ska_tango_base.type_hints.DevVarLongStringArrayType[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.

Note

A call to this method will delete any previous channels allocated to this beam before reconfiguration.

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(antenna_config: dict[str, Any], station_config: dict[str, Any] | None = None, field_station_config: dict[str, Any] | None = None) → ska_tango_base.type_hints.TaskFunctionType[source]

Configure semi static information like position of antennas.

Parameters:

  • station_config – Configuration specification for the station device.

  • field_station_config – Configuration specification for the field station device.

  • antenna_config – Configuration specification for the antenna deviced.

Returns:

The task to be submitted as an LRC.

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)
ConfigureStationForCalibration(argin: str) → ska_tango_base.type_hints.TaskFunctionType[source]

Configure the station for calibration.

Parameters:

argin – JSON-ified argument containing DAQ configuration overrides.

Returns:

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

DeallocateSubarray(subarray_id: int) → ska_tango_base.type_hints.DevVarLongStringArrayType[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) → ska_tango_base.type_hints.DevVarLongStringArrayType[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(pointing_type: str, values: dict | str, **kwargs: Any) → ndarray[source]

Get pointing coefficients.

Parameters:

  • pointing_type – The type of pointing requested.

  • values – The pointing values, either in alt_az or ra_dec.

  • kwargs – Optional arguments, including:

  • interface: The schema version this is running against.

  • reference_time: Time in which coordinates are equal,

    in ISO8601 formatted astropy.Time.

  • time_step: Duration between each time step in seconds.

Returns:

The pointing delays as pairs of (delay, delay rate) in EEP order.

Initialise() → ska_tango_base.type_hints.TaskFunctionType[source]

Initialise this station’s tiles.

Returns:

The task to be submitted as an LRC.

LoadBeamWeights(subarray_beam_id: int, antenna_weights: list[float]) → None[source]

Load beam weights for a specific subarray beam.

Parameters:

  • subarray_beam_id – The ID of the subarray beam for which to load weights.

  • antenna_weights – A list of weights for each antenna in the subarray beam.

LoadCalibrationCoefficients(calibration_id: str | None) → ska_tango_base.type_hints.TaskFunctionType[source]

Load Calibration solutions to the station, but does not apply them.

NOTE for each 384 channels this will select and load a solution specified by the selection rules https://developer.skao.int/projects/ska-low-mccs /en/0.22.0/api/calibration_store/selection_policy/index.html

TODO This method will load calibration solution for each antenna in the station. This will loop round all antenna and load a inverse jones for each 384 channels. A unity calibration is loaded is no solution is found, this sounds dangerous.

Parameters:

calibration_id – Unique identifier for calibration.

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("LoadCalibrationCoefficients")
LoadPointingDelays(delays: ndarray) → ska_tango_base.type_hints.TaskFunctionType[source]

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

Parameters:

delays – 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)
ResetCspIngest() → ska_tango_base.type_hints.DevVarLongStringArrayType[source]

Reset link for beam data packets to CSP to defaults.

Returns:

result code name of ResetCspIngest

Scan(subarray_id: int, scan_id: int, start_time: str | None = None, duration: float = 0.0, **kwargs: Any) → ska_tango_base.type_hints.TaskFunctionType[source]

Start the scan associated with the station_beam.

Parameters:

  • subarray_id – The subarray for which the command applies

  • scan_id – The ID for this scan

  • start_time – UTC time for begin of scan, None for immediate start

  • duration – Scan duration in seconds. 0.0 or omitted means forever

  • kwargs – kwargs

Returns:

The task to be submitted as an LRC.

SetCspIngest(argin: str) → tuple[ska_control_model.ResultCode, str][source]

Configure link for beam data packets to CSP.

Parameters:

argin –

json dictionary with optional keywords:

  • destination_ip - (string) Destination IP

  • source_port - (int) Source port for integrated data streams

  • destination_port - (int) Destination port for integrated data streams

Returns:

result code name of SetCspIngest

StartAcquisition(**kwargs: Any) → ska_tango_base.type_hints.TaskFunctionType[source]

Start data acquisition.

Parameters:

kwargs – optional keywords:

  • start_time - (ISO UTC time) start time

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

Returns:

The task to be submitted as a LRC.

StopTracking(station_beam_id: int) → ska_tango_base.type_hints.TaskFunctionType[source]

Stop the current tracking thread.

Parameters:

station_beam_id – the beam id whose tracking you wish to stop.

Returns:

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

StopTrackingAll() → ska_tango_base.type_hints.TaskFunctionType[source]

Stop all tracking threads.

Returns:

The task to be submitted as a LRC.

TrackObject(pointing_type: str, values: dict | str, reference_time: str, **kwargs: Any) → ska_tango_base.type_hints.DevVarLongStringArrayType[source]

Track an object through the sky.

Parameters:

  • pointing_type – the type of pointing requested

  • values – Coordinates for object to be tracked

  • reference_time – time in which coordinates are equal, in ISO8601 formatted astropy.Time time

  • kwargs – optional kwargs inclusing scan_time, station_beam_number, time_step

Returns:

A tuple containing a return code and a string message. ResultCode.OK if tracking was successfully queued in the pointing manager, ResultCode.REJECTED otherwise.

TriggerAdcEqualisation(target_adc: float, **kwargs: Any) → ska_tango_base.type_hints.DevVarLongStringArrayType[source]

Get the equalised ADC values.

Getting the equalised values takes up to 20 seconds (to get an average to avoid spikes). So we trigger the collection and publish to dbmPowers

Parameters:

  • target_adc – the expected average power received by antennas in ADU units. Has an input minimum of 0, but in code its limited to 4.2e-7 (corresponds to the maxiumum output of 31.75 dB). There is no maximum value, however, 40 ADUs will result in 0 dB with no bias and 1600 ADUs will result in 0 dB with the maxiumum bias allowed of 32dB.

  • kwargs – optional parameters include bias. user specifed bias in dB added to the antenna preadu levels. Bias input value rounded as part of value sanitation and as a result it increases in steps of 0.25. Ranges from -32 to 32 with default 0.

Returns:

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

Example:
>>> dp = tango.DeviceProxy("mccs/station/001")
>>> json_arg = json.dumps({"target_adc" : "18", bias: "1"})
>>> dp.command_inout("TriggerAdcEqualisation", json_arg)
__init__(*args: Any, **kwargs: Any) → None[source]

Initialise this device object.

Parameters:

  • args – positional args to the init

  • kwargs – keyword args to the init

activeBankCalibrationId() → str[source]

Return the calibration id in the active bank.

Returns:

Return the calibration id in use

activeBeamInfo() → str[source]

Return the configuration information of currently active PointingRequests.

List information about each active beam. Info includes pointing type,

values, reference time, scan time, time step, next update time, and whether the scan is complete.

Returns:

A json encoded dictionary with beam IDs as keys and their information as values.

additionalDebug(argin: bool) → None[source]

Set whether additional debug information is enabled.

Parameters:

argin – True to enable additional debug information, False to disable.

antennaMasks(masks: list[bool]) → None[source]

Set the antennaMasks.

This must be executed in EngineeringMode. Value is persisted in the database.

Parameters:

masks – masks to apply

appliedWeightingBlock() → ndarray[source]

Return the last applied weighting block.

Magic number -1 represents that this has not been updated.

Returns:

The last know weightingBlock applied.

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

calibrationLoadDelay(value: float) → None[source]

Set the load time for calls to ApplyCalibration.

Parameters:

value – Number of seconds in the future for the calibration to be applied.

configuredWeightingBlock() → ndarray[source]

Return the configured weighting block.

This will return the Weighting Block configured. Note that this may or may not have been written to the hardware. Please view appliedWeightingBlock for the block applied.

Magic number -1 represents that this has not been updated.

Returns:

The current configured weightingBlock.

create_component_manager() → StationComponentManager[source]

Create and return a component manager for this device.

Returns:

a component manager for this device.

cspIngestConfig() → str[source]

Report the CspIngest configuration in use for this station.

Returns:

json string with CspIngest configuration.

daqJobId() → int[source]

Return the DAQ job id.

Returns:

the DAQ job id

daqPath() → str[source]

Report the DAQ path in use for this station.

Returns:

DAQ path.

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

engineering_to_write(req_type: tango.AttReqType) → bool[source]

Return a flag representing whether we are in Engineering mode.

Parameters:

req_type – the request type

Returns:

True if Tile is in Engineering Mode.

execute_AcquireDataForCalibration(first_channel: int, last_channel: int, daq_mode: str) → ska_tango_base.type_hints.TaskFunctionType[source]

Execute the AcquireDataForCalibration command as a long-running task.

Parameters:

  • first_channel – The first channel to acquire data for

  • last_channel – The last channel to acquire data for

  • daq_mode – Which correlator to start, default TCC.

Returns:

The task to be submitted as a LRC.

execute_ApplyConfiguration(**kwargs: Any) → ska_tango_base.type_hints.TaskFunctionType[source]

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

Parameters:

kwargs – containing optional transaction_id, calibration_id, subarray_id.

Returns:

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

execute_EndScan(subarray_id: int) → ska_tango_base.type_hints.TaskFunctionType[source]

Execute the EndScan command as a long-running task.

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.

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

Turn the station on.

Returns:

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

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

Turn the station on.

Returns:

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

execute_TriggerAdcEqualisation(target_adc: float, **kwargs: Any) → ska_tango_base.type_hints.TaskFunctionType[source]

Execute the TriggerAdcEqualisation command as a long-running task.

Parameters:

  • target_adc – target value for ADC gain.

  • kwargs – optional parameters include bias.

Returns:

The task to be submitted as a LRC.

failedPointingUpdates() → str[source]

Return the tracking beams which are currently unlocked.

Returns:

the tracking beams which are currently unlocked.

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:

“antennas”: (f2f, d2f, d2d),
tuple(int, int, int): Number of antennas failed before health failed,

Number of antennas degraded before health failed, Number of antennas degraded before health degraded

Parameters:

argin – JSON-string of dictionary of health thresholds

inactiveBankCalibrationId() → str[source]

Return the calibration id of the inactive bank.

Returns:

Return the inactive bank calibration id

init_device() → None[source]

Initialise the device.

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.

lastPointingDelays() → str[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.

latestPreferredJobId() → str[source]

Return the user_friendly_name of the most recent preferred calibration job.

Returns:

the user_friendly_name of the most recent preferred job, or an empty string if none exists.

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.

pointingUpdateTimingHistory(history_size: int) → None[source]

Set the number of pointing update timing records to keep in history.

Parameters:

history_size – Number of timing records to keep (must be > 0)

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

staticDelays() → list[float][source]

Report the static delays in use for this station.

Returns:

list of static delays.

stationID() → int[source]

Return the station ID.

Returns:

the numerical station ID.

trackingStatus() → str[source]

Return the tracking status of currently active PointingRequests.

Beam IDs not listed have no active PointingRequest.

Returns:

Tracking status of active PointingRequests.

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

useNewHealthModel(argin: bool) → None[source]

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

Parameters:

argin – a flag indicating whether this station is currently using the new health model.

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

A component manager for a station.

__init__(station_id: int, ref_latitude: float, ref_longitude: float, ref_height: float, field_station_trl: str, antenna_trls: Sequence[str], antenna_station_locations: ndarray, antenna_element_ids: list[int], station_calibrator_trl: str, sps_station_trl: str, calibration_load_delay: float, antenna_masks: list[bool], use_beam_weights: bool, logger: Logger, communication_state_callback: Callable[[ska_control_model.CommunicationStatus], None], component_state_callback: Callable[[...], None], event_serialiser: ska_low_mccs_common.EventSerialiser | None = None) → None[source]

Initialise a new instance.

Parameters:

  • station_id – the id of this station

  • ref_latitude – reference latitude of the station.

  • ref_longitude – reference longitude of the station.

  • ref_height – reference ellipsoidal height of the station.

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

  • antenna_trls – TRLs of the Tango devices and manage this station’s antennas

  • antenna_station_locations – array of the x, y, z positions of the antennas

  • antenna_element_ids – list of the element IDs of the antennas

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

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

  • calibration_load_delay – The amount of seconds in the future for the calibration solutions to be applied.

  • antenna_masks – the antenna masks to be masked from all beams in the beamformerTable.

  • use_beam_weights – True if we are prototyping the beam weighting feature.

  • logger – the logger to be used by this object.

  • communication_state_callback – callback to be called when the status of the communications channel between the component manager and its component changes

  • component_state_callback – callback to be called when the component state changes

  • event_serialiser – the event serialiser to be used by this object.

acquire_data_for_calibration(first_channel: int | None = 64, last_channel: int | None = 72, daq_mode: str = 'TCC', task_callback: Callable | None = None, task_abort_event: Event | None = None) → None[source]

Submit the AcquireDataForCalibration slow task.

This method returns immediately after it is submitted for execution.

Parameters:

  • first_channel – The first channel to acquire data for

  • last_channel – The last channel to acquire data for

  • task_callback – Update task state, defaults to None

  • daq_mode – Which correlator to start, default TCC.

  • task_abort_event – task abort event.

property additional_debug: bool

Return whether additional debug information is enabled.

Returns:

True if additional debug information is enabled, False otherwise.

property antenna_masks: ndarray

Return the antennas masked from all beams.

Returns:

the masked antenna.

apply_calibration(load_time: str = '', task_callback: Callable | None = None, task_abort_event: Event | None = None) → tuple[ska_control_model.ResultCode, str][source]

Apply a solution to the station.

Parameters:

  • task_callback – Update task state, defaults to None

  • load_time – An optional future time to swap the calibration banks.

  • task_abort_event – Abort the task

Returns:

ResultCode and response message

apply_configuration(transaction_id: str | None = None, calibration_id: str | None = None, subarray_id: int | None = None, solution_type: str = 'fitted', task_callback: Callable | None = None, task_abort_event: Event | None = None) → None[source]

Apply the configuration to the SpsStation.

Parameters:

  • transaction_id – the transaction id for the configuration

  • calibration_id – Unique calibration id.

  • subarray_id – ID of the subarray to which the configuration applies, default applies for all subarrays

  • solution_type – "fitted" (default) reconstructs gains from stored phase fit params; "raw" loads the stored solution directly.

  • task_callback – Update task state, defaults to None

  • task_abort_event – Abort the task

apply_pointing_delays(load_time: str, task_callback: Callable | None = None, task_abort_event: Event | None = None) → None[source]

Load the pointing delay at a specified time.

Parameters:

  • load_time – time at which to load the pointing delay

  • task_callback – Update the task state, defaults to None

  • task_abort_event – Abort the task

beamformer_running_for_subarray(subarray_id: int) → bool[source]

Check that the beamformer is running for a given subarray.

Parameters:

subarray_id – subarray ID to check

Returns:

True if the beamformer is running for the given subarray

property beamformer_table: list[list[int]]

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

Returns:

reformatted channel table

cleanup() → None[source]

Cleanup resources held by the component manager.

This includes cleaning up resources held by all sub-component managers.

configure_channels(channel_blocks: list[int]) → ska_control_model.ResultCode[source]

Configure channels for a station beam in the channel table.

Parameters:

channel_blocks – List of defined channel table entries

Returns:

a result code and response string

configure_semi_static(antenna_config: dict, station_config: dict | None, field_station_config: dict | None, task_callback: Callable | None = None, task_abort_event: Event | None = None) → None[source]

Configure the stations children.

This sends off configuration commands to all of the devices that this station manages.

Parameters:

  • station_config – Configuration specification for the station device.

  • field_station_config – Configuration specification for the field station device.

  • antenna_config – Configuration specification for the antenna deviced.

  • task_callback – Update task state, defaults to None

  • task_abort_event – Abort the task

Raises:

ValueError – Station value not correct

configure_station_for_calibration(argin: str, task_callback: Callable | None = None, task_abort_event: Event | None = None) → None[source]

Execute the configure station for calibration method.

Parameters:

  • argin – JSON-ified argument containing DAQ configuration overrides.

  • task_callback – Update task state, defaults to None

  • task_abort_event – Check for abort, defaults to None

deallocate_subarray(subarray_id: int) → ska_control_model.ResultCode[source]

Clear channels for a station beam in the channel table.

Parameters:

subarray_id – subarray_id to clear

Returns:

a result code and response string

do_off(task_callback: Callable | None = None, task_abort_event: Event | None = None) → None[source]

Turn off this station.

The order to turn a station on is: FieldStation, then tiles and antennas.

Parameters:

  • task_callback – Update task state, defaults to None

  • task_abort_event – Abort the task

do_on(task_callback: Callable | None = None, task_abort_event: Event | None = None) → None[source]

Turn on this station.

The order to turn a station on is: FieldStation, then tiles and antennas.

Parameters:

  • task_callback – Update task state, defaults to None

  • task_abort_event – Abort the task

end_scan(subarray_id: int, task_callback: Callable | None = None, task_abort_event: Event | None = None) → None[source]

Execute the EndScan slow task.

Parameters:

  • subarray_id – The subarray for which the command applies

  • task_callback – Update task state, defaults to None

  • task_abort_event – Check for abort, defaults to None

get_pointing_delays(task_callback: Callable | None = None, *, interface: str | None = None, pointing_type: str, values: dict | str, time_step: float = 10.0, reference_time: str | None = None) → ndarray[source]

Get the pointing delays for this station.

Parameters:

  • interface – the schema version this is running against.

  • pointing_type – the type of pointing requested

  • values – the pointing values, either in alt_az or ra_dec

  • reference_time – time in which coordinates are equal, in ISO8601 formatted astropy.Time time

  • time_step – How long between each time step in seconds

  • task_callback – callback to signal end of command

Returns:

list of pointing delays

initialise(task_callback: Callable | None = None, task_abort_event: Event | None = None) → None[source]

Initialise using slow command.

Parameters:

  • task_callback – Update task state, defaults to None

  • task_abort_event – Check for abort, defaults to None

property is_configured: bool

Return whether this station component manager is configured.

Returns:

whether this station component manager is configured.

property latest_preferred_job_id: str

Return the user_friendly_name of the most recent preferred calibration job.

Returns:

the user_friendly_name of the most recent preferred job, or an empty string if none exists.

load_beam_weights(subarray_beam_id: int, antenna_weights: list[complex]) → None[source]

Load beam weights to the weighting matrix.

:TODO Consider a station with fewer than 256 antenna. Or not Do we just mask the rest?

Parameters:

  • subarray_beam_id – The beam we are loading the weights for (1-48)

  • antenna_weights – A list of the 256 antenna weights to apply.

load_calibration_coefficients(calibration_id: str | None = None, subarray_id: int | None = None, solution_type: str = 'fitted', task_callback: Callable | None = None, task_abort_event: Event | None = None) → tuple[ska_control_model.ResultCode, str][source]

Load a calibration solution from the store.

NOTE: Is calibration key specified on a per channel basis?

Parameters:

  • calibration_id – Unique identifier for calibration.

  • subarray_id – ID of the subarray to which the calibration applies, default applies for all subarrays

  • solution_type – "fitted" (default) reconstructs gains from stored phase fit params; "raw" loads the stored solution directly.

  • task_callback – Update task state, defaults to None

  • task_abort_event – Abort the task

Returns:

a result code and string.

load_pointing_delays(delays: ndarray, task_callback: Callable | None = None, task_abort_event: Event | None = None) → None[source]

Load the pointing delays for this station.

Parameters:

  • delays – list of delays

  • task_callback – Update task state, defaults to None

  • task_abort_event – Abort the task

property number_of_channels: int

Return the total number of channels in the beamformer.

Returns:

the total numebr of channels

property pointing_update_timing_history: str

Return the pointing update timing history as JSON.

Returns:

JSON string containing timing history for recent pointing updates.

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.

reset_csp_ingest() → ska_tango_base.type_hints.DevVarLongStringArrayType[source]

Reset link for beam data packets to CSP to defaults.

Returns:

result code of ResetCspIngest

scan(subarray_id: int, scan_id: int, start_time: str | None, duration: float, task_callback: Callable | None = None, task_abort_event: Event | None = None) → None[source]

Execute the Scan slow task.

Parameters:

  • subarray_id – The subarray for which the command applies

  • scan_id – The ID for this scan

  • start_time – UTC time for begin of scan, None for immediate start

  • duration – Scan duration in seconds. 0.0 or omitted means forever

  • task_callback – Update task state, defaults to None

  • task_abort_event – Check for abort, defaults to None

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

set_csp_ingest(argin: str) → tuple[ska_control_model.ResultCode, str][source]

Configure link for beam data packets to CSP.

Parameters:

argin –

json dictionary with optional keywords:

  • destination_ip - (string) Destination IP

  • source_port - (int) Source port for integrated data streams

  • destination_port - (int) Destination port for integrated data streams

Returns:

result code of SetCspIngest

set_pointing_update_timing_history_size(history_size: int) → None[source]

Set the number of pointing update timing records to keep.

Parameters:

history_size – Number of timing records to keep (must be > 0)

setup_pointing_helper() → Pointing[source]

Set up the pointing helper.

Returns:

A Pointing helper object.

start_acquisition(start_time: str | None = None, delay: int | None = 2, task_callback: Callable | None = None, task_abort_event: Event | None = None) → None[source]

Start acquisition using slow command.

Parameters:

  • start_time – the time at which to start data acquisition, defaults to None

  • delay – delay start, defaults to 2

  • task_callback – Update task state, defaults to None

  • task_abort_event – Check for abort, defaults to None

start_communicating() → None[source]

Establish communication with the station components.

stop_communicating() → None[source]

Break off communication with the station components.

stop_tracking(station_beam_id: int, task_callback: Callable | None = None, task_abort_event: Event | None = None) → None[source]

Stop a tracking thread.

Parameters:

  • station_beam_id – the beam id whose tracking you wish to stop.

  • task_callback – Update task state, defaults to None

  • task_abort_event – Abort the task

stop_tracking_all(task_callback: Callable | None = None, task_abort_event: Event | None = None) → None[source]

Stop a tracking thread.

Parameters:

  • task_callback – Update task state, defaults to None

  • task_abort_event – Abort the task

property tileprogrammingstate: tuple[str]

Return the tileprogrammingstate of the SpsStation.

Returns:

the tileprogrammingstate of the SpsStation.

track_object(pointing_type: str, values: dict | str, reference_time: str, station_beam_number: int = 1, scan_time: float = 86400.0, time_step: float = 10.0) → ska_control_model.ResultCode[source]

Track the object in the sky.

Parameters:

  • pointing_type – the type of pointing requested

  • values – Coordinates for object to be tracked

  • scan_time – Time to scan object in seconds

  • reference_time – time in which coordinates are equal, in ISO8601 formatted astropy.Time time

  • station_beam_number – The station beam number to be used

  • time_step – How long between each time step in seconds

Returns:

ResultCode.OK on success, ResultCode.REJECTED on failure

trigger_adc_equalisation(target_adc: float, bias: float | None = 0.0, task_callback: Callable | None = None, task_abort_event: Event | None = None) → None[source]

Trigger Adc Equalisation on the sps station.

This method returns immediately after it is submitted for execution.

Parameters:

  • task_callback – Update task state, defaults to None

  • target_adc – adc value in ADU units. Defaults to 17.

  • bias – user specifed bias in dB added to the antenna preadu levels. Defaults to 0.

  • task_abort_event – the task abort event.

verify_antennas_mapping(task_callback: Callable | None = None) → tuple[ResultCode | TaskStatus, str][source]

Verify that the antenna mapping available is correct.

Parameters:

task_callback – Update task state, defaults to None

Returns:

True if the mapping is correct, False otherwise

verify_values(pointing_type: str, values: dict | str) → bool[source]

Verify if the pointing values are valid.

Parameters:

  • pointing_type – the type of pointing requested

  • values – Coordinates for object to be tracked

Returns:

If the pointing values are valid

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

A health model for a station.

__init__(field_station_trl: str, sps_station_trl: str, antenna_trls: Sequence[str], health_changed_callback: ska_low_mccs_common.health.HealthChangedCallbackProtocol, thresholds: dict[str, float] | None = None) → None[source]

Initialise a new instance.

Parameters:

  • field_station_trl – the TRL of this station’s FieldStation

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

  • antenna_trls – the TRLs of this station’s antennas

  • health_changed_callback – callback to be called whenever there is a change to this this health model’s evaluated health state.

  • thresholds – the threshold parameters for the health rules

antenna_health_changed(antenna_trl: str, antenna_health: ska_control_model.HealthState | None) → None[source]

Handle a change in antenna health.

Parameters:

  • antenna_trl – the TRL of the antenna whose health has changed

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

evaluate_health() → tuple[ska_control_model.HealthState, str][source]

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.

Returns:

an overall health of the station

field_station_health_changed(field_station_trl: str | None = None, field_station_health: ska_control_model.HealthState | None = None) → None[source]

Handle a change in FieldStation health.

Parameters:

  • field_station_health – 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 – the TRL of the FieldStation.

sps_station_health_changed(sps_station_trl: str, sps_station_health: ska_control_model.HealthState | None) → None[source]

Handle a change in SpsStation health.

Parameters:

  • sps_station_trl – the TRL of the SpsStation

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

class StationObsStateModel(logger: Logger, obs_state_changed_callback: Callable[[ska_control_model.ObsState], None])[source]

An observation state model for a station.

__init__(logger: Logger, obs_state_changed_callback: Callable[[ska_control_model.ObsState], None]) → None[source]

Initialise a new instance.

Parameters:

  • logger – a logger for this model to use

  • obs_state_changed_callback – callback to be called when there is a change to this model’s evaluated observation state.

is_configured_changed(is_configured: bool) → None[source]

Handle a change in whether the station is configured.

Parameters:

is_configured – whether the station is configured

update_obs_state() → None[source]

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