Station beam subpackage

This subpackage implements station beam functionality for MCCS.

class MccsStationBeam(*args: Any, **kwargs: Any)

An implementation of a station beam Tango device for MCCS.

Abort() → tuple[list[ska_control_model.ResultCode], list[Optional[str]]]

Abort the current scan associated with the station_beam.

Returns:

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

class AbortCommand(*args: Any, **kwargs: Any)

A class for MccsSubarrayBeam’s Abort() command.

__init__(command_tracker: ska_tango_base.base.CommandTracker, component_manager: StationBeamComponentManager, callback: Callable[[bool], None], logger: Logger | None = None) → None

Initialise a new AbortCommand instance.

Parameters:

• command_tracker – the device’s command tracker

• component_manager – the device’s component manager

• callback – callback to be called when this command states and finishes

• logger – a logger for this command object to yuse

do(*args: Any, **kwargs: Any) → tuple[ska_control_model.ResultCode, str]

Stateless hook for Abort() command functionality.

Parameters:

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

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

Returns:

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

ApplyPointing() → tuple[list[ska_control_model.ResultCode], list[Optional[str]]]

Apply pointing delays to antennas associated with the station_beam.

Returns:

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

AssignResources(argin: str) → tuple[list[ska_control_model.ResultCode], list[Optional[str]]]

Assign resources to the station beam with all relevant parameters.

Parameters:

argin – Assign Resources parameters encoded in a json string

Returns:

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

Configure(argin: str) → tuple[list[ska_control_model.ResultCode], list[Optional[str]]]

Configure the station_beam with all relevant parameters.

Parameters:

argin – Configuration parameters encoded in a json string

Returns:

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

End() → tuple[list[ska_control_model.ResultCode], list[Optional[str]]]

Deconfigure the station_beam.

Returns:

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

EndScan() → tuple[list[ska_control_model.ResultCode], list[Optional[str]]]

Stop the current scan associated with the station_beam.

Returns:

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

class InitCommand(*args: Any, **kwargs: Any)

A class for MccsStationBeam’s Init command.

The do() method below is

called upon MccsStationBeam’s initialisation.

do() → tuple[ska_control_model.ResultCode, str]

Initialise the attributes and properties of the MccsStationBeam.

Returns:

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

ObsReset() → tuple[list[ska_control_model.ResultCode], list[Optional[str]]]

Reset to IDLE the station_beam.

Returns:

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

ReleaseAllResources() → tuple[list[ska_control_model.ResultCode], list[Optional[str]]]

Release all allocated resources from the station_beam.

Returns:

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

Restart() → tuple[list[ska_control_model.ResultCode], list[Optional[str]]]

Restart to EMPTY the station_beam.

Returns:

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

Scan(argin: str) → tuple[list[ska_control_model.ResultCode], list[Optional[str]]]

Start the scan associated with the station_beam.

Parameters:

argin – Configuration parameters encoded in a json string

Returns:

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

ToFault() → tuple[list[ska_control_model.ResultCode], list[Optional[str]]]

Put this station beam in ObsState.FAULT.

This is for use in testing only.

Returns:

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

__init__(*args: Any, **kwargs: Any) → None

Initialise this device object.

Parameters:

• args – positional args to the init

• kwargs – keyword args to the init

antennaWeights() → list[float]

Return the antenna weights configured for this beam.

Returns:

antenna weightd

apertureId() → str

Return the Aperture Id.

Returns:

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

beamId() → int

Return the station beam id.

Returns:

the station beam id

channels() → list[list[int]]

Return the ids of the channels configured for this beam.

Returns:

channel ids

create_component_manager() → StationBeamComponentManager

Create and return a component manager for this device.

Returns:

a component manager for this device.

desiredPointing() → list[float]

Return the desired pointing of this beam.

Returns:

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

hardwareBeamId() → int

Return the timestamp used for the pointing coordinates.

Returns:

the timestamp used for the pointing coordinates.

healthModelParams(argin: str) → None

Set the params for health transition rules.

Parameters:

argin – JSON-string of dictionary of health states

healthReport() → str

Get the health report.

Returns:

the health report.

init_command_objects() → None

Initialise the command handlers for commands supported by this device.

init_device() → None

Initialise the device.

isBeamLocked() → bool

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

Returns:

whether the beam is locked or not

logicalBeamId() → int

Return the logical beam id.

Todo:

this documentation needs to differentiate logical beam id from beam id

Returns:

the logical beam id

phaseCentre() → list[float]

Return the phase centre.

