Station beam subpackage

This subpackage implements station beam functionality for MCCS.

class MccsStationBeam(*args, **kwargs)

An implementation of a station beam Tango device for MCCS.

Abort()

Abort the current scan associated with the station_beam.

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 (StationBeamComponentManager) – 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.

ApplyPointing()

Apply pointing delays to antennas associated with the station_beam.

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.

AssignResources(argin)

Assign resources to the station 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 station_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()

Deconfigure the station_beam.

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()

Stop the current scan associated with the station_beam.

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 MccsStationBeam’s Init command.

The do() method below is

called upon MccsStationBeam’s initialisation.

do()

Initialise the attributes and properties of the MccsStationBeam.

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 to IDLE the station_beam.

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.

ReleaseAllResources()

Release all allocated resources from the station_beam.

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 to EMPTY the station_beam.

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.

Scan(argin)

Start the scan associated with the station_beam.

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.

ToFault()

Put this station beam in ObsState.FAULT.

This is for use in testing only.

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

apertureId()

Return the Aperture Id.

Return type:

str

Returns:

the station beam Aperture Id, in the form APx.y y=substation

beamId()

Return the station beam id.

Return type:

int

Returns:

the station beam id

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:

StationBeamComponentManager

Returns:

a component manager for this device.

desiredPointing()

Return the desired pointing of this beam.

Return type:

list[float]

Returns:

the desired point of this beam, conforming to the Sky Coordinate Set definition

hardwareBeamId()

Return the timestamp used for the pointing coordinates.

Return type:

int

Returns:

the timestamp used for the pointing coordinates.

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()

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

Return type:

bool

Returns:

whether the beam is locked or not

logicalBeamId()

Return the logical beam id.

Todo:

this documentation needs to differentiate logical beam id from beam id

Return type:

int

Returns:

the logical beam id

phaseCentre()

Return the phase centre.

Return type:

list[float]

Returns:

the phase centre

pointingDelay()

Return the pointing delay of this beam.

Return type:

list[float]

Returns:

the pointing delay of this beam

pointingDelayRate()

Return the pointing delay rate of this beam.

Return type:

list[float]

Returns:

the pointing delay rate of this beam

pointingReferenceFrame()

Return the Reference Frame ID for the pointing coordinates.

Return type:

str

Returns:

the Reference Frame ID for the pointing coordinates

pointingTimestamp()

Return the timestamp used for the pointing coordinates.

Return type:

str

Returns:

the timestamp used for the pointing coordinates.

scanId()

Return the scan ID for the current scan.

Return type:

int

Returns:

the scan ID or 0 if not scanning

stationId()

Return the station id.

Return type:

int

Returns:

the station id

stationTrl()

Return the station TRL.

Return type:

str

Returns:

the station TRL

subarrayId()

Return the subarray id.

Return type:

int

Returns:

the subarray id

updateRate()

Return the update rate (in seconds) for this station beam.

Return type:

float

Returns:

the update rate for this station beam

class StationBeamComponentManager(beam_id, logger, max_workers, communication_state_callback, component_state_callback)

A component manager for a station beam.

__init__(beam_id, logger, max_workers, communication_state_callback, component_state_callback)

Initialise a new instance.

Parameters:

  • beam_id (int) – the beam id of this station beam

  • 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]) – a callback to be called whenever the state of the station beam changes.

  • max_workers (int) – Maximum number of workers in the worker pool. 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.

property antenna_weights: list[float]

Return the antenna weights.

Returns:

the antenna weights

property aperture_id: str

Return the aperture id.

Returns:

the aperture id

apply_pointing(task_callback=None, task_abort_event=None)

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

