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.

ApplyPointing() → ska_tango_base.type_hints.TaskFunctionType

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(subarray_id: int, subarray_beam_id: int, station_id: int, station_trl: str, channel_blocks: list[int], hardware_beam: int, aperture_id: str, first_subarray_channel: int, number_of_channels: int, **kwargs: Any) → ska_tango_base.type_hints.TaskFunctionType

Assign resources to the station beam with all relevant parameters.

Parameters:

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

• subarray_beam_id – ID of the subarray beam

• 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

• aperture_id – ID of the aperture in format “APx.y”

• kwargs – optional arguments including: - interface.

Returns:

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

Configure(logical_bands: list[dict[str, int]], **kwargs: Any) → ska_tango_base.type_hints.TaskFunctionType

Configure the station_beam with all relevant parameters.

Parameters:

• logical_bands – Description of observed sky frequency bands

• kwargs – optional arguments including: - update_rate: Update rate for pointing, default never - weighting_key_ref: Antenna weights, default uniform, - field: Pointing direction - calibration_id: Unique calibration id.

Returns:

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

End() → ska_tango_base.type_hints.TaskFunctionType

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() → ska_tango_base.type_hints.TaskFunctionType

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.

ObsReset() → ska_tango_base.type_hints.TaskFunctionType

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() → ska_tango_base.type_hints.TaskFunctionType

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() → ska_tango_base.type_hints.TaskFunctionType

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(scan_id: int, **kwargs: Any) → ska_tango_base.type_hints.TaskFunctionType

Start the scan associated with the station_beam.

Parameters:

• scan_id – The ID for this scan

• kwargs – Optional arguments including: - start_time: UTC time for begin of scan, None for immediate start - duration: Scan duration in seconds. 0.0 or omitted means forever

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[str | None]]

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

beamWeightStoreTrl() → str

Return the beam weight store TRL.

Returns:

the beam weight store TRL

calibrationId() → str

Return the calibration ID set during the last Configure.

Returns:

the calibration ID, or an empty string if none was resolved.

channels() → list[list[int]]

Return the ids of the channels configured for this beam.

Returns:

channel ids

completed_Abort() → None

Abort command completed callback.

completed_AssignResources() → None

Assign Resources command completed callback.

completed_Configure() → None

Configure command completed callback.

completed_ObsReset() → None

Obs Reset command completed callback.

completed_ReleaseAllResources() → None

Release All Resources command completed callback.

completed_Restart() → None

Restart command completed callback.

create_component_manager() → StationBeamComponentManager

Create and return a component manager for this device.

Returns:

a component manager for this device.

desiredPointing() → list[float] | str

Return the desired pointing of this beam.

Returns:

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

failedPointingUpdates() → int | None

Return how many failed pointing updates we have had in a row.

Returns:

how many failed pointing updates we have had in a row.

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

Initialise the device.

isBeamLocked() → bool | None

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

schedule_abort_task(task_callback: ska_tango_base.type_hints.TaskCallbackType) → tuple[ska_control_model.TaskStatus, str]

Schedule an Abort task to begin executing immediately.

Subclasses should override this to change the behaviour of the Abort() command.

Parameters:

task_callback – Notified of progress of the abort command.

Returns:

A tuple containing TaskStatus.IN_PROGRESS and a message

started_Abort() → None

Abort command started callback.

started_AssignResources() → None

Assign Resources command started callback.

started_Configure() → None

Configure command started callback.

started_ObsReset() → None

Obs Reset command started callback.

started_ReleaseAllResources() → None

Release All Resources command started callback.

started_Restart() → None

Restart command started callback.

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, station_trl: str, beam_weight_store_trl: str, logger: Logger, communication_state_callback: Callable[[ska_control_model.CommunicationStatus], None], component_state_callback: Callable[[...], None], event_serialiser: ska_low_mccs_common.EventSerialiser | None = None) → None

Initialise a new instance.

Parameters:

• beam_id – the beam id of this station beam

• station_trl – the TRL of the station to which this beam belongs.

• beam_weight_store_trl – the TRL of the beam weight store.

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

• event_serialiser – the event serialiser to be used by this object.

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[complex]

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.ResultCode, str]

Apply the configured pointing to this station beam’s station.

Parameters:

• task_callback – Update task state, defaults to None

• task_abort_event – Check for abort, defaults to None

Returns:

Result and message. This is used in configure command to report errors.

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.

cleanup() → None

Cleanup resources held by the component manager.

This includes cleaning up resources held by all sub-component managers.

configure(logical_bands: list[dict[str, int]], interface: str | None = None, update_rate: float = 10.0, weighting_key_ref: str | None = 'uniform', field: dict[str, Any] | None = None, calibration_id: str | None = None, sky_coordinates: dict[str, Any] | None = None, task_callback: Callable | None = None, task_abort_event: Event | None = None) → None

Configure this station beam for scanning.

Parameters:

• 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

• field – Pointing direction

• calibration_id – Unique calibration id, if not given, the latest preferred calibration job will be used.

• task_callback – Update task state, defaults to None

• task_abort_event – Check for abort, defaults to None

property desired_pointing: list[float] | str

Return the desired pointing.

Returns:

the desired pointing

do_assign(subarray_id: int, subarray_beam_id: int, aperture_id: str, station_id: int, station_trl: str, channel_blocks: list[int], hardware_beam: int, first_subarray_channel: int, number_of_channels: int, interface: str | None = None, task_callback: Callable | None = None, task_abort_event: Event | None = None) → None

Assign resources to device.

