Antenna subpackage

This subpackage implements antenna functionality for MCCS.

class AntennaComponentManager(field_station_name, antenna_id, tile_trl, tile_channels, logger, max_workers, communication_state_callback, component_state_callback)

A component manager for managing the component of an MCCS antenna Tango device.

Since there is no way to monitor and control an antenna directly, this component manager simply proxies certain commands to the FieldStation and/or tile Tango device.

__init__(field_station_name, antenna_id, tile_trl, tile_channels, logger, max_workers, communication_state_callback, component_state_callback)

Initialise a new instance.

Parameters:

• field_station_name (str) – the TRL of the Tango device for this antenna’s FieldStation.

• antenna_id (int) – the logical id of this antenna (1-256)

• tile_trl (str) – the TRL of the Tango device for this antenna’s tile.

• tile_channels (tuple[int, int]) – the tile ADC channels containing Y and X signal from this antenna.

• logger (Logger) – a logger for this object to use

• max_workers (int) – the maximum worker threads for the slow commands associated with this component manager.

• communication_state_callback (Callable[[CommunicationStatus], None]) – callback to be called when the status of the communications channel between the component manager and its component changes

• component_state_callback (Callable[..., None]) – callback to be called when the component state changes

configure(task_callback=None, *, interface=None, antenna_config, tile_config)

Submit the configure method.

This method returns immediately after it submitted self._configure for execution.

Parameters:

• interface (Optional[str]) – the schema version this is running against.

• antenna_config (dict) – antenna config

• tile_config (dict) – tile config

• task_callback (Optional[Callable]) – Update task state, defaults to None

Return type:

tuple[TaskStatus, str]

Returns:

a result code and response string

property current: float

Return the antenna’s current.

Returns:

the current of this antenna

off(task_callback=None)

Submit the off slow task.

This method returns immediately after it submitted self._off for execution.

Parameters:

task_callback (Optional[Callable]) – Update task state, defaults to None

Return type:

tuple[TaskStatus, str]

Returns:

task status and message

on(task_callback=None)

Submit the on slow task.

This method returns immediately after it submitted self._on for execution.

Parameters:

task_callback (Optional[Callable]) – Update task state, defaults to None

Return type:

tuple[TaskStatus, str]

Returns:

task status and message

property power_state: ska_control_model.PowerState | None

Return my power state.

Returns:

my power state

property power_state_lock: RLock

Return the power state lock of this component manager.

Returns:

the power state lock of this component manager.

reset(task_callback=None)

Reset the antenna; this is not implemented.

This raises NotImplementedError because the antenna is passive hardware and cannot meaningfully be reset.

Parameters:

task_callback (Optional[Callable]) – Update task state, defaults to None

Raises:

NotImplementedError – because the antenna’s power state is not controlled via the Tile device; it is controlled via the FieldStation device.

Return type:

tuple[TaskStatus, str]

standby(task_callback=None)

Put the antenna into standby state; this is not implemented.

This raises NotImplementedError because the antenna has no standby state.

Parameters:

task_callback (Optional[Callable]) – Update task state, defaults to None

Raises:

NotImplementedError – because the antenna has no standby state.

Return type:

tuple[TaskStatus, str]

start_communicating()

Establish communication with the component, then start monitoring.

Return type:

None

stop_communicating()

Cease monitoring the component, and break off all communication with it.

Return type:

None

property temperature: float

Return the antenna’s temperature.

Returns:

the temperature of this antenna

property voltage: float

Return the antenna’s voltage.

Returns:

the voltage of this antenna

class AntennaHealthModel(component_state_callback, thresholds=None)

A health model for an antenna.

__init__(component_state_callback, thresholds=None)

Initialise a new instance.

Parameters:

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

• thresholds (Optional[dict[str, float]]) – the threshold parameters for the health rules

evaluate_health()

Compute overall health of the station.

The overall health is based on the fault and communication status of the station overall, together with the health of the FieldStation, antennas and tiles that it manages.

This implementation simply sets the health of the station to the health of its least healthy component.

Return type:

HealthState

Returns:

an overall health of the station

fieldstation_health_changed(fieldstation_health)

