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, logger: Logger, communication_state_callback: Callable[[ska_control_model.CommunicationStatus], None], component_state_callback: Callable[[...], 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

• 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

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

Submit the AcquireDataForCalibration slow task.

This method returns immediately after it is submitted for execution.

Parameters:

• channel – The channel to acquire data for

• task_callback – Update task state, defaults to None

• task_abort_event – Check for abort, defaults to None

Returns:

Task status and response message

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

Apply the beamformer and calibration configuration to SpsStation.

Parameters:

• transaction_id – the transaction id for the configuration

• 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

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: list[int]) → ska_control_model.ResultCode

Configure channels for a station beam in the channel table.

Parameters:

channel_blocks – List of 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

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, 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_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 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: 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

setup_pointing_helper() → None

Set up the pointing helper.

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

Stop a tracking thread.

Parameters:

• track_id – the id of the 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: 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 = 0, time_step: float | None = 1.0) → tuple[ska_control_model.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.

update_communication_state(communication_state: ska_control_model.CommunicationStatus) → None

Update the status of communication with the component.

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

Parameters:

communication_state – the status of communication with the component