Tile Simulator

An implementation of a Tile simulator.

class DynamicTileSimulator(logger)

A simulator for a TPM, with dynamic value updates to certain attributes.

__init__(logger)

Initialise a new Dynamic Tile simulator instance.

Parameters:

logger (Logger) – a logger for this simulator to use

property board_temperature: float | None

Return the temperature of the TPM.

Returns:

the temperature of the TPM

property current: float

Return the current of the TPM.

Returns:

the current of the TPM

property fpga1_temperature: float

Return the temperature of FPGA 1.

Returns:

the temperature of FPGA 1

property fpga2_temperature: float

Return the temperature of FPGA 2.

Returns:

the temperature of FPGA 2

get_board_temperature()
Return type:

Optional[float]

Returns:

the mocked board temperature.

get_current()
Return type:

Optional[float]

Returns:

the mocked current.

get_fpga0_temperature()
Return type:

Optional[float]

Returns:

the mocked fpga0 temperature.

get_fpga1_temperature()
Return type:

Optional[float]

Returns:

the mocked fpga1 temperature.

get_voltage()
Return type:

Optional[float]

Returns:

the mocked voltage.

property voltage: float

Return the voltage of the TPM.

Returns:

the voltage of the TPM

class MockTpm(logger)

Simulator for a pyfabil::Tpm class.

__init__(logger)

Initialise the MockTPM.

Parameters:

logger (Logger) – a logger for this simulator to use

find_register(string, display=False, info=False)

Find a item in a dictionary.

This is mocking the reading of a register for the purpose of testing TPM_driver

Parameters:
  • string (str) – Regular expression to search against

  • display (bool | None) – True to output result to console

  • info (bool | None) – for linter.

Return type:

List[RegisterInfo | None]

Returns:

registers found at address.

read_address(address, n=1)

Get address value.

Parameters:
  • address (int) – Memory address to read from

  • n (int) – Number of items to read

Return type:

Optional[Any]

Returns:

Values

read_register(register, n=1, offset=0)

Get register value.

Parameters:
  • register (int | str) – Memory register to read from

  • n (int) – Number of words to read

  • offset (int) – Memory address offset to read from

Return type:

Optional[Any]

Returns:

Values

property station_beamf: List[StationBeamformer]

Station beamf.

Returns:

the station_beamf.

property tpm_preadu: List[PreAdu]

Tpm pre adu.

Returns:

the preadu.

write_address(address, values, retry=True)

Write address value.

Parameters:
  • address (int) – Memory address to write

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

  • retry (bool) – retry (does nothing yet.)

Return type:

None

write_register(register, values, offset=0, retry=True)

Set register value.

Parameters:
  • register (int | str) – Register name

  • values (int) – Values to write

  • offset (int) – Memory address offset to write to

  • retry (bool) – retry

Return type:

None

:raises LibraryError:Attempting to set a register not in the memory address.

class PreAdu(logger)

Mock preadu plugin.

__init__(logger)

Initialise mock plugin.

Parameters:

logger (Logger) – a logger for this simulator to use

get_attenuation()

Get preadu attenuation for all channels.

Return type:

list[float]

Returns:

attenuation for all channels.

read_configuration()

Read configuration.

Return type:

None

select_low_passband()

Select low pass band.

Return type:

None

set_attenuation(attenuation, channels=None)

Set preadu channel attenuation.

Parameters:
Return type:

None

write_configuration()

Write configuration to preadu.

Return type:

None

class StationBeamformer

Station beamformer.

__init__()

Initialise the station beamformer object.

define_channel_table(table)

Define station beamformer table.

Defines the station beamformer table. Each entry in the list contains: - start channel - number of channels - hw beam ID - subarray ID - subarray_logical_channel - subarray_beam_id - substation_id - aperture_id

Parameters:

table (list[list[int]]) – table of channel blocks. Entries of 8 items each:

Raises:

ValueError – if wrong value passed.

Return type:

None

get_channel_table()

Get channel table.

Return type:

list[list[int]]

Returns:

channel table

is_running()
Return type:

bool

Returns:

is running.

start()

Start.

Return type:

None

stop()

stop.

Return type:

None

class TileSimulator(logger)

This attempts to simulate pyaavs Tile.

This is used for testing the tpm_driver, it implements __getitem__, __setitem__ so that the TileSimulator can interface with the TPMSimulator in the same way as the AAVS Tile interfaces with the pyfabil TPM. Instead of writing to a register we write to a dictionary. It overwrite read_address, write_address, read_register, write_register for simplicity.