assign_resources(task_callback=None, *, interface=None, subarray_id, subarray_beam_id, aperture_id, station_id, station_trl, channel_blocks, hardware_beam, first_subarray_channel, 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 to which the beam belongs

  • subarray_beam_id (int) – ID of the subarray beam

  • aperture_id (str) – ID of the aperture, of the form “APx.y”

  • station_id (int) – ID of the associated station

  • station_trl (int) – TRL of the associated station

  • channel_blocks (list[int]) – List of the allocated station channel blocks

  • hardware_beam (int) – Allocated station hardware beam

  • first_subarray_channel (int) – First channel

  • number_of_channels (int) – Number of channels

Return type:

tuple[TaskStatus, str]

Returns:

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

property beam_id: int

Return the station beam id.

Returns:

the station beam id

property channels: list[list[int]]

Return the ids of the channels configured for this station beam.

Returns:

the ids of the channels configured for this subarray beam.

configure(task_callback=None, task_abort_event=None, *, interface=None, update_rate=1.0, logical_bands, weighting_key_ref='uniform', 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

  • task_abort_event (Optional[Event]) – Check for abort, defaults to None

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

  • update_rate (Optional[float]) – Update rate for pointing, default never

  • logical_bands (list[dict]) – Description of observed sky frequency bands

  • weighting_key_ref (Optional[str]) – Antenna weights, default uniform,

  • sky_coordinates (dict[str, Any]) – Pointing direction

Return type:

tuple[TaskStatus, str]

Returns:

A return code and a unique command ID.

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:

Task status and response message

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 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

obsreset(task_callback=None, task_abort_event=None)

Submit the obs_reset 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 phase_centre: list[float]

Return the phase centre.

Returns:

the phase centre

property pointing_delay: list[float]

Return the pointing delay.

Returns:

the pointing delay

property pointing_delay_rate: list[float]

Return the pointing delay rate.

Returns:

the pointing delay rate

property power_state: ska_control_model.PowerState | None

Return my power state.

Returns:

my power state

release_all_resources(task_callback=None)

Submit the release__all_resources 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:

Task status and response message

restart(task_callback=None, task_abort_event=None)

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

scan(task_callback=None, *, scan_id, start_time=None, duration=864000.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 component.

Return type:

None

property station_id: int

Return the station id.

Returns:

the station ids

property station_trl: str

Return the station TRL.

If the station TRL is not set, return the empty string.

Returns:

the station TRL

stop_communicating()

Cease monitoring the component, and break off all communication with it.

Return type:

None

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 StationBeamHealthModel(health_changed_callback, thresholds=None)

A health model for station beams.

__init__(health_changed_callback, thresholds=None)

Initialise a new instance.

Parameters:

  • health_changed_callback (Callable) – a callback to be called when the health of the station beam (as evaluated by this model) changes

  • thresholds (Optional[dict[str, float]]) – Thresholds for failed degraded states.

antenna_health_changed(trl, health_state)

Handle change in station health.

This is a callback hook, called by the component manager when the health of a station changes. :type trl: str :param trl: the TRL of the antenna whose health has changed :type health_state: HealthState :param health_state: the new health state of the antenna.

Return type:

None

evaluate_health()

Compute overall health of the station beam.

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

Return type:

HealthState

Returns:

an overall health of the station beam

is_beam_locked_changed(is_beam_locked)

Handle a change in whether the station beam is locked.

This is a callback hook that is called when whether asubarray beam is locked changes.

Parameters:

is_beam_locked (bool) – whether the station beam is locked

Return type:

None

resources_changed(antenna_trls)

Change the resources used for the station beam.

Parameters:

antenna_trls (set[str]) – TRLs of the antennas used for the beam

Return type:

None

station_fault_changed(station_fault)

Handle a change in the fault state of the station beam’s station device.

This is a callback hook that is called when whether a station beam detects that the fault state of its station has changed. This could occur because the station’s health changes, or because the beam changes station.

Parameters:

station_fault (bool) – the fault state of the station that is controlled by this beam.

Return type:

None

station_health_changed(station_health)

Handle a change in the health of the station device that this beam controls.

This is a callback hook that is called when whether a station beam detects that the health of its station has changed. This could occur because the station’s health changes, or because the beam changes station.

Parameters:

station_health (HealthState) – the health of the station that is controlled by this beam

Return type:

None