Station subpackage

This subpackage implements station functionality for MCCS.

class SpsStation(*args, **kwargs)

An implementation of an SPS Station Tango device for MCCS.

ApplyCalibration(argin)

Load the calibration coefficients at the specified time delay.

Parameters:: argin (str) – switch time, in ISO formatted time. Default: now
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:

>>> dp = tango.DeviceProxy("mccs/tile/01")
>>> dp.command_inout("ApplyCalibration", "")

ApplyPointingDelays(argin)

Set the pointing delay parameters of this Station’s Tiles.

Parameters:: argin (str) – switch time, in ISO formatted time. Default: now
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:

>>> dp = tango.DeviceProxy("mccs/station/01")
>>> time_string = switch time as ISO formatted time
>>> dp.command_inout("ApplyPointingDelays", time_string)

ConfigureIntegratedBeamData(argin)

Configure the transmission of integrated beam data.

Using the provided integration time, the first channel and the last channel. The data are sent continuously until the StopIntegratedData command is run.

Parameters:: argin (str) – json dictionary with optional keywords:

integration_time - (float) in seconds (default = 0.5)
first_channel - (int) default 0
last_channel - (int) default 191

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:

>>> dp = tango.DeviceProxy("mccs/tile/01")
>>> dict = {"integration_time": 0.2, "first_channel":0, "last_channel": 191}
>>> jstr = json.dumps(dict)
>>> dp.command_inout("ConfigureIntegratedBeamData", jstr)

ConfigureIntegratedChannelData(argin)

Configure and start the transmission of integrated channel data.

Using the provided integration time, first channel and last channel. Data are sent continuously until the StopIntegratedData command is run.

Parameters:: argin (str) – json dictionary with optional keywords:

integration_time - (float) in seconds (default = 0.5)
first_channel - (int) default 0
last_channel - (int) default 511

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:

>>> dp = tango.DeviceProxy("mccs/tile/01")
>>> dict = {"integration_time": 0.2, "first_channel":0, "last_channel": 191}
>>> jstr = json.dumps(dict)
>>> dp.command_inout("ConfigureIntegratedChannelData", jstr)

ConfigureTestGenerator(argin)

Set the test signal generator.

Parameters:: argin (str) – json dictionary with keywords:

tone_frequency: first tone frequency, in Hz. The frequency
is rounded to the resolution of the generator. If this is not specified, the tone generator is disabled.
tone_amplitude: peak tone amplitude, normalized to 31.875 ADC
units. The amplitude is rounded to 1/8 ADC unit. Default is 1.0. A value of -1.0 keeps the previously set value.
tone_2_frequency: frequency for the second tone. Same
as ToneFrequency.
tone_2_amplitude: peak tone amplitude for the second tone.
Same as ToneAmplitude.
noise_amplitude: RMS amplitude of the pseudorandom Gaussian
white noise, normalized to 26.03 ADC units.
pulse_frequency: frequency of the periodic pulse. A code
in the range 0 to 7, corresponding to (16, 12, 8, 6, 4, 3, 2) times the ADC frame frequency.
pulse_amplitude: peak amplitude of the periodic pulse, normalized
to 127 ADC units. Default is 1.0. A value of -1.0 keeps the previously set value.
set_time: time at which the generator is set, for synchronization
among different TPMs. In UTC ISO format (string)
adc_channels: list of adc channels which will be substituted with
the generated signal. It is a 32 integer, with each bit representing an input channel. Default: all if at least q source is specified, none otherwises.

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:

>>> dp = tango.DeviceProxy("mccs/tile/01")
>>> dict = {"tone_frequency": 150e6, "tone_amplitude": 0.1,
        "noise_amplitude": 0.9, "pulse_frequency": 7,
        "set_time": "2022-08-09T12:34:56.7Z"}
>>> jstr = json.dumps(dict)
>>> values = dp.command_inout("ConfigureTestGenerator", jstr)

class InitCommand(*args, **kwargs)

A class for SpsStation’s Init command.

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

do(*args, **kwargs)

Initialise the SpsStation.

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.

Initialise()

Initialise the station.

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:

>>> dp = tango.DeviceProxy("mccs/station/001")
>>> dp.command_inout("Initialise")

LoadCalibrationCoefficients(argin)

Load the calibration coefficients, but does not apply them.

This is performed by apply_calibration. The calibration coefficients may include any rotation matrix (e.g. the parallactic angle), but do not include the geometric delay.

Parameters:: argin (list[float]) – list comprises:

antenna - (int) is the antenna to which the coefficients will be applied.
calibration_coefficients - [array] a bidimensional complex array comprising
calibration_coefficients[channel, polarization], with each element representing a normalized coefficient, with (1.0, 0.0) being the normal, expected response for an ideal antenna.
- channel - (int) channel is the index specifying the channels at the
  beamformer output, i.e. considering only those channels actually processed and beam assignments.
