Station Component Manager

This module implements component management for stations.

exception DatabaseSolutionStructureError[source]

Raise when solution has unexpected structure.

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