Handle a change in FieldStation health.

Parameters:

fieldstation_health (Optional[HealthState]) – the health state of the FieldStation

Return type:

None

tile_adc_rms_changed(tile_adc_rms)

Handle a change in tile adc reading.

Parameters:

tile_adc_rms (Optional[tuple[float, float]]) – the adc reading from the tile

Return type:

None

tile_health_changed(tile_health)

Handle a change in tile health.

Parameters:

tile_health (Optional[HealthState]) – the health state of the tile

Return type:

None

class MccsAntenna(*args, **kwargs)

An implementation of an antenna Tango device for MCCS.

Configure(argin)

Configure the antenna device attributes.

Also configures children device that are connected to the antenna.

Parameters:

argin (str) – Configuration parameters encoded in a json string

Return type:

tuple[list[ResultCode], list[Optional[str]]]

Returns:

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

Example:

class InitCommand(*args, **kwargs)

Class that implements device initialisation for the MCCS antenna device.

do(*args, **kwargs)

Stateless hook for device initialisation.

Initialises the attributes and properties of the MccsAntenna.

Parameters:

• args (Any) – positional args to the component manager method

• kwargs (Any) – keyword args to the component manager method

Return type:

tuple[ResultCode, str]

Returns:

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

__init__(*args, **kwargs)

Initialise this device object.

Parameters:

• args (Any) – positional args to the init

• kwargs (Any) – keyword args to the init

altitude()

Return the altitude attribute.

Return type:

float

Returns:

the altitude of the antenna

antennaId()

Return the antenna ID attribute.

Return type:

int

Returns:

antenna ID

bandpassCoefficient()

Return the bandpassCoefficient attribute.

Return type:

list[float]

Returns:

bandpass coefficients

calibrationCoefficient()

Get the Calibration coefficients.

The coefficients to be applied for the next frequency channel in the calibration cycle.

Return type:

list[float]

Returns:

the calibration coefficients

create_component_manager()

Create and return a component manager for this device.

Return type:

AntennaComponentManager

Returns:

a component manager for this device.

current()

Return the current attribute.

Return type:

float

Returns:

the current

delayRates()

Return the delayRates attribute.

Return type:

list[float]

Returns:

delay rate for each beam

delays()

Return the delays attribute.

Return type:

list[float]

Returns:

delay for each beam

fieldNodeLatitude()

Return the fieldNodeLatitude attribute.

Return type:

float

Returns:

the Latitude of field node centre

fieldNodeLongitude()

Return the fieldNodeLongitude attribute.

Return type:

float

Returns:

the Longitude of field node centre

first()

Return the first attribute.

Return type:

bool

Returns:

the first faulty flag

gain()

Return the gain attribute.

Return type:

float

Returns:

the gain

healthModelParams(argin)

Set the params for health transition rules.

Parameters:

argin (str) – JSON-string of dictionary of health states

Return type:

None

init_command_objects()

Initialise the command handlers for commands supported by this device.

Return type:

None

init_device()

Initialise the device.

This is overridden here to change the Tango serialisation model.

Return type:

None

is_On_allowed()

Check if command On is allowed in the current device state.

Return type:

bool

Returns:

True if the command is allowed

logicalAntennaId()

Return the logical antenna ID attribute.

Return type:

int

Returns:

the logical antenna ID

pointingCoefficient()

Return the pointingCoefficient attribute.

Return type:

list[float]

Returns:

the pointing coefficients

position()

Return the position attribute.

Return type:

list[float]

Returns:

positions

rms()

Return the measured RMS of the antenna.

Return type:

float

Returns:

the measured rms

simulationMode(value)

Set the simulation mode of this device.

Parameters:

value (SimulationMode) – the new simulation mode

Raises:

ValueError – because this device cannot be put into simulation mode.

Return type:

None

spectrumX()

Return the spectrumX attribute.

Return type:

list[float]

Returns:

x spectrum

spectrumY()

Return the spectrumY attribute.

Return type:

list[float]

Returns:

y spectrum

temperature()

Return the temperature attribute.

Return type:

float

Returns:

the temperature