- polarization index ranges from 0 to 3.
  0: X polarization direct element
  
  1: X->Y polarization cross element
  
  2: Y->X polarization cross element
  
  3: Y polarization direct element

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.
Raises:: ValueError – if parameters are illegal or inconsistent
Example:

>>> antenna = 2
>>> complex_coefficients = [[complex(3.4, 1.2), complex(2.3, 4.1),
>>>            complex(4.6, 8.2), complex(6.8, 2.4)]]*5
>>> inp = list(itertools.chain.from_iterable(complex_coefficients))
>>> out = ([v.real, v.imag] for v in inp]
>>> coefficients = list(itertools.chain.from_iterable(out))
>>> coefficients.insert(0, float(antenna))
>>> input = list(itertools.chain.from_iterable(coefficients))
>>> dp = tango.DeviceProxy("mccs/tile/01")
>>> dp.command_inout("LoadCalibrationCoefficients", input)

LoadPointingDelays(argin)

Set the pointing delay parameters of this Station’s Tiles.

Parameters:: argin (list[float]) – an array containing a beam index followed by pairs of antenna delays + delay rates, delay in seconds and the delay rate in seconds/second
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.
Raises:: ValueError – if parameters are illegal or inconsistent
Example:

>>> # example delays: 256 values from -32 to +32 ns, rates = 0
>>> delays = [step * 0.25e-9 for step in list(range(-128, 128))]
>>> rates = [0.0]*256
>>> beam = 0.0
>>> dp = tango.DeviceProxy("mccs/station/01")
>>> arg = [beam]
>>> for i in range(256)
>>>   arg.append(delays[i])
>>>   arg.append(rates[i])
>>> dp.command_inout("LoadPointingDelays", arg)

RunTest(argin)

Run a self-check test and optional amount of times.

Parameters:

argin (str) – json-ified args, containing a required ‘test_name’, and optional ‘count’

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:

>>> dp = tango.DeviceProxy("low-mccs/spsstation/aavs3")
>>> dp.RunTest(json.dumps({"test_name" : "my_test", "count" : 5}))

SelfCheck()

Run all the self-check tests once.

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:

>>> dp = tango.DeviceProxy("low-mccs/spsstation/aavs3")
>>> dp.SelfCheck()

SendDataSamples(argin)

Transmit a snapshot containing raw antenna data.

Parameters:: argin (str) – json dictionary with optional keywords:

data_type - type of snapshot data (mandatory): “raw”, “channel”,
“channel_continuous”, “narrowband”, “beam”
start_time - Time (UTC string) to start sending data. Default immediately
seconds - (float) Delay if timestamp is not specified. Default 0.2 seconds

Depending on the data type: raw:

sync: bool: send synchronised samples for all antennas, vs. round robin
larger snapshot from each antenna

channel:

n_samples: Number of samples per channel, default 1024
first_channel - (int) first channel to send, default 0
last_channel - (int) last channel to send, default 511

channel_continuous

channel_id - (int) channel_id (Mandatory)
n_samples - (int) number of samples to send per packet, default 128

narrowband:

frequency - (int) Sky frequency for band centre, in Hz (Mandatory)
round_bits - (int) Specify whow many bits to round
n_samples - (int) number of spectra to send

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.
Raises:: ValueError – if mandatory parameters are missing
Example:

>>> dp = tango.DeviceProxy("mccs/tile/01")
>>> dict = {"data_type": "raw", "Sync":True, "Seconds": 0.2}
>>> jstr = json.dumps(dict)
>>> dp.command_inout("SendDataSamples", jstr)

SetBeamFormerRegions(argin)

Set the frequency regions which are going to be beamformed into each beam.

region_array is defined as a flattened 2D array, for a maximum of 48 regions. Total number of channels must be <= 384.

Parameters:: argin (list[int]) – list of regions. Each region comprises:

start_channel - (int) region starting channel, must be even in range 0 to 510
num_channels - (int) size of the region, must be a multiple of 8
beam_index - (int) beam used for this region with range 0 to 47
subarray_id - (int) Subarray
subarray_logical_channel - (int) logical channel # in the subarray
subarray_beam_id - (int) ID of the subarray beam
substation_id - (int) Substation
aperture_id: ID of the aperture (station*100+substation?)

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.
Raises:: ValueError – if parameters are illegal or inconsistent
Example:

>>> regions = [[4, 24, 0, 0, 0, 3, 1, 101], [26, 40, 1, 0, 24, 4, 2, 102]]
>>> input = list(itertools.chain.from_iterable(regions))
>>> dp = tango.DeviceProxy("mccs/tile/01")
>>> dp.command_inout("SetBeamFormerRegions", input)

SetBeamFormerTable(argin)

Set the beamformer table which are going to be beamformed into each beam.

region_array is defined as a flattened 2D array, for a maximum of 48 entries. Each entry corresponds to 8 consecutive frequency channels. This is equivalent to SetBeamformerRegions, with a different way to specify the bandwidth of each spectral region. Input is consistent with the beamformerTable attribute

