Subarray Beam Component Manager
This module implements component management for subarray beams.
- class SubarrayBeamComponentManager(*args: Any, **kwargs: Any)[source]
A component manager for a subarray beam.
- __init__(logger: Logger, obs_command_timeout: int, communication_state_callback: Callable[[ska_control_model.CommunicationStatus], None], component_state_callback: Callable[[...], None], event_serialiser: ska_low_mccs_common.EventSerialiser | None = None, _station_beams: list[str] | None = None) None[source]
Initialise a new instance.
- Parameters:
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
obs_command_timeout – the default timeout for obs commands in seconds.
component_state_callback – callback to be called when the component state changes
event_serialiser – the event serialiser to be used by this object.
_station_beams – a optional injected station beams for testing purposes only. defaults to None
- abort(task_callback: Callable | None = None) tuple[ska_control_model.TaskStatus, str][source]
Abort the observation.
- Parameters:
task_callback – callback to be called when the status of the command changes
- Returns:
A task status and response message.
- abort_device(task_callback: Callable | None = None, task_abort_event: Event | None = None) Any[source]
Abort only this device, for use in RestartSubarray().
- Parameters:
task_callback – callback to be called when the status of the command changes
task_abort_event – Check for abort, defaults to None
- Returns:
A task status and response message.
- property channels: list[list[int]]
Return the ids of the channels configured for this subarray beam.
- Returns:
the ids of the channels configured for this subarray beam.
- cleanup() None[source]
Cleanup resources held by the component manager.
This includes cleaning up resources held by all sub-component managers.
- configure(update_rate: float, logical_bands: list[dict], apertures: list[dict], subarray_beam_id: int, subarray_id: int, sky_coordinates: dict[str, Any] | None = None, field: dict[str, Any] | None = None, task_callback: Callable | None = None, task_abort_event: Event | None = None) None[source]
Implement :Configure command.
- Parameters:
update_rate – Update rate for pointing in seconds
logical_bands – list of spectral bands to observe
apertures – list of aperture definition
subarray_id – ID of the subarray to which this
subarray_beam_id – ID of (this) subarray beam
field – Direction to point
sky_coordinates – Direction to point
task_callback – Update task state, defaults to None
task_abort_event – Task abort, defaults to None
- property desired_pointing: list[float] | str
Return the desired pointing.
- Returns:
the desired pointing
- do_assign_resources(subarray_id: int, subarray_beam_id: int, first_subarray_channel: int, number_of_channels: int, apertures: dict, task_callback: Callable | None = None, task_abort_event: Event | None = None) None[source]
Assign resources to device.
- Parameters:
subarray_id – ID of the subarray to which the beam belongs
subarray_beam_id – Id of (this) beam
first_subarray_channel – First logical channel assigned to subarray
number_of_channels – Number of channels assigned to beam
apertures – list of dictionaries iwith each entry containing * station_id: (int) in range 1-512 * aperture_id: (str) with format APx.y; x must match station_ID * station_beam_trl: (str) * channel_blocks: Allocated channel blocks for this station * hardware_beam: Allocated hardware beam for this station
task_callback – Update task state, defaults to None
task_abort_event – Task abort, defaults to None
- end(task_callback: Callable | None = None, task_abort_event: Event | None = None) None[source]
Implement End command.
- Parameters:
task_callback – Update task state, defaults to None
task_abort_event – Task abort, defaults to None
- end_scan(task_callback: Callable | None = None, task_abort_event: Event | None = None) None[source]
Execute the EndScan slow task.
- Parameters:
task_callback – Update task state, defaults to None
task_abort_event – Check for abort, defaults to None
- property first_channel: int
Return the first assigned subarray channel.
- Returns:
the first assigned subarray channel.
- get_coords(sky_coordinates: dict[str, Any] | None, field: dict[str, Any] | None) dict[str, Any][source]
Get the coordinates from either the sky_coordinates or field.
- Parameters:
sky_coordinates – Direction to point
field – Direction to point
- Returns:
The coordinates required
- property is_beam_locked: bool
Return whether the beam is locked.
- Returns:
whether the beam is locked
- property number_of_channels: int
Return the number of allocated subarray channels.
- Returns:
the number of channels
- obsreset(task_callback: Callable | None = None, task_abort_event: Event | None = None) None[source]
Implement ObsReset command.
- Parameters:
task_callback – Update task state, defaults to None
task_abort_event – Task abort, defaults to None
- release_all_resources(task_callback: Callable | None = None, task_abort_event: Event | None = None) None[source]
Implement ReleaseAllResources command.
- Parameters:
task_callback – Update task state, defaults to None
task_abort_event – Task abort, defaults to None
- restart(task_callback: Callable | None = None, task_abort_event: Event | None = None) None[source]
Implement Restart command.
- Parameters:
task_callback – Update task state, defaults to None
task_abort_event – Task abort, defaults to None
- scan(scan_id: int, start_time: str | None = None, duration: float = 0.0, task_callback: Callable | None = None, task_abort_event: Event | None = None) None[source]
Execute the Scan slow task.
- Parameters:
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