timestampOfLastSpectrum()

Return the timestampOfLastSpectrum attribute.

Return type:

str

Returns:

the timestamp of the last spectrum

voltage()

Return the voltage attribute.

Return type:

float

Returns:

the voltage

xDisplacement()

Return the horizontal displacement east attribute.

Return type:

float

Returns:

the horizontal displacement eastwards from station reference position

xPolarisationFaulty()

Return the xPolarisationFaulty attribute.

Return type:

bool

Returns:

the x-polarisation faulty flag

xPolarisationScalingFactor()

Return the logical antenna ID attribute.

Return type:

list[int]

Returns:

the x polarisation scaling factor

yDisplacement()

Return the horizontal displacement north attribute.

Return type:

float

Returns:

the horizontal displacement northwards from station reference position

yPolarisationFaulty()

Return the yPolarisationFaulty attribute.

Return type:

bool

Returns:

the y-polarisation faulty flag

yPolarisationScalingFactor()

Return the yPolarisationScalingFactor attribute.

Return type:

list[int]

Returns:

the y polarisation scaling factor

zDisplacement()

Return the vertical displacement attribute.

Return type:

float

Returns:

the vertical displacement upwards from station reference position

• Antenna component manager
  • AntennaComponentManager
    • AntennaComponentManager.__init__()
    • AntennaComponentManager.configure()
    • AntennaComponentManager.current
    • AntennaComponentManager.off()
    • AntennaComponentManager.on()
    • AntennaComponentManager.power_state
    • AntennaComponentManager.power_state_lock
    • AntennaComponentManager.reset()
    • AntennaComponentManager.standby()
    • AntennaComponentManager.start_communicating()
    • AntennaComponentManager.stop_communicating()
    • AntennaComponentManager.temperature
    • AntennaComponentManager.voltage
• Antenna device
  • MccsAntenna
    • MccsAntenna.Configure()
    • MccsAntenna.InitCommand
      • MccsAntenna.InitCommand.do()
    • MccsAntenna.__init__()
    • MccsAntenna.altitude()
    • MccsAntenna.antennaId()
    • MccsAntenna.bandpassCoefficient()
    • MccsAntenna.calibrationCoefficient()
    • MccsAntenna.create_component_manager()
    • MccsAntenna.current()
    • MccsAntenna.delayRates()
    • MccsAntenna.delays()
    • MccsAntenna.fieldNodeLatitude()
    • MccsAntenna.fieldNodeLongitude()
    • MccsAntenna.first()
    • MccsAntenna.gain()
    • MccsAntenna.healthModelParams()
    • MccsAntenna.init_command_objects()
    • MccsAntenna.init_device()
    • MccsAntenna.is_On_allowed()
    • MccsAntenna.logicalAntennaId()
    • MccsAntenna.pointingCoefficient()
    • MccsAntenna.position()
    • MccsAntenna.rms()
    • MccsAntenna.simulationMode()
    • MccsAntenna.spectrumX()
    • MccsAntenna.spectrumY()
    • MccsAntenna.temperature()
    • MccsAntenna.timestampOfLastSpectrum()
    • MccsAntenna.voltage()
    • MccsAntenna.xDisplacement()
    • MccsAntenna.xPolarisationFaulty()
    • MccsAntenna.xPolarisationScalingFactor()
    • MccsAntenna.yDisplacement()
    • MccsAntenna.yPolarisationFaulty()
    • MccsAntenna.yPolarisationScalingFactor()
    • MccsAntenna.zDisplacement()
  • main()
• Antenna health model
  • AntennaHealthModel
    • AntennaHealthModel.__init__()
    • AntennaHealthModel.evaluate_health()
    • AntennaHealthModel.fieldstation_health_changed()
    • AntennaHealthModel.tile_adc_rms_changed()
    • AntennaHealthModel.tile_health_changed()
• Antenna health rules
  • AntennaHealthRules
    • AntennaHealthRules.default_thresholds
    • AntennaHealthRules.degraded_rule()
    • AntennaHealthRules.failed_rule()
    • AntennaHealthRules.healthy_rule()
    • AntennaHealthRules.unknown_rule()