Parameters:: argin (list[int]) – list of regions. Each region comprises:

start_channel - (int) region starting channel, must be even in range 0 to 510
beam_index - (int) beam used for this region with range 0 to 47
subarray_id - (int) Subarray
subarray_logical_channel - (int) logical channel # in the subarray
subarray_beam_id - (int) ID of the subarray beam
substation_id - (int) Substation
aperture_id: ID of the aperture (station*100+substation?)

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.
Raises:: ValueError – if parameters are illegal or inconsistent
Example:

>>> regions = [[4, 0, 0, 0, 3, 1, 101], [26, 1, 0, 24, 4, 2, 102]]
>>> input = list(itertools.chain.from_iterable(regions))
>>> dp = tango.DeviceProxy("mccs/station/01")
>>> dp.command_inout("SetBeamFormerRegions", input)

SetChanneliserRounding(channeliser_rounding)

Set the ChanneliserRounding to all Tiles in this Station.

Number of LS bits dropped in each channeliser frequency channel. Valid values 0-7 Same value applies to all antennas and polarizations

Parameters:

channeliser_rounding (ndarray) – list of 512 values, one per channel. this will apply to all Tiles in this station.

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:

>>> dp = tango.DeviceProxy("low-mccs/station/aavs3")
>>> dp.command_inout("SetChanneliserRounding", np.array([2]*512))

SetCspIngest(argin)

Configure link for beam data packets to CSP.

Parameters:

argin (str) –

json dictionary with optional keywords:

destination_ip - (string) Destination IP
source_port - (int) Source port for integrated data streams
destination_port - (int) Destination port for integrated data streams

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:

>>> dp = tango.DeviceProxy("mccs/tile/01")
>>> dict = {"destination_ip"="10.0.1.23"}
>>> jstr = json.dumps(dict)
>>> dp.command_inout("SetCspIngest", jstr)

SetLmcDownload(argin)

Specify whether control data will be transmitted over 1G or 40G networks.

Parameters:

argin (str) –

json dictionary with optional keywords:

mode - (string) ‘1G’ or ‘10G’ (Mandatory) (use ‘10G’ for 40G also)
payload_length - (int) SPEAD payload length for channel data
destination_ip - (string) Destination IP.
source_port - (int) Source port for integrated data streams
destination_port - (int) Destination port for integrated data streams

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:

>> dp = tango.DeviceProxy(“mccs/tile/01”) >> dict = {“mode”: “1G”, “payload_length”: 1024, “destination_ip”: “10.0.1.23”} >> jstr = json.dumps(dict) >> dp.command_inout(“SetLmcDownload”, jstr)

SetLmcIntegratedDownload(argin)

Configure link and size for integrated data packets, for all tiles.

Parameters:

argin (str) –

json dictionary with optional keywords:

mode - (string) ‘1G’ ‘10G’ ‘40G’ - default 40G
channel_payload_length - (int) SPEAD payload length for integrated
channel data
beam_payload_length - (int) SPEAD payload length for integrated beam data
destination_ip - (string) Destination IP
source_port - (int) Source port for integrated data streams
destination_port - (int) Destination port for integrated data streams

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:

>>> dp = tango.DeviceProxy("mccs/tile/01")
>>> dict = {"mode": "1G", "channel_payload_length":4,
            "beam_payload_length": 1024, "destination_ip"="10.0.1.23"}
>>> jstr = json.dumps(dict)
>>> dp.command_inout("SetLmcIntegratedDownload", jstr)

StartAcquisition(argin)

Start the acquisition synchronously for all tiles, checks for synchronisation.

Parameters:

argin (str) – Start acquisition time in ISO9601 format

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:

>>> dp = tango.DeviceProxy("mccs/station/001")
>>> dp.command_inout("StartAcquisition", "20230101T12:34:55.000Z")

StartBeamformer(argin)

Start the beamformer at the specified time delay.

Parameters:: argin (str) – json dictionary with optional keywords:

start_time - (str, ISO UTC time) start time
duration - (int) if > 0 is a duration in seconds
if < 0 run forever
subarray_beam_id - (int)Subarray beam ID of the channels to be started
Command affects only beamformed channels for given subarray ID Default -1: all channels
scan_id - (int) The unique ID for the started scan. Default 0

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:

>>> dp = tango.DeviceProxy("mccs/tile/01")
>>> dict = {"StartTime": "2022-01-02T34:56:08.987Z", "Duration": 30.0}
>>> jstr = json.dumps(dict)
>>> dp.command_inout("StartBeamformer", jstr)

StopBeamformer()

Stop the beamformer.

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:

>>> dp = tango.DeviceProxy("mccs/tile/01")
>>> dp.command_inout("StopBeamformer")

StopDataTransmission()

Stop data transmission from board.

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:

>>> dp = tango.DeviceProxy("mccs/tile/01")
>>> dp.command_inout("StopDataTransmission")

