Subarray beam subpackage

This subpackage implements subarray beam functionality for MCCS.

class MccsSubarrayBeam(*args, **kwargs)

An implementation of a subarray beam Tango device for MCCS.

Abort()

Abort any long-running command such as Configure() or Scan().

Return type:

tuple[list[ResultCode], list[Optional[str]]]

Returns:

A tuple containing a return code and a string message indicating status. The message is for information purpose only.

class AbortCommand(command_tracker, component_manager, callback, logger=None)

A class for MccsSubarrayBeam’s Abort() command.

__init__(command_tracker, component_manager, callback, logger=None)

Initialise a new AbortCommand instance.

Parameters:

  • command_tracker (CommandTracker) – the device’s command tracker

  • component_manager (SubarrayBeamComponentManager) – the device’s component manager

  • callback (Callable[[bool], None]) – callback to be called when this command states and finishes

  • logger (Optional[Logger]) – a logger for this command object to yuse

do(*args, **kwargs)

Stateless hook for Abort() command functionality.

Parameters:

  • args (Any) – positional arguments to the command. This command does not take any, so this should be empty.

  • kwargs (Any) – keyword arguments to the command. This command does not take any, so this should be empty.

Return type:

tuple[ResultCode, str]

Returns:

A tuple containing a return code and a string message indicating status. The message is for information purpose only.

AbortDevice()

Abort any long-running command such as Configure() or Scan().

This will only cancel commands on this device, not further down the hierarchy, use Abort() for that use case.

Return type:

tuple[list[ResultCode], list[Optional[str]]]

Returns:

A tuple containing a return code and a string message indicating status. The message is for information purpose only.

class AbortDeviceCommand(command_tracker, component_manager, callback, logger=None)

A class for MccsSubarray’s AbortDevice() command.

__init__(command_tracker, component_manager, callback, logger=None)

Initialise a new AbortCommand instance.

Parameters:

  • command_tracker (CommandTracker) – the device’s command tracker

  • component_manager (SubarrayBeamComponentManager) – the device’s component manager

  • callback (Callable[[bool], None]) – callback to be called when this command states and finishes

  • logger (Optional[Logger]) – a logger for this command object to yuse

do(*args, **kwargs)

Stateless hook for AbortDevice() command functionality.

Parameters:

  • args (Any) – positional arguments to the command. This command does not take any, so this should be empty.

  • kwargs (Any) – keyword arguments to the command. This command does not take any, so this should be empty.

Return type:

tuple[ResultCode, str]

Returns:

A tuple containing a return code and a string message indicating status. The message is for information purpose only.

AssignResources(argin)

Assign resources to the subarray_beam with all relevant parameters.

Parameters:

argin (str) – Assign Resources parameters encoded in a json string

Return type:

tuple[list[ResultCode], list[Optional[str]]]

Returns:

A tuple containing a return code and a string message indicating status. The message is for information purpose only.

Configure(argin)

Configure the subarray_beam with all relevant parameters.

Parameters:

argin (str) – Configuration parameters encoded in a json string

Return type:

tuple[list[ResultCode], list[Optional[str]]]

Returns:

A tuple containing a return code and a string message indicating status. The message is for information purpose only.

End()

End the scan block (deconfigure).

Return type:

tuple[list[ResultCode], list[Optional[str]]]

Returns:

A tuple containing a return code and a string message indicating status. The message is for information purpose only.

EndScan()

End the scan.

Return type:

tuple[list[ResultCode], list[Optional[str]]]

Returns:

A tuple containing a return code and a string message indicating status. The message is for information purpose only.

class InitCommand(*args, **kwargs)

A class for MccsSubarrayBeam’s Init command.

The do() method below is called upon MccsSubarrayBeam’s initialisation.

do()

Initialise the attributes and properties of the MccsSubarrayBeam.

State is managed under the hood; the basic sequence is:

  1. Device state is set to INIT

  2. The do() method is run

  3. Device state is set to the OFF

Return type:

tuple[ResultCode, str]

Returns:

A tuple containing a return code and a string message indicating status. The message is for information purpose only.

ObsReset()

Reset the current observation process.

To modify behaviour for this command, modify the do() method of the command class.

Return type:

tuple[list[ResultCode], list[Optional[str]]]

Returns:

A tuple containing a result code and the unique ID of the command

ReleaseAllResources()

Release all resources.

Return type:

tuple[list[ResultCode], list[Optional[str]]]

Returns:

A tuple containing a return code and a string message indicating status. The message is for information purpose only.

Restart()

Restart the subarray. That is, deconfigure and release all resources.

To modify behaviour for this command, modify the do() method of the command class.

Return type:

tuple[list[ResultCode], list[Optional[str]]]

Returns:

A tuple containing a result code and the unique ID of the command

Scan(argin)

Start a scan on the subarray_beam.

Parameters:

argin (str) – Scan parameters encoded in a json string

Return type:

tuple[list[ResultCode], list[Optional[str]]]

Returns:

A tuple containing a return code and a string message indicating status. The message is for information purpose only.

__init__(*args, **kwargs)

Initialise this device object.

Parameters:

  • args (Any) – positional args to the init

  • kwargs (Any) – keyword args to the init

antennaWeights()

Return the antenna weights configured for this beam.

Return type:

list[float]

Returns:

antenna weightd

channels()

Return the ids of the channels configured for this beam.

Return type:

list[list[int]]

Returns:

channel ids

create_component_manager()

Create and return a component manager for this device.

Return type:

SubarrayBeamComponentManager

Returns:

a component manager for this device.

desiredPointing(values)

Set the desired pointing of this beam.

  • activation time (s) – value range 0-10^37

  • azimuth position (deg) – value range 0-360

  • azimuth speed (deg/s) – value range 0-10^37

  • elevation position (deg) – value range 0-90

  • elevation rate (deg/s) – value range 0-10^37

Parameters:

values (list[float]) – the desired pointing of this beam, expressed as a sky coordinate set

Return type:

None

firstSubarrayChannel()

Return the first logical channel allocated to the beam.

Return type:

int

Returns:

the first logical channel allocated to the beam.

healthModelParams(argin)

Set the params for health transition rules.

Parameters:

argin (str) – JSON-string of dictionary of health states

Return type:

None

init_command_objects()

Initialise the command handlers for commands supported by this device.

Return type:

None

init_device()

Initialise the device.

Return type:

None

isBeamLocked(value)

Set a flag indicating whether the beam is locked or not.

Parameters:

value (bool) – whether the beam is locked or not

Return type:

None

logicalBeamId(logical_beam_id)

Set the logical beam id.

Parameters:

logical_beam_id (int) – the logical beam id

Return type:

None

numberOfChannels()

Return the first logical channel allocated to the beam.

Return type:

int

Returns:

the first logical channel allocated to the beam.

phaseCentre()

Return the phase centre.

Return type:

list[float]

Returns:

the phase centre

stationBeamIds(station_beam_ids)

Set the station beam ids.

Parameters:

station_beam_ids (list[str]) – ids of the station beams for this subarray beam

Return type:

None

stationIds(station_ids)

Set the station ids.

Parameters:

station_ids (list[int]) – ids of the stations for this beam

Return type:

None

subarrayBeamId()

Return the subarray beam id.

Return type:

int

Returns:

the subarray beam id

subarrayId()

Return the subarray id.

Return type:

int

Returns:

the subarray id

updateRate()

Return the update rate (in hertz) for this subarray beam.

Return type:

float

Returns:

the update rate for this subarray beam

class SubarrayBeamComponentManager(logger, communication_state_callback, component_state_callback, _station_beams=None)

A component manager for a subarray beam.

__init__(logger, communication_state_callback, component_state_callback, _station_beams=None)

Initialise a new instance.

Parameters:

  • logger (Logger) – the logger to be used by this object.

  • 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

  • _station_beams (Optional[dict[str, Any]]) – a optional injected station beams for testing purposes only. defaults to None

abort(task_callback=None)

Abort the observation.

Parameters:

task_callback (Optional[Callable]) – callback to be called when the status of the command changes

Return type:

tuple[TaskStatus, str]

Returns:

A task status and response message.

abort_device(task_callback=None)

Abort only this device, for use in RestartSubarray().

Parameters:

task_callback (Optional[Callable]) – callback to be called when the status of the command changes

Return type:

tuple[TaskStatus, str]

Returns:

A task status and response message.

property antenna_weights: list[float]

Return the antenna weights.

Returns:

the antenna weights

assign_resources(task_callback=None, *, interface=None, subarray_id, subarray_beam_id=0, apertures, first_subarray_channel=0, number_of_channels)

Submit the assign_resources slow task.

This method returns immediately after it is submitted for execution.

Parameters:

  • task_callback (Optional[Callable]) – Update task state, defaults to None

  • interface (Optional[str]) – the schema version this is running against.

  • subarray_id (int) – ID of the subarray which controls this beam

  • subarray_beam_id (Optional[int]) – Id of (this) beam

  • apertures (Iterable) – Allocated apertures

  • first_subarray_channel (Optional[int]) – First channel (default 0)

  • number_of_channels (int) – Number of allocated channels

Return type:

tuple[TaskStatus, str]

Returns:

A tuple containing a task status and a unique id string to identify the command

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.

configure(task_callback=None, *, interface=None, subarray_id, subarray_beam_id, update_rate=0.0, logical_bands, apertures, sky_coordinates)

Submit the configure slow task.

This method returns immediately after it is submitted for execution.

Parameters:

  • task_callback (Optional[Callable]) – Update task state, defaults to None

  • interface (Optional[str]) – the schema version this is running against.

  • subarray_id (int) – ID of the subarray to which this

  • subarray_beam_id (int) – ID of (this) subarray beam

  • update_rate (Optional[float]) – Update rate for pointing in seconds

  • logical_bands (list[dict]) – list of spectral bands to observe

  • apertures (list[dict]) – list of aperture definition

  • sky_coordinates (dict[str, Any]) – Direction to point

