Subarray beam subpackage

This subpackage implements subarray beam functionality for MCCS.

class BeamMetricsBuilder(logger: Logger)

Builder for constructing valid beam QA metrics structures.

__init__(logger: Logger) → None

Initialize the builder.

Parameters:

logger – logger

build() → dict[str, Any]

Build the complete beam QA metrics dictionary.

Returns:

Valid QA metrics structure

check_all_locked() → bool

Check if all tracked apertures are locked.

Returns:

True if all apertures are locked, False otherwise

reset() → BeamMetricsBuilder

Reset the builder for reuse.

Returns:

Self for method chaining

set_beam_locked(is_locked: bool) → BeamMetricsBuilder

Set the overall beam locked status.

Parameters:

is_locked – Whether the beam is locked

Returns:

Self for method chaining

update_aperture_metrics(aperture_id: str, is_beam_locked: bool) → BeamMetricsBuilder

Update metrics for an aperture.

Parameters:

• aperture_id – Identifier for the aperture

• is_beam_locked – Whether beam is locked for this aperture

Returns:

Self for method chaining

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

An implementation of a subarray beam Tango device for MCCS.

AbortDevice() → ska_tango_base.type_hints.TaskFunctionType

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.

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, first_subarray_channel: int, number_of_channels: int, apertures: dict) → ska_tango_base.type_hints.TaskFunctionType

Assign resources to the subarray_beam with all relevant parameters.

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

Returns:

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

Configure(**kwargs: Any) → ska_tango_base.type_hints.TaskFunctionType

Configure the subarray_beam with all relevant parameters.

Parameters:

kwargs –

Required entries:

subarray_id:

integer in range 1-16.

subarray_beam_id:

integer in range 1-48.

logical_bands:

list of band objects, each with: :start_channel: even integer in range 2-504. :number_of_channels: integer in range 8-384, multiple of 8.

Optional entries:

interface:

expected schema version URI.

update_rate:

update rate in Hz (minimum 0.0).

apertures:

list of aperture objects, each supporting aperture_id (required): APXXX.YY format, station_beam_trl, weighting_key_ref, calibration_id

sky_coordinates:

pointing definition with required reference_frame and optional timestamp, target_name, c1, c1_rate, c2, c2_rate, line1, line2.

field:

pointing definition with required reference_frame and optional timestamp, target_name, and attrs where attrs may include c1, c1_rate, c2, c2_rate, line1, line2.

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

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

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

Abort device 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() → SubarrayBeamComponentManager

Create and return a component manager for this device.

Returns:

a component manager for this device.

desiredPointing(values: list[float]) → None

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 – the desired pointing of this beam, expressed as a sky coordinate set

firstSubarrayChannel() → int

Return the first logical channel allocated to the beam.

Returns:

the first logical channel allocated to the beam.

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

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

Parameters:

value – whether the beam is locked or not

logicalBeamId(logical_beam_id: int) → None

Set the logical beam id.

Parameters:

logical_beam_id – the logical beam id

missedEvents() → int

Get the amount of missed change events.

Some commands rely on change events from sub-devices, sometime we miss these events, this attribute keeps track of how many we know we have missed.

Returns:

the amount of missed change events

numberOfChannels() → int

Return the first logical channel allocated to the beam.

Returns:

the first logical channel allocated to the beam.

phaseCentre() → list[float]

Return the phase centre.

Returns:

the phase centre

qualityAssuranceMetrics() → str

Expose quality assurance metrics.

Structure: >>> { >>> “apertures”: { >>> <aperture_id>: { >>> “is_beam_locked”: <bool> >>> }, >>> … >>> }, >>> “is_beam_locked”: <bool> >>> } :return: A json serialised dictionary

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

Abort device 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.

stationBeamIds(station_beam_ids: list[str]) → None

Set the station beam ids.

Parameters:

station_beam_ids – ids of the station beams for this subarray beam

stationIds(station_ids: list[int]) → None

Set the station ids.

Parameters:

station_ids – ids of the stations for this beam

subarrayBeamId() → int

Return the local subarray beam id (scoped per subarray, 1-48).

Returns:

the local subarray beam id

subarrayId() → int

Return the subarray id.

Returns:

the subarray id

targetName() → str

Get the Name of the current target.

Returns:

the name of the object being pointed at.

updateRate() → float

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

Returns:

the update rate for this subarray beam

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

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

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]

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

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 antenna_weights: list[complex]

Return the antenna weights.

Returns:

the antenna weights

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

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

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

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

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

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]

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

Implement ObsReset command.

Parameters:

• task_callback – Update task state, defaults to None

• task_abort_event – Task abort, defaults to None

property phase_centre: list[float]

Return the phase centre.

Returns:

the phase centre

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

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

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

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 subarray beam.

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

Break off communication with the subarray beam.

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

A health model for subarray beams.

__init__(station_beam_trls: Sequence[str], component_state_callback: Callable[[...], None], ignore_power_state: bool = False, thresholds: dict[str, float] | None = None) → None

Initialise a new instance.

Parameters:

• station_beam_trls – the TRLs of this subarray beam’s station beams

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

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

• thresholds – the threshold parameters for the health rules

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

Compute overall health of the subarray beam.

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

Returns:

an overall health of the subarray beam

is_beam_locked_changed(is_beam_locked: bool) → None

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

remove_station_beam(station_beam_trl: str) → None

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