StopIntegratedData()

Stop the integrated data.

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.

TriggerAdcEqualisation()

Get the equalised ADC values.

Getting the equalised values takes up to 20 seconds (to get an average to avoid spikes). So we trigger the collection and publish to dbmPowers

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:

>>> dp = tango.DeviceProxy("mccs/station/001")
>>> dp.command_inout("TriggerAdcEqualisation")

UpdateStaticDelays()

Update static delays from TelModel.

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.

__init__(*args, **kwargs)

Initialise this device object.

Parameters:

args (Any) – positional args to the init
kwargs (Any) – keyword args to the init

adcPower()

Get the ADC RMS input levels for all input signals.

Returns an array of 2 values (X and Y polarizations) per antenna, 32 per tile, 512 per station

Return type:: Optional[list[float]]
Returns:: the ADC RMS input levels, in ADC units

antennaInfo()

Return antenna information.

Returns a json string representing a dictionary coded: by antenna number and presenting that antenna’s station_id, tile_id and location information.

Return type:: str
Returns:: json string containing antenna information.

antennasMapping()

Return the mappings of the antennas.

Returns a mapping of antenna number to: TPM port number.

Return type:: str
Returns:: json string containing antenna mappings

beamformerTable()

Get beamformer region table.

Bidimensional array of one row for each 8 channels, with elements: 0. start physical channel 1. beam number 2. subarray ID 3. subarray_logical_channel 4. subarray_beam_id 5. substation_id 6. aperture_id

Each row is a set of 7 consecutive elements in the list.

Return type:: list[int]
Returns:: list of up to 7*48 values

boardTemperaturesSummary()

Get summary of board temperatures (minimum, average, maximum).

Return type:: list[float]
Returns:: minimum, average, maximum board temperatures, in deg Celsius

channeliserRounding()

Channeliser rounding.

Number of LS bits dropped in each channeliser frequency channel. Valid values 0-7 Same value applies to all antennas and polarizations

Return type:: ndarray
Returns:: A list of 512 values for every tile, one per channel.

clockPresentSummary()

Get summary of clock present status for all tiles.

Return type:: bool
Returns:: True if 10 MHz clock signal is present in all tiles

create_component_manager()

Create and return a component manager for this device.

Return type:: SpsStationComponentManager
Returns:: a component manager for this device.

cspIngestAddress()

Get CSP ingest IP address.

CSP ingest address and port are set by the SetCspIngest command

Return type:: str
Returns:: IP net address for CSP ingest port

cspIngestPort()

Get CSP ingest port.

CSP ingest address and port are set by the SetCspIngest command

Return type:: int
Returns:: UDP port for the CSP ingest port

cspRounding(rounding)

Set CSP formatter rounding.

Parameters:: rounding (list[int]) – list of up to 384 values in the range 0-7. Current hardware supports only a single value, thus oly 1st value is used
Return type:: None

cspSourcePort()

Get CSP source port.

CSP source port is set by the SetCspIngest command

Return type:: int
Returns:: UDP port for the CSP source port

daqTRL(value)

Set the Tango Resource Locator for this SpsStation’s DAQ instance.

Parameters:: value (str) – The new DAQ TRL.
Return type:: None

fortyGbNetworkAddress()

Get 40Gb network address for this station.

Return type:: str
Returns:: IP subnet address

fortyGbNetworkErrors()

Get number of network errors for all 40 Gb interfaces.

Return type:: list[int]
Returns:: Total number of errors on each interface (2 per tile)

fpgaTemperaturesSummary()

Get summary of FPGA temperatures (minimum, average, maximum).

Return type:: list[float]
Returns:: minimum, average, maximum board temperatures, in deg Celsius

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()

Set up the handler objects for Commands.

Return type:: None

init_device()

Initialise the device.

This is overridden here to change the Tango serialisation model.

Return type:: None

isBeamformerRunning()

Get the state of the test generator.

Return type:: bool
Returns:: true if the test generator is active in at least one tile

isCalibrated()

Return a flag indicating whether this station is currently calibrated or not.

Return type:: bool
Returns:: a flag indicating whether this station is currently calibrated or not.

isConfigured()

Return a flag indicating whether this station is currently configured or not.

Return type:: bool
Returns:: a flag indicating whether this station is currently configured or not.

isProgrammed()

Return a flag indicating whether of not the TPM boards are programmed.

Attribute is False if at least one TPM is not programmed.

Return type:: bool
Returns:: whether of not the TPM boards are programmed

is_On_allowed()

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

Return type:: bool
Returns:: True if the command is allowed

is_Standby_allowed()

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

Return type:: bool
Returns:: True if the command is allowed

pllLockedSummary()

Get summary of PLL locked status for all tiles.

Return type:: bool
Returns:: True if PLL is locked to reference in all tiles

ppsDelayCorrections(delays)

Set PPS delay correction, one per tile.

