Subarray Beam Device

This module implements the MCCS subarray beam device.

class MccsSubarrayBeam(*args: Any, **kwargs: Any)[source]

An implementation of a subarray beam Tango device for MCCS.

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

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)[source]

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[source]

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][source]

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]]][source]

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)[source]

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[source]

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][source]

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]]][source]

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]]][source]

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]]][source]

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]]][source]

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)[source]

A class for MccsSubarrayBeam’s Init command.

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

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

Initialise the attributes and properties of the MccsSubarrayBeam.

State is managed under the hood; the basic sequence is:

Device state is set to INIT
The do() method is run
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]]][source]

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]]][source]

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]]][source]

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]]][source]

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[source]

Initialise this device object.

Parameters:

args – positional args to the init
kwargs – keyword args to the init

antennaWeights() → list[float][source]

Return the antenna weights configured for this beam.

Returns:: antenna weightd

channels() → list[list[int]][source]

Return the ids of the channels configured for this beam.

Returns:: channel ids

create_component_manager() → SubarrayBeamComponentManager[source]

Create and return a component manager for this device.

Returns:: a component manager for this device.

desiredPointing(values: list[float]) → None[source]

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[source]

Return the first logical channel allocated to the beam.

Returns:: the first logical channel allocated to the beam.

healthModelParams(argin: str) → None[source]

Set the params for health transition rules.

Parameters:: argin – JSON-string of dictionary of health states

healthReport() → str[source]

Get the health report.

Returns:: the health report.

init_command_objects() → None[source]: Initialise the command handlers for commands supported by this device.

init_device() → None[source]: Initialise the device.

isBeamLocked(value: bool) → None[source]

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[source]

Set the logical beam id.

Parameters:: logical_beam_id – the logical beam id

missedEvents() → int[source]

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[source]

Return the first logical channel allocated to the beam.

Returns:: the first logical channel allocated to the beam.

phaseCentre() → list[float][source]

Return the phase centre.

Returns:: the phase centre

stationBeamIds(station_beam_ids: list[str]) → None[source]

Set the station beam ids.

Parameters:: station_beam_ids – ids of the station beams for this subarray beam

stationIds(station_ids: list[int]) → None[source]

Set the station ids.

Parameters:: station_ids – ids of the stations for this beam

subarrayBeamId() → int[source]

Return the subarray beam id.

Returns:: the subarray beam id

subarrayId() → int[source]

Return the subarray id.

Returns:: the subarray id

updateRate() → float[source]

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

Returns:: the update rate for this subarray beam

main(*args: str, **kwargs: str) → int[source]

Entry point for module.

Parameters:

args – positional arguments
kwargs – named arguments

Returns:

exit code