Return type:

tuple[TaskStatus, str]

Returns:

A tuple containing a task status and a unique id string to identify the command

property desired_pointing: list[float]

Return the desired pointing.

Returns:

the desired pointing

end(task_callback=None)

Submit the End slow task.

This method returns immediately after it is submitted for execution.

Parameters:

task_callback (Optional[Callable]) – Update task state, defaults to None

Return type:

tuple[TaskStatus, str]

Returns:

A tuple containing a task status and a unique id string to identify the command

end_scan(task_callback=None, task_abort_event=None)

Submit the EndScan 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

Return type:

tuple[TaskStatus, str]

Returns:

Task status and response message

property first_channel: int

Return the first assigned subarray channel.

Returns:

the first assigned subarray channel.

property is_beam_locked: bool

Return whether the beam is locked.

Returns:

whether the beam is locked

property logical_beam_id: int

Return the logical beam id.

Returns:

the logical beam id

property number_of_channels: int

Return the number of allocated subarray channels.

Returns:

the number of channels

obsreset(task_callback=None)

Submit the ObsReset command.

This method returns immediately after it is submitted for execution.

Parameters:

task_callback (Optional[Callable]) – Update task state, defaults to None

Return type:

tuple[TaskStatus, str]

Returns:

A tuple containing a task status and a unique id string to identify the command

property phase_centre: list[float]

Return the phase centre.

Returns:

the phase centre

release_all_resources(task_callback=None)

Submit the ReleaseAllResources slow task.

This method returns immediately after it is submitted for execution.

Parameters:

task_callback (Optional[Callable]) – Update task state, defaults to None

Return type:

tuple[TaskStatus, str]

Returns:

A tuple containing a task status and a unique id string to identify the command

release_resources(task_callback=None)

Submit the ReleaseResources slow task.

This method returns immediately after it is submitted for execution.

Parameters:

task_callback (Optional[Callable]) – Update task state, defaults to None

Return type:

tuple[TaskStatus, str]

Returns:

A tuple containing a task status and a unique id string to identify the command

restart(task_callback=None)

Submit the Restart command.

This method returns immediately after it is submitted for execution.

Parameters:

task_callback (Optional[Callable]) – Update task state, defaults to None

Return type:

tuple[TaskStatus, str]

Returns:

A tuple containing a task status and a unique id string to identify the command

scan(task_callback=None, *, scan_id, start_time=None, duration=0.0)

Submit the Scan slow task.

This method returns immediately after it is submitted for execution.

Parameters:

  • 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

start_communicating()

Establish communication with the subarray beam.

Return type:

None

property station_beam_ids: list[str]

Return the station beam ids.

Returns:

list of beam IDs

property station_ids: list[int]

Return the station ids.

Returns:

the station ids

stop_communicating()

Break off communication with the subarray beam.

Return type:

None

property subarray_beam_id: int

Return the subarray beam id.

Returns:

the subarray beam id

property subarray_id: int

Return the subarray id.

Returns:

the subarray id

property update_rate: float

Return the update rate.

Returns:

the update rate

class SubarrayBeamHealthModel(station_beam_trls, component_state_callback, thresholds=None)

A health model for subarray beams.

__init__(station_beam_trls, component_state_callback, thresholds=None)

Initialise a new instance.

Parameters:

  • station_beam_trls (Sequence[str]) – the TRLs of this subarray beam’s station beams

  • component_state_callback (Callable[..., None]) – callback to be called whenever there is a change to this component’s state, including the health model’s evaluated health state.

  • thresholds (Optional[dict[str, float]]) – the threshold parameters for the health rules

evaluate_health()

Compute overall health of the subarray beam.

The overall health is based on the whether the beam is locked or not.

Return type:

HealthState

Returns:

an overall health of the subarray beam

is_beam_locked_changed(is_beam_locked)

Handle a change in whether the subarray beam is locked.

This is a callback hook called when the subarray beam lock changes.

Parameters:

is_beam_locked (bool) – whether the subarray beam is locked

Return type:

None

remove_station_beam(station_beam_trl)

Remove a station beam from the station beams used to compute health.

Parameters:

station_beam_trl (str) – the TRL of the station beam to remove

Return type:

None

station_beam_health_changed(station_beam_trl, station_beam_health)

Handle a change in station beam health.

If the station beam TRL is not in the internal dictionary of station beam healths, it is added with the provided health

Parameters:

  • station_beam_trl (str) – the TRL of the station beam whose health has changed or whose health is to be added

  • station_beam_health (Optional[HealthState]) – the health state of the specified station beam, or None if the station beam’s admin mode indicates that its health should not be rolled up.

Return type:

None

update_station_beams(new_station_beams)

Add and remove station beams so that they are up to date.

Parameters:

new_station_beams (set[str]) – the new set of station beams

Return type:

None