Note: this will be set in the next initialisation.

Parameters:: delays (list[int]) – PPS delay correction in nanoseconds, one value per tile. Values are internally rounded to 1.25 ns units.
Return type:: None

ppsDelaySummary()

Get summary of PPS delay (minimum, average, maximum).

Return type:: list[float]
Returns:: minimum, average, maximum board temperatures, in deg Celsius

ppsDelays()

Get PPS delay correction, one per tile.

Return type:: list[int]
Returns:: Array of PPS delay in nanoseconds, one value per tile.

ppsPresentSummary()

Get summary of PPS present status for all tiles.

Return type:: bool
Returns:: True if PPS signal is present in all tiles

preaduLevels(levels)

Set attenuator level of preADU channels, one per input channel.

Parameters:: levels (list[float]) – attenuator level of preADU channels, one per input channel (2 per antenna, 32 per tile, 512 total), in dB
Return type:: None

staticTimeDelays(delays)

Set static time delay.

Parameters:: delays (list[float]) – Delay in samples (positive = increase the signal delay) to correct for static delay mismathces, e.g. cable length. 2 values per antenna (pol. X and Y), 32 values per tile, 512 total.
Return type:: None

sysrefPresentSummary()

Get summary of sysrf present status for all tiles.

Return type:: bool
Returns:: True if SYSREF signal is present in all tiles

testGeneratorActive()

Get the state of the test generator.

Return type:: bool
Returns:: true if the test generator is active in at least one tile

testList()

Get the list of self-check tests available.

Return type:: list[str]
Returns:: the list of self-check tests available.

testLogs()

Get logs of the most recently run self-check test.

Return type:: str
Returns:: the logs of the most recently run self-check test.

testReport()

Get the report for the most recently run self-check test set.

Return type:: str
Returns:: the report for the most recently run self-check test set.

tileProgrammingState()

Get the tile programming state.

Return type:: list[str]
Returns:: a list of strings describing the programming state of the tiles

xPolBandpass()

Read the last bandpass plot data for the x-polarisation.

Return type:: ndarray
Returns:: The last block of x-polarised bandpass data.

yPolBandpass()

Read the last bandpass plot data for the y-polarisation.

Return type:: ndarray
Returns:: The last block of y-polarised bandpass data.

class SpsStationComponentManager(station_id, subrack_fqdns, tile_fqdns, daq_trl, sdn_first_interface, sdn_gateway, antenna_config_uri, logger, max_workers, communication_state_changed_callback, component_state_changed_callback, tile_health_changed_callback, subrack_health_changed_callback)

A component manager for a station.

__init__(station_id, subrack_fqdns, tile_fqdns, daq_trl, sdn_first_interface, sdn_gateway, antenna_config_uri, logger, max_workers, communication_state_changed_callback, component_state_changed_callback, tile_health_changed_callback, subrack_health_changed_callback)

Initialise a new instance.

Parameters:

station_id (int) – the id of this station
subrack_fqdns (Sequence[str]) – FQDNs of the Tango devices which manage this station’s subracks
tile_fqdns (Sequence[str]) – FQDNs of the Tango devices which manage this station’s TPMs
daq_trl (str) – The TRL of this Station’s DAQ Receiver.
sdn_first_interface (str) – CIDR-style IP address with mask, for the first interface in the block assigned for science data For example, “10.130.0.1/25” means “address 10.130.0.1 on network 10.130.0.0/25”.
sdn_gateway (str) – IP address of the SDN gateway, or None if the network has no gateway.
antenna_config_uri (Optional[list[str]]) – location of the antenna mapping file
logger (Logger) – the logger to be used by this object.
max_workers (int) – the maximum worker threads for the slow commands associated with this component manager.
communication_state_changed_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_changed_callback (Callable[..., None]) – callback to be called when the component state changes
tile_health_changed_callback (Callable[[str, Optional[HealthState]], None]) – callback to be called when a tile’s health changed
subrack_health_changed_callback (Callable[[str, Optional[HealthState]], None]) – callback to be called when a subrack’s health changed

adc_power()

Get input RMS levels.

Return type:: list[float]
Returns:: list of RMS levels of ADC inputs

apply_calibration(switch_time)

Switch the calibration bank.

(i.e. apply the calibration coefficients previously loaded by load_calibration_coefficients()).

Parameters:: switch_time (str) – an optional time at which to perform the switch
Return type:: None

apply_pointing_delays(load_time)

Load the pointing delay at a specified time.

Parameters:: load_time (str) – time at which to load the pointing delay
Return type:: None

property beamformer_table: list[list[int]]

Get beamformer region table.

Bidimensional array of one row for each 8 channels, with elements: 0. start physical channel 1. beam number 2. subarray ID 3. subarray_logical_channel 4. subarray_beam_id 5. substation_id 6. aperture_id

Each row is a set of 7 consecutive elements in the list.

Returns:: list of up to 7*48 values

board_temperature_summary()

