Tpm Driver

An implementation of a Tile component manager that drives a real TPM.

The class is basically a wrapper around the Tile class, in order to have a consistent interface for driver and simulator. This is an initial version. Some methods are still simulated. A warning is issued in this case, or a NotImplementedError exception raised.

class TpmDriver(logger, tile_id, tile, tpm_version, communication_state_changed_callback, component_state_changed_callback)[source]

Hardware driver for a TPM.

__init__(logger, tile_id, tile, tpm_version, communication_state_changed_callback, component_state_changed_callback)[source]

Initialise a new TPM driver instance passing in the Tile object.

Parameters
  • logger (logging.Logger) – a logger for this simulator to use

  • tile_id (int) – the unique ID for the tile

  • tile (pyaavs.tile.Tile) – the tile driven by this TpmDriver

  • tpm_version (str) – TPM version: “tpm_v1_2” or “tpm_v1_6”

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

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

property adc_rms: list[float]

Return the last measured RMS power of the TPM’s analog-to-digital converter.

Return type

list[float]

Returns

the RMS power of the TPM’s ADC

apply_calibration(switch_time=0)[source]

Switch the calibration bank.

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

Parameters

switch_time (typing.Optional[int]) – an optional time at which to perform the switch

Return type

None

apply_pointing_delays(load_time)[source]

Load the pointing delay at a specified time.

Parameters

load_time (int) – time at which to load the pointing delay

Return type

None

property arp_table: dict[int, list[int]]

Check that ARP table has been populated in 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

dict[int, list[int]]

Returns

list of core id and arp table populated

property beamformer_table: list[list[int]]

Fetch internal beamformer table.

Fetch table used by the hardware beamformer to define beams and logical bands :rtype: list[list[int]] :return: bidimensional table, with 48 entries, one every 8 channels

  • start physical channel

  • tile hardware beam

  • subarray ID

  • subarray start logical channel

  • subarray_beam_id - (int) ID of the subarray beam

  • substation_id - (int) Substation

  • aperture_id: ID of the aperture (station*100+substation?)

property board_temperature: float

Return the temperature of the TPM.

Return type

float

Returns

the temperature of the TPM

property channeliser_truncation: list[int]

Read the cached value for the channeliser truncation.

Return type

list[int]

Returns

cached value for the channeliser truncation

property clock_present: bool

Check if 10 MHz clock signal is present.

Return type

bool

Returns

True if 10 MHz clock is present. Checked in poll loop, cached

configure_40g_core(core_id, arp_table_entry, src_mac, src_ip, src_port, dst_ip, dst_port)[source]

Configure the 40G code.

Parameters
  • core_id (int) – id of the core

  • arp_table_entry (int) – ARP table entry to use

  • src_mac (int) – MAC address of the source

  • src_ip (str) – IP address of the source

  • src_port (int) – port of the source

  • dst_ip (str) – IP address of the destination

  • dst_port (int) – port of the destination

Return type

None

configure_integrated_beam_data(integration_time=0.5, first_channel=0, last_channel=191)[source]

Configure and start the transmission of integrated channel data.

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

Parameters
  • integration_time (float) – integration time in seconds, defaults to 0.5

  • first_channel (int) – first channel

  • last_channel (int) – last channel

Return type

None

configure_integrated_channel_data(integration_time=0.5, first_channel=0, last_channel=511)[source]

Configure and start the transmission of integrated channel data.

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

Parameters
  • integration_time (float) – integration time in seconds, defaults to 0.5

  • first_channel (int) – first channel

  • last_channel (int) – last channel

Return type

None

configure_test_generator(frequency0, amplitude0, frequency1, amplitude1, amplitude_noise, pulse_code, amplitude_pulse, load_time=0)[source]

Test generator setting.