Parameters:

station_beam_trl – the TRL of the station beam to remove

station_beam_health_changed(station_beam_trl: str, station_beam_health: ska_control_model.HealthState | None) → None

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 – the TRL of the station beam whose health has changed or whose health is to be added

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

update_station_beams(new_station_beams: set[str]) → None

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

Parameters:

new_station_beams – the new set of station beams

• Subarray beam component manager
  • SubarrayBeamComponentManager
    • SubarrayBeamComponentManager.__init__()
    • SubarrayBeamComponentManager.abort()
    • SubarrayBeamComponentManager.abort_device()
    • SubarrayBeamComponentManager.antenna_weights
    • SubarrayBeamComponentManager.channels
    • SubarrayBeamComponentManager.cleanup()
    • SubarrayBeamComponentManager.configure()
    • SubarrayBeamComponentManager.desired_pointing
    • SubarrayBeamComponentManager.do_assign_resources()
    • SubarrayBeamComponentManager.end()
    • SubarrayBeamComponentManager.end_scan()
    • SubarrayBeamComponentManager.first_channel
    • SubarrayBeamComponentManager.get_coords()
    • SubarrayBeamComponentManager.is_beam_locked
    • SubarrayBeamComponentManager.logical_beam_id
    • SubarrayBeamComponentManager.number_of_channels
    • SubarrayBeamComponentManager.obsreset()
    • SubarrayBeamComponentManager.phase_centre
    • SubarrayBeamComponentManager.release_all_resources()
    • SubarrayBeamComponentManager.restart()
    • SubarrayBeamComponentManager.scan()
    • SubarrayBeamComponentManager.start_communicating()
    • SubarrayBeamComponentManager.station_beam_ids
    • SubarrayBeamComponentManager.station_ids
    • SubarrayBeamComponentManager.stop_communicating()
    • SubarrayBeamComponentManager.subarray_beam_id
    • SubarrayBeamComponentManager.subarray_id
    • SubarrayBeamComponentManager.update_rate
• Subarray beam device
  • MccsSubarrayBeam
    • MccsSubarrayBeam.AbortDevice()
    • MccsSubarrayBeam.AssignResources()
    • MccsSubarrayBeam.Configure()
    • MccsSubarrayBeam.End()
    • MccsSubarrayBeam.EndScan()
    • MccsSubarrayBeam.ObsReset()
    • MccsSubarrayBeam.ReleaseAllResources()
    • MccsSubarrayBeam.Restart()
    • MccsSubarrayBeam.Scan()
    • MccsSubarrayBeam.__init__()
    • MccsSubarrayBeam.antennaWeights()
    • MccsSubarrayBeam.channels()
    • MccsSubarrayBeam.completed_Abort()
    • MccsSubarrayBeam.completed_AbortDevice()
    • MccsSubarrayBeam.completed_AssignResources()
    • MccsSubarrayBeam.completed_Configure()
    • MccsSubarrayBeam.completed_ObsReset()
    • MccsSubarrayBeam.completed_ReleaseAllResources()
    • MccsSubarrayBeam.completed_Restart()
    • MccsSubarrayBeam.create_component_manager()
    • MccsSubarrayBeam.desiredPointing()
    • MccsSubarrayBeam.firstSubarrayChannel()
    • MccsSubarrayBeam.healthModelParams()
    • MccsSubarrayBeam.healthReport()
    • MccsSubarrayBeam.init_device()
    • MccsSubarrayBeam.isBeamLocked()
    • MccsSubarrayBeam.logicalBeamId()
    • MccsSubarrayBeam.missedEvents()
    • MccsSubarrayBeam.numberOfChannels()
    • MccsSubarrayBeam.phaseCentre()
    • MccsSubarrayBeam.qualityAssuranceMetrics()
    • MccsSubarrayBeam.schedule_abort_task()
    • MccsSubarrayBeam.started_Abort()
    • MccsSubarrayBeam.started_AbortDevice()
    • MccsSubarrayBeam.started_AssignResources()
    • MccsSubarrayBeam.started_Configure()
    • MccsSubarrayBeam.started_ObsReset()
    • MccsSubarrayBeam.started_ReleaseAllResources()
    • MccsSubarrayBeam.started_Restart()
    • MccsSubarrayBeam.stationBeamIds()
    • MccsSubarrayBeam.stationIds()
    • MccsSubarrayBeam.subarrayBeamId()
    • MccsSubarrayBeam.subarrayId()
    • MccsSubarrayBeam.targetName()
    • MccsSubarrayBeam.updateRate()
  • main()
• Subarray beam health model
  • SubarrayBeamHealthModel
    • SubarrayBeamHealthModel.__init__()
    • SubarrayBeamHealthModel.evaluate_health()
    • SubarrayBeamHealthModel.is_beam_locked_changed()
    • SubarrayBeamHealthModel.remove_station_beam()
    • SubarrayBeamHealthModel.station_beam_health_changed()
    • SubarrayBeamHealthModel.update_station_beams()
• Subarray beam health rules
  • SubarrayBeamHealthRules
    • SubarrayBeamHealthRules.default_thresholds
    • SubarrayBeamHealthRules.degraded_rule()
    • SubarrayBeamHealthRules.failed_rule()
    • SubarrayBeamHealthRules.healthy_rule()
    • SubarrayBeamHealthRules.unknown_rule()