Get summary of board temperatures.

Return type:: list[float]
Returns:: minimum, average and maximum of board temperatures

property channeliser_rounding: numpy.ndarray

Channeliser rounding.

Number of LS bits dropped in each channeliser frequency channel. Valid values 0-7 Same value applies to all antennas and polarizations

Returns:: list of 512 values for each Tile, one per channel.

clock_present_summary()

Get summary of 10 MHz clock presence.

Return type:: bool
Returns:: TRUE if 10 MHz clock is present in all tiles

configure_integrated_beam_data(integration_time, first_channel, last_channel)

Configure and start the transmission of integrated channel data.

Configure with the provided integration time, first channel and last channel. Data are sent continuously until the StopIntegratedData command is run.

Parameters:

integration_time (float) – integration time in seconds, defaults to 0.5
first_channel (int) – first channel
last_channel (int) – last channel

Return type:

None

configure_integrated_channel_data(integration_time, first_channel, last_channel)

Configure and start the transmission of integrated channel data.

Configure with the provided integration time, first channel and last channel. Data are sent continuously until the StopIntegratedData command is run.

Parameters:

integration_time (float) – integration time in seconds, defaults to 0.5
first_channel (int) – first channel
last_channel (int) – last channel

Return type:

None

configure_test_generator(argin)

Distribute to tiles command configure_test_generator.

Parameters:: argin (str) – Json encoded parameter List
Return type:: None

property csp_ingest_address: str

Get 40Gb CSP address.

Returns:: IP address for CSP ingest port

property csp_ingest_port: int

Get 40Gb CSP ingest port.

Returns:: UDP port for CSP ingest port

property csp_rounding: list[int] | None

CSP formatter rounding.

Rounding from 16 to 8 bits in final stage of the station beamformer, before sending data to CSP. Array of (up to) 384 values, one for each logical channel. Range 0 to 7, as number of discarded LS bits.

Returns:: CSP formatter rounding for each logical channel.

property csp_source_port: int

Get 40Gb CSP source port.

Returns:: UDP port for CSP source port

property forty_gb_network_address: str

Get 40Gb network address.

Returns:: IP network address for station network

forty_gb_network_errors()

Get summary of network errors.

Return type:: list[int]
Returns:: list of 40Gb network errors for all tiles

fpga_temperature_summary()

Get summary of FPGAs temperatures.

Return type:: list[float]
Returns:: minimum, average and maximum of FPGAs temperatures

initialise(task_callback=None)

Submit the _initialise method.

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

Parameters:: task_callback (Optional[Callable]) – Update task state, defaults to None
Return type:: tuple[TaskStatus, str]
Returns:: a task status and response message

property is_beamformer_running: bool

Get station beamformer state.

Returns:: Get station beamformer state

property is_configured: bool

Return whether this station component manager is configured.

Returns:: whether this station component manager is configured.

property is_programmed: bool

Get TPM programming state.

Returns:: True if all TPMs are programmed

load_calibration_coefficients(calibration_coefficients)

Load calibration coefficients.

These may include any rotation matrix (e.g. the parallactic angle), but do not include the geometric delay.

Parameters:: calibration_coefficients (list[float]) – a bidirectional complex array of coefficients, flattened into a list
Return type:: None

load_pointing_delays(delay_list)

Specify the delay in seconds and the delay rate in seconds/second.

The delay_array specifies the delay and delay rate for each antenna. beam_index specifies which beam is desired (range 0-47, limited to 7 in the current firmware)

Parameters:: delay_list (list[float]) – delay in seconds, and delay rate in seconds/second
Return type:: None

off(task_callback=None)

Submit the _off method.

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:: a result code and response message

on(task_callback=None)

Submit the _on method.

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:: a task status and response message

pll_locked_summary()

Get summary of PLL lock state.

Return type:: bool
Returns:: TRUE if PLL locked in all tiles

property pps_delay_corrections: list[int]

Get the PPS delay correction.

Returns:: Array of pps delay corrections, one value per tile, in nanoseconds

pps_delay_summary()

Get summary of PPS delays.

Return type:: list[float]
Returns:: minimum, average and maximum of PPS delays

property pps_delays: list[int]

Get PPS delay.

Array of one value per tile. Returns the PPS delay, Values are internally rounded to 1.25 ns steps

Returns:: Array of one value per tile, in nanoseconds

pps_present_summary()

Get summary of PPS presence.

Return type:: bool
Returns:: TRUE if PPS is present in all tiles

property preadu_levels: list[float]

Get attenuator level of preADU channels, one per input channel.

Returns:: Array of one value per antenna/polarization (32 per tile)

run_test(task_callback=None, *, count=1, test_name)

Submit the _run_test method.

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

Parameters:

task_callback (Optional[Callable]) – Update task state, defaults to None
count (Optional[int]) – how many times to run the test, default is 1.
test_name (str) – which test to run.

Return type:

tuple[TaskStatus, str]