__init__(logger)

Initialise a new TPM simulator instance.

Parameters:

logger (Logger) – a logger for this simulator to use

beamformer_is_running()

Beamformer is running.

Return type:

bool

Returns:

is the beam is running

check_arp_table(timeout=30.0)

Check arp table.

Parameters:

timeout (float) – Timeout in seconds

Return type:

bool

Returns:

a bool representing if arp table is healthy.

check_global_status_alarms()

Check global status alarms.

Return type:

dict[str, int]

Returns:

a dictionary with the simulated alarm status.

check_pending_data_requests()
Return type:

bool

Returns:

the pending data requess flag.

compute_calibration_coefficients()

Compute calibration coefficients.

Return type:

None

configure_40g_core(core_id=0, arp_table_entry=0, src_mac=None, src_ip=None, src_port=None, dst_ip=None, dst_port=None, rx_port_filter=None, netmask=None, gateway_ip=None)

Configure the 40G code.

The dst_mac parameter is ignored in true 40G core (ARP resolution used instead)

Parameters:
  • core_id (int) – id of the core

  • arp_table_entry (int) – ARP table entry to use

  • src_mac (Optional[int]) – MAC address of the source

  • src_ip (Optional[str]) – IP address of the source

  • src_port (Optional[int]) – port of the source

  • dst_ip (Optional[str]) – IP address of the destination

  • dst_port (Optional[int]) – port of the destination

  • rx_port_filter (Optional[int]) – Filter for incoming packets

  • netmask (Optional[int]) – Netmask

  • gateway_ip (Optional[int]) – Gateway IP

Raises:

ValueError – when the core_id is not [0,1]

Return type:

None

configure_integrated_beam_data(integration_time=0.5, first_channel=0, last_channel=192)

Configure and start continuous integrated beam data.

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=512)

Configure and start continuous integrated channel data.

TODO Implement generation of integrated packets :type integration_time: float :param integration_time: integration time in seconds, defaults to 0.5 :type first_channel: int :param first_channel: first channel :type last_channel: int :param last_channel: last channel

Return type:

None

connect(initialise=False, load_plugin=True, enable_ada=False, enable_adc=True, dsp_core=True, adc_mono_channel_14_bit=False, adc_mono_channel_sel=0)

Attempt to form a connection with TPM.

Parameters:
  • initialise (bool) – Initialises the TPM object

  • load_plugin (bool) – loads software plugins

  • enable_ada (bool) – Enable ADC amplifier (usually not present)

  • enable_adc (bool) – Enable ADC

  • dsp_core (bool) – Enable loading of DSP core plugins

  • adc_mono_channel_14_bit (bool) – Enable ADC mono channel 14bit mode

  • adc_mono_channel_sel (int) – Select channel in mono channel mode (0=A, 1=B)

Return type:

None

current_tile_beamformer_frame()
Return type:

int

Returns:

beamformer frame.

define_spead_header(station_id, subarray_id, nof_antennas, ref_epoch=-1, start_time=0)

Define the SPEAD header for the given parameters.

Parameters:
  • station_id (int) – The ID of the station.

  • subarray_id (int) – The ID of the subarray.

  • nof_antennas (int) – Number of antennas in the station

  • ref_epoch (int) – Unix time of epoch. -1 uses value defined in set_epoch

  • start_time (Optional[int]) – start time

Return type:

bool

Returns:

a bool representing if command executed without error.

erase_fpga(force=True)

Erase the fpga firmware.

Parameters:

force (bool) – force the erase.

Return type:

None

get_40g_core_configuration(core_id=-1, arp_table_entry=0)

Return a 40G configuration.

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

  • arp_table_entry (int) – ARP table entry to use

Return type:

UnionType[dict, list[dict], None]

Returns:

core configuration or list of core configurations or none

get_adc_rms(sync=False)

Get ADC power, immediate.

Parameters:

sync (Optional[bool]) – Synchronise RMS read

Return type:

list[float]

Returns:

the mock ADC rms values.

get_arp_table()

Get arp table.

Return type:

dict[int, list[int]]

Returns:

the app table

get_firmware_list()
Return type:

List[dict[str, Any]]

Returns:

firmware list.

get_fpga_time(device)
Parameters:

device (Device) – device.

Return type:

