Station subpackage

This subpackage implements station functionality for MCCS.

class SpsStation(*args, **kwargs)

An implementation of an SPS Station Tango device for MCCS.

AcquireDataForCalibration(first_channel, last_channel, start_time=None, daq_mode='TCC', nof_samples=1835008)

Start acquiring data for calibration.

A JSON string containing the keys:

Parameters:

  • first_channel (int) – the first channel to acquire, in the range 0-511

  • last_channel (int) – the last channel to acquire, in the range 0-511

  • start_time (Optional[str]) – optional start acquisition time in ISO9601 format defaults to ‘now’

  • daq_mode (str) – the DAQ mode to use for acquisition, either “TCC” or “DSS” defaults to “TCC”

  • nof_samples (int) – the number of samples to acquire defaults to 1835008 (corresponding to 1 subband of data at 8kHz resolution)

Return type:

TaskFunctionType

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/ci-1")
>>> argin = json.dumps({"first_channel": 64, "last_channel": 448})
>>> dp.command_inout("AcquireDataForCalibration", argin)
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)
BandpassdaqTRL(value)

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

Parameters:

value (str) – The new DAQ TRL.

Return type:

None

BeamformerRunningForChannels(channel_groups=None)

Check whether the beamformer is running for the given channel groups.

Parameters:

channel_groups (Optional[list[int]]) – list of channel groups to check. If None, check all channel groups.

  • channel_groups - (list) List of channel groups

Return type:

bool

Returns:

Whether the beamformer is running

Example:

>>> dp = tango.DeviceProxy("mccs/station/01")
>>> dict = {"channel_groups": [0,1,4,5]}
>>> jstr = json.dumps(dict)
>>> running = dp.command_inout("BeamformerRunningForChannels", jstr)
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)
ConfigureStationForCalibration(daq_config)

Configure the station for calibration.

Parameters:

daq_config (str) – a JSON string containing optional additions/overrides to default DAQ configuration.

Return type:

TaskFunctionType

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/ci-1")
>>> json_arg = json.dumps({"description" : "Calibration data for s8-2"})
>>> dp.command_inout("ConfigureStationForCalibration", json_arg)
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. 32 bit integer, with each bit representing an input channel. Default: all if at least 1 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)
DescribeTest(test_name)

Fetch the docstring of a given test.

Parameters:

test_name (str) – the name of the test you wish to fetch the details of.

Return type:

str

Returns:

the docstring of a given test.

Initialise()

Initialise the station.

Return type:

TaskFunctionType

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")
LMCdaqTRL(value)

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

Parameters:

value (str) – The new DAQ TRL.

Return type:

None

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)
LoadCalibrationCoefficientsForChannels(calibration_coefficients)

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:

calibration_coefficients (list[float]) – list comprises:

  • first_channel - (int) is the first channel in the block of channels

    to which the coefficients will be applied.

  • calibration_coefficients - [array] a flatteded tridimensional complex

    array comprising calibration_coefficients[channel, antenna, pol], 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.

    • antenna - (int) antenna index ranging from 0 to 255, in

      (tile, antenna) order

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

TaskFunctionType

Returns:

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

Example:

>>> first_channel = 8
>>> complex_coefficients = [[complex(3.4, 1.2), complex(2.3, 4.1),
>>>            complex(4.6, 8.2), complex(6.8, 2.4)]]*256*8
>>> 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(first_channel))
>>> input = list(itertools.chain.from_iterable(coefficients))
>>> dp = tango.DeviceProxy("mccs/tile/01")
>>> dp.command_inout("LoadCalibrationCoefficientsForChannels", 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. In order of antenna EEP.

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)
OnWorkaround(value)

Set the status of the OnWorkaroundFlag.

Parameters:

value (bool) – The new status of the OnWorkaroundFlag.

Return type:

None

ReInitialise(start_bandpasses=None, global_reference_time=None)

Reinitialise the station with overridable defaults.