Parameters
  • frequency0 (float) – Tone frequency in Hz of DDC 0

  • amplitude0 (float) – Tone peak amplitude, normalized to 31.875 ADC units, resolution 0.125 ADU

  • frequency1 (float) – Tone frequency in Hz of DDC 1

  • amplitude1 (float) – Tone peak amplitude, normalized to 31.875 ADC units, resolution 0.125 ADU

  • amplitude_noise (float) – Amplitude of pseudorandom noise normalized to 26.03 ADC units, resolution 0.102 ADU

  • pulse_code (int) – Code for pulse frequency. Range 0 to 7: 16,12,8,6,4,3,2 times frame frequency

  • amplitude_pulse (float) – pulse peak amplitude, normalized to 127.5 ADC units, resolution 0.5 ADU

  • load_time (int) – Time to start the generator.

Return type

None

property csp_rounding: list[int]

Read the cached value for the final rounding in the CSP samples.

Need to be specfied only for the last tile :rtype: list[int] :return: Final rounding for the CSP samples. Up to 384 values

property current_tile_beamformer_frame: int

Return current tile beamformer frame, in units of 256 ADC frames.

Return type

int

Returns

current tile beamformer frame

property currents: dict[str, Any]

Return a dictionary of all current values available in the TPM.

Return type

dict[str, typing.Any]

Returns

currents in the TPM

download_firmware(bitfile, task_callback=None)[source]

Download firmware bitfile onto the TPM as a long runnning command.

Parameters
Return type

tuple[ska_control_model.TaskStatus, str]

Returns

TaskStatus and message

property dsp: dict[str, Any]

Return the tile beamformer and station beamformer status in the TPM.

Return type

dict[str, typing.Any]

Returns

dsp status in the TPM

erase_fpga()[source]

Erase FPGA programming to reduce FPGA power consumption.

Return type

None

property firmware_available: list[dict[str, Any]]

Return the list of the firmware loaded in the system.

Return type

list[dict[str, typing.Any]]

Returns

the firmware list

property firmware_name: str

Return the name of the firmware that this TPM simulator is running.

Return type

str

Returns

firmware name

property firmware_version: str

Return the name of the firmware that this TPM simulator is running.

Return type

str

Returns

firmware version (major.minor)

property fpga1_temperature: float

Return the temperature of FPGA 1.

Return type

float

Returns

the temperature of FPGA 1

property fpga2_temperature: float

Return the temperature of FPGA 2.

Return type

float

Returns

the temperature of FPGA 2

property fpga_current_frame: int

Return the FPGA current frame counter.

Return type

int

Returns

the FPGA_1 current frame counter

Raises

ConnectionError – if communication with tile failed

property fpga_reference_time: int

Return the FPGA reference time.

Required to map the FPGA timestamps, expressed in frames to UTC time

Return type

int

Returns

the FPGA_1 reference time, in Unix seconds

property fpgas_time: list[int]

Return the FPGAs clock time.

Useful for detecting clock skew, propagation delays, contamination delays, etc.

Return type

list[int]

Returns

the FPGAs clock time

Raises

ConnectionError – if communication with tile failed

get_40g_configuration(core_id=-1, arp_table_entry=0)[source]

Return a 40G configuration.

Parameters
  • core_id (int) – id of the core for which a configuration is to be return. Defaults to -1, in which case all cores configurations are returned

  • arp_table_entry (int) – ARP table entry to use

Return type

list[dict]

Returns

core configuration or list of core configurations

get_tile_id()[source]

Get the tile ID stored in the FPGA.

Return type

int

Returns

tile ID

Raises

LibraryError

property hardware_version: str

Return whether this TPM is 1.2 or 1.6.

Return type

str

Returns

TPM hardware version. 120 or 160

initialise(task_callback=None)[source]

Download firmware, if not already downloaded, and initializes tile.

Parameters

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

Return type

tuple[ska_control_model.TaskStatus, str]

Returns

A tuple containing a task status and a message

initialise_beamformer(start_channel, nof_channels, is_first, is_last)[source]

