Station Component Manager

This module implements component management for stations.

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