Returns:

a task status and response message

self_check(task_callback=None)

Submit the _self_check method.

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

Parameters:: task_callback (Optional[Callable]) – Update task state, defaults to None
Return type:: tuple[TaskStatus, str]
Returns:: a task status and response message

send_data_samples(argin)

Front end for send_xxx_data methods.

Parameters:: argin (str) – Json encoded parameter List
Return type:: None

set_beamformer_table(beamformer_table)

Set the frequency regions to be beamformed into a single beam.

Parameters:: beamformer_table (list[list[int]]) – a list encoding up to 48 regions, with each region containing a start channel, the size of the region (which must be a multiple of 8), and a beam index (between 0 and 7) and a substation ID (not used)
Return type:: None

set_channeliser_rounding(channeliser_rounding, task_callback=None)

Set the channeliserRounding in all Tiles.

Parameters:

channeliser_rounding (ndarray) – the number of LS bits dropped in each channeliser frequency channel.
task_callback (Optional[Callable]) – Update task state, defaults to None

Return type:

tuple[TaskStatus, str]

Returns:

a task status and response message

set_csp_ingest(dst_ip, src_port, dst_port)

Configure link for CSP ingest channel.

Parameters:

dst_ip (str) – Destination IP, defaults to None
src_port (int) – source port, defaults to 0xF0D0
dst_port (int) – destination port, defaults to 4660

Return type:

None

set_lmc_download(mode, payload_length, dst_ip, src_port=61648, dst_port=4660)

Configure link and size of LMC channel.

Parameters:

mode (str) – ‘1G’ or ‘10G’
payload_length (int) – SPEAD payload length for LMC packets
dst_ip (str) – Destination IP, defaults to None
src_port (int) – source port, defaults to 0xF0D0
dst_port (int) – destination port, defaults to 4660

Return type:

None

set_lmc_integrated_download(mode, channel_payload_length, beam_payload_length, dst_ip='', src_port=61648, dst_port=4660)

Configure link and size of integrated LMC channel.

Parameters:

mode (str) – ‘1G’ or ‘10G’
channel_payload_length (int) – SPEAD payload length for integrated channel data
beam_payload_length (int) – SPEAD payload length for integrated beam data
dst_ip (str) – Destination IP, defaults to None
src_port (int) – source port, defaults to 0xF0D0
dst_port (int) – destination port, defaults to 4660

Return type:

None

set_power_state(power_state, fqdn=None)

Set the power_state of the component.

TODO:

Power state should be set in the component mananger and then the device updated. Current design sets the component power state from the device component_state_changed callback. This should be corrected

Parameters:

power_state (PowerState) – the value of PowerState to be set.
fqdn (Optional[str]) – the fqdn of the component’s device.

Raises:

ValueError – fqdn not found

Return type:

None

standby(task_callback=None)

Submit the _standby method.

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

Parameters:: task_callback (Optional[Callable]) – Update task state, defaults to None
Return type:: tuple[TaskStatus, str]
Returns:: a task status and response message

start_acquisition(argin, task_callback=None)

Submit the start acquisition method.

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

Parameters:: argin (str) – json dictionary with optional keywords

start_time - (str) start time
delay - (int) delay start

Parameters:: task_callback (Optional[Callable]) – Update task state, defaults to None
Return type:: tuple[TaskStatus, str]
Returns:: a task status and response message

start_beamformer(start_time, duration, subarray_beam_id, scan_id)

Start the beamformer at the specified time.

Parameters:

start_time (str) – time at which to start the beamformer, defaults to 0
duration (float) – duration for which to run the beamformer, defaults to -1 (run forever)
subarray_beam_id (int) – ID of the subarray beam to start. Default = -1, all
scan_id (int) – ID of the scan which is started.

Return type:

None

start_communicating()

Establish communication with the station components.

Return type:: None

property static_delays: list[float]

Get static time delay correction.

Array of one value per antenna/polarization (32 per tile), in range +/-124. Delay in samples (positive = increase the signal delay) to correct for static delay mismathces, e.g. cable length.

Returns:: Array of one value per antenna/polarization (32 per tile)

stop_beamformer()

Stop the beamformer.

Return type:: None

stop_communicating()

Break off communication with the station components.

Return type:: None

stop_data_transmission()

Stop data transmission for send_channelised_data_continuous.

Return type:: None

stop_integrated_data()

Stop the integrated data.

Return type:: None

subscribe_to_attributes(task_callback=None, task_abort_event=None)

Subscribe to attributes of interest.

This will form subscriptions to attributes on subdevices MccsTile and MccsSubrack if attribute not already subscribed.

Parameters:

task_callback (Optional[Callable]) – Update task state, defaults to None
task_abort_event (Optional[Event]) – Abort the task

Return type:

None

sysref_present_summary()

Get summary of SYSREF presence.

Return type:: bool
Returns:: TRUE if SYSREF is present in all tiles

property test_generator_active: bool

