Station Component Manager

This module implements component management for stations.

class StationComponentManager(*args: Any, **kwargs: Any)

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

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(task_callback: Callable, *, first_channel: int, last_channel: int) → tuple[ska_control_model.TaskStatus, str]

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

Returns:

Task status and response message

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: Callable | None = None) → tuple[ska_control_model.TaskStatus, str]

Apply the loaded solution to the station.

This method returns immediately after it is submitted for execution.

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:

Task status and response message

apply_configuration(task_callback: Callable | None = None, task_abort_event: Event | None = None, *, transaction_id: str | None = None, calibration_id: str | None = None, subarray_id: str | None = None) → tuple[ska_control_model.TaskStatus, str]

Apply the beamformer and calibration configuration to SpsStation.

This method returns immediately after it is submitted for execution.

Parameters:

• task_callback – Update task state, defaults to None

• task_abort_event – Check for abort, defaults to None

• transaction_id – the transaction id for the configuration

• calibration_id – Unique calibration id.

• subarray_id – Subarray affected by command, default all.

• task_callback – Update task state, defaults to None

Returns:

a result code and response string

apply_pointing_delays(load_time: str, task_callback: Callable | None = None) → tuple[ska_control_model.TaskStatus, str]

Submit the apply_pointing method.

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

Parameters:

• load_time – time at which to load the pointing delay

• task_callback – Update task state, defaults to None

Returns:

a task status and response message

beamformer_running_for_subarray(subarray_id: int) → bool

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

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

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(task_callback: Callable | None = None, *, interface: str | None = None, station_config: dict, field_station_config: dict | None, antenna_config: dict | None) → tuple[ska_control_model.TaskStatus, str]

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 – the schema version this is running against.

• 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

Returns:

a result code and response string

configure_station_for_calibration(argin: str, task_callback: Callable | None = None) → tuple[ska_control_model.TaskStatus, str]

Submit the configure station for calibration method.

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

Parameters:

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

• task_callback – Update task state, defaults to None

Returns:

a task staus and response message

deallocate_subarray(subarray_id: int) → ska_control_model.ResultCode

Clear channels for a station beam in the channel table.

Parameters:

subarray_id – subarray_id to clear

Returns:

a result code and response string

end_scan(subarray_id: int, task_callback: Callable | None = None, task_abort_event: Event | None = None) → tuple[ska_control_model.TaskStatus, str]

Submit the EndScan slow task.

This method returns immediately after it is submitted for execution.

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

Returns:

Task status and response message

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

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) → tuple[ska_control_model.TaskStatus, str]

Submit the initialise method.

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

Parameters:

task_callback – Update task state, defaults to None

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_beam_weights(subarray_beam_id: int, antenna_weights: list[float]) → None

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, task_callback: Callable | None = None, task_abort_event: Callable | None = None) → tuple[ska_control_model.TaskStatus, str]

Load a calibration solution from the store to the stations TPM.

This method returns immediately after it is submitted for execution.

Parameters:

• calibration_id – Unique identifier for calibration.

• task_callback – Update task state, defaults to None

• task_abort_event – Abort the task

Returns:

Task status and response message

load_pointing_delays(delays: ndarray, task_callback: Callable | None = None) → tuple[ska_control_model.TaskStatus, str]

Load the pointing delays for this station.

Parameters:

• delays – list of delays

• task_callback – Update task state, defaults to None

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: Callable | None = None) → tuple[ska_control_model.TaskStatus, str]

Submit the _off method.

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

Parameters:

task_callback – Update task state, defaults to None

Returns:

a result code and response message

on(task_callback: Callable | None = None) → tuple[ska_control_model.TaskStatus, str]

Submit the _on method.

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

Parameters:

task_callback – Update task state, defaults to None

Returns:

a task staus and response message

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

Reset link for beam data packets to CSP to defaults.

Returns:

result code of ResetCspIngest

scan(task_callback: Callable | None = None, *, interface: str | None = None, subarray_id: int, scan_id: int, start_time: str | None = None, duration: float | None = 0.0) → tuple[ska_control_model.TaskStatus, str]

Submit the Scan slow task.

This method returns immediately after it is submitted for execution.

Parameters:

• interface – the schema version this is running against.

• subarray_id – The subarray for whic 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

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

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

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

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

Set up the pointing helper.

Returns:

A Pointing helper object.

start_acquisition(task_callback: Callable | None = None, *, start_time: str | None = None, delay: int | None = 2) → tuple[ska_control_model.TaskStatus, str]

Submit the start acquisition method.

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

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

Returns:

a task staus and response message

start_communicating() → None

Establish communication with the station components.

stop_communicating() → None

Break off communication with the station components.

stop_tracking(station_beam_id: int, task_callback: Callable | None = None) → tuple[ska_control_model.TaskStatus, str]

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

Returns:

a result code and list of pointing delays

stop_tracking_all(task_callback: Callable | None = None) → tuple[ska_control_model.TaskStatus, str]

Stop all tracking threads.

Parameters:

task_callback – Update task state, defaults to None

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: Callable | None = None, task_abort_event: threading.Event | None = None, *, interface: str | None = None, pointing_type: str, values: dict, scan_time: float, reference_time: str | None = None, station_beam_number: int | None = 1, time_step: float | None = 10.0) → tuple[ResultCode | TaskStatus, str]

Submit the track_object slow task.

This method returns immediately after it is submitted for execution.

Parameters:

• task_callback – Update task state, defaults to None

• task_abort_event – Check for abort, defaults to None

• interface – the schema version this is running against.

• 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:

A return code and a unique command ID.

trigger_adc_equalisation(task_callback: Callable, *, target_adc: float, bias: float | None = 0.0) → tuple[ska_control_model.TaskStatus, str]

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.

Returns:

Task status and response message