int

Returns:

the fpga_time.

Raises:

LibraryError – If invalid device specified.

get_fpga_timestamp(device=pyfabil.base.definitions.Device.FPGA_1)

Get timestamp from FPGA.

Parameters:

device (Device) – device.

Return type:

int

Returns:

the simulated timestamp.

Raises:

LibraryError – Invalid device specified

get_health_status(**kwargs)

Get the health state of the tile.

Parameters:

kwargs (Any) – Any kwargs to identify health group. see aavs-system.Tile

Return type:

dict[str, Any]

Returns:

mocked fetch of health.

get_phase_terminal_count()

Get PPS phase terminal count.

Return type:

int

Returns:

the simulated phase terminal count.

get_pps_delay(enable_correction=True)

Get the pps delay.

Parameters:

enable_correction (bool) – enable correction.

Return type:

int

Returns:

the pps delay.

get_preadu_levels()

Get preADU attenuation levels.

Return type:

list[float]

Returns:

Attenuation levels corresponding to each ADC channel, in dB.

get_station_id()

Get station ID.

Returns:

station ID programmed in HW

Return type:

int

get_tile_id()
Return type:

int

Returns:

the mocked tile_id.

initialise(station_id=0, tile_id=0, lmc_use_40g=False, lmc_dst_ip=None, lmc_dst_port=4660, lmc_integrated_use_40g=False, src_ip_fpga1=None, src_ip_fpga2=None, dst_ip_fpga1=None, dst_ip_fpga2=None, src_port=4661, dst_port=4660, dst_port_single_port_mode=4662, rx_port_single_port_mode=4662, netmask_40g=None, gateway_ip_40g=None, active_40g_ports_setting='port1-only', enable_adc=True, enable_ada=False, enable_test=False, use_internal_pps=False, pps_delay=0, time_delays=0, is_first_tile=False, is_last_tile=False, qsfp_detection='auto', adc_mono_channel_14_bit=False, adc_mono_channel_sel=0)

Initialise tile.

Parameters:
  • station_id (int) – station ID

  • tile_id (int) – Tile ID in the station

  • lmc_use_40g (bool) – if True use 40G interface to transmit LMC data, otherwise use 1G

  • lmc_dst_ip (Optional[str]) – destination IP address for LMC data packets

  • lmc_dst_port (int) – destination UDP port for LMC data packets

  • lmc_integrated_use_40g (bool) – if True use 40G interface to transmit LMC integrated data, otherwise use 1G

  • src_ip_fpga1 (Optional[str]) – source IP address for FPGA1 40G interface

  • src_ip_fpga2 (Optional[str]) – source IP address for FPGA2 40G interface

  • dst_ip_fpga1 (Optional[str]) – destination IP address for beamformed data from FPGA1 40G interface

  • dst_ip_fpga2 (Optional[str]) – destination IP address for beamformed data from FPGA2 40G interface

  • src_port (int) – source UDP port for beamformed data packets

  • dst_port (int) – destination UDP port for beamformed data packets

  • enable_ada (bool) – enable adc amplifier, Not present in most TPM versions

  • enable_adc (bool) – Enable ADC

  • active_40g_ports_setting (str) – placeholder docstring

  • dst_port_single_port_mode (int) – placeholder docstring

  • gateway_ip_40g (Optional[str]) – placeholder docstring

  • netmask_40g (Optional[str]) – placeholder docstring

  • rx_port_single_port_mode (int) – placeholder docstring

  • enable_test (bool) – setup internal test signal generator instead of ADC

  • use_internal_pps (bool) – use internal PPS generator synchronised across FPGAs

  • pps_delay (int) – PPS delay correction in 625ps units

  • time_delays (int) – time domain delays for 32 inputs

  • is_first_tile (bool) – True if this tile is the first tile in the beamformer chain

  • is_last_tile (bool) – True if this tile is the last tile in the beamformer chain

  • qsfp_detection (str) – “auto” detects QSFP cables automatically, “qsfp1”, force QSFP1 cable detected, QSFP2 cable not detected “qsfp2”, force QSFP1 cable not detected, QSFP2 cable detected “all”, force QSFP1 and QSFP2 cable detected “flyover_test”, force QSFP1 and QSFP2 cable detected and adjust polarity for board-to-board cable “none”, force no cable not detected

  • adc_mono_channel_14_bit (bool) – Enable ADC mono channel 14bit mode

  • adc_mono_channel_sel (int) – Select channel in mono channel mode (0=A, 1=B)

