Subarray beam subpackage

This subpackage implements subarray beam functionality for MCCS.

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

An implementation of a subarray beam Tango device for MCCS.

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

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

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: SubarrayBeamComponentManager, 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.

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

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.

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

A class for MccsSubarray’s AbortDevice() command.

__init__(command_tracker: ska_tango_base.base.CommandTracker, component_manager: SubarrayBeamComponentManager, 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 AbortDevice() 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.

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

Assign resources to the subarray_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 subarray_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]]]

End the scan block (deconfigure).

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

End the scan.

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

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

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

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

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 the current observation process.

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

Returns:

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

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

Release all resources.

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 the subarray. That is, deconfigure and release all resources.

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

Returns:

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

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

Start a scan on the subarray_beam.

Parameters:

argin – Scan 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.

__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

channels() → list[list[int]]

Return the ids of the channels configured for this beam.

Returns:

channel ids

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

Initialise the command handlers for commands supported by this device.

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

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

Returns:

the subarray beam id

subarrayId() → int

Return the subarray id.

Returns:

the subarray id

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], _station_beams: dict[str, Any] | 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

• _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) → tuple[ska_control_model.TaskStatus, str]

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

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

assign_resources(task_callback: Callable | None = None, *, interface: str | None = None, subarray_id: int, subarray_beam_id: int | None = 0, apertures: Iterable, first_subarray_channel: int | None = 0, 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 which controls this beam

• subarray_beam_id – Id of (this) beam

• apertures – Allocated apertures

• first_subarray_channel – First channel (default 0)

• number_of_channels – Number of allocated channels

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: Callable | None = None, *, interface: str | None = None, subarray_id: int, subarray_beam_id: int, update_rate: float | None = 0.0, logical_bands: list[dict], apertures: list[dict], 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

• interface – the schema version this is running against.

• subarray_id – ID of the subarray to which this

• subarray_beam_id – ID of (this) subarray beam

• update_rate – Update rate for pointing in seconds

• logical_bands – list of spectral bands to observe

• apertures – list of aperture definition

• sky_coordinates – Direction to point

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

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

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 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 max_executing_tasks: int

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

Returns:

max number of simultaneously executing tasks.

property number_of_channels: int

Return the number of allocated subarray channels.

Returns:

the number of channels

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

Submit the ObsReset command.

This method returns immediately after it is submitted for execution.

Parameters:

task_callback – Update task state, defaults to None

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

Submit the ReleaseAllResources slow task.

This method returns immediately after it is submitted for execution.

Parameters:

task_callback – Update task state, defaults to None

Returns:

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

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

Submit the ReleaseResources slow task.

This method returns immediately after it is submitted for execution.

Parameters:

task_callback – Update task state, defaults to None

Returns:

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

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

Submit the Restart command.

This method returns immediately after it is submitted for execution.

Parameters:

task_callback – Update task state, defaults to None

Returns:

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

scan(task_callback: Callable | None = None, *, scan_id: int, start_time: str | None = None, duration: float | None = 0.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 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.assign_resources()
    • SubarrayBeamComponentManager.channels
    • SubarrayBeamComponentManager.configure()
    • SubarrayBeamComponentManager.desired_pointing
    • SubarrayBeamComponentManager.end()
    • SubarrayBeamComponentManager.end_scan()
    • SubarrayBeamComponentManager.first_channel
    • SubarrayBeamComponentManager.is_beam_locked
    • SubarrayBeamComponentManager.logical_beam_id
    • SubarrayBeamComponentManager.max_executing_tasks
    • SubarrayBeamComponentManager.number_of_channels
    • SubarrayBeamComponentManager.obsreset()
    • SubarrayBeamComponentManager.phase_centre
    • SubarrayBeamComponentManager.release_all_resources()
    • SubarrayBeamComponentManager.release_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.Abort()
    • MccsSubarrayBeam.AbortCommand
      • MccsSubarrayBeam.AbortCommand.__init__()
      • MccsSubarrayBeam.AbortCommand.do()
    • MccsSubarrayBeam.AbortDevice()
    • MccsSubarrayBeam.AbortDeviceCommand
      • MccsSubarrayBeam.AbortDeviceCommand.__init__()
      • MccsSubarrayBeam.AbortDeviceCommand.do()
    • MccsSubarrayBeam.AssignResources()
    • MccsSubarrayBeam.Configure()
    • MccsSubarrayBeam.End()
    • MccsSubarrayBeam.EndScan()
    • MccsSubarrayBeam.InitCommand
      • MccsSubarrayBeam.InitCommand.do()
    • MccsSubarrayBeam.ObsReset()
    • MccsSubarrayBeam.ReleaseAllResources()
    • MccsSubarrayBeam.Restart()
    • MccsSubarrayBeam.Scan()
    • MccsSubarrayBeam.__init__()
    • MccsSubarrayBeam.antennaWeights()
    • MccsSubarrayBeam.channels()
    • MccsSubarrayBeam.create_component_manager()
    • MccsSubarrayBeam.desiredPointing()
    • MccsSubarrayBeam.firstSubarrayChannel()
    • MccsSubarrayBeam.healthModelParams()
    • MccsSubarrayBeam.healthReport()
    • MccsSubarrayBeam.init_command_objects()
    • MccsSubarrayBeam.init_device()
    • MccsSubarrayBeam.isBeamLocked()
    • MccsSubarrayBeam.logicalBeamId()
    • MccsSubarrayBeam.missedEvents()
    • MccsSubarrayBeam.numberOfChannels()
    • MccsSubarrayBeam.phaseCentre()
    • MccsSubarrayBeam.stationBeamIds()
    • MccsSubarrayBeam.stationIds()
    • MccsSubarrayBeam.subarrayBeamId()
    • MccsSubarrayBeam.subarrayId()
    • 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()