Parameters:

• 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 in format “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

• task_callback – Update task state, defaults to None

• task_abort_event – Task abort, defaults to None

• interface – interface to use. Currently not implemented.

end(task_callback: Callable | None = None, task_abort_event: Event | None = None) → None

Deconfigure the station_beam.

Parameters:

• task_callback – Update task state, defaults to None

• task_abort_event – Check for abort, defaults to None

end_scan(task_callback: Callable | None = None, task_abort_event: Event | None = None) → None

Execute the EndScan slow task.

Parameters:

• task_callback – Update task state, defaults to None

• task_abort_event – Check for abort, defaults to None

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

Reset (to IDLE) the station_beam.

Parameters:

• task_callback – Update task state, defaults to None

• task_abort_event – Check for abort, defaults to None

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, task_abort_event: Event | None = None) → None

Release all allocated resources from the station_beam.

Parameters:

• task_callback – Update task state, defaults to None

• task_abort_event – Check for abort, defaults to None

restart(task_callback: Callable | None = None, task_abort_event: Event | None = None) → None

Restart (to EMPTY) the station_beam.

Parameters:

• task_callback – Update task state, defaults to None

• task_abort_event – Check for abort, defaults to None

scan(scan_id: int, start_time: str | None = None, duration: float = 864000.0, task_callback: Callable | None = None, task_abort_event: Event | None = None) → None

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

start_communicating() → None

Establish communication with the component.

property station_id: int

Return the station id.

Returns:

the station ids

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, beam_locked_callback: Callable[[bool | None], None], 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.

• beam_locked_callback – callback to call when the beam locked status changes. We say the beam is unlocked if we are over the FAILED limit of failed pointing updates in a row in the thresholds.

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

failed_pointing_updates_changed(failed_pointing_updates: int | None) → None

Handle a change in how many pointing updates have failed in a row.

Parameters:

failed_pointing_updates – how many pointing updates have failed in a row.

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.beam_id
    • StationBeamComponentManager.channels
    • StationBeamComponentManager.cleanup()
    • StationBeamComponentManager.configure()
    • StationBeamComponentManager.desired_pointing
    • StationBeamComponentManager.do_assign()
    • StationBeamComponentManager.end()
    • StationBeamComponentManager.end_scan()
    • 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.stop_communicating()
    • StationBeamComponentManager.subarray_id
    • StationBeamComponentManager.update_rate
• Station beam device
  • MccsStationBeam
    • MccsStationBeam.ApplyPointing()
    • MccsStationBeam.AssignResources()
    • MccsStationBeam.Configure()
    • MccsStationBeam.End()
    • MccsStationBeam.EndScan()
    • MccsStationBeam.ObsReset()
    • MccsStationBeam.ReleaseAllResources()
    • MccsStationBeam.Restart()
    • MccsStationBeam.Scan()
    • MccsStationBeam.ToFault()
    • MccsStationBeam.__init__()
    • MccsStationBeam.antennaWeights()
    • MccsStationBeam.apertureId()
    • MccsStationBeam.beamId()
    • MccsStationBeam.beamWeightStoreTrl()
    • MccsStationBeam.calibrationId()
    • MccsStationBeam.channels()
    • MccsStationBeam.completed_Abort()
    • MccsStationBeam.completed_AssignResources()
    • MccsStationBeam.completed_Configure()
    • MccsStationBeam.completed_ObsReset()
    • MccsStationBeam.completed_ReleaseAllResources()
    • MccsStationBeam.completed_Restart()
    • MccsStationBeam.create_component_manager()
    • MccsStationBeam.desiredPointing()
    • MccsStationBeam.failedPointingUpdates()
    • MccsStationBeam.hardwareBeamId()
    • MccsStationBeam.healthModelParams()
    • MccsStationBeam.healthReport()
    • MccsStationBeam.init_device()
    • MccsStationBeam.isBeamLocked()
    • MccsStationBeam.logicalBeamId()
    • MccsStationBeam.phaseCentre()
    • MccsStationBeam.pointingDelay()
    • MccsStationBeam.pointingDelayRate()
    • MccsStationBeam.pointingReferenceFrame()
    • MccsStationBeam.pointingTimestamp()
    • MccsStationBeam.scanId()
    • MccsStationBeam.schedule_abort_task()
    • MccsStationBeam.started_Abort()
    • MccsStationBeam.started_AssignResources()
    • MccsStationBeam.started_Configure()
    • MccsStationBeam.started_ObsReset()
    • MccsStationBeam.started_ReleaseAllResources()
    • MccsStationBeam.started_Restart()
    • MccsStationBeam.stationId()
    • MccsStationBeam.stationTrl()
    • MccsStationBeam.subarrayId()
    • MccsStationBeam.updateRate()
  • main()
• Station beam health model
  • StationBeamHealthModel
    • StationBeamHealthModel.__init__()
    • StationBeamHealthModel.antenna_health_changed()
    • StationBeamHealthModel.evaluate_health()
    • StationBeamHealthModel.failed_pointing_updates_changed()
    • StationBeamHealthModel.resources_changed()
    • StationBeamHealthModel.station_fault_changed()
    • StationBeamHealthModel.station_health_changed()
• Station beam health rules
  • StationBeamHealthRules
    • StationBeamHealthRules.__init__()
    • StationBeamHealthRules.default_thresholds
    • StationBeamHealthRules.degraded_rule()
    • StationBeamHealthRules.failed_rule()
    • StationBeamHealthRules.healthy_rule()
    • StationBeamHealthRules.unknown_rule()