Station Device

This module implements the MCCS station device.

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.

main(*args, **kwargs)

Entry point for module.

Parameters:

  • args (str) – positional arguments

  • kwargs (str) – named arguments

Return type:

int

Returns:

exit code