Return type:

None

initialise_beamformer(start_channel, nof_channels)

Mock set the beamformer parameters.

Parameters:
  • start_channel (int) – start_channel

  • nof_channels (int) – nof_channels

Raises:

ValueError – For out of range values.

Return type:

None

is_programmed()

Return whether the mock has been implemented.

Return type:

bool

Returns:

the mocked programmed state

load_antenna_tapering(beam, tapering_coefficients)

Load antenna tapering.

Parameters:
  • beam (int) – beam

  • tapering_coefficients (list[int]) – tapering coefficients

Return type:

None

load_beam_angle(angle_coefficients)

Load beam angle.

Parameters:

angle_coefficients (list[float]) – angle coefficients.

Return type:

None

load_calibration_coefficients(antenna, calibration_coefficients)

Load calibration coefficients.

calibration_coefficients is a bi-dimensional complex array of the form calibration_coefficients[channel, polarization], with each element representing a normalized coefficient, with (1.0, 0.0) the normal, expected response for an ideal antenna. Channel is the index specifying the channels at the beamformer output, i.e. considering only those channels actually processed and beam assignments. The 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 The calibration coefficients may include any rotation matrix (e.g. the parallitic angle), but do not include the geometric delay.

Parameters:
  • antenna (int) – Antenna number (0-15)

  • calibration_coefficients (list[complex]) – Calibration coefficient array

Return type:

None

load_pointing_delay(load_time=0, load_delay=64)

Load pointing delay.

Parameters:
  • load_time (int) – load time

  • load_delay (int) – delay in (in ADC frames/256) to apply when load_time == 0

Return type:

None

mock_off()

Fake a connection by constructing the TPM.

NOTE:

This method exists in the Simulator Only

Return type:

None

mock_on()

Fake a connection by constructing the TPM.

NOTE:

This method exists in the Simulator Only

Return type:

None

program_fpgas(bitfile)

Mock programmed state to True.

Parameters:

bitfile (str) – the name of the bitfile to download

Raises:

LibraryError – if bitfile is of type None.

Return type:

None

reset_eth_errors()

Reset Ethernet errors.

Return type:

None

send_beam_data(timeout=0, timestamp=0, seconds=0.2)

Send beam data.

Parameters:
  • timeout (int) – timeout

  • timestamp (int) – timestamp

  • seconds (float) – When to synchronise

Return type:

None

send_channelised_data(number_of_samples=1024, first_channel=0, last_channel=511, timestamp=None, seconds=0.2)

Send channelised data from the TPM.

Parameters:
  • number_of_samples (int) – Number of spectra to send

  • first_channel (int) – First channel to send

  • last_channel (int) – Last channel to send

  • timestamp (Optional[int]) – When to start transmission

  • seconds (float) – When to synchronise

Return type:

None

send_channelised_data_continuous(channel_id, number_of_samples=128, wait_seconds=0, timestamp=None, seconds=0.2)

Continuously send channelised data from a single channel.

Parameters:
  • channel_id (int) – Channel ID

  • number_of_samples (int) – Number of spectra to send

  • wait_seconds (int) – Wait time before sending data

  • timestamp (Optional[int]) – When to start

  • seconds (float) – When to synchronise

Return type:

None

send_channelised_data_narrowband(frequency, round_bits, number_of_samples=128, wait_seconds=0, timestamp=None, seconds=0.2)

Continuously send channelised data from a single channel.

Parameters:
  • frequency (int) – Sky frequency to transmit

  • round_bits (int) – Specify which bits to round

  • number_of_samples (int) – Number of spectra to send

  • wait_seconds (int) – Wait time before sending data

  • timestamp (Optional[int]) – When to start

  • seconds (float) – When to synchronise

Return type:

None

send_raw_data(sync=False, timestamp=None, seconds=0.2, fpga_id=None)

Send raw data.

Parameters:
Return type:

None

set_beamformer_regions(region_array)

Set beamformer region_array.

Parameters:

region_array (list[list[int]]) – region_array

Return type:

None

set_channeliser_truncation(trunc, signal=None)

Set the channeliser coefficients to modify the bandpass.

Parameters:
  • trunc (list[int]) – list with M values, one for each of the frequency channels. Same truncation is applied to the corresponding frequency channels in all inputs.

  • signal (Optional[int]) – Input signal, 0 to 31. If None, apply to all