Get test generator state.

Returns:: True if at least one TPM uses test generator

property test_list: list[str]

Get list of self-check tests available.

Returns:: list of self-check tests available.

property test_logs: str

Get logs of most recently run self-check test set.

Returns:: logs of most recently run self-check test set.

property test_report: str

Get report of most recently run self-check test set.

Returns:: report of most recently run self-check test set.

tile_programming_state()

Get TPM programming state.

Return type:: list[str]
Returns:: list of programming state for all TPMs

trigger_adc_equalisation(task_callback=None)

Submit the trigger adc equalisation method.

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

Parameters:: task_callback (Optional[Callable]) – Update task state, defaults to None
Return type:: tuple[TaskStatus, str]
Returns:: a task status and response message

class SpsStationHealthModel(subrack_fqdns, tile_fqdns, health_changed_callback, thresholds=None)

A health model for a Sps station.

__init__(subrack_fqdns, tile_fqdns, health_changed_callback, thresholds=None)

Initialise a new instance.

Parameters:

subrack_fqdns (Sequence[str]) – the FQDNs of this station’s subracks
tile_fqdns (Sequence[str]) – the FQDNs of this station’s tiles
health_changed_callback (HealthChangedCallbackProtocol) – callback to be called whenever there is a change to this this 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 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

subrack_health_changed(subrack_fqdn, subrack_health)

Handle a change in subrack health.

Parameters:

subrack_fqdn (str) – the FQDN of the tile whose health has changed
subrack_health (Optional[HealthState]) – the health state of the specified tile, or None if the subrack’s admin mode indicates that its health should not be rolled up.

Return type:

None

tile_health_changed(tile_fqdn, tile_health)

Handle a change in tile health.

Parameters:

tile_fqdn (str) – the FQDN of the tile whose health has changed
tile_health (Optional[HealthState]) – the health state of the specified tile, or None if the tile’s admin mode indicates that its health should not be rolled up.

Return type:

None

class SpsStationHealthRules(*args, **kwargs)

A class to handle transition rules for station.

property default_thresholds: dict[str, float]

Get the default thresholds for this device.

Returns:: the default thresholds

degraded_rule(subrack_healths, tile_healths)

Test whether DEGRADED is valid for the station.

Parameters:

subrack_healths (dict[str, HealthState | None]) – dictionary of subrack healths
tile_healths (dict[str, HealthState | None]) – dictionary of tile healths

Return type:

bool

Returns:

True if DEGRADED is a valid state

failed_rule(subrack_healths, tile_healths)

Test whether FAILED is valid for the station.

Parameters:

subrack_healths (dict[str, HealthState | None]) – dictionary of subrack healths
tile_healths (dict[str, HealthState | None]) – dictionary of tile healths

Return type:

bool

Returns:

True if FAILED is a valid state

healthy_rule(subrack_healths, tile_healths)

Test whether OK is valid for the station.

Parameters:

subrack_healths (dict[str, HealthState | None]) – dictionary of subrack healths
tile_healths (dict[str, HealthState | None]) – dictionary of tile healths

Return type:

bool

Returns:

True if OK is a valid state

unknown_rule(subrack_healths, tile_healths)

Test whether UNKNOWN is valid for the station.

Parameters:

subrack_healths (dict[str, HealthState | None]) – dictionary of subrack healths
tile_healths (dict[str, HealthState | None]) – dictionary of tile healths

Return type:

bool

Returns:

True if UNKNOWN is a valid state

class SpsStationObsStateModel(logger, obs_state_changed_callback)

An observation state model for a station.

__init__(logger, obs_state_changed_callback)

Initialise a new instance.

Parameters:

logger (Logger) – a logger for this model to use
obs_state_changed_callback (Callable[[ObsState], None]) – callback to be called when there is a change to this model’s evaluated observation state.

is_configured_changed(is_configured)

Handle a change in whether the station is configured.

Parameters:: is_configured (bool) – whether the station is configured
Return type:: None

update_obs_state()

Update the observation state, ensuring that the callback is called.

Return type:: None

class SpsStationSelfCheckManager(logger, tile_trls, subrack_trls, daq_trl, component_manager)

A class for initiating station self-check procedures.

__init__(logger, tile_trls, subrack_trls, daq_trl, component_manager)

Initialise a new instance.

Parameters:

logger (Logger) – a logger for this model to use.
tile_trls (list[str]) – trls of tiles the station has.
subrack_trls (list[str]) – trls of subracks the station has.
daq_trl (str) – trl of the daq the station has.
component_manager (SpsStationComponentManager) – SpsStation component manager under test.

run_test(test_name, count)

Run a specific test, with an optional count parameter to run multiple times.

Parameters:

test_name (str) – name of the test to run.
count (int) – number of times to run test.

Return type:

list[TestResult]

Returns:

results of the tests.

run_tests()

Run all self check tests.

Return type:: list[TestResult]
Returns:: results of the test set.