Initialise the beamformer.

Parameters
  • start_channel (int) – the start channel

  • nof_channels (int) – number of channels

  • is_first (bool) – whether this is the first (?)

  • is_last (bool) – whether this is the last (?)

Return type

None

property io: dict[str, Any]

Return a dictionary of I/O Interfaces status available in the TPM.

Return type

dict[str, typing.Any]

Returns

io in the TPM

property is_beamformer_running: bool

Whether the beamformer is currently running.

Return type

bool

Returns

whether the beamformer is currently running

property is_programmed: bool

Return whether this TPM is programmed (i.e. firmware has been downloaded to it).

Return type

bool

Returns

whether this TPM is programmed

load_calibration_coefficients(antenna, calibration_coefficients)[source]

Load calibration coefficients.

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

Parameters
  • antenna (int) – the antenna to which the coefficients apply

  • calibration_coefficients (list[complex]) – a bidirectional complex array of coefficients, flattened into a list

Return type

None

load_pointing_delays(delay_array, beam_index)[source]

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
  • delay_array (list[list[float]]) – delay in seconds, and delay rate in seconds/second

  • beam_index (int) – the beam to which the pointing delay should be applied

Return type

None

property pending_data_requests: bool

Check for pending data requests.

Return type

bool

Returns

whether there are pending send data requests

property phase_terminal_count: int

Return the phase terminal count.

Return type

int

Returns

the phase terminal count

property pll_locked: bool

Check if ADC clock PLL is locked.

Return type

bool

Returns

True if PLL is locked. Checked in poll loop, cached

post_synchronisation()[source]

Perform post tile configuration synchronization.

TODO Private method or must be available externally?

Return type

None

property pps_delay: float

Return the last measured PPS delay of the TPM.

Return type

float

Returns

PPS delay correction in nanoseconds. Rounded to 1.25 ns units

property pps_present: bool

Check if PPS signal is present.

Return type

bool

Returns

True if PPS is present. Checked in poll loop, cached

property preadu_levels: list[int]

Get preadu levels in dB.

Return type

list[int]

Returns

cached values of Preadu attenuation level in dB

read_address(address, nvalues)[source]

Return a list of values from a given address.

Parameters
  • address (int) – address of start of read

  • nvalues (int) – number of values to read

Return type

list[int]

Returns

values at the address

read_register(register_name)[source]

Read the values in a named register.

Parameters

register_name (str) – name of the register

Return type

list[int]

Returns

values read from the register

property register_list: list[str]

Return a list of registers available on each device.

Return type

list[str]

Returns

list of registers

send_data_samples(data_type='', timestamp=0, seconds=0.2, n_samples=1024, sync=False, first_channel=0, last_channel=511, channel_id=128, frequency=150000000.0, round_bits=3)[source]

Front end for send_xxx_data methods.

Parameters
  • data_type (str) – sample type. “raw”, “channel”, “channel_continuous”, “narrowband”, “beam”

  • timestamp (int) – Timestamp for start sending data. Default 0 start now

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

  • n_samples (int) – number of samples to send per packet

  • sync (bool) – (raw) send synchronised antenna samples, vs. round robin

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

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

  • channel_id (int) – (channel_continuous) channel to send

  • frequency (float) – (narrowband) Sky frequency for band centre, in Hz

  • round_bits (int) – (narrowband) how many bits to round

Raises

ValueError – if values wrong

Return type

None

set_beamformer_regions(regions)[source]

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

Parameters

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

Return type

None

set_lmc_download(mode, payload_length=1024, dst_ip=None, src_port=61648, dst_port=4660)[source]

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

Parameters
  • mode (str) – “1G” or “10G”

  • payload_length (int) – SPEAD payload length for integrated channel data, defaults to 1024

  • dst_ip (typing.Optional[str]) – destination IP, defaults to None

  • src_port (typing.Optional[int]) – sourced port, defaults to 0xF0D0

  • dst_port (typing.Optional[int]) – destination port, defaults to 4660