Return type:

TaskFunctionType

Returns:

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

A json string adhering to the initialise schema:

Parameters:

  • start_bandpasses (Optional[bool]) – whether to start the bandpasses after initialisation defaults to true

  • global_reference_time (Optional[str]) – the global synchronization time, in ISO9660 format or “” to use current time, defaults to “”

{
    "$schema": "https://json-schema.org/draft/2020-12/schema",
    "$id": "https://skao.int/SpsStation_Initialise.json",
    "title": "SpsStation Initialise schema",
    "description": "Schema for SpsStations's Initialise command",
    "type": "object",
    "properties": {
        "global_reference_time": {
            "description": "Common global reference time for all TPMs, needs to be some time in the last 2 weeks. If not provided, 8am on the most recent Monday AWST will be used.",
            "type": "string"
        },
        "start_bandpasses": {
            "description": "Whether to configure the TPMs to send integrated channel data to the bandpass DAQ.",
            "type": "boolean"
        }
    },
    "required": []
}
Example:
>>> dp = tango.DeviceProxy("mccs/station/001")
>>> dp.command_inout("Initialise")
ResetCspIngest()

Reset link for beam data packets to CSP to defaults.

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.

RunTest(test_name, count=1)

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

A json-string containing a required ‘test_name’, and optional

‘count’

Parameters:

  • test_name (str) – the name of the test to run, should be one of the available tests in testList attribute.

  • count (int) – the number of times to run the test, defaults to 1 if not specified.

Return type:

TaskFunctionType

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:

TaskFunctionType

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

  • force - (bool) Whether or not to cancel ongoing data requests.

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("SetBeamFormerTable", 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 (list[int]) – list of 512 values, one per channel. this will apply to all Tiles in this station.

Return type:

TaskFunctionType

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(start_time=None, delay=None)

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

If a start time isn’t given, it will default to ‘now’.

Parameters:

  • start_time (Optional[str]) – Start acquisition time in ISO9601 format

  • delay (Optional[int]) – Optional delay in seconds before starting acquisition

Return type:

TaskFunctionType

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(start_time=None, duration=-1, channel_groups=None, scan_id=0)

Start the beamformer at the specified time delay.

A json dictionary with optional keywords:

Parameters:

  • start_time (Optional[str]) – (str, ISO UTC time) start time

  • duration (int) – (int) if > 0 is a duration in seconds if < 0 run forever

  • channel_groups (Optional[list[int]]) – (list(int)) : list of channel groups to be started Command affects only beamformed channels for given groups Default: all channels

  • scan_id (int) – (int) The unique ID for the started scan. Default 0

Return type:

TaskFunctionType

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 for all channel groups.

Return type:

TaskFunctionType

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")
StopBeamformerForChannels(channel_groups=None)

Stop the beamformer for given channel groups.

A json dictionary with optional keywords:

Parameters:

channel_groups (Optional[list[int]]) – (list(int)) : list of channel groups to be started Command affects only beamformed channels for given groups Default: all channels

Return type:

TaskFunctionType

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 = {"channel_groups": [0,1,4] }
>>> jstr = json.dumps(dict)
>>> dp.command_inout("StopBeamformerForChannels", jstr)
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(target_adc=17.0, bias=0.0)

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

A JSON string containing:

Parameters:

  • target_adc (float) – the expected average power received by antennas in ADU units. Has an input minimum of 0, but in code its limited to 4.2e-7 (corresponds to the maxiumum output of 31.75 dB). There is no maximum value, however, 40 ADUs will result in 0 dB with no bias and 1600 ADUs will result in 0 dB with the maxiumum bias allowed of 32dB. Defaults to 17.

  • bias (float) – user specifed bias in dB added to the antenna preadu levels. Bias input value rounded as part of value sanitation and as a result it increases in steps of 0.25. Ranges from -32 to 32 with default 0.

Return type:

TaskFunctionType

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")
>>> json_arg = json.dumps({"target_adc" : "18"})
>>> dp.command_inout("TriggerAdcEqualisation", json_arg)
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

beamformerRegions()

Get beamformer region table.

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

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

Return type:

Optional[list[int]]

Returns:

list of up to 8*48 values

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:

Optional[list[int]]

Returns:

list of up to 7*48 values

boardTemperaturesSummary()

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

Return type:

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

cspIngestConfig()

Report the CspIngest configuration in use for this station.

Return type:

str

Returns:

json string with CspIngest configuration.

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

cspSpeadFormat(spead_format)

Set CSP SPEAD format.

CSP format is: AAVS for the format used in AAVS2-AAVS3 system, using a reference Unix time specified in the header. SKA for the format defined in SPS-CBF ICD, based on TAI2000 epoch.

Parameters:

spead_format (str) – format used in CBF SPEAD header: “AAVS” or “SKA”

Return type:

None

daqPath()

Get DAQ data path.

Return type:

str

Returns:

Path where DAQ data files are stored.

dataReceivedResult()

Read the result of the receiving of data.

Return type:

Optional[tuple[str, str]]

Returns:

A tuple containing the data mode of transmission and a json string with any additional data about the data such as the file name.

executeAsync(execute_async)

Set whether to execute MccsTile methods asynchronously.

We can either execute MccsTile methods in serial or sequence, this attribute dictates which.

Parameters:

execute_async (bool) – whether to execute MccsTile methods asynchronously.

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:

Optional[list[float]]

Returns:

minimum, average, maximum board temperatures, in deg Celsius

globalReferenceTime(reference_time)

Set the global global synchronization timestamp.

Parameters:

reference_time (str) – the synchronization time, in ISO9660 format, or “”

Raises:

ValueError – if specified time not in ISO format or < TAI2000

Return type:

None

healthModelParams(argin)

Set the params for health transition rules.

These are the thresholds for the old health model.

Parameters:

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

  • argin – JSON-string of dictionary of health thresholds

Return type:

None

healthReport()

Get the health report.

Return type:

str

Returns:

the health report.

healthThresholds(argin)

Set the params for health transition rules.

Default health thresholds:

“pps_delta_degraded”: 4,

int: PPS delay spread in 1.25ns units that triggers degraded health.

“pps_delta_failed”: 9,

int: PPS delay spread in 1.25ns units that triggers failed health.

“subracks”: (f2f, d2f, d2d),
tuple(int, int, int): Number of subracks failed before health failed,

Number of subracks degraded before health failed, Number of subracks degraded before health degraded

“tiles”: (f2f, d2f, d2d),
tuple(int, int, int): Number of tiles failed before health failed,

Number of tiles degraded before health failed, Number of tiles degraded before health degraded.

Parameters:

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

Return type:

None

init_device()

Initialise the device.

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

keepTestData(keep_test_data)

Set whether to keep test data.

We can either keep or discard test data after tests are run.

Parameters:

keep_test_data (bool) – whether to keep test data.

Return type:

None

lastPointingDelays()

Return last pointing delays applied to the tiles.

Values are initialised to 0.0 if they haven’t been set. These values are in antenna EEP order.

Return type:

list

Returns:

last pointing delays applied to the tiles.

pllLockedSummary()

Get summary of PLL locked status for all tiles.

Return type:

bool

Returns:

True if PLL is locked to reference in all tiles

pointingDelays()

Read the last pointing delays received from HW.

Return type:

ndarray

Returns:

the last pointing delays received from HW.

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

ppsDelaySpread()

Get difference between maximum and minimum delays.

Returns the difference between max and min delays used for this station. This can be used to detect “drifting” delays.

Return type:

int

Returns:

Difference between maximum and minimum delays.

ppsDelaySummary()

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

Return type:

Optional[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 nanoseconds (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

useNewHealthModel(argin)

Set a flag indicating whether this station is using the new health model.

Parameters:

argin (bool) – a flag indicating whether this station is currently using the new health model.

Return type:

None

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, lmc_daq_trl, bandpass_daq_trl, sdn_first_interface, sdn_gateway, csp_ingest_ip, channeliser_rounding, csp_rounding, antenna_config_uri, start_bandpasses_in_initialise, bandpass_integration_time, logger, communication_state_changed_callback, component_state_changed_callback, tile_health_changed_callback, subrack_health_changed_callback, on_workaround_flag=False, event_serialiser=None)

A component manager for a station.

__init__(station_id, subrack_fqdns, tile_fqdns, lmc_daq_trl, bandpass_daq_trl, sdn_first_interface, sdn_gateway, csp_ingest_ip, channeliser_rounding, csp_rounding, antenna_config_uri, start_bandpasses_in_initialise, bandpass_integration_time, logger, communication_state_changed_callback, component_state_changed_callback, tile_health_changed_callback, subrack_health_changed_callback, on_workaround_flag=False, event_serialiser=None)

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

  • lmc_daq_trl (str) – The TRL of this Station’s DAQ Receiver for general LMC use. Could be empty if the device property is not set.

  • bandpass_daq_trl (str) – The TRL of this Station’s DAQ Receiver for bandpasses. Could be empty if the device property is not set.

  • sdn_first_interface (IPv4Interface) – 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 (Optional[IPv4Address]) – IP address of the SDN gateway, or None if the network has no gateway.

  • csp_ingest_ip (Optional[IPv4Address]) – IP address of the CSP ingest for this station.

  • channeliser_rounding (Optional[list[int]]) – The channeliser rounding to use for this station.

  • csp_rounding (int) – The CSP rounding to use for this station. An integer value between 0 and 7. Currently the underlying library accepts a list of 384 values (one for each coarse channel sent to CSP) but it actually only uses the first one of these. Until it is updated to support a full list, we restrict this interface to one integer.

  • antenna_config_uri (Optional[list[str]]) – location of the antenna mapping file

  • start_bandpasses_in_initialise (bool) – whether to start bandpasses in initialise.

  • bandpass_integration_time (float) – the integration time for channelised data capture started in initialise.

  • logger (Logger) – the logger to be used by this object.

  • 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

  • on_workaround_flag (bool) – whether to enable the workaround

  • event_serialiser (Optional[EventSerialiser]) – the event serialiser to be used by this object.

acquire_data_for_calibration(first_channel, last_channel, start_time=None, daq_mode='TCC', nof_samples=1835008, task_callback=None, task_abort_event=None)

Acquire data for calibration.

Parameters:

  • start_time (Optional[str]) – UTC Time for start sending data.

  • first_channel (int) – first channel to calibrate for

  • last_channel (int) – last channel to calibrate for

  • daq_mode (str) – the correlator mode to start, xGPU or TCC.

  • nof_samples (int) – the number of samples to integrate, only variable in TCC.

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

  • task_abort_event (Optional[Event]) – Check for abort, defaults to None

Return type:

None

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:

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.

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:

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.

property beamformer_regions: list[list[int]]

Get beamformer region table.

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

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

Returns:

list of up to 8*48 values

beamformer_running_for_channels(channel_groups)

Check if the beamformer is running in a list of channel blocks.

Parameters:

channel_groups (Optional[list[int]]) – List of channel blocks to check

Return type:

bool

Returns:

True if the beamformer is running

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:

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

cleanup()

Cleanup resources held by the component manager.

This includes cleaning up resources held by all sub-component managers.

Return type:

None

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:

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.

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:

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.

configure_station_for_calibration(task_callback=None, task_abort_event=None, **daq_config)

Configure station for calibration.

Parameters:

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

  • task_abort_event (Optional[Event]) – Check for abort, defaults to None

  • daq_config (Any) – any extra config to configure DAQ with

Return type:

None

configure_test_generator(argin)

Distribute to tiles command configure_test_generator.

Parameters:

argin (str) – Json encoded parameter List

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.

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

Get CSP SPEAD format.

CSP format is: AAVS for the format used in AAVS2-AAVS3 system, using a reference Unix time specified in the header. SKA for the format defined in SPS-CBF ICD, based on TAI2000 epoch.

Returns:

CSP Spead format. AAVS or SKA

describe_test(test_name)

Return the doc string of a given self-check test.

Parameters:

test_name (str) – name of the test.

Return type:

str

Returns:

the doc string of a given self-check test.

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:

Optional[list[float]]

Returns:

minimum, average and maximum of FPGAs temperatures

property global_reference_time: str

Return the UTC time used as global synchronization time.

Returns:

UTC time in ISOT format used as global synchronization time

initialise(start_bandpasses=None, global_reference_time=None, task_callback=None, task_abort_event=None)

Initialise this station.

The order to turn a station on is: subrack, then tiles

Parameters:

  • start_bandpasses (Optional[bool]) – Whether to configure TPMs to send integrated data.

  • global_reference_time (Optional[str]) – Common global reference time for all TPMs, needs to be some time in the last 2 weeks. If not provided, 8am on the most recent Monday AWST will be used.

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

  • task_abort_event (Optional[Event]) – Abort the task

Return type:

None

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

property keep_test_data: bool

Get whether test data will be kept from the self_check_manager.

Returns:

whether the test data will be kept.

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_calibration_coefficients_for_channels(calibration_coefficients, task_callback=None, task_abort_event=None)

Load calibration coefficients for all antennas and specific channels.

Parameters:

  • calibration_coefficients (list[float]) – a tridimensional complex array of coefficients, indexed by channels, antennas, polarizations, flattened into a list. First element is first channel. Dimension of antennas is 256, in tile antenna order, Dimension of polarizations is 4.

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

  • task_abort_event (Optional[Event]) – Check for abort, defaults to None

Raises:

ValueError – when the number of coefficients is insufficient or not a multiple of 2048 (i.e. 8 coefficients per channel per antenna)

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. First element in the array is beam_index: specifies which beam is desired (range 0-47, limited to 7 in the current firmware) Other elements are pairs of delay and delay rate for each antenna, in EEP antenna order. delay_list[2*eep_index -1] is delay, in seconds, and delay_list[2*eep] is the delay rate, in second/second, for antenna with given EEP index (range 1-256)

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

property pps_delay_spread: int

Get PPS delay delta.

Returns the difference between the maximum and minimum delays so that users can track an observed pps drift. Returns a result in samples, each sample is 1.25ns.

Returns:

Maximum delay difference between tiles in samples.

pps_delay_summary()

Get summary of PPS delays.

Return type:

Optional[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(count, test_name, task_callback=None, task_abort_event=None)

Run a specific self check test.

Parameters:

  • count (int) – how many times to run the test.

  • test_name (str) – which test to run.

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

  • task_abort_event (Optional[Event]) – Check for abort, defaults to None

Return type:

None

self_check(task_callback=None, task_abort_event=None)

Run all self check tests.

Parameters:

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

  • task_abort_event (Optional[Event]) – Check for abort, defaults to None

Return type:

None

send_data_samples(argin, force=False)

Front end for send_xxx_data methods.

Parameters:

  • argin (str) – Json encoded parameter List

  • force (bool) – whether to cancel ongoing requests first.

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.

set_beamformer_table(beamformer_table)

Set the frequency regions to be beamformed into each beam.

Parameters:

beamformer_table (list[list[int]]) – a list encoding up to 48 regions, with each region corresponding to 8 channels. Entry items specify:

  • start physical channel

  • beam_index: subarray beam used for this region, range [0:48)

  • subarray_id: ID of the subarray [1:48]

  • subarray_logical_channel: Logical channel in the subarray

    it is the same for all (sub)stations in the subarray Defaults to station logical channel

  • subarray_beam_id: ID of the subarray beam

    Defaults to beam index

  • substation_ID: ID of the substation

    Defaults to 0 (no 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.

set_channeliser_rounding(channeliser_rounding, task_callback=None, task_abort_event=None)

Set the channeliserRounding in all Tiles.

Parameters:

  • channeliser_rounding (np.ndarray | list[int]) – the number of LS bits dropped in each channeliser frequency channel.

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

  • task_abort_event (Optional[threading.Event]) – Check for abort, defaults to None

Return type:

None

set_csp_ingest(dst_ip, src_port, dst_port)

Configure last tile 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:

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

Returns:

tuple containing Resultcode and message.

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:

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.

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

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

  • lock_mode (bool) – whether this call should lock integrated-mode auto refresh.

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.

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(start_time=None, delay=2, task_callback=None, task_abort_event=None)

Start acquisition using slow command.

Parameters:

  • start_time (Optional[str]) – the time at which to start data acquisition, defaults to None

  • delay (Optional[int]) – delay start, defaults to 2

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

  • task_abort_event (Optional[Event]) – Check for abort, defaults to None

Return type:

None

start_adcs()

Start ADCs on all tiles.

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.

start_beamformer(start_time=None, duration=-1, channel_groups=None, scan_id=0, task_callback=None, task_abort_event=None)

Start the beamformer at the specified time.

Parameters:

  • start_time (Optional[str]) – time at which to start the beamformer, defaults to 0

  • duration (int) – duration for which to run the beamformer, defaults to -1 (run forever)

  • channel_groups (Optional[list[int]]) – Channel groups to which the command applies.

  • scan_id (int) – ID of the scan which is started.

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

  • task_abort_event (Optional[Event]) – Check for abort, defaults to None

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 nanoseconds (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_adcs()

Stop ADCs on all tiles.

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.

stop_beamformer(channel_groups, task_callback=None, task_abort_event=None)

Stop the beamformer.

Parameters:

  • channel_groups (Optional[list[int]]) – Channel groups to which the command applies.

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

  • task_abort_event (Optional[Event]) – Check for abort, defaults to None

Return type:

None

stop_beamformer_for_channels(task_callback=None, task_abort_event=None, *, channel_groups=None)

Run the stop_beamformer method.

Parameters:

  • channel_groups (Optional[list[int]]) – Channel groups to which the command applies.

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

  • task_abort_event (Optional[Event]) – Check for abort, defaults to None

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:

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.

stop_integrated_data()

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.

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(target_adc=17.0, bias=0.0, task_callback=None, task_abort_event=None)

Equalise adc using slow command.

Parameters:

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

  • task_abort_event (Optional[Event]) – Check for abort, defaults to None

  • target_adc (float) – adc value in ADU units. Defaults to 17.

  • bias (float) – user specifed bias in dB added to the antenna preadu levels. Defaults to 0.

Return type:

None

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:

tuple[HealthState, str]

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, station_state)

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

  • station_state (dict[str, Any | None]) – dictionary of station state attributes.

Return type:

tuple[bool, str]

Returns:

True if DEGRADED is a valid state, along with a text report.

failed_rule(subrack_healths, tile_healths, station_state)

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

  • station_state (dict[str, Any | None]) – dictionary of station state attributes.

Return type:

tuple[bool, str]

Returns:

True if FAILED is a valid state, along with a text report.

healthy_rule(subrack_healths, tile_healths, station_state)

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

  • station_state (dict[str, Any | None]) – dictionary of station state attributes.

Return type:

tuple[bool, str]

Returns:

True if OK is a valid state, along with a text report.

unknown_rule(subrack_healths, tile_healths, station_state)

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

  • station_state (dict[str, Any | None]) – dictionary of station state attributes.

Return type:

tuple[bool, str]

Returns:

True if UNKNOWN is a valid state, along with a text report.

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.

property keep_test_data: bool

Return whether or not to keep any test data.

Returns:

whether or not to keep any test data.

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.