Tile Device

This module implements the MCCS Tile device.

class MccsTile(*args, **kwargs)

An implementation of a Tile Tango device for MCCS.

ApplyCalibration(argin)

Load the calibration coefficients at the specified time delay.

Parameters:

argin (str) – switch time, in ISO formatted time

Return type:

tuple[list[ResultCode], list[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", "")

class ApplyCalibrationCommand(component_manager, logger=None)

Class for handling the ApplyCalibration(argin) command.

__init__(component_manager, logger=None)

Initialise a new ApplyCalibrationCommand instance.

Parameters:

• component_manager (TileComponentManager) – the device to which this command belongs.

• logger (Optional[Logger]) – a logger for this command to use.

do(*args, **kwargs)

Implement MccsTile.ApplyCalibration() command functionality.

Parameters:

• args (Any) – switch time

• kwargs (Any) – unspecified keyword arguments. This should be empty and is provided for type hinting only

Return type:

tuple[ResultCode, str]

Returns:

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

ApplyPointingDelays(argin)

Apply the pointing delays at the specified time delay.

Parameters:

argin (str) – time delay (default = 0)

Return type:

tuple[list[ResultCode], list[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("ApplyPointingDelays", "")

class ApplyPointingDelaysCommand(component_manager, logger=None)

Class for handling the ApplyPointingDelays(argin) command.

__init__(component_manager, logger=None)

Initialise a new ApplyPointingDelayommand instance.

Parameters:

• component_manager (TileComponentManager) – the device to which this command belongs.

• logger (Optional[Logger]) – a logger for this command to use.

do(*args, **kwargs)

Implement MccsTile.ApplyPointingDelays() command functionality.

Parameters:

• args (Any) – load time

• kwargs (Any) – unspecified keyword arguments. This should be empty and is provided for type hinting only

Return type:

tuple[ResultCode, str]

Returns:

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

Configure(argin)

Configure the tile device attributes.

Parameters:

argin (str) – the configuration for the device in stringified json format

Return type:

None

Configure40GCore(argin)

Configure 40g core_id with specified parameters.

Parameters:

argin (str) –

json dictionary with only optional keywords:

• core_id - (int) core id

• arp_table_entry - (int) ARP table entry ID

• source_mac - (int) mac address

• source_ip - (string) IP dot notation.

• source_port - (int) source port

• destination_ip - (string) IP dot notation

• destination_port - (int) destination port

• netmask - (int) 40g (science data) subnet mask

• gateway_ip - (int) IP address of 40g (science) subnet gateway

Return type:

tuple[list[ResultCode], list[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 = {"core_id":2, "arp_table_entry":0, "source_mac":0x62000a0a01c9,
            "source_ip":"10.0.99.3", "source_port":4000,
            "destination_ip":"10.0.99.3", "destination_port":5000}
>>> jstr = json.dumps(dict)
>>> dp.command_inout("Configure40GCore", jstr)

class Configure40GCoreCommand(component_manager, logger=None)

Class for handling the Configure40GCore() command.

This command takes as input a JSON string that conforms to the following schema:

{
    "$schema": "https://json-schema.org/draft/2020-12/schema",
    "$id": "https://skao.int/MccsTile_Configure40gCore.json",
    "title": "MccsTile Configure40gCore schema",
    "description": "Schema for MccsTile's Configure40gCore command",
    "type": "object",
    "properties": {
        "core_id": {
            "description": "Core ID",
            "type": "integer"
        },
        "arp_table_entry": {
            "description": "ARP table entry",
            "type": "integer"
        },
        "source_mac": {
            "description": "Source MAC address",
            "type": "integer"
        },
        "source_ip": {
            "description": "Source IP address",
            "type": "string",
            "format": "ipv4"
        },
        "source_port": {
            "description": "Source port",
            "type": "integer",
            "minimum": 1
        },
        "destination_ip": {
            "description": "Destinate IP address",
            "type": "string",
            "format": "ipv4"
        },
        "destination_port": {
            "description": "Destination port",
            "type": "integer",
            "minimum": 1
        },
        "rx_port_filter": {
            "description": "Receive ports filter",
            "type": "integer",
            "minimum": 1
        },
        "netmask": {
            "description": "Integer netmask for the 40g (science data) subnet",
            "type": "integer",
            "minimum": 0,
            "maximum": 4294967296
        },
        "gateway_ip": {
            "description": "Integer IP address of the 40g (science data) subnet gateway",
            "minimum": 0,
            "maximum": 4294967296
        }
    },
    "additionalProperties": false
}

__init__(component_manager, logger=None)

Initialise a new Configure40GCoreCommand instance.

Parameters:

• component_manager (TileComponentManager) – the device to which this command belongs.

• logger (Optional[Logger]) – a logger for this command to use.

do(*args, **kwargs)

Implement MccsTile.Configure40GCore() command functionality.

Parameters:

• args (Any) – Positional arguments. This should be empty and is provided for type hinting purposes only.

• kwargs (Any) – keyword arguments unpacked from the JSON argument to the command.

Return type:

tuple[ResultCode, str]

Returns:

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

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

class ConfigureIntegratedBeamDataCommand(component_manager, logger=None)

Class for handling the ConfigureIntegratedBeamData() command.

This command takes as input a JSON string that conforms to the following schema:

{
    "$schema": "https://json-schema.org/draft/2020-12/schema",
    "$id": "https://skao.int/MccsTile_ConfigureIntegratedBeamData.json",
    "title": "MccsTile ConfigureIntegratedBeamData schema",
    "description": "Schema for MccsTile's ConfigureIntegratedBeamData command",
    "type": "object",
    "properties": {
        "integration_time": {
            "description": "Integration time in seconds",
            "type": "number",
            "minimum": 0.0
        },
        "first_channel": {
            "description": "First channel",
            "type": "integer",
            "minimum": 0,
            "maximum": 511
        },
        "last_channel": {
            "description": "Last channel",
            "type": "integer",
            "minimum": 0,
            "maximum": 511
        }
    }
}

__init__(component_manager, logger=None)

Initialise a new ConfigureIntegratedBeamDataCommand instance.

Parameters:

• component_manager (TileComponentManager) – the device to which this command belongs.

• logger (Optional[Logger]) – a logger for this command to use.

do(*args, **kwargs)

Implement MccsTile.ConfigureIntegratedBeamData() commands.

Parameters:

• args (Any) – Positional arguments. This should be empty and is provided for type hinting purposes only.

• kwargs (Any) – keyword arguments unpacked from the JSON argument to the command.

Return type:

tuple[ResultCode, str]

Returns:

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

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

class ConfigureIntegratedChannelDataCommand(component_manager, logger=None)

Class for handling the ConfigureIntegratedChannelData(argin) command.

This command takes as input a JSON string that conforms to the following schema:

{
    "$schema": "https://json-schema.org/draft/2020-12/schema",
    "$id": "https://skao.int/MccsTile_ConfigureIntegratedChannelData.json",
    "title": "MccsTile ConfigureIntegratedChannelData schema",
    "description": "Schema for MccsTile's ConfigureIntegratedChannelData command",
    "type": "object",
    "properties": {
        "integration_time": {
            "description": "Integration time in seconds",
            "type": "number",
            "minimum": 0.0
        },
        "first_channel": {
            "description": "First channel",
            "type": "integer",
            "minimum": 0,
            "maximum": 511
        },
        "last_channel": {
            "description": "Last channel",
            "type": "integer",
            "minimum": 0,
            "maximum": 511
        }
    }
}

__init__(component_manager, logger=None)

Initialise a new ConfigureIntegratedChannelDataCommand instance.

Parameters:

• component_manager (TileComponentManager) – the device to which this command belongs.

• logger (Optional[Logger]) – a logger for this command to use.

do(*args, **kwargs)

Implement MccsTile.ConfigureIntegratedChannelData() commands.

Parameters:

• args (Any) – Positional arguments. This should be empty and is provided for type hinting purposes only.

• kwargs (Any) – keyword arguments unpacked from the JSON argument to the command.

Return type:

tuple[ResultCode, str]

Returns:

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

ConfigureStationBeamformer(argin)

Initialise and start the station beamformer.

Initial configuration of the tile-station beamformer. Optionally set the observed region, Default is 6.25 MHz starting at 150 MHz, and set whether the tile is the first or last in the beamformer chain.

Parameters:

argin (str) –

json dictionary with mandatory keywords:

• start_channel - (int) start channel of the observed region default = 192 (150 MHz)

• n_channels - (int) is the number of channels in the observed region default = 8 (6.25 MHz)

• is_first - (bool) whether the tile is the first one in the station default False

• is_last - (bool) whether the tile is the last one in the station default False

Return type:

tuple[list[ResultCode], list[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 = {“start_channel”:64, “n_channels”:10, “is_first”:True, >> “is_last:True} >> jstr = json.dumps(dict) >> dp.command_inout(“ConfigureStationBeamformer”, jstr)

class ConfigureStationBeamformerCommand(component_manager, logger=None)

Class for handling the ConfigureStationBeamformer() command.

This command takes as input a JSON string that conforms to the following schema:

{
    "$schema": "https://json-schema.org/draft/2020-12/schema",
    "$id": "https://skao.int/MccsTile_ConfigureStationBeamformer.json",
    "title": "MccsTile ConfigureStationBeamformer schema",
    "description": "Schema for MccsTile's ConfigureStationBeamformer command",
    "type": "object",
    "properties": {
        "start_channel": {
            "description": "Start channel",
            "type": "integer",
            "minimum": 2,
            "maximum": 504
        },
        "n_channels": {
            "description": "Number of channels",
            "type": "integer",
            "minimum": 0,
            "maximum": 511
        },
        "is_first": {
            "description": "Is first",
            "type": "boolean"
        },
        "is_last": {
            "description": "Is last",
            "type": "boolean"
        }
    }
}

__init__(component_manager, logger=None)

Initialise a new ConfigureStationBeamformerCommand instance.

Parameters:

• component_manager (TileComponentManager) – the device to which this command belongs.

• logger (Optional[Logger]) – a logger for this command to use.

do(*args, **kwargs)

Implement MccsTile.ConfigureStationBeamformer() commands.

Parameters:

• args (Any) – Positional arguments. This should be empty and is provided for type hinting purposes only.

• kwargs (Any) – keyword arguments unpacked from the JSON argument to the command.

Return type:

tuple[ResultCode, str]

Returns:

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

Raises:

ValueError – if the argin argument does not have the right length / structure

ConfigureTestGenerator(argin)

Set the test signal generator.

Parameters:

argin (str) – json dictionary with keywords:

• tone_frequency: first tone frequency, in Hz. The frequency

  is rounded to the resolution of the generator. If this is not specified, the tone generator is disabled.

• tone_amplitude: peak tone amplitude, normalized to 31.875 ADC

  units. The amplitude is rounded to 1/8 ADC unit. Default is 1.0. A value of -1.0 keeps the previously set value.

• tone_2_frequency: frequency for the second tone. Same

  as ToneFrequency.

• tone_2_amplitude: peak tone amplitude for the second tone.

  Same as ToneAmplitude.

• noise_amplitude: RMS amplitude of the pseudorandom Gaussian

  white noise, normalized to 26.03 ADC units.

• pulse_frequency: frequency of the periodic pulse. A code

  in the range 0 to 7, corresponding to (16, 12, 8, 6, 4, 3, 2) times the ADC frame frequency.

• pulse_amplitude: peak amplitude of the periodic pulse, normalized

  to 127 ADC units. Default is 1.0. A value of -1.0 keeps the previously set value.

• set_time: time at which the generator is set, for synchronization

  among different TPMs. In UTC ISO format (string)

• adc_channels: list of adc channels which will be substituted with

  the generated signal. It is a 32 integer, with each bit representing an input channel. Default: all if at least q source is specified, none otherwises.

Return type:

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

Returns:

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

Example:

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

class ConfigureTestGeneratorCommand(component_manager, logger=None)

Class for handling the ConfigureTestGenerator() command.

This command takes as input a JSON string that conforms to the following schema:

{
    "$schema": "https://json-schema.org/draft/2020-12/schema",
    "$id": "https://skao.int/MccsTile_ConfigureTestGenerator.json",
    "title": "MccsTile ConfigureTestGenerator schema",
    "description": "Schema for MccsTile's ConfigureTestGenerator command",
    "type": "object",
    "properties": {
        "set_time": {
            "description": "Time to start the generator, in UTC ISO format",
            "type": "string"
        },
        "tone_frequency": {
            "description": "Tone 1 frequency in Hz of DDC 0",
            "type": "number"
        },
        "tone_amplitude": {
            "description": "Tone 1 peak amplitude, normalized to 31.875 ADC units, resolution 0.125 ADU",
            "type": "number"
        },
        "tone_2_frequency": {
            "description": "Tone 2 frequency in Hz of DDC 1",
            "type": "number"
        },
        "tone_2_amplitude": {
            "description": "Tone 2 peak amplitude, normalized to 31.875 ADC units, resolution 0.125 ADU",
            "type": "number"
        },
        "noise_amplitude": {
            "description": "Amplitude of pseudorandom noise normalized to 26.03 ADC units, resolution 0.102 ADU",
            "type": "number"
        },
        "pulse_frequency": {
            "description": "Code for pulse frequency. Range 0 to 7: 16,12,8,6,4,3,2 times frame frequency",
            "type": "integer",
            "minimum": 0,
            "maximum": 7
        },
        "pulse_amplitude": {
            "description": "pulse peak amplitude, normalized to 127.5 ADC units, resolution 0.5 ADU",
            "type": "number"
        },
        "adc_channels": {
            "description": "ADC channels",
            "type": "array",
            "items": {
                "type": "integer"
            }
        }
    }
}

__init__(component_manager, logger=None)

Initialise a new ConfigureTestGeneratorCommand instance.

Parameters:

• component_manager (TileComponentManager) – the device to which this command belongs.

• logger (Optional[Logger]) – a logger for this command to use.

do(*args, **kwargs)

Implement MccsTile.ConfigureTestGenerator() commands.

Parameters:

• args (Any) – Positional arguments. This should be empty and is provided for type hinting purposes only.

• kwargs (Any) – keyword arguments unpacked from the JSON argument to the command.

Return type:

tuple[ResultCode, str]

Returns:

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

DownloadFirmware(argin)

Download the firmware contained in bitfile to all FPGAs on the board.

This should also update the internal register mapping, such that registers become available for use.

Parameters:

argin (str) – can either be the design name returned from GetFirmwareAvailable() command, or a path to a file

Return type:

tuple[list[ResultCode], list[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("DownloadFirmware", "/tmp/firmware/bitfile")

Get40GCoreConfiguration(argin)

Get 40g core configuration for core_id.

This is required to chain up TPMs to form a station.

Parameters:

argin (str) – json dictionary with optional keywords:

• core_id - (int) core id

• arp_table_entry - (int) ARP table entry ID to use

Return type:

str

Returns:

the configuration is a json string describilg a list (possibly empty) Each list entry comprising: core_id, arp_table_entry, source_mac, source_ip, source_port, destination_ip, destination_port, netmask, gateway_ip

Example:

>>> dp = tango.DeviceProxy("mccs/tile/01")
>>> core_id = 2
>>> arp_table_entry = 0
>>> argout = dp.command_inout(Get40GCoreConfiguration, core_id, arp_table_entry)
>>> params = json.loads(argout)

class Get40GCoreConfigurationCommand(component_manager, logger=None)

Class for handling the Get40GCoreConfiguration() command.

This command takes as input a JSON string that conforms to the following schema:

{
    "$schema": "https://json-schema.org/draft/2020-12/schema",
    "$id": "https://skao.int/MccsTile_Get40gCoreConfiguration.json",
    "title": "MccsTile Get40gCoreConfiguration schema",
    "description": "Schema for MccsTile's Get40gCoreConfiguration command",
    "type": "object",
    "properties": {
        "core_id": {
            "description": "Core ID",
            "type": "integer"
        },
        "arp_table_entry": {
            "description": "ARP table entry",
            "type": "integer"
        }
    }
}

__init__(component_manager, logger=None)

Initialise a new Get40GCoreConfigurationCommand instance.

Parameters:

• component_manager (TileComponentManager) – the device to which this command belongs.

• logger (Optional[Logger]) – a logger for this command to use.

do(*args, **kwargs)

Implement MccsTile.Get40GCoreConfiguration() commands.

Parameters:

• args (Any) – Positional arguments. This should be empty and is provided for type hinting purposes only.

• kwargs (Any) – keyword arguments unpacked from the JSON argument to the command.

Return type:

str

Returns:

json string with configuration

Raises:

ValueError – if the argin is an invalid code id

GetArpTable()

Return a dictionary with populated ARP table for all used cores.

40G interfaces use cores 0 (fpga0) and 1(fpga1) and ARP ID 0 for beamformer, 1 for LMC. 10G interfaces use cores 0,1 (fpga0) and 4,5 (fpga1) for beamforming, and 2, 6 for LMC with only one ARP.

Return type:

str

Returns:

a JSON-encoded dictionary of coreId and populated arpID table

Example:

>>> argout = dp.command_inout("GetArpTable")
>>> dict = json.loads(argout)
>>>    {
>>>    "core_id0": [0, 1],
>>>    "core_id1": [0],
>>>    "core_id3": [],
>>>    }

class GetArpTableCommand(component_manager, logger=None)

Class for handling the GetArpTable() command.

__init__(component_manager, logger=None)

Initialise a new GetArpTableCommand instance.

Parameters:

• component_manager (TileComponentManager) – the device to which this command belongs.

• logger (Optional[Logger]) – a logger for this command to use.

do(*args, **kwargs)

Implement MccsTile.GetArpTable() commands.

Parameters:

• args (Any) – unspecified positional arguments. This should be empty and is provided for type hinting only

• kwargs (Any) – unspecified keyword arguments. This should be empty and is provided for type hinting only

Return type:

str

Returns:

a JSON-encoded dictionary of coreId and populated arpID table

GetFirmwareAvailable()

Get available firmware.

Return a dictionary containing the following information for each firmware stored on the board (such as in Flash memory).

For each firmware, a dictionary containing the following keys with their respective values should be provided: ‘design’, which is a textual name for the firmware, ‘major’, which is the major version number, and ‘minor’.

Return type:

str

Returns:

a JSON-encoded dictionary of firmware details

Example:

>>> dp = tango.DeviceProxy("mccs/tile/01")
>>> jstr = dp.command_inout("GetFirmwareAvailable")
>>> dict = json.load(jstr)
{
"firmware1": {"design": "model1", "major": 2, "minor": 3},
"firmware2": {"design": "model2", "major": 3, "minor": 7},
"firmware3": {"design": "model3", "major": 2, "minor": 6},
}

class GetFirmwareAvailableCommand(component_manager, logger=None)

Class for handling the GetFirmwareAvailable() command.

__init__(component_manager, logger=None)

Initialise a new GetFirmwareAvailableCommand instance.

Parameters:

• component_manager (TileComponentManager) – the device to which this command belongs.

• logger (Optional[Logger]) – a logger for this command to use.

do(*args, **kwargs)

Implement MccsTile.GetFirmwareAvailable() command functionality.

Parameters:

• args (Any) – unspecified positional arguments. This should be empty and is provided for type hinting only

• kwargs (Any) – unspecified keyword arguments. This should be empty and is provided for type hinting only

Return type:

str

Returns:

json encoded string containing list of dictionaries

GetRegisterList()

Return a list of descriptions of the exposed firmware (and CPLD) registers.

Return type:

list[str]

Returns:

a list of register names

Example:

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

class GetRegisterListCommand(component_manager, logger=None)

Class for handling the GetRegisterList() command.

__init__(component_manager, logger=None)

Initialise a new GetRegisterListCommand instance.

Parameters:

• component_manager (TileComponentManager) – the device to which this command belongs.

• logger (Optional[Logger]) – a logger for this command to use.

do(*args, **kwargs)

Implement MccsTile.GetRegisterList() command functionality.

Parameters:

• args (Any) – unspecified positional arguments. This should be empty and is provided for type hinting only

• kwargs (Any) – unspecified keyword arguments. This should be empty and is provided for type hinting only

Return type:

list[str]

Returns:

a list of firmware & cpld registers

class InitCommand(*args, **kwargs)

Class that implements device initialisation for the MCCS Tile device.

do(*args, **kwargs)

Initialise the attributes and properties of the MCCS Tile device.

Parameters:

• args (Any) – unspecified positional arguments. This should be empty and is provided for type hinting only

• kwargs (Any) – unspecified keyword arguments. This should be empty and is provided for type hinting only

Return type:

tuple[ResultCode, str]

Returns:

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

Initialise()

Perform all required initialisation.

(switches on on-board devices, locks PLL, performs synchronisation and other operations required to start configuring the signal processing functions of the firmware, such as channelisation and beamforming)

Return type:

tuple[list[ResultCode], list[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("Initialise")

LoadCalibrationCoefficients(argin)

Load the calibration coefficients, but does not apply them.

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

Parameters:

argin (list[float]) – list comprises:

• antenna - (int) is the antenna to which the coefficients will be applied.

• calibration_coefficients - [array] a bidimensional complex array comprising

  calibration_coefficients[channel, polarization], with each element representing a normalized coefficient, with (1.0, 0.0) being the normal, expected response for an ideal antenna.

  • channel - (int) channel is the index specifying the channels at the

    beamformer output, i.e. considering only those channels actually processed and beam assignments.

  • polarization index ranges from 0 to 3.

    • 0: X polarization direct element

    • 1: X->Y polarization cross element

    • 2: Y->X polarization cross element

    • 3: Y polarization direct element

Return type:

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

Returns:

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

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)

class LoadCalibrationCoefficientsCommand(component_manager, logger=None)

Class for handling the LoadCalibrationCoefficients(argin) command.

__init__(component_manager, logger=None)

Initialise a new LoadCalibrationCoefficientsCommand instance.

Parameters:

• component_manager (TileComponentManager) – the device to which this command belongs.

• logger (Optional[Logger]) – a logger for this command to use.

do(argin, *args, **kwargs)

Implement MccsTile.LoadCalibrationCoefficients() commands.

Parameters:

• argin (list[float]) – calibration coefficients

• args (Any) – unspecified positional arguments. This should be empty and is provided for type hinting only

• kwargs (Any) – unspecified keyword arguments. This should be empty and is provided for type hinting only

Return type:

tuple[ResultCode, str]

Returns:

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

Raises:

ValueError – if the argin argument does not have the right length / structure

LoadPointingDelays(argin)

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

The delay_array specifies the delay and delay rate for each antenna. beam_index specifies which beam is desired (range 0-7)

Parameters:

argin (list[float]) – An array containing: beam index, the delay in seconds and the delay rate in seconds/second, for each antenna.

Return type:

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

Returns:

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

Example:

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

class LoadPointingDelaysCommand(component_manager, logger)

Class for handling the LoadPointingDelays(argin) command.

__init__(component_manager, logger)

Initialise a new LoadPointingDelaysCommand instance.

Parameters:

• component_manager (TileComponentManager) – the device to which this command belongs.

• logger (Logger) – the logger to be used by this Command. If not provided, then a default module logger will be used.

do(argin, *args, **kwargs)

Implement MccsTile.LoadPointingDelays() command functionality.

Parameters:

• argin (list[float]) – an array containing a beam index and antenna delays

• args (Any) – unspecified positional arguments. This should be empty and is provided for type hinting only

• kwargs (Any) – unspecified keyword arguments. This should be empty and is provided for type hinting only

Return type:

tuple[ResultCode, str]

Returns:

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

Raises:

ValueError – if the argin argument does not have the right length / structure

ReadAddress(argin)

Read n 32-bit values from address.

Parameters:

argin (list[int]) – [0] = address to read from [1] = number of values to read, default 1

Return type:

list[int]

Returns:

list of values

Example:

>>> dp = tango.DeviceProxy("mccs/tile/01")
>>> reglist = dp.command_inout("ReadAddress", [address, nvalues])

class ReadAddressCommand(component_manager, logger=None)

Class for handling the ReadAddress(argin) command.

__init__(component_manager, logger=None)

Initialise a new ReadAddressCommand instance.

Parameters:

• component_manager (TileComponentManager) – the device to which this command belongs.

• logger (Optional[Logger]) – a logger for this command to use.

do(argin, *args, **kwargs)

Implement MccsTile.ReadAddress() command functionality.

Parameters:

• argin (list[int]) – sequence of length two, containing an address and a value

• args (Any) – unspecified positional arguments. This should be empty and is provided for type hinting only

• kwargs (Any) – unspecified keyword arguments. This should be empty and is provided for type hinting only

Return type:

tuple[ResultCode, str]

Returns:

[values, ]

Raises:

ValueError – if the argin argument has the wrong length or structure

ReadRegister(register_name)

Return the value(s) of the specified register.

Parameters:

register_name (str) – full hyerarchic register name

Return type:

list[int]

Returns:

a list of register values

Example:

>>> dp = tango.DeviceProxy("fpga1./tile/01")
>>> values = dp.command_inout("ReadRegister", "test-reg1")

class ReadRegisterCommand(component_manager, logger=None)

Class for handling the ReadRegister(argin) command.

__init__(component_manager, logger=None)

Initialise a new ReadRegisterCommand instance.

Parameters:

• component_manager (TileComponentManager) – the device to which this command belongs.

• logger (Optional[Logger]) – a logger for this command to use.

do(*args, **kwargs)

Implement MccsTile.ReadRegister() command functionality.

Parameters:

• args (Any) – the register name

• kwargs (Any) – unspecified keyword arguments. This should be empty and is provided for type hinting only

Return type:

list[int]

Returns:

list of register values

Raises:

ValueError – if the name is invalid

SendDataSamples(argin)

Transmit a snapshot containing raw antenna data.

Parameters:

argin (str) – json dictionary with optional keywords:

• data_type - type of snapshot data (mandatory): “raw”, “channel”,

  “channel_continuous”, “narrowband”, “beam”

• start_time - Time (UTC string) to start sending data. Default immediately

• seconds - (float) Delay if timestamp is not specified. Default 0.2 seconds

Depending on the data type: raw:

• sync: bool: send synchronised samples for all antennas, vs. round robin

  larger snapshot from each antenna

channel:

• n_samples: Number of samples per channel, default 1024

• first_channel - (int) first channel to send, default 0

• last_channel - (int) last channel to send, default 511

channel_continuous

• channel_id - (int) channel_id (Mandatory)

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

narrowband:

• frequency - (int) Sky frequency for band centre, in Hz (Mandatory)

• round_bits - (int) Specify whow many bits to round

• n_samples - (int) number of spectra to send

Return type:

tuple[list[ResultCode], list[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 = {"data_type": "raw", "Sync":True, "Seconds": 0.2}
>>> jstr = json.dumps(dict)
>>> dp.command_inout("SendDataSamples", jstr)

class SendDataSamplesCommand(component_manager, logger=None)

Class for handling the SendDataSamples() command.

This command takes as input a JSON string that conforms to the following schema:

{
    "$schema": "https://json-schema.org/draft/2020-12/schema",
    "$id": "https://skao.int/MccsTile_SendDataSamples.json",
    "title": "MccsTile SendDataSamples schema",
    "description": "Schema for MccsTile's SendDataSamples command",
    "type": "object",
    "properties": {
        "data_type": {
            "description": "type of snapshot data (mandatory): raw, channel, channel_continuous, narrowband, beam",
            "type": "string",
            "enum": ["raw", "channel", "channel_continuous", "narrowband", "beam"]
        }
    },
    "required": [ "data_type" ],
    "allOf": [
        {
            "if": {
                "properties": {
                    "data_type": {
                        "const": "channel"
                    }
                }
            },
            "then": {
                "properties": {
                    "first_channel": {
                        "description": "First channel to send",
                        "type": "integer",
                        "minimum": 0,
                        "maximum": 511
                    },
                    "last_channel": {
                        "description": "Last channel to send",
                        "type": "integer",
                        "minimum": 0,
                        "maximum": 511
                    }
                }
            }
        },
        {
            "if": {
                "properties": {
                    "data_type": {
                        "const": "channel_continuous"
                    }
                }
            },
            "then": {
                "properties": {
                    "channel_id": {
                        "description": "Channel ID",
                        "type": "integer",
                        "minimum": 1,
                        "maximum": 511
                    }
                },
                "required": [ "channel_id" ]
            }
        },
        {
            "if": {
                "properties": {
                    "data_type": {
                        "const": "narrowband"
                    }
                }
            },
            "then": {
                "properties": {
                    "frequency": {
                        "description": "Sky frequency of band centre, in Hz",
                        "type": "number",
                        "minimum": 1e6,
                        "maximum": 399e6
                    }
                },
                "required": [ "frequency" ]
            }
        }
    ]
}

__init__(component_manager, logger=None)

Initialise a new SendDataSamplesCommand instance.

Parameters:

• component_manager (TileComponentManager) – the device to which this command belongs.

• logger (Optional[Logger]) – a logger for this command to use.

do(*args, **kwargs)

Implement MccsTile.SendDataSamples() command functionality.

Parameters:

• args (Any) – Positional arguments. This should be empty and is provided for type hinting purposes only.

• kwargs (Any) – keyword arguments unpacked from the JSON argument to the command.

Return type:

tuple[ResultCode, str]

Returns:

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

Raises:

ValueError – if mandatory parameters are missing

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

Returns:

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

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)

class SetBeamFormerRegionsCommand(component_manager, logger=None)

Class for handling the SetBeamFormerRegions(argin) command.

__init__(component_manager, logger=None)

Initialise a new SetBeamFormerRegionsCommand instance.

Parameters:

• component_manager (TileComponentManager) – the device to which this command belongs.

• logger (Optional[Logger]) – a logger for this command to use.

do(argin, *args, **kwargs)

Implement MccsTile.SetBeamFormerRegions() command functionality.

Parameters:

• argin (list[int]) – a region array

• args (Any) – unspecified positional arguments. This should be empty and is provided for type hinting only

• kwargs (Any) – unspecified keyword arguments. This should be empty and is provided for type hinting only

Return type:

tuple[ResultCode, str]

Returns:

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

Raises:

ValueError – if the argin argument does not have the right length / structure

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

• netmask_40g - (int) 40g (science data) subnet mask

• gateway_40g - (int) IP address of 40g (science) subnet gateway

Return type:

tuple[list[ResultCode], list[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”: 4, “destination_ip”: “10.0.1.23”} >> jstr = json.dumps(dict) >> dp.command_inout(“SetLmcDownload”, jstr)

class SetLmcDownloadCommand(component_manager, logger=None)

Class for handling the SetLmcDownload() command.

This command takes as input a JSON string that conforms to the following schema:

{
    "$schema": "https://json-schema.org/draft/2020-12/schema",
    "$id": "https://skao.int/MccsTile_SetLmcDownload.json",
    "title": "MccsTile SetLmcDownload schema",
    "description": "Schema for MccsTile's SetLmcDownload command",
    "type": "object",
    "properties": {
        "mode": {
            "description": "Mode: 1G or 10G",
            "type": "string",
            "enum": [
                "1g",
                "1G",
                "10g",
                "10G"
            ]
        },
        "payload_length": {
            "description": "SPEAD payload length for channel data",
            "type": "integer",
            "minimum": 0
        },
        "destination_ip": {
            "description": "Destination IP address",
            "type": "string",
            "format": "ipv4"
        },
        "source_port": {
            "description": "Source port for integrated data streams",
            "type": "integer",
            "minimum": 0
        },
        "destination_port": {
            "description": "Destination port for integrated data streams",
            "type": "integer",
            "minimum": 0
        },
        "netmask_40g": {
            "description": "Integer netmask for the 40g (science data) subnet",
            "type": "integer",
            "minimum": 0,
            "maximum": 4294967296
        },
        "gateway_40g": {
            "description": "Integer IP address of the 40g (science data) subnet gateway",
            "minimum": 0,
            "maximum": 4294967296
        }
    },
    "required": [
        "mode"
    ],
    "additionalProperties": false
}

__init__(component_manager, logger=None)

Initialise a new SetLmcDownloadCommand instance.

Parameters:

• component_manager (TileComponentManager) – the device to which this command belongs.

• logger (Optional[Logger]) – a logger for this command to use.

do(*args, **kwargs)

Implement MccsTile.SetLmcDownload() command functionality.

Parameters:

• args (Any) – Positional arguments. This should be empty and is provided for type hinting purposes only.

• kwargs (Any) – keyword arguments unpacked from the JSON argument to the command.

Return type:

tuple[ResultCode, str]

Returns:

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

SetLmcIntegratedDownload(argin)

Configure link and size of control data.

Parameters:

argin (str) –

json dictionary with optional keywords:

• mode - (string) ‘1G’ or ‘10G’ (Mandatory)

• 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

• netmask_40g - (int) 40g (science data) subnet mask

• gateway_40g - (int) IP address of 40g (science) subnet gateway

Return type:

tuple[list[ResultCode], list[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)

class SetLmcIntegratedDownloadCommand(component_manager, logger=None)

Class for handling the SetLmcIntegratedDownload() command.

This command takes as input a JSON string that conforms to the following schema:

{
    "$schema": "https://json-schema.org/draft/2020-12/schema",
    "$id": "https://skao.int/MccsTile_SetLmcIntegratedDownload.json",
    "title": "MccsTile SetLmcIntegratedDownload schema",
    "description": "Schema for MccsTile's SetLmcIntegratedDownload command",
    "type": "object",
    "properties": {
        "mode": {
            "description": "Mode: 1G or 10G",
            "type": "string",
            "enum": [
                "1g",
                "1G",
                "10g",
                "10G"
            ]
        },
        "channel_payload_length": {
            "description": "SPEAD payload length for integrated channel data",
            "type": "integer",
            "minimum": 0
        },
        "beam_payload_length": {
            "description": "SPEAD payload length for integrated beam data",
            "type": "integer",
            "minimum": 0
        },
        "destination_ip": {
            "description": "Destination IP address",
            "type": "string",
            "format": "ipv4"
        },
        "source_port": {
            "description": "Source port for integrated data streams",
            "type": "integer",
            "minimum": 0
        },
        "destination_port": {
            "description": "Destination port for integrated data streams",
            "type": "integer",
            "minimum": 0
        },
        "netmask_40g": {
            "description": "Integer netmask for the 40g (science data) subnet",
            "type": "integer",
            "minimum": 0,
            "maximum": 4294967296
        },
        "gateway_40g": {
            "description": "Integer IP address of the 40g (science data) subnet gateway",
            "minimum": 0,
            "maximum": 4294967296
        }
    },
    "required": [
        "mode"
    ],
    "additionalProperties": false
}

__init__(component_manager, logger=None)

Initialise a new SetLmcIntegratedDownloadCommand instance.

Parameters:

• component_manager (TileComponentManager) – the device to which this command belongs.

• logger (Optional[Logger]) – a logger for this command to use.

do(*args, **kwargs)

Implement MccsTile.SetLmcIntegratedDownload() commands.

Parameters:

• args (Any) – Positional arguments. This should be empty and is provided for type hinting purposes only.

• kwargs (Any) – keyword arguments unpacked from the JSON argument to the command.

Return type:

tuple[ResultCode, str]

Returns:

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

StartAcquisition(argin)

Start data acquisition.

Parameters:

argin (str) – json dictionary with optional keywords:

• start_time - (ISO UTC time) start time

• delay - (int) delay start if StartTime is not specified, default 0.2s

Return type:

tuple[list[ResultCode], list[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 = {"start_time":"2021-11-22, "delay":20}
>>> jstr = json.dumps(dict)
>>> dp.command_inout("StartAcquisition", jstr)

class StartAcquisitionCommand(command_tracker, component_manager, callback=None, logger=None)

Class for handling the StartAcquisition() command.

This command takes as input a JSON string that conforms to the following schema:

{
    "$schema": "https://json-schema.org/draft/2020-12/schema",
    "$id": "https://skao.int/MccsTile_StartAcquisition.json",
    "title": "MccsTile StartAcquisition schema",
    "description": "Schema for MccsTile's StartAcquisition command",
    "type": "object",
    "properties": {
        "start_time": {
            "description": "",
            "type": "string",
            "format": "time"
        },
        "delay": {
            "description": "An acquisition start delay",
            "type": "integer"
        }
    }
}

__init__(command_tracker, component_manager, callback=None, logger=None)

Initialise a new instance.

Parameters:

• command_tracker (CommandTracker) – the device’s command tracker

• component_manager (TileComponentManager) – the device’s component manager

• callback (Optional[Callable]) – an optional callback to be called when this command starts and finishes.

• logger (Optional[Logger]) – a logger for this command to log with.

StartBeamformer(argin)

Start the beamformer at the specified time delay.

Parameters:

argin (str) – json dictionary with optional keywords:

• start_time - (str, ISO UTC time) start time

• duration - (int) if > 0 is a duration in CSP frames (2211.84 us)

  if == -1 run forever

• subarray_beam_id - (int)Subarray beam ID of the channels to be started

  Command affects only beamformed channels for given subarray ID Default -1: all channels

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

Return type:

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

Returns:

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

Example:

>>> dp = tango.DeviceProxy("mccs/tile/01")
>>> dict = {"StartTime":10, "Duration":20}
>>> jstr = json.dumps(dict)
>>> dp.command_inout("StartBeamformer", jstr)

class StartBeamformerCommand(component_manager, logger=None)

Class for handling the StartBeamformer(argin) command.

This command takes as input a JSON string that conforms to the following schema:

{
    "$schema": "https://json-schema.org/draft/2020-12/schema",
    "$id": "https://skao.int/MccsTile_StartBeamformer.json",
    "title": "MccsTile StartBeamformer schema",
    "description": "Schema for MccsTile's StartBeamformer command",
    "type": "object",
    "properties": {
        "start_time": {
            "description": "",
            "oneOf": [
                {"type": "string", "format": "time"},
                {"type": "null"}
            ]
        },
        "duration": {
            "description": "Duration in CSP frames (or -1 to run forever)",
            "oneOf": [
                {"type": "integer", "minimum": -1},
                {"type": "null"}
            ]
        },
        "subarray_beam_id": {
            "description": "Subarray beam ID of the changes to be started.",
            "type": "integer",
            "minimum": -1
        },
        "scan_id": {
            "description": "The unique ID for the started scan",
            "type": "integer",
            "minimum": 0
        }
    }
}

__init__(component_manager, logger=None)

Initialise a new StartBeamformerCommand instance.

Parameters:

• component_manager (TileComponentManager) – the device to which this command belongs.

• logger (Optional[Logger]) – a logger for this command to use.

do(*args, **kwargs)

Implement MccsTile.StartBeamformer() command functionality.

Parameters:

• args (Any) – Positional arguments. This should be empty and is provided for type hinting purposes only.

• kwargs (Any) – keyword arguments unpacked from the JSON argument to the command.

Return type:

tuple[ResultCode, str]

Returns:

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

StopBeamformer()

Stop the beamformer.

Return type:

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

Returns:

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

Example:

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

class StopBeamformerCommand(component_manager, logger=None)

Class for handling the StopBeamformer() command.

__init__(component_manager, logger=None)

Initialise a new StopBeamformerCommand instance.

Parameters:

• component_manager (TileComponentManager) – the device to which this command belongs.

• logger (Optional[Logger]) – a logger for this command to use.

do(*args, **kwargs)

Implement MccsTile.StopBeamformer() command functionality.

Parameters:

• args (Any) – unspecified positional arguments. This should be empty and is provided for type hinting only

• kwargs (Any) – unspecified keyword arguments. This should be empty and is provided for type hinting only

Return type:

tuple[ResultCode, str]

Returns:

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

StopDataTransmission()

Stop data transmission from board.

Return type:

tuple[list[ResultCode], list[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")

class StopDataTransmissionCommand(component_manager, logger=None)

Class for handling the StopDataTransmission() command.

__init__(component_manager, logger=None)

Initialise a new StopDataTransmissionCommand instance.

Parameters:

• component_manager (TileComponentManager) – the device to which this command belongs.

• logger (Optional[Logger]) – a logger for this command to use.

do(*args, **kwargs)

Implement MccsTile.StopDataTransmission() command functionality.

Parameters:

• args (Any) – unspecified positional arguments. This should be empty and is provided for type hinting only

• kwargs (Any) – unspecified keyword arguments. This should be empty and is provided for type hinting only

Return type:

tuple[ResultCode, str]

Returns:

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

StopIntegratedData()

Stop the integrated data.

Return type:

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

Returns:

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

class StopIntegratedDataCommand(component_manager, logger=None)

Class for handling the StopIntegratedData command.

__init__(component_manager, logger=None)

Initialise a new StopIntegratedDataCommand instance.

Parameters:

• component_manager (TileComponentManager) – the device to which this command belongs.

• logger (Optional[Logger]) – a logger for this command to use.

do(*args, **kwargs)

Implement MccsTile.StopIntegratedData() command functionality.

Parameters:

• args (Any) – unspecified positional arguments. This should be empty and is provided for type hinting only

• kwargs (Any) – unspecified keyword arguments. This should be empty and is provided for type hinting only

Return type:

tuple[ResultCode, str]

Returns:

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

WriteAddress(argin)

Write list of values at address.

Parameters:

argin (list[int]) – [0] = address to write to [1..n] = list of values to write

Return type:

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

Returns:

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

Example:

>>> values = [.....]
>>> address = 0xfff
>>> dp = tango.DeviceProxy("mccs/tile/01")
>>> dp.command_inout("WriteAddress", [address, values])

class WriteAddressCommand(component_manager, logger=None)

Class for handling the WriteAddress(argin) command.

__init__(component_manager, logger=None)

Initialise a new WriteAddressCommand instance.

Parameters:

• component_manager (TileComponentManager) – the device to which this command belongs.

• logger (Optional[Logger]) – a logger for this command to use.

do(argin, *args, **kwargs)

Implement MccsTile.WriteAddress() command functionality.

Parameters:

• argin (list[int]) – sequence of length two, containing an address and a value

• args (Any) – unspecified positional arguments. This should be empty and is provided for type hinting only

• kwargs (Any) – unspecified keyword arguments. This should be empty and is provided for type hinting only

Return type:

tuple[ResultCode, str]

Returns:

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

Raises:

ValueError – if the argin has the wrong length/structure

WriteRegister(argin)

Write values to the specified register.

Parameters:

argin (str) –

json dictionary with mandatory keywords:

• register_name - (string) register fully qualified string representation

• values - (list) is a list containing the 32-bit values to write

Return type:

tuple[list[ResultCode], list[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 = {"register_name": "test-reg1", "values": values,
            "offset": 0}
>>> jstr = json.dumps(dict)
>>> dp.command_inout("WriteRegister", jstr)

class WriteRegisterCommand(component_manager, logger=None)

Class for handling the WriteRegister() command.

This command takes as input a JSON string that conforms to the following schema:

{
    "$schema": "https://json-schema.org/draft/2020-12/schema",
    "$id": "https://skao.int/MccsTile_WriteRegister.json",
    "title": "MccsTile WriteRegister schema",
    "description": "Schema for MccsTile's WriteRegister command",
    "type": "object",
    "properties": {
        "register_name": {
            "description": "Register fully qualified string representation",
            "type": "string"
        },
        "values": {
            "description": "List of 32-bit values to write",
            "type": "array"
        }
    },
    "required": ["register_name", "values"]
}

__init__(component_manager, logger=None)

Initialise a new WriteRegisterCommand instance.

Parameters:

• component_manager (TileComponentManager) – the device to which this command belongs.

• logger (Optional[Logger]) – a logger for this command to use.

do(*args, **kwargs)

Implement MccsTile.WriteRegister() command functionality.

Parameters:

• args (Any) – Positional arguments. This should be empty and is provided for type hinting purposes only.

• kwargs (Any) – keyword arguments unpacked from the JSON argument to the command.

Return type:

tuple[ResultCode, str]

Returns:

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

__init__(*args, **kwargs)

Initialise this device object.

Parameters:

• args (Any) – positional args to the init

• kwargs (Any) – keyword args to the init

adcHealth()

Read the ADC Health State of the device.

This is an aggregated quantity representing if any of the ADC monitoring points are outside of their thresholds. This is used to compute the overall healthState of the tile.

Return type:

HealthState

Returns:

ADC Health State of the device

adcPower()

Return the RMS power of every ADC signal.

(so a TPM processes 16 antennas, this should return 32 RMS value.

Return type:

list[float]

Returns:

RMP power of ADC signals

adcs()

Return the ADC status.

Return type:

str

Returns:

the ADC status

alarmHealth()

Read the alarm Health State of the device.

This is an aggregated quantity representing if any of the alarm monitoring points are outside of their thresholds. This is used to compute the overall healthState of the tile.

Return type:

HealthState

Returns:

alarm Health State of the device

alarms()

Return the TPM’s alarm status.

Return type:

str

Returns:

the TPM’s alarm status

antennaIds(antenna_ids)

Set the antenna IDs.

Parameters:

antenna_ids (list[int]) – the antenna IDs

Return type:

None

beamformerTable()

Get beamformer region table.

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

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

Return type:

list[int]

Returns:

list of up to 7*48 values

boardTemperature()

Return the board temperature.

Return type:

tuple[Optional[float], float, AttrQuality]

Returns:

the board temperature

channeliserRounding(truncation)

Set channeliser rounding.

Parameters:

truncation (list[int]) – List with either a single value (applies to all channels) or a list of 512 values. Range 0 (no truncation) to 7

Return type:

None

clockPresent()

Report if 10 MHz clock signal is present at the TPM input.

Return type:

bool

Returns:

presence of 10 MHz clock signal

create_component_manager()

Create and return a component manager for this device.

Return type:

TileComponentManager

Returns:

a component manager for this device.

cspDestinationIp()

Return the cspDestinationIp attribute.

Return type:

str

Returns:

the IP address of the csp destination

cspDestinationMac()

Return the cspDestinationMac attribute.

Return type:

str

Returns:

the MAC address of the csp destination

cspDestinationPort()

Return the cspDestinationMac attribute.

Return type:

int

Returns:

the port of the csp destination

cspRounding(rounding)

Set CSP formatter rounding.

Parameters:

rounding (ndarray) – 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

currentFrame()

Return current frame.

in units of 256 ADC frames (276,48 us) Currently this is required, not sure if it will remain so.

Return type:

int

Returns:

current frame

currentHealth()

Read the current Health State of the device.

This is an aggregated quantity representing if any of the current monitoring points are outside of their thresholds. This is used to compute the overall healthState of the tile.

Return type:

HealthState

Returns:

current Health State of the device

currentTileBeamformerFrame()

Return current frame.

in units of 256 ADC frames (276,48 us) Currently this is required, not sure if it will remain so.

Return type:

int

Returns:

current frame

currents()

Return all the currents values available.

Return type:

str

Returns:

currents available

dev_state()

Calculate this device state.

The base device offers some automatic state discovery. However we have some attributes that require explicit analysis as to whether they are in ALARM or not,

e.g. DevBoolean

Return type:

DevState

Returns:

the ‘tango.DevState’ calculated

dsp()

Return the tile beamformer and station beamformer status.

Return type:

str

Returns:

the tile beamformer and station beamformer status

dspHealth()

Read the dsp Health State of the device.

This is an aggregated quantity representing if any of the dsp monitoring points do not have a permitted value. This is used to compute the overall healthState of the tile.

Return type:

HealthState

Returns:

dsp Health State of the device

firmwareName(value)

Set the firmware name.

Parameters:

value (str) – firmware name

Return type:

None

firmwareVersion(value)

Set the firmware version.

Parameters:

value (str) – firmware version

Return type:

None

fortyGbDestinationIps()

Return the destination IPs for all 40Gb ports on the tile.

Return type:

list[str]

Returns:

IP addresses

fortyGbDestinationPorts()

Return the destination ports for all 40Gb ports on the tile.

Return type:

list[int]

Returns:

ports

fpga1Temperature()

Return the temperature of FPGA 1.

Return type:

tuple[Optional[float], float, AttrQuality]

Returns:

the temperature of FPGA 1

fpga2Temperature()

Return the temperature of FPGA 2.

Return type:

tuple[Optional[float], float, AttrQuality]

Returns:

the temperature of FPGA 2

fpgaFrameTime()

Return the FPGA synchronization timestamp.

Return type:

str

Returns:

the FPGA timestamp, in UTC format

fpgaReferenceTime()

Return the FPGA synchronization timestamp.

Return type:

str

Returns:

the FPGA timestamp, in UTC format

fpgaTime()

Return the FPGA internal time.

Return type:

str

Returns:

the FPGA time, in UTC format

fpgasUnixTime()

Return the time for FPGAs.

Return type:

list[int]

Returns:

the time for FPGAs

healthModelParams(argin)

Set the params for health transition rules.

Parameters:

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

Return type:

None

init_command_objects()

Set up the handler objects for Commands.

Return type:

None

init_device()

Initialise the device.

Return type:

None

io()

Return a dictionary of I/O interfaces status available.

Return type:

str

Returns:

I/O interfaces status

ioHealth()

Read the io Health State of the device.

This is an aggregated quantity representing if any of the io monitoring points do not have a permitted value. This is used to compute the overall healthState of the tile.

Return type:

HealthState

Returns:

io Health State of the device

isBeamformerRunning()

Check if beamformer is running.

Return type:

bool

Returns:

whether the beamformer is running

isProgrammed()

Return a flag indicating whether of not the board is programmed.

Return type:

bool

Returns:

whether of not the board is programmed

is_ConfigureTestGenerator_allowed()

Check if command is allowed.

It is allowed only in maintenance mode.

Return type:

bool

Returns:

whether the command is allowed

is_Off_allowed()

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

Return type:

bool

Returns:

True if the command is allowed

is_On_allowed()

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

Return type:

bool

Returns:

True if the command is allowed

logicalTileId(value)

Set the logicalTileId attribute.

The logical tile id is the id of the tile in the station.

Parameters:

value (int) – the new logical tile id

Return type:

None

pendingDataRequests()

Check for pending data requests.

Return type:

bool

Returns:

whether there are data requests pending

phaseTerminalCount(value)

Set the phase terminal count.

Parameters:

value (int) – the phase terminal count

Return type:

None

pllLocked()

Report if ADC clock PLL is in locked state.

Return type:

bool

Returns:

PLL lock state

ppsDelay()

Return the delay between PPS and 10 MHz clock.

Return type:

int

Returns:

Return the PPS delay in nanoseconds

ppsDelayCorrection(pps_delay_correction)

Set a correction to make to the pps delay.

Note: will be applied during next initialisation.

Parameters:

pps_delay_correction (int) – a correction to apply to the pps_delay.

Return type:

None

ppsPresent()

Report if PPS signal is present at the TPM input.

Return type:

tuple[Optional[bool], float, AttrQuality]

Returns:

a tuple with attribute_value, time, quality

preaduLevels(levels)

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

Parameters:

levels (ndarray) – ttenuator level of preADU channels, one per input channel, in dB

Return type:

None

simulationMode(value)

Set the simulation mode.

Writing this attribute is deliberately unimplemented. The simulation mode should instead be set by setting the device’s SimulationConfig property at launch.

Parameters:

value (SimulationMode) – The simulation mode, as a SimulationMode value

Return type:

None

staticTimeDelays(delays)

Set static time delay.

Parameters:

delays (list[float]) – Delay in samples (positive = increase the signal delay) to correct for static delay mismathces, e.g. cable length.

Return type:

None

stationId(value)

Set the id of the station to which this tile is assigned.

Parameters:

value (int) – the station id

Return type:

None

sysrefPresent()

Report if SYSREF signal is present at the FPGA.

Return type:

bool

Returns:

presence of SYSREF signal

temperatureHealth()

Read the temperature Health State of the device.

This is an aggregated quantity representing if any of the temperature monitoring points are outside of their thresholds. This is used to compute the overall healthState of the tile.

Return type:

HealthState

Returns:

temperature Health State of the device

temperatures()

Return all the temperatures values available.

Return type:

str

Returns:

temperatures available

testGeneratorActive()

Report if the test generator is used for some channels.

Return type:

bool

Returns:

test generator status

testMode(value)

Set the test mode.

Writing this attribute is deliberately unimplemented. The test mode should instead be set by setting the device’s TestConfig property at launch.

Parameters:

value (int) – The test mode, as a TestMode value

Return type:

None

tileProgrammingState()

Get the tile programming state.

Return type:

str

Returns:

a string describing the programming state of the tile

timing()

Return a dictionary of the timing signals status.

Return type:

str

Returns:

timing signals status

timingHealth()

Read the timing Health State of the device.

This is an aggregated quantity representing if any of the timing monitoring points do not have a permitted value. This is used to compute the overall healthState of the tile.

Return type:

HealthState

Returns:

timing Health State of the device

unpack_monitoring_point(health_structure, dictionary_path)

Unpack the monitoring point value from dictionary.

Parameters:

• health_structure (dict[str, Any]) – A nested health_structure dictionary

• dictionary_path (list[str]) – A list of strings used to traverse the dictionary.

Example:

>> tile_health = {‘timing’: { ‘pps’: {‘status’: False}}} >> pps=[‘timing’, ‘pps’, ‘status’] >> value = unpack_monitoring_point(tile_health, pps) >> print(value) -> False

Return type:

Optional[Any]

Returns:

the monitoring point value or None.

update_tile_health_attributes()

Update TANGO attributes from the tile health structure dictionary.

Return type:

None

voltageHealth()

Read the voltage Health State of the device.

This is an aggregated quantity representing if any of the voltage monitoring points are outside of their thresholds. This is used to compute the overall healthState of the tile.

Return type:

HealthState

Returns:

voltage Health State of the device

voltageMon()

Return the internal 5V supply of the TPM.

Return type:

float

Returns:

Internal supply of the TPM

voltages()

Return all the voltage values available.

Return type:

str

Returns:

voltages available

main(*args, **kwargs)

Entry point for module.

Parameters:

• args (str) – positional arguments

• kwargs (str) – named arguments

Return type:

int

Returns:

exit code