Return type

None

set_lmc_integrated_download(mode, channel_payload_length, beam_payload_length, dst_ip=None, src_port=61648, dst_port=4660)[source]

Configure link and size of control data.

Parameters
  • mode (str) – ‘1G’ or ‘10G’

  • channel_payload_length (int) – SPEAD payload length for integrated channel data

  • beam_payload_length (int) – SPEAD payload length for integrated beam data

  • dst_ip (typing.Optional[str]) – Destination IP, defaults to None

  • src_port (int) – source port, defaults to 0xF0D0

  • dst_port (int) – destination port, defaults to 4660

Return type

None

start_acquisition(start_time=None, delay=2)[source]

Start data acquisition.

This must be run as a long running command

Parameters
Return type

bool

Returns

if data acquisition started correctly

start_beamformer(start_time=0, duration=-1, subarray_beam_id=1, scan_id=0)[source]

Start the beamformer at the specified time.

Parameters
  • start_time (int) – time at which to start the beamformer, defaults to 0

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

  • subarray_beam_id (int) – ID of the subarray beam to start. Default = -1, all

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

Return type

None

start_communicating()[source]

Establish communication with the TPM.

Return type

None

start_connection()[source]

Try to connect tile to tpm.

Return type

None

Returns

None

property static_delays: list[float]

Read the cached value for the static delays, in sample.

Return type

list[float]

Returns

static delay, in samples one per TPM input

property station_id: int

Get the Station ID.

Return type

int

Returns

assigned station Id value

stop_beamformer()[source]

Stop the beamformer.

Return type

None

stop_communicating()[source]

Stop communicating with the TPM.

Todo

is there a better way to do this? Should Tile16 have a disconnect() method that we can call here?

Return type

None

stop_data_transmission()[source]

Stop data transmission for send_channelised_data_continuous.

Return type

None

stop_integrated_data()[source]

Stop the integrated data.

Return type

None

sync_fpgas()[source]

Synchronise the FPGAs.

TODO Method appears to be mostly internal (private).

Return type

None

property sysref_present: bool

Check if SYSREF signal is present.

Return type

bool

Returns

True if SYSREF is present. Checked in poll loop, cached

property temperatures: dict[str, Any]

Return a dictionary of all temperature values available in the TPM.

Return type

dict[str, typing.Any]

Returns

temperatures in the TPM

property test_generator_active: bool

Check if the test generator is active.

Return type

bool

Returns

whether the test generator is active

test_generator_input_select(inputs=0)[source]

Specify ADC inputs which are substitute to test signal.

Specified using a 32 bit mask, with LSB for ADC input 0.

Parameters

inputs (int) – Bit mask of inputs using test signal

Return type

None

property tile_id: int

Get the Tile ID.

Return type

int

Returns

assigned tile Id value

property timing: dict[str, Any]

Return a dictionary of the Timing Signals status available in the TPM.

Return type

dict[str, typing.Any]

Returns

timings in the TPM

tpm_connected()[source]

Tile connected to tpm.

Return type

None

tpm_disconnected()[source]

Tile disconnected to tpm.

Return type

None

property tpm_status: TpmStatus

Return the TPM status.

Return type

ska_low_mccs_spshw.tile.tpm_status.TpmStatus

Returns

the TPM status

property voltage_mon: float

Return the internal 5V supply of the TPM.

Return type

float

Returns

the internal 5V supply of the TPM

property voltages: dict[str, Any]

Return a dictionary of all voltage values available in the TPM.

Return type

dict[str, typing.Any]

Returns

voltages in the TPM

write_address(address, values)[source]

Write a list of values to a given address.

Parameters
  • address (int) – address of start of read

  • values (list[int]) – values to write

Return type

None

write_register(register_name, values)[source]

Read the values in a register.

Parameters
Return type

None