"""
The lower-level driver for the QICK library. Contains classes for interfacing with the SoC.
"""
import os
import mmap
from pynq.overlay import Overlay
import xrfclk
import xrfdc
import numpy as np
import time
import queue
import logging
from collections import OrderedDict, defaultdict
from fractions import Fraction
from . import bitfile_path, obtain, get_version
from .ip import SocIP, QickMetadata
from .parser import parse_to_bin
from .streamer import DataStreamer
from .qick_asm import QickConfig
from .asm_v1 import QickProgram
from .asm_v2 import QickProgramV2
from .drivers.generator import *
from .drivers.readout import *
from .drivers.tproc import *
from .drivers.peripherals import *
from .drivers.xcom import *
logger = logging.getLogger(__name__)
[docs]class AxisSwitch(SocIP):
"""
AxisSwitch class to control Xilinx AXI-Stream switch IP
:param nslave: Number of slave interfaces
:type nslave: int
:param nmaster: Number of master interfaces
:type nmaster: int
"""
bindto = ['xilinx.com:ip:axis_switch:1.1']
def _init_config(self, description):
# Number of slave interfaces.
self.NSL = int(description['parameters']['NUM_SI'])
# Number of master interfaces.
self.NMI = int(description['parameters']['NUM_MI'])
self.REGISTERS = {'ctrl': 0x0, 'mix_mux': 0x040}
def _init_firmware(self):
# Init axis_switch.
self.ctrl = 0
self.disable_ports()
[docs] def disable_ports(self):
"""
Disables ports
"""
for ii in range(self.NMI):
offset = self.REGISTERS['mix_mux'] + 4*ii
self.write(offset, 0x80000000)
[docs] def sel(self, mst=0, slv=0):
"""
Digitally connects a master interface with a slave interface
:param mst: Master interface
:type mst: int
:param slv: Slave interface
:type slv: int
"""
# Sanity check.
if slv > self.NSL-1:
print("%s: Slave number %d does not exist in block." %
__class__.__name__)
return
if mst > self.NMI-1:
print("%s: Master number %d does not exist in block." %
__class__.__name__)
return
# Disable register update.
self.ctrl = 0
# Disable all MI ports.
self.disable_ports()
# MI[mst] -> SI[slv]
offset = self.REGISTERS['mix_mux'] + 4*mst
self.write(offset, slv)
# Enable register update.
self.ctrl = 2
[docs]class RFDC(SocIP, xrfdc.RFdc):
"""
Extends the xrfdc driver.
Calling xrfdc functions is slow (typically ~8 ms per call).
We therefore cache parameters that need to be set in program initialization, such as Nyquist zone and frequency.
"""
bindto = ["xilinx.com:ip:usp_rf_data_converter:2.3",
"xilinx.com:ip:usp_rf_data_converter:2.4",
"xilinx.com:ip:usp_rf_data_converter:2.6"]
# consts from https://github.com/Xilinx/embeddedsw/blob/master/XilinxProcessorIPLib/drivers/rfdc/src/xrfdc.h
XRFDC_CAL_BLOCK_OCB1 = 0
XRFDC_CAL_BLOCK_OCB2 = 1
XRFDC_CAL_BLOCK_GCB = 2
XRFDC_CAL_BLOCK_TSCB = 3
XRFDC_GEN3 = 2
# name, ID, number of coefficients
ADC_CAL_BLOCKS = {'OCB1': (XRFDC_CAL_BLOCK_OCB1, 4),
'OCB2': (XRFDC_CAL_BLOCK_OCB2, 4),
'GCB' : (XRFDC_CAL_BLOCK_GCB, 4),
'TSCB': (XRFDC_CAL_BLOCK_TSCB, 8),
}
def _init_config(self, description):
# Nyquist zone for each channel
self.nqz_dict = {'dac': {}, 'adc': {}}
# Rounded NCO frequency for each channel
self.mixer_dict = {'dac': {}, 'adc': {}}
ip_params = description['parameters']
# which generation RFSoC we are using
self.cfg['ip_type'] = int(ip_params['C_IP_Type'])
# quad or dual RF-ADC
self.cfg['hs_adc'] = (ip_params['C_High_Speed_ADC'] == '1')
# dicts of RFDC tiles and channels
self.cfg['tiles'] = {'dac':{}, 'adc':{}}
self.cfg['dacs'] = OrderedDict()
self.cfg['adcs'] = OrderedDict()
# list the enabled DAC+ADC tiles and blocks, and enumerate the "channel name" and tile/block indices for each block
# the channel name is a 2-digit string that gets used in RFDC port and parameter names
# the indices are what you need to index into the dac_tiles/adc_tiles structures
for tiletype in ['dac', 'adc']:
for iTile in range(4):
if ip_params['C_%s%d_Enable' % (tiletype.upper(), iTile)] != '1': continue
tilecfg = {}
self['tiles'][tiletype][iTile] = tilecfg
# some firmwares (older versions of Vivado?) do not have link coupling params for the DAC
if ('C_%s%d_Link_Coupling' % (tiletype.upper(), iTile)) in ip_params:
tilecfg['coupling'] = ['AC', 'DC'][int(ip_params['C_%s%d_Link_Coupling' % (tiletype.upper(), iTile)])]
f_fabric = float(ip_params['C_%s%d_Fabric_Freq' % (tiletype.upper(), iTile)])
f_out = float(ip_params['C_%s%d_Outclk_Freq' % (tiletype.upper(), iTile)])
fs = float(ip_params['C_%s%d_Sampling_Rate' % (tiletype.upper(), iTile)])*1000
tilecfg['fabric_div'] = int(fs/f_fabric)
tilecfg['out_div'] = int(f_fabric/f_out)
#out_div = Fraction(f_fabric/f_out).limit_denominator()
tilecfg['blocks'] = []
for block in range(4):
# pack the indices for the tile/block structure "channel name"
chname = "%d%d" % (iTile, block)
if tiletype == 'adc' and self['hs_adc']:
if block%2 != 0: continue
iBlock = block//2
else:
iBlock = block
# check whether this block is enabled
if ip_params['C_%s_Slice%s_Enable' % (tiletype.upper(), chname)] != 'true': continue
tilecfg['blocks'].append(chname)
self[tiletype+'s'][chname] = {'index': [iTile, iBlock]}
# read the clock settings and block configs
self._read_freqs()
def _get_tile(self, tiletype, iTile):
tiles = {'dac':self.dac_tiles, 'adc':self.adc_tiles}[tiletype]
return tiles[iTile]
def _get_block(self, blocktype, blockname):
iTile, iBlock = self[blocktype+'s'][blockname]['index']
tiles = {'dac':self.dac_tiles, 'adc':self.adc_tiles}[blocktype]
return tiles[iTile].blocks[iBlock]
def _read_freqs(self):
for tiletype in ['dac', 'adc']:
for iTile, tilecfg in self['tiles'][tiletype].items():
#tilecfg.clear()
tile = self._get_tile(tiletype, iTile)
pllcfg = tile.PLLConfig
tilecfg['f_ref'] = pllcfg['RefClkFreq']
tilecfg['ref_div'] = pllcfg['RefClkDivider']
tilecfg['fs_mult'] = pllcfg['FeedbackDivider']
tilecfg['fs_div'] = pllcfg['RefClkDivider']*pllcfg['OutputDivider']
# we could use SampleRate here, but it's the same
tilecfg['fs'] = tilecfg['f_ref']*tilecfg['fs_mult']/tilecfg['fs_div']
tilecfg['f_fabric'] = tilecfg['fs']/tilecfg['fabric_div']
tilecfg['f_out'] = tilecfg['f_fabric']/tilecfg['out_div']
"""
# lookup table for deciding whether the AXI-S interface uses IQ or real data
# this only covers the cases we actually use in our generators/ROs
mixer2iq = {}
mixer2iq['dac'] = {
(xrfdc.MIXER_TYPE_FINE, xrfdc.MIXER_MODE_C2R): 2,
(xrfdc.MIXER_TYPE_COARSE, xrfdc.MIXER_MODE_R2R): 1,
}
mixer2iq['adc'] = {
(xrfdc.MIXER_TYPE_COARSE, xrfdc.MIXER_MODE_R2C): 2,
(xrfdc.MIXER_TYPE_COARSE, xrfdc.MIXER_MODE_R2R): 1,
}
fabric_divs = {k:{iTile:[] for iTile in v.keys()} for k,v in self['tiles'].items()}
"""
for tiletype in ['dac', 'adc']:
for chname, chcfg in self[tiletype+'s'].items():
iTile, iBlock = chcfg['index']
#chcfg.clear()
#chcfg['index'] = [iTile, iBlock]
# copy the tile info
chcfg.update(self['tiles'][tiletype][iTile])
# clean up parameters that are only used at the tile level
del chcfg['ref_div'], chcfg['out_div'], chcfg['fabric_div'], chcfg['f_out'], chcfg['blocks']
block = self._get_tile(tiletype, iTile).blocks[iBlock]
"""
# now we compute the ratio between the sample and fabric clocks
# this is surprisingly annoying to do in full generality
# https://docs.amd.com/r/en-US/pg269-rf-data-converter/RF-DAC-Interface-Data-and-Clock-Rates
# https://docs.amd.com/r/en-US/pg269-rf-data-converter/RF-ADC-Interface-Data-and-Clock-Rates
# note that this ratio has to be the same for all channels in a tile
if tiletype == 'dac':
data_width = block.FabWrVldWords
else:
data_width = block.FabRdVldWords
mixer_settings = block.MixerSettings
iq = mixer2iq[tiletype][tuple(mixer_settings[k] for k in ['MixerType', 'MixerMode'])]
"""
if tiletype == 'dac':
chcfg['interpolation'] = block.InterpolationFactor
if self['ip_type'] == self.XRFDC_GEN3:
chcfg['datapath'] = block.DataPathMode
#chcfg['fabric_div'] = data_width*chcfg['interpolation']//iq
else:
chcfg['decimation'] = block.DecimationFactor
#chcfg['fabric_div'] = data_width*iq//chcfg['decimation']
#fabric_divs[tiletype][iTile].append(chcfg['fabric_div'])
#chcfg['f_fabric'] = chcfg['fs']/chcfg['fabric_div']
"""
for tiletype in ['dac', 'adc']:
for iTile, tiledivs in fabric_divs[tiletype].items():
assert len(set(tiledivs)) == 1
tilecfg = self['tiles'][tiletype][iTile]
tilecfg['fabric_div'] = tiledivs[0]
tilecfg['f_out'] = tilecfg['fs']/tilecfg['fabric_div']/tilecfg['out_div']
"""
[docs] def map_clocks(self, soc):
"""Map the clock networks driving the various IP blocks, and determine the resulting constraints on the RFDC sampling rates.
This method gets run early in QickSoc initialization because it's needed to check validity of a requested set of sampling freqs.
This code assumes that the RFDC configuration dictionary has been filled (this happens in RFDC driver initialization).
It does not assume that configure_connections() has been run on all drivers.
"""
# first, gather information
# search for IP blocks with trace_clocks() methods - typically this is just the tProc
# we run trace_clocks() here
# it will also run as part of QickSoc init (via configure_connections()), but that's after sampling rate modification
clk_groups = defaultdict(list)
for blockname, blockdict in soc.ip_dict.items():
if hasattr(blockdict['driver'], 'trace_clocks'):
ip = soc._get_block(blockname)
ip.trace_clocks(soc)
for clkname, clkcfg in ip['clk_srcs'].items():
clkid = (blockname, clkname)
clk_groups[clkcfg['source']].append([clkid, clkcfg['src_range']])
# check all RFDC inputs and outputs
for tiletype, direction in [('dac', 's'), ('adc', 'm')]:
for iTile in self['tiles'][tiletype].keys():
clkcfg = soc.metadata.trace_clk_back(self['fullpath'],'%s%d_axis_aclk'%(direction, iTile))
clkid = (tiletype, iTile)
clk_groups[clkcfg['source']].append([clkid, clkcfg['src_range']])
# now, analyze clock groups
self.cfg['clk_groups'] = []
for clk_src, clk_dests in clk_groups.items():
# find RFDC tiles whose fabric clocks come from this source
fs_group = [x[0] for x in clk_dests]
# find limits imposed on the clock source freq
src_ranges = [x[1] for x in clk_dests if x[1] is not None]
if clk_src[0] in ['dac', 'adc']:
if fs_group:
self['clk_groups'].append(fs_group)
tilecfg = self['tiles'][clk_src[0]][clk_src[1]]
# cross-check
# this isn't a fundamental rule, we might end up making a firmware that violates it
# for now, this assumption simplifies thinking about clock groups
if clk_src not in fs_group:
raise RuntimeError("%s tile %d drives logic, but not its own fabric clock. There may be a problem with this firmware design."%(clk_src[0].upper(), clk_src[1]))
# add the output clock limits to the tile info
if src_ranges:
tilecfg['outclk_limits'] = [max([x[0] for x in src_ranges]), min([x[1] for x in src_ranges])]
def clocks_locked(self):
lockdict = {}
for tiletype in ['dac', 'adc']:
lockdict[tiletype] = {}
for i in self['tiles'][tiletype]:
lockdict[tiletype][i] = (self._get_tile(tiletype, i).PLLLockStatus == 2)
return lockdict
def tile_states(self):
status = self.IPStatus
statedict = {}
for tiletype in ['dac', 'adc']:
statedict[tiletype] = {}
for i in self['tiles'][tiletype]:
statedict[tiletype][i] = status[tiletype.upper()+"TileStatus"][i]['TileState']
return statedict
[docs] def restart_all_tiles(self):
"""
Restart all DAC and ADC tiles using XRFdc_StartUp.
This will re-lock tile PLLs, reset DDS phases to random values, and reset all logic driven by RF clocks.
It won't reset RFDC registers (e.g. if you set custom sampling rates, those will stay in place).
"""
fails = []
for tiletype in ['dac', 'adc']:
for i in self['tiles'][tiletype]:
try:
self._get_tile(tiletype, i).StartUp()
except Exception as e:
logger.error("failed to restart %s tile %d" % (tiletype.upper(), i))
fails.append(e.args)
if fails:
err = ("Some DAC or ADC tiles failed to start up, with the following error messages " +
"(see https://docs.amd.com/r/en-US/pg269-rf-data-converter/Power-on-Sequence-Steps for state numbers):\n" +
"\n".join(["\n".join(fail) for fail in fails])
)
raise RuntimeError(err)
[docs] def valid_sample_rates(self, tiletype, tile):
"""
Return an array of valid sample rates.
This code is based on XRFdc_SetPLLConf().
"""
if tiletype not in ['dac', 'adc']:
raise RuntimeError('tiletype must be "dac" or "adc"')
if tile not in self['tiles'][tiletype]:
raise RuntimeError('specified tile is not enabled in this firmware')
tilecfg = self['tiles'][tiletype][tile]
# reference clock after the PLL reference divider
# this divider can't be changed by software, and Xilinx recommends keeping it at 1 for best phase noise
refclk = tilecfg['f_ref']/tilecfg['ref_div']
words_per_axi = tilecfg['fabric_div']
# Allowed divider values, see PG269 "PLL Parameters"
# https://docs.amd.com/r/en-US/pg269-rf-data-converter/PLL-Parameters
# Note that the values in PG269 (in comments) are rounded and can't be used literally.
# The values here are the VCO_RANGE_* constants used by XRFdc_SetPLLConf().
Fb_div_vals = np.arange(13,161, dtype=int)
if self['ip_type'] == self.XRFDC_GEN3 and tiletype=='dac':
M_vals = np.concatenate([[1,2,3], np.arange(4,66,2)])
VCO_range = [7800, 13800] # [7863, 13760]
else:
M_vals = np.concatenate([[2,3], np.arange(4,66,2)])
VCO_range = [8500, 13200]
VCO_possible = refclk * Fb_div_vals
Fb_div_possible = Fb_div_vals[(VCO_possible>=VCO_range[0]) & (VCO_possible<=VCO_range[1])]
fs_possible = refclk*(Fb_div_possible.T/M_vals[:,np.newaxis]).ravel()
if 'outclk_limits' in tilecfg:
fs_range = [x*tilecfg['out_div']*tilecfg['fabric_div'] for x in tilecfg['outclk_limits']]
fs_possible = fs_possible[fs_possible >= fs_range[0]]
fs_possible = fs_possible[fs_possible <= fs_range[1]]
# See DS926 "RF-ADC/RF-DAC to PL Interface Performance"
# https://docs.amd.com/r/en-US/ds926-zynq-ultrascale-plus-rfsoc/RF-ADC/RF-DAC-to-PL-Interface-Switching-Characteristics
if self['ip_type'] == self.XRFDC_GEN3:
max_axi_clk = 614 # MHz
else:
max_axi_clk = 520 # MHz
fs_possible = fs_possible[fs_possible <= words_per_axi*max_axi_clk]
# Allowed ranges of sampling freqs
# https://docs.amd.com/r/en-US/ds926-zynq-ultrascale-plus-rfsoc/RF-DAC-Electrical-Characteristics
# https://docs.amd.com/r/en-US/ds926-zynq-ultrascale-plus-rfsoc/RF-ADC-Electrical-Characteristics
if tiletype=='adc' and self['hs_adc']:
fs_min = 1000
else:
fs_min = 500
fs_possible = fs_possible[fs_possible >= fs_min]
if tiletype=='dac':
if self['ip_type'] == self.XRFDC_GEN3:
fs_max = 9850
else:
fs_max = 6554
else:
if self['ip_type'] < self.XRFDC_GEN3:
fs_max = 4096 # ZCU111
else:
if self['hs_adc']:
fs_max = 5000
else:
fs_max = 2500
fs_possible = fs_possible[fs_possible <= fs_max]
# special rules for Gen3 RFSoC DACs
if self['ip_type'] == self.XRFDC_GEN3 and tiletype=='dac':
# forbidden "hole" for Gen3 RFSoC DAC PLL
# https://docs.amd.com/r/en-US/ds926-zynq-ultrascale-plus-rfsoc/RF-Converters-Clocking-Characteristics
# actually, this is a consequence of the VCO range
fs_possible = fs_possible[(fs_possible<=6882) | (fs_possible>=7863)]
# in datapath mode 1, Gen3 DACs can't go above 7 Gsps
# https://docs.amd.com/r/en-US/pg269-rf-data-converter/RF-DAC-High-Sampling-Rates-Mode-Gen-3/DFE
# https://docs.amd.com/r/en-US/ds926-zynq-ultrascale-plus-rfsoc/RF-DAC-Electrical-Characteristics
datapaths = [self[tiletype+'s'][chname]['datapath'] for chname in tilecfg['blocks']]
if any([x==1 for x in datapaths]):
fs_possible = fs_possible[fs_possible<=7000]
fs_possible.sort()
return fs_possible
[docs] def round_sample_rate(self, tiletype, tile, fs_target):
"""
Return the closest achievable sample rate to the requested value.
"""
if tiletype not in ['dac', 'adc']:
raise RuntimeError('tiletype must be "dac" or "adc"')
if tile not in self['tiles'][tiletype]:
raise RuntimeError('specified tile is not enabled in this firmware')
fs_possible = self.valid_sample_rates(tiletype, tile)
fs_best = fs_possible[np.argmin(np.abs(fs_possible - fs_target))]
return fs_best
def _set_sample_rate(self, tiletype, tile, fs):
"""
Set the sample rate of a tile.
It's assumed that the requested frequency has already been validated and rounded to a valid value.
Parameters
----------
tiletype : str
'dac' or 'adc'
tile : int
Tile number (0-3)
fs : float
Requested sample rate, in Msps
"""
self.logger.info('programming %s tile %d to %.3f Msps'%(tiletype.upper(), tile, fs))
f_ref = self['tiles'][tiletype][tile]['f_ref']
self._get_tile(tiletype, tile).DynamicPLLConfig(source=xrfdc.CLK_SRC_PLL, ref_clk_freq=f_ref, samp_rate=fs)
[docs] def set_mixer_freq(self, blockname, f, blocktype='dac', phase_reset=True, force=False):
"""
Set the NCO frequency that will be mixed with the generator output (for DAC) or raw ADC data (for ADC).
Note that the RFdc driver does its own math to round the frequency to the NCO's frequency step.
If you want predictable behavior, the frequency you use here should already be rounded.
Rounding is normally done for you as part of AbsQickProgram.declare_gen().
blockname : int
channel ID (2-digit string)
f : float
NCO frequency (MHz)
blocktype : str
'dac' or 'adc'
force: bool
force update, even if the setting is the same
phase_reset : bool
if we change the frequency, also reset the NCO's phase accumulator
"""
if not force and f == self.get_mixer_freq(blockname, blocktype):
return
blk = self._get_block(blocktype, blockname)
tile, channel = self[blocktype+'s'][blockname]['index']
# Make a copy of mixer settings.
blk_mixer = blk.MixerSettings
if blk_mixer['MixerType'] != xrfdc.MIXER_TYPE_FINE:
raise RuntimeError("tried to set mixer freq for %s %s, but mixer is not enabled" % (blocktype.upper(), blockname))
new_mixcfg = blk_mixer.copy()
# Update the copy
new_mixcfg.update({
'EventSource': xrfdc.EVNT_SRC_IMMEDIATE,
'Freq': f,
'MixerType': xrfdc.MIXER_TYPE_FINE,
'PhaseOffset': 0})
# Update settings.
blk.MixerSettings = new_mixcfg
blk.UpdateEvent(xrfdc.EVENT_MIXER)
# The phase reset is mostly important when setting the frequency to 0: you want the NCO to end up at 1 instead of a complex value.
# So we apply the reset after setting the new frequency (otherwise you accumulate some rotation before stopping the NCO).
if phase_reset: blk.ResetNCOPhase()
self.mixer_dict[blocktype][blockname] = f
def get_mixer_freq(self, blockname, blocktype='dac'):
try:
return self.mixer_dict[blocktype][blockname]
except KeyError:
blk_mixer = self._get_block(blocktype, blockname).MixerSettings
if blk_mixer['MixerType'] != xrfdc.MIXER_TYPE_FINE:
raise RuntimeError("tried to get mixer freq for %s %s, but mixer is not enabled" % (blocktype.upper(), blockname))
self.mixer_dict[blocktype][blockname] = blk_mixer['Freq']
return self.mixer_dict[blocktype][blockname]
[docs] def set_nyquist(self, blockname, nqz, blocktype='dac', force=False):
"""
Sets channel to operate in Nyquist zone nqz.
This setting doesn't change the DAC output frequencies:
you will always have some power at both the demanded frequency and its image(s).
Setting the NQZ to 2 increases output power in the 2nd/3rd Nyquist zones.
See "RF-DAC Nyquist Zone Operation" in PG269.
:param blockname: channel ID (2-digit string)
:type blockname: int
:param nqz: Nyquist zone (1 or 2)
:type nqz: int
:param blocktype: 'dac' or 'adc'
:type blocktype: str
:param force: force update, even if the setting is the same
:type force: bool
"""
if nqz not in [1,2]:
raise RuntimeError("Nyquist zone must be 1 or 2")
if blocktype not in ['dac','adc']:
raise RuntimeError("Block type must be adc or dac")
if not force and self.get_nyquist(blockname, blocktype) == nqz:
return
blk = self._get_block(blocktype, blockname)
blk.NyquistZone = nqz
self.nqz_dict[blocktype][blockname] = nqz
[docs] def get_nyquist(self, blockname, blocktype='dac'):
"""
Get the current Nyquist zone setting for a channel.
Parameters
----------
blockname : str
Channel ID (2-digit string)
blocktype : str
'dac' or 'adc'
Returns
-------
int
NQZ setting (1 or 2)
"""
if blocktype not in ['dac','adc']:
raise RuntimeError("Block type must be adc or dac")
try:
return self.nqz_dict[blocktype][blockname]
except KeyError:
blk = self._get_block(blocktype, blockname)
self.nqz_dict[blocktype][blockname] = blk.NyquistZone
return self.nqz_dict[blocktype][blockname]
[docs] def get_adc_attenuator(self, blockname):
"""Read the ADC's built-in step attenuator.
Only available for RFSoC Gen 3 (ZCU216, RFSoC4x2).
Parameters
----------
blockname : str
Channel ID (2-digit string)
Returns
-------
float
Attenuation value (dB)
"""
if self['ip_type'] < self.XRFDC_GEN3:
raise RuntimeError("you tried to access the RF-ADC attenuator, but this only exists on Gen 3 RFSoC (ZCU216, RFSoC4x2).")
adc = self._get_block('adc', blockname)
return adc.DSA['Attenuation']
[docs] def set_adc_attenuator(self, blockname, attenuation):
"""Set the ADC's built-in step attenuator.
The requested value will be rounded to the nearest valid value (0-27 dB inclusive, 1 dB steps).
Only available for RFSoC Gen 3 (ZCU216, RFSoC4x2).
Parameters
----------
blockname : str
Channel ID (2-digit string)
attenuation : float
Attenuation value (dB)
"""
if self['ip_type'] < self.XRFDC_GEN3:
raise RuntimeError("you tried to access the RF-ADC attenuator, but this only exists on Gen 3 RFSoC (ZCU216, RFSoC4x2).")
adc = self._get_block('adc', blockname)
attenuation = np.round(attenuation)
adc.DSA['Attenuation'] = attenuation
return attenuation
[docs] def get_adc_cal(self, blockname):
"""Get the current calibration coefficients for an ADC.
Parameters
----------
blockname : str
Channel ID (2-digit string)
Returns
-------
dict of list
Calibration coefficients
"""
adc = self._get_block('adc', blockname)
a = xrfdc._ffi.new("XRFdc_Calibration_Coefficients *")
cal = {}
for name, (const, n) in self.ADC_CAL_BLOCKS.items():
adc.GetCalCoefficients(const, a)
cal[name] = [getattr(a, 'Coeff%d'%(i)) for i in range(n)]
return cal
[docs] def set_adc_cal(self, blockname, cal, calblocks):
"""Set calibration coefficients for an ADC.
See the Xilinx documentation for explanations and cautions:
https://docs.amd.com/r/en-US/pg269-rf-data-converter/Getting/Setting-Calibration-Coefficients
Parameters
----------
blockname : str
Channel ID (2-digit string)
cal : dict of list
Calibration coefficients
calblocks : list of str
List of calibration blocks to configure.
Valid values are OCB1, OCB2, GCB, TSCB.
Returns
-------
dict of list
Calibration coefficients
"""
adc = self._get_block('adc', blockname)
for name, (const, n) in self.ADC_CAL_BLOCKS.items():
a = xrfdc._ffi.new("XRFdc_Calibration_Coefficients *")
for i in range(n):
setattr(a, 'Coeff%d'%(i), cal[name][i])
adc.SetCalCoefficients(const, a)
[docs] def restart_adc_tile(self, tile):
"""Restart an ADC tile.
This is useful as a way to rerun the OCB2 offset calibration, though of course it resets all of the calibrations.
WHatever voltage the ADCs on this tile are seeing when you run this, that will be 0 ADU.
Parameters
----------
tile : int
ADC tile number (0-3)
"""
self.adc_tiles[tile].Reset()
[docs] def freeze_adc_cal(self, blockname):
"""Freeze an ADC's calibration (stop the background calibration).
See the Xilinx documentation:
https://docs.amd.com/r/en-US/pg269-rf-data-converter/Background-Calibration-Process
Parameters
----------
blockname : str
Channel ID (2-digit string)
"""
adc = self._get_block('adc', blockname)
adc.CalFreeze['FreezeCalibration'] = 1
[docs] def unfreeze_adc_cal(self, blockname, calblocks=None):
"""Unfreeze an ADC's calibration (resume the background calibration).
See the Xilinx documentation:
https://docs.amd.com/r/en-US/pg269-rf-data-converter/Background-Calibration-Process
Parameters
----------
blockname : str
Channel ID (2-digit string)
"""
adc = self._get_block('adc', blockname)
adc.CalFreeze['FreezeCalibration'] = 0
if calblocks is None:
if self['ip_type'] < self.XRFDC_GEN3:
calblocks = ['OCB2', 'GCB', 'TSCB']
else:
calblocks = ['OCB1', 'OCB2', 'GCB', 'TSCB']
for calblock in calblocks:
adc.DisableCoefficientsOverride(self.ADC_CAL_BLOCKS[calblock][0])
[docs]class QickSoc(Overlay, QickConfig):
"""
This class loads, initializes, and provides access to the QICK firmware.
Parameters
----------
bitfile : str
Path to the firmware bitfile. This should end with .bit, and the corresponding .hwh file must be in the same directory.
download : bool
Load the bitfile into the FPGA logic. If you are certain that the bitfile you specified is already running, you can use False here.
no_tproc : bool
Use if this is a special firmware that doesn't have a tProcessor.
no_rf : bool
Use if this is a special firmware that doesn't have an RF data converter.
force_init_clks : bool
Re-initialize the board clocks regardless of whether they appear to be locked. Specifying (as True or False) the clk_output or external_clk options will also force clock initialization.
clk_output: bool or None
If true, output a copy of the RF reference. This option is supported for the ZCU111 (get 122.88 MHz from J108) and ZCU216 (get 245.76 MHz from OUTPUT_REF J10).
external_clk: bool or None
If true, lock the board clocks to an external reference. This option is supported for the ZCU111 (put 12.8 MHz on External_REF_CLK J109), ZCU216 (put 10 MHz on INPUT_REF_CLK J11), and RFSoC 4x2 (put 10 MHz on CLK_IN).
dac_sample_rates : dict[int, float] or None
Sample rates to override the values compiled into the firmware.
This should be a dictionary mapping DAC tiles to sample rates (in megasamples per second).
adc_sample_rates : dict[int, float] or None
Sample rates to override the values compiled into the firmware.
This should be a dictionary mapping ADC tiles to sample rates (in megasamples per second).
"""
# The following constants are no longer used. Some of the values may not match the bitfile.
# fs_adc = 384*8 # MHz
# fs_dac = 384*16 # MHz
# pulse_mem_len_IQ = 65536 # samples for I, Q
# ADC_decim_buf_len_IQ = 1024 # samples for I, Q
# ADC_accum_buf_len_IQ = 16384 # samples for I, Q
#tProc_instruction_len_bytes = 8
#tProc_prog_mem_samples = 8000
#tProc_prog_mem_size_bytes_tot = tProc_instruction_len_bytes*tProc_prog_mem_samples
#tProc_data_len_bytes = 4
#tProc_data_mem_samples = 4096
#tProc_data_mem_size_bytes_tot = tProc_data_len_bytes*tProc_data_mem_samples
#tProc_stack_len_bytes = 4
#tProc_stack_samples = 256
#tProc_stack_size_bytes_tot = tProc_stack_len_bytes*tProc_stack_samples
#phase_resolution_bits = 32
#gain_resolution_signed_bits = 16
# Constructor.
def __init__(self, bitfile=None, download=True, no_tproc=False, no_rf=False, force_init_clks=False, clk_output=None, external_clk=None, dac_sample_rates=None, adc_sample_rates=None, **kwargs):
if bitfile is None:
bitfile = bitfile_path()
# 1. read the config from the HWH file and (optionally) download the bitstream into the FPGA with Overlay.__init__()
# 2. check and (if necessary) configure the reference clocks with QickSoc.config_clocks() - there must be a loaded bitstream at this point, to check the clocks
# 2a. if we configure the clocks, we re-download the bitstream
# 3. initialize IP blocks and map connections with QickSoc.map_signal_paths() - this must be done after download, otherwise the IPs will get reset by download
# NOTE: the exception to this is the RFDC - we initialize that IP in step 2, because we need to check for PLL lock, but it doesn't seem to do anything stateful in its init
# Read the bitstream configuration from the HWH file.
# If download=True, we also program the FPGA.
Overlay.__init__(self, bitfile, ignore_version=True, download=download, **kwargs)
# Initialize the configuration
self._cfg = {}
QickConfig.__init__(self)
self['board'] = os.environ["BOARD"]
self['sw_version'] = get_version()
# a space to dump any additional lines of config text which you want to print in the QickConfig
self['extra_description'] = []
# Extract the IP connectivity information from the HWH parser and metadata.
self.metadata = QickMetadata(self)
self['fw_timestamp'] = self.metadata.timestamp
# list of objects that need to be registered for autoproxying over Pyro
self.autoproxy = []
# Initialize lists of IP blocks.
# Signal generators (anything driven by the tProc)
self.gens = []
# Constant generators
self.iqs = []
# Average + Buffer blocks.
self.avg_bufs = []
# Readout blocks.
self.readouts = []
# Time-tagger blocks.
self.time_taggers = []
if not no_rf:
# RF data converter (for configuring ADCs and DACs, and setting NCOs)
self.rf = self.usp_rf_data_converter_0
self['rf'] = self.rf.cfg
# map the clock networks, so we can validate the requested sampling rates
self.rf.map_clocks(self)
# Examine the RFDC config to find the reference clock frequency.
refclks = []
for tiletype in ['dac', 'adc']:
refclks.extend([v['f_ref'] for k,v in self.rf['tiles'][tiletype].items()])
if len(set(refclks)) != 1:
raise RuntimeError("This firmware wants RF reference clocks %s, but they must all be equal"%(refclks))
self['refclk_freq'] = refclks[0]
# Configure xrfclk reference clocks
self.config_clocks(force_init_clks, clk_output, external_clk)
# Update the ADC sample rate if specified
if dac_sample_rates or adc_sample_rates:
self.rf.configure_sample_rates(dac_sample_rates, adc_sample_rates)
self.map_signal_paths(no_tproc)
if not no_tproc:
#self.tnet = self.qick_net_0
self._streamer = DataStreamer(self)
self.autoproxy.extend([self.streamer, self.tproc])
@property
def tproc(self):
return self._tproc
@property
def streamer(self):
return self._streamer
def _get_block(self, fullpath):
"""Return the IP block specified by its full path.
Blocks inside hierarchies have slash-separated fullpaths e.g. "rfb_control/attn_spi"
PYNQ has two ways to access blocks inside hierarchies: recursively (soc.rfb_control.attn_spi) or directly (getattr(soc, "rfb_control/attn_spi")).
Annoyingly, these are separate instances of the driver object so you have to be consistent about which you use.
For some types of blocks (e.g. memories) the direct access uses a modified path with slashes removed (e.g. ddr4ddr4_0 instead of ddr4/ddr4_0).
To avoid these complications, we use this method, which consistently uses recursive access.
"""
block = self
# recurse into hierarchies, if present
for x in fullpath.split('/'):
block = getattr(block, x)
return block
[docs] def map_signal_paths(self, no_tproc):
"""
Make lists of signal generator, readout, and buffer blocks in the firmware.
Also map the switches connecting the generators and buffers to DMA.
Fill the config dictionary with parameters of the DAC and ADC channels.
"""
# Use the HWH parser to trace connectivity and deduce the channel numbering.
# Some blocks (e.g. DDR4) are inside hierarchies.
# We access these through the hierarchy (e.g. self.ddr4.axis_buffer_ddr_0)
# but list them using ip_dict, which has all blocks, even those inside hierarchies
for key, val in self.ip_dict.items():
if hasattr(val['driver'], 'configure_connections'):
self._get_block(val['fullpath']).configure_connections(self)
if not no_tproc:
# tProcessor, 64-bit instruction, 32-bit registers, x8 channels.
if 'axis_tproc64x32_x8_0' in self.ip_dict:
self.TPROC_VERSION = 1
self._tproc = self.axis_tproc64x32_x8_0
self._tproc.configure(self.axi_bram_ctrl_0, self.axi_dma_tproc)
elif 'qick_processor_0' in self.ip_dict:
self.TPROC_VERSION = 2
self._tproc = self.qick_processor_0
self._tproc.configure(self.axi_dma_tproc)
else:
raise RuntimeError('No tProcessor found')
self['tprocs'] = [self.tproc.cfg]
else:
self.TPROC_VERSION = 0
# temporary lists for blocks that we only expect to see once
ddr4_buf = []
mr_buf = []
# Populate the lists with the registered IP blocks.
for key, val in self.ip_dict.items():
if issubclass(val['driver'], AbsPulsedSignalGen):
self.gens.append(self._get_block(key))
elif val['driver'] == AxisConstantIQ:
self.iqs.append(self._get_block(key))
elif issubclass(val['driver'], AbsReadout):
self.readouts.append(self._get_block(key))
elif issubclass(val['driver'], AxisAvgBuffer):
self.avg_bufs.append(self._get_block(key))
elif issubclass(val['driver'], AxisBufferDdrV1):
ddr4_buf.append(self._get_block(key))
elif issubclass(val['driver'], MrBufferEt):
mr_buf.append(self._get_block(key))
elif val['driver'] == QICK_Time_Tagger:
self.time_taggers.append(self._get_block(key))
# AxisReadoutV3 isn't a PYNQ-registered IP block, so we add it here
for buf in self.avg_bufs:
if buf.readout not in self.readouts:
self.readouts.append(buf.readout)
# Sort the lists.
# We order gens by the tProc port number and tProc mux port number (if present).
# We order buffers by the switch port number.
# Those orderings are important, since those indices get used in programs.
self.gens.sort(key=lambda x:(x['tproc_ch'], x._cfg.get('tmux_ch')))
self.avg_bufs.sort(key=lambda x: x.switch_ch)
# The IQ and readout orderings aren't critical for anything.
self.iqs.sort(key=lambda x: x['dac'])
self.readouts.sort(key=lambda x: x['adc'])
# Configure the drivers.
for i, gen in enumerate(self.gens):
gen.configure(i, self.rf)
for i, iq in enumerate(self.iqs):
iq.configure(i, self.rf)
for readout in self.readouts:
readout.configure(self.rf)
# Filter the list of MR buffers to select the ones wired to readout blocks (discarding those wired directly to ADCs or other blocks).
# There should only be one.
mr_buf = [x for x in mr_buf if x['readouts']]
if len(mr_buf) == 1:
self.mr_buf = mr_buf[0]
self['mr_buf'] = self.mr_buf.cfg
elif len(mr_buf) > 1:
raise RuntimeError("found multiple MR buffers wired to readouts, which is not currently supported by the software")
# Find the DDR4 controller and buffer, if present.
if len(ddr4_buf) == 1:
self.ddr4_buf = ddr4_buf[0]
self['ddr4_buf'] = self.ddr4_buf.cfg
elif len(ddr4_buf) > 1:
raise RuntimeError("found multiple DDR4 buffers, which is not currently supported by the software")
# Fill the config dictionary with driver parameters.
self['gens'] = [gen.cfg for gen in self.gens]
self['iqs'] = [iq.cfg for iq in self.iqs]
self['time_taggers'] = [x.cfg for x in self.time_taggers]
# In the config, we define a "readout" as the chain of ADC+readout+buffer.
def merge_cfgs(bufcfg, rocfg):
merged = {**bufcfg, **rocfg}
for k in set(bufcfg.keys()) & set(rocfg.keys()):
del merged[k]
merged["avgbuf_"+k] = bufcfg[k]
merged["ro_"+k] = rocfg[k]
return merged
self['readouts'] = [merge_cfgs(buf.cfg, buf.readout.cfg) for buf in self.avg_bufs]
[docs] def config_clocks(self, force_init_clks, clk_output, external_clk):
"""
Configure PLLs if requested, or if any ADC/DAC is not locked.
The ADC/DAC PLL lock status is read through the RFDC IP, so this assumes that the bitstream has already been downloaded.
The reference clock frequency must already have been read from the firmware config.
"""
# if we're changing the clock config, we must set the clocks to apply the config
if force_init_clks or (external_clk is not None) or (clk_output is not None):
print("configuring reference clock chips, as requested")
self.set_all_clks(clk_output, external_clk)
else:
# only set clocks if the RFDC isn't locked
if not self.clocks_locked():
print("RFSoC PLLs are not locked, configuring reference clock chips (this is normal after power cycle)")
self.set_all_clks(clk_output, external_clk)
# Check if all DAC and ADC PLLs are locked.
if not self.clocks_locked():
print(
"Not all DAC and ADC PLLs are locked. The FPGA may not be getting a good reference clock from the on-board clock chips.")
[docs] def clocks_locked(self):
"""
Checks whether the DAC and ADC PLLs are locked.
This can only be run after the bitstream has been downloaded.
A failure usually means the FPGA is not getting a good reference clock from the on-board clock chips.
:return: clock status
:rtype: bool
"""
lockdict = self.rf.clocks_locked()
for tiletype in ['dac', 'adc']:
if not all(list(lockdict[tiletype].values())):
return False
return True
[docs] def set_all_clks(self, clk_output, external_clk):
"""
Resets all the board clocks
"""
if self['board'] == 'ZCU111':
# master clock generator is LMK04208, always outputs 122.88
# DAC/ADC are clocked by LMX2594
# available: 102.4, 204.8, 409.6, 737.0
lmk_freq = 122.88
lmx_freq = self['refclk_freq']
print("LMK04208 clock reference = %.3f MHz, LMX2594 clock synth = %.3f MHz" % (lmk_freq, lmx_freq))
if hasattr(xrfclk, "xrfclk"): # pynq 2.7
# load the default clock chip configurations from file, so we can then modify them
xrfclk.xrfclk._find_devices()
xrfclk.xrfclk._read_tics_output()
if clk_output:
# change the register for the LMK04208 chip's 5th output, which goes to J108
# we need this for driving the RF board
xrfclk.xrfclk._Config['lmk04208'][lmk_freq][6] = 0x00140325
if external_clk:
# default value is 0x2302886D
xrfclk.xrfclk._Config['lmk04208'][lmk_freq][14] = 0x2302826D
else: # pynq 2.6
if clk_output:
# change the register for the LMK04208 chip's 5th output, which goes to J108
# we need this for driving the RF board
xrfclk._lmk04208Config[lmk_freq][6] = 0x00140325
else: # restore the default
xrfclk._lmk04208Config[lmk_freq][6] = 0x80141E05
if external_clk:
xrfclk._lmk04208Config[lmk_freq][14] = 0x2302826D
else: # restore the default
xrfclk._lmk04208Config[lmk_freq][14] = 0x2302886D
xrfclk.set_all_ref_clks(lmx_freq)
elif self['board'] == 'ZCU216':
# master clock generator is LMK04828, which is used for DAC/ADC clocks
# only 245.76 available by default
# LMX2594 is not used
# available: 102.4, 204.8, 409.6, 491.52, 737.0
lmk_freq = self['refclk_freq']
lmx_freq = self['refclk_freq']*2
print("LMK04828 clock reference = %.3f MHz, LMX2594 clock synth = %.3f MHz" % (lmk_freq, lmx_freq))
assert hasattr(xrfclk, "xrfclk") # ZCU216 only has a pynq 2.7 image
xrfclk.xrfclk._find_devices()
xrfclk.xrfclk._read_tics_output()
if external_clk:
# default value is 0x01471A
xrfclk.xrfclk._Config['lmk04828'][lmk_freq][80] = 0x01470A
if clk_output:
# default value is 0x012C22
xrfclk.xrfclk._Config['lmk04828'][lmk_freq][55] = 0x012C02
xrfclk.set_ref_clks(lmk_freq=lmk_freq, lmx_freq=lmx_freq)
elif self['board'] == 'RFSoC4x2':
# master clock generator is LMK04828, always outputs 245.76
# DAC/ADC are clocked by LMX2594
# available: 102.4, 204.8, 409.6, 491.52, 737.0
lmk_freq = 245.76
lmx_freq = self['refclk_freq']
print("LMK04828 clock reference = %.3f MHz, LMX2594 clock synth = %.3f MHz" % (lmk_freq, lmx_freq))
assert hasattr(xrfclk, "xrfclk") # RFSoC4x2 only has a pynq 3.0 image
xrfclk.xrfclk._find_devices()
xrfclk.xrfclk._read_tics_output()
if external_clk:
# default value is 0x01471A
xrfclk.xrfclk._Config['lmk04828'][lmk_freq][80] = 0x01470A
xrfclk.set_ref_clks(lmk_freq=lmk_freq, lmx_freq=lmx_freq)
# wait for the clock chips to lock
time.sleep(1.0)
# initialize the FPGA
self.download()
# or: force the tile PLLs to relock
#self.rf.restart_all_tiles()
# or: reset PL, wait for reset
#self.pl_reset(reinit=False)
#time.sleep(1.0)
[docs] def pl_reset(self, reinit=True):
"""Reset all firmware IP blocks.
This pulses the pl_resetn0 line from the PS (also known as the "fabric reset" or the "PS-PL reset").
Every firmware block's reset logic is triggered by this pulse.
The main visible effect of this reset is to reset the start times of all the phase-coherent DDS oscillators.
Some firmware blocks have startup code in their __initialize__() to configure the block after the firmware is loaded.
That configuration gets wiped out by a PL reset, so this method also re-configures those blocks.
More info on the PL reset:
The reset sequence is defined in psu_ps_pl_reset_config_data() in psu_init.c.
You can find this in the BSP or the firmware project files.
The reset sequence and the memory addresses invovled appear to be the same for all RFSoCs.
https://support.xilinx.com/s/article/68962
https://support.xilinx.com/s/question/0D52E00006lLhBnSAK/zynq-ultrascale-howto-reset-the-pl
https://docs.amd.com/r/en-US/ug1137-zynq-ultrascale-mpsoc-swdev/GPIO-Reset-to-PL
https://docs.amd.com/r/en-US/pg201-zynq-ultrascale-plus-processing-system/Fabric-Reset-Enable
Parameters
----------
reinit : bool
Reinitialize firmware blocks. False is OK if you haven't run map_signal_paths() yet (which triggers block initialization).
"""
base_addr = 0xFF0A0000
logger.debug("base addr: %#010x" % (base_addr))
length = 0x400 # bytes
# Align the base address with the pages
virt_base = base_addr & ~(mmap.PAGESIZE - 1)
# Calculate base address offset w.r.t the base address
virt_offset = base_addr - virt_base
mmap_file = os.open("/dev/mem", os.O_RDWR | os.O_SYNC)
mem = mmap.mmap(
mmap_file,
length + virt_offset,
mmap.MAP_SHARED,
mmap.PROT_READ | mmap.PROT_WRITE,
offset=virt_base,
)
os.close(mmap_file)
array = np.frombuffer(mem, np.uint32, length >> 2, virt_offset)
logger.debug(array)
def mask_write(offset, mask, val):
index = (offset - base_addr) >> 2
regval = array[index]
logger.debug("initial\t%#010x" % (regval))
regval &= ~(mask)
logger.debug("masked\t%#010x" % (regval))
regval |= (val & mask)
logger.debug("final\t%#010x" % (regval))
array[index] = regval
GPIO_MASK_DATA_5_MSW_OFFSET = 0XFF0A002C
GPIO_DIRM_5_OFFSET = 0XFF0A0344
GPIO_OEN_5_OFFSET = 0XFF0A0348
GPIO_DATA_5_OFFSET = 0XFF0A0054
mask_write(GPIO_MASK_DATA_5_MSW_OFFSET, 0xFFFF0000, 0x80000000)
mask_write(GPIO_DIRM_5_OFFSET, 0xFFFFFFFF, 0x80000000)
mask_write(GPIO_OEN_5_OFFSET, 0xFFFFFFFF, 0x80000000)
mask_write(GPIO_DATA_5_OFFSET, 0xFFFFFFFF, 0x80000000)
mask_write(GPIO_DATA_5_OFFSET, 0xFFFFFFFF, 0x00000000)
mask_write(GPIO_DATA_5_OFFSET, 0xFFFFFFFF, 0x80000000)
if reinit:
for k,v in self.ip_dict.items():
if v['type'].startswith("xilinx.com:ip:axi_dma"):
dma = getattr(self,k)
dma.set_up_tx_channel()
dma.set_up_rx_channel()
for x in self.readouts:
x.freq_reg = 0
x.phase_reg = 0
x.nsamp_reg = 10
x.outsel_reg = 0
x.mode_reg = 1
x.update()
[docs] def get_sample_rates(self):
"""
Produce dictionaries of the current sample rates of the DAC and ADC tiles.
A dictionary of this form can be used to configure the sample rates at SoC initialization.
Returns
-------
dict[str, dict[int, float]]
A pair of dictionaries mapping DAC/ADC tiles to sample rates (in Msps).
"""
return {tiletype: {k: v['fs'] for k,v in self['rf']['tiles'][tiletype].items()} for tiletype in ['dac', 'adc']}
[docs] def valid_sample_rates(self, tiletype, tile):
"""
Return an array of valid sample rates.
This does not account for dependencies due to clock groups,
or the restriction that you're not allowed to raise the sample rate.
Parameters
----------
tiletype : str
'dac' or 'adc'
tile : int
Tile number (0-3)
Returns
-------
numpy.ndarray
Array of sample rates, in Msps
"""
return self.rf.valid_sample_rates(tiletype, tile)
[docs] def round_sample_rate(self, tiletype, tile, fs_target):
"""
Return the closest achievable sample rate to the requested value.
This does not account for dependencies due to clock groups,
or the restriction that you're not allowed to raise the sample rate.
Parameters
----------
tiletype : str
'dac' or 'adc'
tile : int
Tile number (0-3)
fs_target : float
Requested sample rate, in Msps
Returns
-------
float
Closest valid sample rate, in Msps
"""
return self.rf.round_sample_rate(tiletype, tile, fs_target)
[docs] def get_decimated(self, ch, address=0, length=None):
"""
Acquires data from the readout decimated buffer
:param ch: ADC channel
:type ch: int
:param address: Address of data
:type address: int
:param length: Buffer transfer length
:type length: int
:return: List of I and Q decimated arrays
:rtype: list of numpy.ndarray
"""
if length is None:
# this default will always cause a RuntimeError
# TODO: remove the default, or pick a better fallback value
length = self.avg_bufs[ch]['buf_maxlen']
# request data from DMA
return self.avg_bufs[ch].transfer_buf(address, length)
[docs] def get_accumulated(self, ch, address=0, length=None):
"""
Acquires data from the readout accumulated buffer
:param ch: ADC channel
:type ch: int
:param address: Address of data
:type address: int
:param length: Buffer transfer length
:type length: int
:returns:
- di[:length] (:py:class:`list`) - list of accumulated I data
- dq[:length] (:py:class:`list`) - list of accumulated Q data
"""
if length is None:
# this default will always cause a RuntimeError
# TODO: remove the default, or pick a better fallback value
length = self.avg_bufs[ch]['avg_maxlen']
# request data from DMA
return self.avg_bufs[ch].transfer_avg(address, length)
[docs] def config_avg(
self, ch, address=0, length=1,
edge_counting=False, high_threshold=1000, low_threshold=0):
"""Configure accumulated buffer; must then enable using enable_buf()
Parameters
----------
ch : int
Readout channel to configure
address : int
Starting address of buffer
length : int
length of buffer (how many samples to integrate)
"""
avg_buf = self.avg_bufs[ch]
if avg_buf['has_edge_counter']:
avg_buf.config_avg(
address, length,
edge_counting=edge_counting, high_threshold=high_threshold, low_threshold=low_threshold)
else:
avg_buf.config_avg(address, length)
[docs] def config_buf(self, ch, address=0, length=1):
"""Configure decimated buffer; must then enable using enable_buf()
Parameters
----------
ch : int
Readout channel to configure
address : int
Starting address of buffer
length : int
length of buffer (how many samples to take)
"""
avg_buf = self.avg_bufs[ch]
avg_buf.config_buf(address, length)
[docs] def enable_buf(self, ch, enable_avg=True, enable_buf=True):
"""Enable capture of accumulated and/or decimated data for a buffer
Parameters
----------
ch : int
Readout channel to configure
enable_avg : bool
Enable accumulated data capture
enable_buf : bool
Enable decimated data capture
"""
avg_buf = self.avg_bufs[ch]
avg_buf.enable(avg=enable_avg, buf=enable_buf)
[docs] def load_weights(self, ch, data, addr=0):
"""Load weights array to a weighted buffer.
Parameters
----------
ch : int
Readout channel to configure
data : numpy.ndarray of int16
array of 16-bit (I, Q) values for weights
address : int
starting address
"""
# we may have converted to list for pyro compatiblity, so convert back to ndarray
data = np.array(data, dtype=np.int16)
self.avg_bufs[ch].load_weights(data, addr)
[docs] def load_envelope(self, ch, data, addr):
"""Load envelope data into a signal generator.
Parameters
----------
ch: int
Generator channel to configure
data: numpy.ndarray of int16
Array of (I, Q) values for pulse envelope
addr: int
Starting address
"""
# we may have converted to list for pyro compatiblity, so convert back to ndarray
data = np.array(data, dtype=np.int16)
self.gens[ch].load(xin=data, addr=addr)
[docs] def set_nyquist(self, ch, nqz, force=False):
"""
Sets DAC channel ch to operate in Nyquist zone nqz mode.
:param ch: DAC channel (index in 'gens' list)
:type ch: int
:param nqz: Nyquist zone
:type nqz: int
"""
self.gens[ch].set_nyquist(nqz)
[docs] def set_mixer_freq(self, ch, f, ro_ch=None, phase_reset=True):
"""
Set mixer frequency for a signal generator.
If the generator does not have a mixer, you will get an error.
Parameters
----------
ch : int
DAC channel (index in 'gens' list)
f : float
Mixer frequency (in MHz)
ro_ch : int
readout channel (index in 'readouts' list) for frequency matching
use None if you don't want mixer freq to be rounded to a valid readout frequency
phase_reset : bool
if this changes the frequency, also reset the phase (so if we go to freq=0, we end up on the real axis)
"""
if self.gens[ch].HAS_MIXER:
self.gens[ch].set_mixer_freq(f, ro_ch, phase_reset=phase_reset)
elif f != 0:
raise RuntimeError("tried to set a mixer frequency, but this channel doesn't have a mixer")
[docs] def set_adc_attenuator(self, blockname, attenuation):
"""Set the RFSoC ADC's built-in step attenuator.
The requested value will be rounded to the nearest valid value (0-27 dB inclusive, 1 dB steps).
Only available for RFSoC Gen 3 (ZCU216, RFSoC4x2).
See https://docs.amd.com/r/en-US/pg269-rf-data-converter/Digital-Step-Attenuator-Gen-3/DFE.
Parameters
----------
blockname : str
RF-ADC ID (2-digit string).
This is the concatenation of the tile and block numbers displayed in the firmware configuration:
in other words, for "ADC tile 2, blk 1" the blockname is "21".
attenuation : float
Attenuation value (dB)
Returns
-------
float
The rounded attenuation value that was actually set (dB)
"""
return self.rf.set_adc_attenuator(blockname, attenuation)
[docs] def get_adc_attenuator(self, blockname):
"""Read the RFSoC ADC's built-in step attenuator.
Only available for RFSoC Gen 3 (ZCU216, RFSoC4x2).
See https://docs.amd.com/r/en-US/pg269-rf-data-converter/Digital-Step-Attenuator-Gen-3/DFE.
Parameters
----------
blockname : str
RF-ADC ID (2-digit string).
This is the concatenation of the tile and block numbers displayed in the firmware configuration:
in other words, for "ADC tile 2, blk 1" the blockname is "21".
Returns
-------
float
Attenuation value (dB)
"""
return self.rf.get_adc_attenuator(blockname)
[docs] def freeze_adc_cals(self, blocknames):
"""Freeze the calibrations (stop the background calibration) of a list of RFSoC ADCs.
See the Xilinx documentation:
https://docs.amd.com/r/en-US/pg269-rf-data-converter/Background-Calibration-Process
Parameters
----------
blocknames : list of str
List of RF-ADC IDs (2-digit strings).
An ADC's ID is the concatenation of the tile and block numbers displayed in the firmware configuration:
in other words, for "ADC tile 2, blk 1" the blockname is "21".
"""
for adc in blocknames: self.rf.freeze_adc_cal(adc)
[docs] def unfreeze_adc_cals(self, blocknames):
"""Unfreeze the calibrations (resume the background calibration) of a list of RFSoC ADCs.
See the Xilinx documentation:
https://docs.amd.com/r/en-US/pg269-rf-data-converter/Background-Calibration-Process
Parameters
----------
blocknames : list of str
List of RF-ADC IDs (2-digit strings).
An ADC's ID is the concatenation of the tile and block numbers displayed in the firmware configuration:
in other words, for "ADC tile 2, blk 1" the blockname is "21".
"""
for adc in blocknames: self.rf.unfreeze_adc_cal(adc)
[docs] def config_mux_gen(self, ch, tones):
"""Set up a list of tones all at once, using raw (integer) units.
If the supplied list of tones is shorter than the number supported, the extra tones will have their gains set to 0.
Parameters
----------
ch : int
generator channel (index in 'gens' list)
tones : list of dict
Tones to configure.
This is generated by QickConfig.calc_muxgen_regs().
"""
self.gens[ch].set_tones_int(tones)
[docs] def config_mux_readout(self, pfbpath, cfgs, sel=None):
"""Set up a list of readout frequencies all at once, using raw (integer) units.
Parameters
----------
pfbpath : str
Firmware path of the PFB readout being configured.
cfgs : list of dict
Readout chains to configure.
This is generated by QickConfig.calc_pfbro_regs().
sel : str
Output selection (if supported), default to 'product'
"""
pfb = getattr(self, pfbpath)
if pfb.HAS_OUTSEL:
if sel is None: sel = 'product'
pfb.set_out(sel)
else:
if sel is not None:
raise RuntimeError("this readout doesn't support configuring sel, you have sel=%s" % (sel))
for cfg in cfgs:
pfb.set_freq_int(cfg)
[docs] def set_iq(self, ch, f, i, q, ro_ch=None, phase_reset=True):
"""
Set frequency, I, and Q for a constant-IQ output.
Parameters
----------
ch : int
DAC channel (index in 'gens' list)
f : float
frequency (in MHz)
i : float
I value (in range -1 to 1)
q : float
Q value (in range -1 to 1)
ro_ch : int
readout channel (index in 'readouts' list) for frequency matching
use None if you don't want freq to be rounded to a valid readout frequency
phase_reset : bool
if this changes the frequency, also reset the phase (so if we go to freq=0, we end up on the real axis)
"""
self.iqs[ch].set_mixer_freq(f)
self.iqs[ch].set_iq(i, q)
[docs] def load_bin_program(self, binprog, load_mem=True):
"""Write the program to the tProc program memory.
Parameters
----------
binprog : numpy.ndarray or dict
compiled program (format depends on tProc version)
load_mem : bool
write waveform and data memory now (can do this later with reload_mem())
"""
binprog = obtain(binprog)
# cast to ndarray
if self.TPROC_VERSION == 1:
binprog = np.array(binprog, dtype=np.uint64)
elif self.TPROC_VERSION == 2:
for mem_sel in ['pmem', 'dmem', 'wmem']:
if binprog[mem_sel] is not None:
binprog[mem_sel] = np.array(binprog[mem_sel], dtype=np.int32)
self.tproc.load_bin_program(binprog, load_mem=load_mem)
[docs] def reload_mem(self):
"""Reload the waveform and data memory, overwriting any changes made by running the program.
"""
if self.TPROC_VERSION == 2:
self.tproc.reload_mem()
[docs] def load_mem(self, data, mem_sel='dmem', addr=0):
"""
Write a block of the selected tProc memory.
For tProc v1 only the data memory ("dmem") is valid.
For tProc v2 the program, data, and waveform memory are all accessible.
Parameters
----------
data : numpy.ndarray of int
Data to be loaded
32-bit array of shape (n, 8) for pmem and wmem, (n) for dmem
mem_sel : str
"pmem", "dmem", "wmem"
addr : int
Starting write address
"""
data = np.array(data, dtype=np.int32)
if self.TPROC_VERSION == 1:
if mem_sel=='dmem':
self.tproc.load_dmem(data, addr)
else:
raise RuntimeError("invalid mem_sel: %s"%(mem_sel))
elif self.TPROC_VERSION == 2:
self.tproc.load_mem(mem_sel, data, addr)
[docs] def read_mem(self, length, mem_sel='dmem', addr=0):
"""
Read a block of the selected tProc memory.
For tProc v1 only the data memory ("dmem") is valid.
For tProc v2 the program, data, and waveform memory are all accessible.
Parameters
----------
length : int
Number of words to read
mem_sel : str
"pmem", "dmem", "wmem"
addr : int
Starting read address
Returns
-------
numpy.ndarray
32-bit array of shape (n, 8) for pmem and wmem, (n) for dmem
"""
if self.TPROC_VERSION == 1:
if mem_sel=='dmem':
return self.tproc.read_dmem(addr, length)
else:
raise RuntimeError("invalid mem_sel: %s"%(mem_sel))
elif self.TPROC_VERSION == 2:
return self.tproc.read_mem(mem_sel, length, addr)
[docs] def start_src(self, src):
"""
Sets the start source of tProc
:param src: start source "internal" or "external"
:type src: str
"""
self.tproc.start_src(src)
[docs] def start_tproc(self):
"""
Start the tProc.
If the tProc is configured for external start, this does nothing (the tProc will start on the first start signal it sees after external start is enabled).
"""
if self.TPROC_VERSION == 1:
self.tproc.start()
elif self.TPROC_VERSION == 2:
if self.tproc.get_start_src() == 'internal':
self.tproc.start()
[docs] def stop_tproc(self, lazy=False):
"""
Stop the tProc.
This is somewhat slow (tens of ms) for tProc v1.
Parameters
----------
lazy : bool
Only stop the tProc if it's easy (i.e. do nothing for v1)
"""
if self.TPROC_VERSION == 1:
if not lazy:
# there's no easy way to stop v1 - we need to reset and reload
self.tproc.reset()
# reload the program (since the reset will have wiped it out)
self.tproc.reload_program()
elif self.TPROC_VERSION == 2:
self.tproc.stop()
[docs] def clear_tproc_counter(self, addr):
"""
Initialize the tProc shot counter.
For tProc v1, the data memory at the specified address is zeroed.
For tProc v2, the tProc is reset, which zeroes all registers (the address is ignored).
Typical tProc v2 programs will also initialize the counter registers at the beginning of the program, but zeroing the counter now is important to distinguish "program waiting for external start" from "program complete."
Parameters
----------
addr : int
Counter address
"""
if self.TPROC_VERSION == 1:
self.tproc.single_write(addr=addr, data=0)
elif self.TPROC_VERSION == 2:
self.tproc.reset()
[docs] def get_tproc_counter(self, addr):
"""
Read the tProc shot counter.
For tProc V1, this accesses the data memory at the given address.
For tProc V2, this accesses one of the two special AXI-readable registers.
Parameters
----------
addr : int
Counter address
Returns
-------
int
Counter value
"""
if self.TPROC_VERSION == 1:
return self.tproc.single_read(addr=addr)
elif self.TPROC_VERSION == 2:
self.tproc.read_sel=1
reg = {1:'axi_r_dt1', 2:'axi_r_dt2'}[addr]
return getattr(self.tproc, reg)
[docs] def reset_gens(self):
"""
Reset the tProc and run a minimal tProc program that drives all signal generators with 0's.
Useful for stopping any periodic or stdysel="last" outputs that may have been driven by a previous program.
"""
# list channel numbers for all generators capable of playing arbitrary envelopes
# (what we actually care about is whether they can play periodic pulses, but it's the same set of gens)
gen_chs = [i for i, gen in enumerate(self.gens) if isinstance(gen, AbsArbSignalGen)]
if self.TPROC_VERSION == 1:
prog = QickProgram(self)
for gen in gen_chs:
prog.set_pulse_registers(ch=gen, style="const", mode="oneshot", freq=0, phase=0, gain=0, length=3)
prog.pulse(ch=gen,t=0)
prog.end()
elif self.TPROC_VERSION == 2:
prog = QickProgramV2(self)
for gen in gen_chs:
prog.pulse(ch=gen, name="dummypulse", t=0)
prog.end()
self.tproc.reset()
# this should always run with internal trigger
prog.run(self, start_src="internal")
[docs] def start_readout(self, total_shots, counter_addr=1, ch_list=None, reads_per_shot=1, stride=None):
"""
Start a streaming readout of the accumulated buffers.
:param total_shots: Final value expected for the shot counter
:type total_shots: int
:param counter_addr: Data memory address for the shot counter
:type counter_addr: int
:param ch_list: List of readout channels
:type ch_list: list of int
:param reads_per_shot: Number of data points to expect per counter increment
:type reads_per_shot: list of int
:param stride: Default number of measurements to transfer at a time.
:type stride: int
"""
ch_list = obtain(ch_list)
reads_per_shot = obtain(reads_per_shot)
if ch_list is None: ch_list = [0, 1]
if isinstance(reads_per_shot, int):
reads_per_shot = [reads_per_shot]*len(ch_list)
streamer = self.streamer
if not streamer.readout_worker.is_alive():
print("restarting readout worker")
streamer.start_worker()
print("worker restarted")
# if there's still a readout job running, stop it
if streamer.readout_running():
print("cleaning up previous readout: stopping tProc and streamer loop")
# stop the tProc
self.stop_tproc()
# tell the readout to stop (this will break the readout loop)
streamer.stop_readout()
streamer.done_flag.wait()
# push a dummy packet into the data queue to halt any running poll_data(), and wait long enough for the packet to be read out
streamer.data_queue.put((0, None))
time.sleep(0.1)
print("streamer stopped")
streamer.stop_flag.clear()
if streamer.data_available():
# flush all the data in the streamer buffer
print("clearing streamer buffer")
# read until the queue times out, discard the data
self.poll_data(totaltime=-1, timeout=0.1)
print("buffer cleared")
streamer.total_count = total_shots
streamer.count = 0
streamer.done_flag.clear()
streamer.job_queue.put((total_shots, counter_addr, ch_list, reads_per_shot, stride))
[docs] def poll_data(self, totaltime=0.1, timeout=None):
"""
Get as much data as possible from the streamer data queue.
Stop when any of the following conditions are met:
* all the data has been transferred (based on the total_count)
* we got data, and it has been totaltime seconds since poll_data was called
* timeout is defined, and the timeout expired without getting new data in the queue
If there are errors in the error queue, raise the first one.
:param totaltime: How long to acquire data (negative value = ignore total time and total count, just read until timeout)
:type totaltime: float
:param timeout: How long to wait for the next data packet (None = wait forever)
:type timeout: float
:return: list of (data, stats) pairs, oldest first
:rtype: list
"""
streamer = self.streamer
time_end = time.time() + totaltime
new_data = []
while (totaltime < 0) or (streamer.count < streamer.total_count and time.time() < time_end):
try:
raise RuntimeError("exception in readout loop") from streamer.error_queue.get(block=False)
except queue.Empty:
pass
try:
length, data = streamer.data_queue.get(block=True, timeout=timeout)
# if we stopped the readout while we were waiting for data, break out and return
if streamer.stop_flag.is_set() or data is None:
break
streamer.count += length
new_data.append((length, data))
except queue.Empty:
break
return new_data
[docs] def prepare_round(self):
"""This runs before a program starts running.
This is called by acquire/acquire_decimated/run_rounds; user code should not call it.
By default this does nothing, but a subclass of QickSoc may override this.
"""
pass
[docs] def cleanup_round(self):
"""This runs after a program has finished running.
This is called by acquire/acquire_decimated/run_rounds; user code should not call it.
By default this does nothing, but a subclass of QickSoc may override this.
"""
pass
[docs] def clear_ddr4(self, length=None):
"""Clear the DDR4 buffer, filling it with 0's.
This is not necessary (the buffer will overwrite old data), but may be useful for debugging.
Clearing the full buffer (4 GB) typically takes 4-5 seconds.
Parameters
----------
length : int
Number of samples to clear (starting at the beginning of the buffer). If None, clear the entire buffer.
"""
self.ddr4_buf.clear_mem(length)
[docs] def get_ddr4(self, nt, start=None):
"""Get data from the DDR4 buffer.
The first samples (typically 401 or 801) of the buffer are always stale data from the previous acquisition.
Parameters
----------
nt : int
Number of data transfers (each transfer is 128 or 256 decimated samples) to retrieve.
If start=None, the amount of data will be reduced (see below).
start : int
Number of samples to skip at the beginning of the buffer.
If a value is specified, the end address of the transfer window will also be incremented.
If None, the junk at the start of the buffer will be skipped but the end address will not be incremented.
This reduces the amount of data, giving you exactly the block of valid data from a DDR4 trigger with the same value of nt.
"""
return self.ddr4_buf.get_mem(nt, start)
[docs] def arm_ddr4(self, ch, nt, force_overwrite=False):
"""Prepare the DDR4 buffer to take data.
This must be called before starting a program that triggers the buffer.
Once the buffer is armed, the first trigger it receives will cause the buffer to record the specified amount of data.
Later triggers will have no effect.
Parameters
----------
ch : int
The readout channel to record (index in 'readouts' list).
nt : int
Number of data transfers to record; the number of IQ samples/transfer (128 or 256) is printed in the QickSoc config.
Note that the amount of useful data is less (see ``get_ddr4``)
force_overwrite : bool
Allow a DDR4 acqusition that exceeds the DDR4 memory capacity. The memory will be used as a circular buffer:
later transfers will wrap around to the beginning of the memory and overwrite older data.
"""
self.ddr4_buf.set_switch(self['readouts'][ch]['avgbuf_fullpath'])
self.ddr4_buf.arm(nt, force_overwrite)
[docs] def arm_mr(self, ch):
"""Prepare the Multi-Rate buffer to take data.
This must be called before starting a program that triggers the buffer.
Once the buffer is armed, the first trigger it receives will cause the buffer to record until the buffer is filled.
Later triggers will have no effect.
Parameters
----------
ch : int
The readout channel to record (index in 'readouts' list).
"""
self.mr_buf.set_switch(self['readouts'][ch]['avgbuf_fullpath'])
self.mr_buf.disable()
self.mr_buf.enable()
[docs] def get_mr(self, start=None):
"""Get data from the multi-rate buffer.
The first 8 samples are always stale data from the previous acquisition.
The transfer window always extends to the end of the buffer.
Parameters
----------
start : int
Number of samples to skip at the beginning of the buffer.
If None, the junk at the start of the buffer is skipped.
"""
return self.mr_buf.transfer(start)
[docs] def tt_arm(self, blk):
"""Start data capture on the specified time-tagger block.
Parameters
----------
blk : int
The time tagger block to arm (index in `time_taggers' list).
"""
self.time_taggers[blk].disarm()
self.time_taggers[blk].arm()
[docs] def tt_disarm(self, blk):
"""Stop data capture on the specified time-tagger block.
Parameters
----------
blk : int
The time tagger block to arm (index in `time_taggers' list).
"""
self.time_taggers[blk].disarm()
[docs] def tt_readmem(self, blk, mem):
"""Read one of the specified time-tagger block's memories.
Reading a time-tagger memory clears it.
Parameters
----------
blk : int
The time tagger block to read (index in `time_taggers' list).
mem : str
"ARM", "SMP", "TAG0"/"TAG1"/"TAG2"/"TAG3"
"""
return self.time_taggers[blk].read_mem(mem)
[docs] def tt_reset(self, blk):
"""Reset and flush the memories of the specified time-tagger block.
Parameters
----------
blk : int
The time tagger block to reset (index in `time_taggers' list).
"""
#self.time_taggers[blk].reset()
self.time_taggers[blk].flush_mems()
[docs] def tt_config(self, blk, threshold, wr_smp=32, filt=False, slope=False, invert=False, interp=0, deadtime=5):
"""Configure the specified time-tagger block.
Parameters
----------
blk : int
The time tagger block to flush (index in `time_taggers' list).
threshold : int
Tag threshold (-2^15 through 2^15-1).
wr_smp : int
Number of 8-sample chunks to capture per tag in the SMP memory (1 through 32).
If this time tagger has no SMP memory, this is ignored.
filt : bool
Trigger on signal after a 2-sample smoothing filter.
slope : bool
Trigger on slope (difference between consecutive samples), not level.
invert : bool
Trigger on inverted signal.
interp : int
Number of bits for interpolation (0 through 7).
deadtime : int
Minimum time (in fabric ticks, 8 samples) between one tag and the next (5 through 255).
This should be larger than the expected pulse width, otherwise you will get double-triggering on the same pulse.
"""
filt = 1 if filt else 0
slope = 1 if slope else 0
invert = 1 if invert else 0
self.time_taggers[blk].set_config(
filt=filt,
slope=slope,
invert=invert,
wr_smp=wr_smp,
interp=interp
)
self.time_taggers[blk].set_dead_time(deadtime)
self.time_taggers[blk].set_threshold(threshold)