Station Beam Device

@startuml class StationBeamHealthEvaluator StationBeamHealthEvaluator : +evaluate_health() class StationBeamDriver StationBeamDriver : +__init__() StationBeamDriver : +connection_status() StationBeamDriver : +is_locked() class StationBeamHardwareFactory StationBeamHardwareFactory : +__init__() StationBeamHardwareFactory : +hardware() class StationBeamHardwareManager StationBeamHardwareManager : +__init__() StationBeamHardwareManager : +is_locked() class MccsStationBeam MccsStationBeam : +__init__() MccsStationBeam : +do() MccsStationBeam : -_initialise_connections() MccsStationBeam : -_initialise_hardware_management() MccsStationBeam : -_initialise_health_monitoring() MccsStationBeam : +interrupt() MccsStationBeam : +init_command_objects() MccsStationBeam : +always_executed_hook() MccsStationBeam : +delete_device() MccsStationBeam : +health_changed() MccsStationBeam : +beamId() MccsStationBeam : +stationIds() MccsStationBeam : +stationTrl() MccsStationBeam : +logicalBeamId() MccsStationBeam : +updateRate() MccsStationBeam : +isBeamLocked() MccsStationBeam : +channels() MccsStationBeam : +desiredPointing() MccsStationBeam : +pointingDelay() MccsStationBeam : +pointingDelayRate() MccsStationBeam : +antennaWeights() MccsStationBeam : +Configure() StationBeamDriver : -_is_locked StationBeamHardwareFactory : -_hardware MccsStationBeam : -_thread MccsStationBeam : -_lock MccsStationBeam : -_interrupt MccsStationBeam : -_health_state MccsStationBeam : -_station_ids MccsStationBeam : -_station_trl MccsStationBeam : -_logical_beam_id MccsStationBeam : -_desired_pointing MccsStationBeam : -_antenna_weights HardwareHealthEvaluator <|-- StationBeamHealthEvaluator HardwareDriver <|-- StationBeamDriver HardwareFactory <|-- StationBeamHardwareFactory HardwareManager <|-- StationBeamHardwareManager SKAObsDevice <|-- MccsStationBeam StationBeamHardwareFactory -- StationBeamDriver StationBeamHardwareManager -- StationBeamHardwareFactory StationBeamHardwareManager -- StationBeamHealthEvaluator MccsStationBeam -- StationBeamHardwareManager @enduml

This module implements the MCCS station beam device.

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

An implementation of a station beam Tango device for MCCS.

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

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

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

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

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

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

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

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

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

A class for MccsStationBeam’s Init command.

The do() method below is

called upon MccsStationBeam’s initialisation.

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

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

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

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

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

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

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

apertureId() → str[source]

Return the Aperture Id.

Returns:

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

beamId() → int[source]

Return the station beam id.

Returns:

the station beam id

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

Return the ids of the channels configured for this beam.

Returns:

channel ids

create_component_manager() → StationBeamComponentManager[source]

Create and return a component manager for this device.

Returns:

a component manager for this device.

desiredPointing() → list[float][source]

Return the desired pointing of this beam.

Returns:

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

hardwareBeamId() → int[source]

Return the timestamp used for the pointing coordinates.

Returns:

the timestamp used for the pointing coordinates.

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() → bool[source]

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

Returns:

whether the beam is locked or not

logicalBeamId() → int[source]

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

Return the phase centre.

Returns:

the phase centre

pointingDelay() → list[float][source]

Return the pointing delay of this beam.

Returns:

the pointing delay of this beam

pointingDelayRate() → list[float][source]

Return the pointing delay rate of this beam.

Returns:

the pointing delay rate of this beam

pointingReferenceFrame() → str[source]

Return the Reference Frame ID for the pointing coordinates.

Returns:

the Reference Frame ID for the pointing coordinates

pointingTimestamp() → str[source]

Return the timestamp used for the pointing coordinates.

Returns:

the timestamp used for the pointing coordinates.

scanId() → int[source]

Return the scan ID for the current scan.

Returns:

the scan ID or 0 if not scanning

stationId() → int[source]

Return the station id.

Returns:

the station id

stationTrl() → str[source]

Return the station TRL.

Returns:

the station TRL

subarrayId() → int[source]

Return the subarray id.

Returns:

the subarray id

updateRate() → float[source]

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

Returns:

the update rate for this station beam

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

Entry point for module.

Parameters:

  • args – positional arguments

  • kwargs – named arguments

Returns:

exit code