Returns:

the phase centre

pointingDelay() → list[float]

Return the pointing delay of this beam.

Returns:

the pointing delay of this beam

pointingDelayRate() → list[float]

Return the pointing delay rate of this beam.

Returns:

the pointing delay rate of this beam

pointingReferenceFrame() → str

Return the Reference Frame ID for the pointing coordinates.

Returns:

the Reference Frame ID for the pointing coordinates

pointingTimestamp() → str

Return the timestamp used for the pointing coordinates.

Returns:

the timestamp used for the pointing coordinates.

scanId() → int

Return the scan ID for the current scan.

Returns:

the scan ID or 0 if not scanning

stationId() → int

Return the station id.

Returns:

the station id

stationTrl() → str

Return the station TRL.

Returns:

the station TRL

subarrayId() → int

Return the subarray id.

Returns:

the subarray id

updateRate() → float

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

Returns:

the update rate for this station beam

class StationBeamComponentManager(*args: Any, **kwargs: Any)

A component manager for a station beam.

__init__(beam_id: int, logger: Logger, communication_state_callback: Callable[[ska_control_model.CommunicationStatus], None], component_state_callback: Callable[[...], None]) → None

Initialise a new instance.

Parameters:

• beam_id – the beam id of this station beam

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

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

Abort the observation.

Parameters:

task_callback – callback to be called when the status of the command changes

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: Callable | None = None, task_abort_event: Event | None = None) → tuple[ska_control_model.TaskStatus, str]

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

Returns:

Task status and response message

assign_resources(task_callback: Callable | None = None, *, interface: str | None = None, subarray_id: int, subarray_beam_id: int, aperture_id: str, station_id: int, station_trl: int, channel_blocks: list[int], hardware_beam: int, first_subarray_channel: int, number_of_channels: int) → tuple[ska_control_model.TaskStatus, str]

Submit the assign_resources slow task.

This method returns immediately after it is submitted for execution.

Parameters:

• task_callback – Update task state, defaults to None

• interface – the schema version this is running against.

• subarray_id – ID of the subarray to which the beam belongs

• subarray_beam_id – ID of the subarray beam

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

• station_id – ID of the associated station

• station_trl – TRL of the associated station

• channel_blocks – List of the allocated station channel blocks

• hardware_beam – Allocated station hardware beam

• first_subarray_channel – First channel

• number_of_channels – Number of channels

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: Callable | None = None, task_abort_event: Event | None = None, *, interface: str | None = None, update_rate: float | None = 1.0, logical_bands: list[dict], weighting_key_ref: str | None = 'uniform', sky_coordinates: dict[str, Any]) → tuple[ska_control_model.TaskStatus, str]

Submit the configure 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.

• update_rate – Update rate for pointing, default never

• logical_bands – Description of observed sky frequency bands

• weighting_key_ref – Antenna weights, default uniform,

• sky_coordinates – Pointing direction

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

Submit the end slow task.

This method returns immediately after it is submitted for execution.

Parameters:

task_callback – Update task state, defaults to None

Returns:

Task status and response message