Return type:

None

set_csp_rounding(rounding)

Set the final rounding in the CSP samples, one value per beamformer channel.

Parameters:

rounding (list[int]) – Number of bits rounded in final 8 bit requantization to CSP

Return type:

bool

Returns:

true is write a success.

set_first_last_tile(is_first, is_last)

Set first last tile in chain.

Parameters:
  • is_first (bool) – true if first

  • is_last (bool) – true if last

Return type:

bool

Returns:

a bool representing if command executed without error.

set_lmc_download(mode, payload_length=1024, dst_ip=None, src_port=61648, dst_port=4660, netmask_40g=None, gateway_ip_40g=None)

Specify where the control data will be transmitted.

With the simulator no traffic will leave the cluster. To transmit data from the pod hosting the simulator to the DAQ (data acquisition) receiver, a Kubernetes service is required. Therefore dst_ip is the name of the service to use rather than the IP.

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

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

  • dst_ip (Optional[str]) – destination service.

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

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

  • netmask_40g (Optional[str]) – the mask to apply

  • gateway_ip_40g (Optional[str]) – the gateway ip.

Return type:

None

set_lmc_integrated_download(mode, channel_payload_length, beam_payload_length, dst_ip=None, src_port=61648, dst_port=4660, netmask_40g=None, gateway_ip_40g=None)

Configure link and size of control data for integrated LMC packets.

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 (Optional[str]) – Destination IP

  • src_port (int) – Source port for integrated data streams

  • dst_port (int) – Destination port for integrated data streams

  • netmask_40g (Optional[str]) – the mask to apply to the 40g.

  • gateway_ip_40g (Optional[str]) – the gateway ip for the 40g.

Return type:

None

set_pointing_delay(delay_array, beam_index)

Set pointing delay.

Parameters:
  • delay_array (list[list[float]]) – delay array

  • beam_index (int) – beam index

Return type:

None

set_preadu_levels(levels)

Set preADU attenuation levels.

Parameters:

levels (list[float]) – Desired attenuation levels for each ADC channel, in dB.

Return type:

None

set_station_id(station_id, tile_id)

Set mock registers to some value.

Parameters:
  • tile_id (int) – tile_id

  • station_id (int) – station_id

Return type:

None

set_test_generator_pulse(freq_code, amplitude=0.0)

Set test generator pulse.

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

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

Return type:

None

set_time_delays(delays)

Set coarse zenith delay for input ADC streams.

Parameters:

delays (list[float]) – the delay in input streams, specified in nanoseconds. A positive delay adds delay to the signal stream

Return type:

bool

Returns:

True if command executed to completion.

start_acquisition(start_time=None, delay=2, tpm_start_time=None)

Start data acquisition.

Parameters:
  • start_time (Optional[int]) – Time for starting (frames)

  • delay (int) – delay after start_time (frames)

  • tpm_start_time (Optional[int]) – TPM will act as if it is started at this time (seconds)

Return type:

None

start_beamformer(start_time=0, duration=-1, scan_id=0, mask=1099511627775)

Start beamformer.

Parameters:
  • start_time (int) – start time UTC

  • duration (int) – duration

  • scan_id (int) – ID of the scan, to be specified in the CSP SPEAD header

  • mask (int) – Bitmask of the channels to be started. Unsupported by FW

Return type:

bool

Returns:

true if the beamformer was started successfully.

stop_beamformer()

Stop beamformer.

Return type:

None

stop_data_transmission()

Stop data transmission.

Return type:

None

stop_integrated_data()

Stop integrated data.

Return type:

None

switch_calibration_bank(switch_time=0)

Switch calibration bank.

Parameters:

switch_time (int) – switch time

Return type:

None

test_generator_input_select(inputs)

Test generator input select.

Parameters:

inputs (int) – inputs

Return type:

None

test_generator_set_noise(amplitude_noise=0.0, load_time=0)

Set generator test noise.

Parameters:
  • amplitude_noise (float) – amplitude of noise

  • load_time (int) – load time

Return type:

None

test_generator_set_tone(generator, frequency=100000000.0, amplitude=0.0, phase=0.0, load_time=0)

Test generator tone setting.

Parameters:
  • generator (int) – generator select. 0 or 1

  • frequency (float) – Tone frequency in Hz

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

  • phase (float) – Initial tone phase, in turns

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

Return type:

None