end_scan(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:

• task_callback – Update task state, defaults to None

• task_abort_event – Check for abort, defaults to None

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

property max_executing_tasks: int

Get the max number of tasks that can be executing at once.

Returns:

max number of simultaneously executing tasks.

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

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

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

Submit the release__all_resources slow task.

This method returns immediately after it is submitted for execution.

Parameters:

task_callback – Update task state, defaults to None

Returns:

Task status and response message

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

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

Returns:

Task status and response message

scan(task_callback: Callable | None = None, *, scan_id: int, start_time: str | None = None, duration: float | None = 864000.0) → tuple[ska_control_model.TaskStatus, str]

Submit the Scan slow task.

This method returns immediately after it is submitted for execution.

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

Returns:

Task status and response message

start_communicating() → None

Establish communication with the component.

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

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

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(*args: Any, **kwargs: Any)

A health model for station beams.

__init__(health_changed_callback: Callable, ignore_power_state: bool = False, thresholds: dict[str, float] | None = None) → None

Initialise a new instance.

Parameters:

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

• ignore_power_state – whether the health model should ignore the power state when computing health.

• thresholds – Thresholds for failed degraded states.

antenna_health_changed(trl: str, health_state: ska_control_model.HealthState) → None

Handle change in station health.

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

evaluate_health() → tuple[ska_control_model.HealthState, str]

Compute overall health of the station beam.

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

Returns:

an overall health of the station beam

is_beam_locked_changed(is_beam_locked: bool) → None

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 – whether the station beam is locked

resources_changed(antenna_trls: set[str]) → None

Change the resources used for the station beam.

Parameters:

antenna_trls – TRLs of the antennas used for the beam

station_fault_changed(station_fault: bool) → None

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 – the fault state of the station that is controlled by this beam.

station_health_changed(station_health: ska_control_model.HealthState) → None

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 – the health of the station that is controlled by this beam

• Station beam component manager
  • StationBeamComponentManager
    • StationBeamComponentManager.__init__()
    • StationBeamComponentManager.abort()
    • StationBeamComponentManager.antenna_weights
    • StationBeamComponentManager.aperture_id
    • StationBeamComponentManager.apply_pointing()
    • StationBeamComponentManager.assign_resources()
    • StationBeamComponentManager.beam_id
    • StationBeamComponentManager.channels
    • StationBeamComponentManager.configure()
    • StationBeamComponentManager.desired_pointing
    • StationBeamComponentManager.end()
    • StationBeamComponentManager.end_scan()
    • StationBeamComponentManager.is_beam_locked
    • StationBeamComponentManager.logical_beam_id
    • StationBeamComponentManager.max_executing_tasks
    • StationBeamComponentManager.obsreset()
    • StationBeamComponentManager.phase_centre
    • StationBeamComponentManager.pointing_delay
    • StationBeamComponentManager.pointing_delay_rate
    • StationBeamComponentManager.power_state
    • StationBeamComponentManager.release_all_resources()
    • StationBeamComponentManager.restart()
    • StationBeamComponentManager.scan()
    • StationBeamComponentManager.start_communicating()
    • StationBeamComponentManager.station_id
    • StationBeamComponentManager.station_trl
    • StationBeamComponentManager.stop_communicating()
    • StationBeamComponentManager.subarray_id
    • StationBeamComponentManager.update_rate
• Station beam device
  • MccsStationBeam
    • MccsStationBeam.Abort()
    • MccsStationBeam.AbortCommand
      • MccsStationBeam.AbortCommand.__init__()
      • MccsStationBeam.AbortCommand.do()
    • MccsStationBeam.ApplyPointing()
    • MccsStationBeam.AssignResources()
    • MccsStationBeam.Configure()
    • MccsStationBeam.End()
    • MccsStationBeam.EndScan()
    • MccsStationBeam.InitCommand
      • MccsStationBeam.InitCommand.do()
    • MccsStationBeam.ObsReset()
    • MccsStationBeam.ReleaseAllResources()
    • MccsStationBeam.Restart()
    • MccsStationBeam.Scan()
    • MccsStationBeam.ToFault()
    • MccsStationBeam.__init__()
    • MccsStationBeam.antennaWeights()
    • MccsStationBeam.apertureId()
    • MccsStationBeam.beamId()
    • MccsStationBeam.channels()
    • MccsStationBeam.create_component_manager()
    • MccsStationBeam.desiredPointing()
    • MccsStationBeam.hardwareBeamId()
    • MccsStationBeam.healthModelParams()
    • MccsStationBeam.healthReport()
    • MccsStationBeam.init_command_objects()
    • MccsStationBeam.init_device()
    • MccsStationBeam.isBeamLocked()
    • MccsStationBeam.logicalBeamId()
    • MccsStationBeam.phaseCentre()
    • MccsStationBeam.pointingDelay()
    • MccsStationBeam.pointingDelayRate()
    • MccsStationBeam.pointingReferenceFrame()
    • MccsStationBeam.pointingTimestamp()
    • MccsStationBeam.scanId()
    • MccsStationBeam.stationId()
    • MccsStationBeam.stationTrl()
    • MccsStationBeam.subarrayId()
    • MccsStationBeam.updateRate()
  • main()
• Station beam health model
  • StationBeamHealthModel
    • StationBeamHealthModel.__init__()
    • StationBeamHealthModel.antenna_health_changed()
    • StationBeamHealthModel.evaluate_health()
    • StationBeamHealthModel.is_beam_locked_changed()
    • StationBeamHealthModel.resources_changed()
    • StationBeamHealthModel.station_fault_changed()
    • StationBeamHealthModel.station_health_changed()
• Station beam health rules
  • StationBeamHealthRules
    • StationBeamHealthRules.default_thresholds
    • StationBeamHealthRules.degraded_rule()
    • StationBeamHealthRules.failed_rule()
    • StationBeamHealthRules.healthy_rule()
    • StationBeamHealthRules.unknown_rule()