"""
Drivers for readouts (FPGA blocks that receive data from ADCs) and buffers (blocks that receive data from readouts).
"""
from pynq.buffer import allocate
import numpy as np
from qick.ip import SocIP, QickIP, DummyIP
RO_TYPES = ["axis_readout_v2", "axis_readout_v3", "axis_pfb_readout_v2", "axis_pfb_readout_v3", "axis_pfb_readout_v4", "axis_dyn_readout_v1"]
BUF_TYPES = ['axis_avg_buffer', 'axis_weighted_buffer']
[docs]class AbsReadout(QickIP):
# Downsampling ratio (RFDC samples per decimated readout sample)
DOWNSAMPLING = 1
# Number of bits in the phase register
B_PHASE = None
# Some readouts put a small nonzero offset on the I and Q values due to rounding.
# For muxed readouts, this offset is usually different between even and odd channels.
IQ_OFFSET = 0.0
# Configure this driver with the sampling frequency.
def configure(self, rf):
self.rf = rf
# Sampling frequency.
if self.B_PHASE is not None: self.cfg['b_phase'] = self.B_PHASE
adccfg = self.rf['adcs'][self['adc']]
for p in ['fs', 'fs_mult', 'fs_div', 'decimation', 'f_fabric']:
self.cfg[p] = adccfg[p]
# decimation reduces the DDS range
self.cfg['f_dds'] = self['fs']/self['decimation']
self.cfg['fdds_div'] = self['fs_div']*self['decimation']
self.cfg['f_output'] = self['fs']/(self['decimation']*self.DOWNSAMPLING)
self.cfg['b_dds'] = self.B_DDS
self.cfg['iq_offset'] = self.IQ_OFFSET
self.cfg['has_outsel'] = self.HAS_OUTSEL
[docs] def update(self):
"""
Push the register values to the readout logic.
"""
pass
[docs]class AxisReadoutV2(SocIP, AbsReadout):
"""
AxisReadoutV2 class
Registers.
FREQ_REG : 32-bit.
PHASE_REG : 32-bit.
NSAMP_REG : 16-bit.
OUTSEL_REG : 2-bit.
* 0 : product.
* 1 : dds.
* 2 : input (bypass).
MODE_REG : 1-bit.
* 0 : NSAMP.
* 1 : Periodic.
WE_REG : enable/disable to perform register update.
:param fs: sampling frequency in MHz
:type fs: float
"""
bindto = ['user.org:user:axis_readout_v2:1.0',
'QICK:QICK:axis_readout_v2:1.0']
# Bits of DDS.
B_DDS = 32
B_PHASE = 32
# This is actually the only current readout that doesn't have an offset.
IQ_OFFSET = 0.0
# Downsampling ratio (RFDC samples per decimated readout sample)
DOWNSAMPLING = 8
# this readout is not controlled by the tProc.
tproc_ch = None
# Output mode selection is supported.
HAS_OUTSEL = True
def _init_config(self, description):
self.REGISTERS = {'freq_reg': 0, 'phase_reg': 1, 'nsamp_reg': 2,
'outsel_reg': 3, 'mode_reg': 4, 'we_reg': 5}
def _init_firmware(self):
# Default registers.
self.freq_reg = 0
self.phase_reg = 0
self.nsamp_reg = 10
self.outsel_reg = 0
self.mode_reg = 1
# Register update.
self.update()
[docs] def update(self):
"""
Update register values
"""
self.we_reg = 1
self.we_reg = 0
[docs] def set_all_int(self, regs):
"""Set all readout parameters using a dictionary computed by QickConfig.calc_ro_regs().
"""
self.outsel_reg = {"product": 0, "dds": 1, "input": 2}[regs['sel']]
self.freq_reg = regs['f_int'] % 2**self.B_DDS
self.phase_reg = regs['phase_int'] % 2**self.B_PHASE
self.nsamp_reg = 10
self.mode_reg = 1
self.update()
# sometimes it seems that we need to update the readout an extra time to make it configure everything correctly?
# this has only really been seen with setting a downconversion freq of 0.
self.update()
[docs] def set_all(self, f, sel='product', gen_ch=None, phase=0):
"""Set up the readout directly.
This method is not normally used, it's only for debugging and testing.
Normally the PFB is configured based on parameters supplied in QickProgram.declare_readout().
"""
cfg = self.soc.calc_ro_regs(self.cfg, phase, sel)
ro_pars = {'freq': f,
'gen_ch': gen_ch
}
mixer_freq = 0
if gen_ch is not None and self.soc.gens[gen_ch].HAS_MIXER:
mixer_freq = self.soc.gens[gen_ch].get_mixer_freq()
self.soc.calc_ro_freq(self.cfg, ro_pars, cfg, False, mixer_freq)
self.set_all_int(cfg)
[docs]class AbsPFBReadout(SocIP, AbsReadout):
# Bits of DDS.
B_DDS = 32
# based on testing this seems like it might really be some weird value like -0.48, even though this makes no sense
IQ_OFFSET = -0.5
# this readout is not controlled by the tProc.
tproc_ch = None
def _init_config(self, description):
# Downsampling ratio (RFDC samples per decimated readout sample)
self.DOWNSAMPLING = self.NCH//2
# index of the PFB channel that is centered around DC.
self.CH_OFFSET = self.NCH//2
self.cfg['pfb_nch'] = self.NCH
self.cfg['pfb_nout'] = self.NOUT
self.cfg['pfb_ch_offset'] = self.CH_OFFSET
self.cfg['pfb_dds_on_output'] = self.DDS_ON_OUTPUT
def configure(self, rf):
super().configure(rf)
# The DDS range is reduced by both the RF-ADC decimation and the PFB.
# The PFB decimation ratio is half the number of channels because it is
# an overlap 50 % structure.
self.cfg['f_dds'] /= self.DOWNSAMPLING
self.cfg['fdds_div'] *= self.DOWNSAMPLING
[docs] def set_ch(self, f, out_ch, sel='product', gen_ch=None, phase=0):
"""Set up a single PFB output.
This method is not normally used, it's only for debugging and testing.
Normally the PFB is configured based on parameters supplied in QickProgram.declare_readout().
"""
cfg = self.soc.calc_ro_regs(self.cfg, phase, sel)
ro_pars = {'freq': f,
'gen_ch': gen_ch
}
mixer_freq = 0
if gen_ch is not None and self.soc.gens[gen_ch].HAS_MIXER:
mixer_freq = self.soc.gens[gen_ch].get_mixer_freq()
self.soc.calc_ro_freq(self.cfg, ro_pars, cfg, False, mixer_freq)
cfg['pfb_port'] = out_ch
if self.HAS_OUTSEL:
self.set_out(sel)
self.set_freq_int(cfg)
[docs]class AxisPFBReadoutV2(AbsPFBReadout):
"""
AxisPFBReadoutV2 class.
This readout block contains a polyphase filter bank with 8 channels.
Channel i mixes the input signal down by a fixed frequency f = i * fs/16,
then by a programmable DDS with a range of +/- fs/16.
The PFB channels can be freely mapped to the 4 outputs of the readout block.
Registers.
FREQ[0-7]_REG : 32-bit frequency of each channel.
OUTSEL_REG : 2-bit.
* 0 : product.
* 1 : input (bypass).
* 2 : dds.
CH[0-3]SEL_REG : 3-bit ID mapping an output channel to an input.
"""
bindto = ['user.org:user:axis_pfb_readout_v2:1.0',
'QICK:QICK:axis_pfb_readout_v2:1.0']
# Number of PFB channels.
NCH = 8
# Number of outputs.
NOUT = 4
# The DDS is per-channel, not per-output.
DDS_ON_OUTPUT = False
# Output mode selection is supported.
HAS_OUTSEL = True
def _init_config(self, description):
super()._init_config(description)
self.REGISTERS = {'freq0_reg': 0,
'freq1_reg': 1,
'freq2_reg': 2,
'freq3_reg': 3,
'freq4_reg': 4,
'freq5_reg': 5,
'freq6_reg': 6,
'freq7_reg': 7,
'outsel_reg': 8,
'ch0sel_reg': 9,
'ch1sel_reg': 10,
'ch2sel_reg': 11,
'ch3sel_reg': 12,
}
[docs] def set_out(self, sel='product'):
"""
Select readout signal output
:param sel: select mux control
:type sel: int
"""
self.outsel_reg = {"product": 0, "input": 1, "dds": 2}[sel]
def set_freq_int(self, cfg):
# it's assumed that channel collisions have already been checked in config_readouts()
# we don't check here, so a collision will break the previously set channel
# phase_int is ignored
if 'phase_int' in cfg:
raise RuntimeError("this muxed readout does not support setting the phase")
# wire the selected PFB channel to the output
setattr(self, "ch%dsel_reg"%(cfg['pfb_port']), cfg['pfb_ch'])
# set the PFB channel's DDS frequency
setattr(self, "freq%d_reg"%(cfg['pfb_ch']), cfg['f_int'])
[docs]class AxisPFBReadoutV3(AbsPFBReadout):
"""
AxisPFBReadoutV3 class.
This readout block contains a polyphase filter bank with 64 channels.
Channel i mixes the input signal down by a fixed frequency f = i * fs/64,
then by a programmable DDS with a range of +/- fs/32.
The PFB channels can be freely mapped to the 4 outputs of the readout block.
DDS blocks are Phase-Coherent. The same PFB channel can be sent to multiple outputs.
For channel selection, channels are streamed out the PFB using TDM, with L=8 parallel
channels each clock. The number of packets is N/L = 64/8 = 8. The IDx_REG should be
mapped as follows:
* IDx_REG : lower 8 bits are the "packet" field, from 0 .. 7 (N/L).
: upper 8 bits are the "index" field, from 0 .. 7 (L-1).
There are 4 IDx_REG, one per selectable output.
Registers.
ID[0-3]_REG : 16-bit channel selection.
FREQ[0-3]_REG : 32-bit frequency of each output channel.
PHASE[0-3]_REG : 32-bit phase of each output channel.
"""
bindto = ['user.org:user:axis_pfb_readout_v3:1.0',
'QICK:QICK:axis_pfb_readout_v3:1.0']
# Bits of DDS.
B_PHASE = 32
# Number of lanes of PFB output.
L_PFB = 8
# Number of outputs.
NOUT = 4
# The DDS is per-output.
DDS_ON_OUTPUT = True
# No output mode selection.
HAS_OUTSEL = False
def _init_config(self, description):
# Generics.
self.NCH = int(description['parameters']['N'])
super()._init_config(description)
# define the register map
self.REGISTERS = {}
iReg = 0
for i in range(self.NOUT): self.REGISTERS['id%d_reg'%(i)] = i + iReg
iReg += self.NOUT
for i in range(self.NOUT):
self.REGISTERS['freq%d_reg'%(i)] = 2*i + iReg
self.REGISTERS['phase%d_reg'%(i)] = 2*i + iReg + 1
def set_freq_int(self, cfg):
# There are 4 outputs. Any PFB channel can be assigned to any output.
# No need to check for collisions are they are all truly independent.
pfb_ch = cfg['pfb_ch']
out_ch = cfg['pfb_port']
# Check pfb channel is within allowed range.
if pfb_ch not in range(self.NCH):
raise RuntimeError("Invalid PFB channel: %d. It must be within [0, %d]"%(pfb_ch, self.NCH-1))
# Check output channel is within allowed range.
if out_ch not in range(self.NOUT):
raise RuntimeError("Invalid %d output channel. It must be within [0, %d]"%(out_ch, self.NOUT-1))
# Compute packet and index fields from pfb channel.
packet = int(pfb_ch/self.L_PFB)
index = int(pfb_ch % self.L_PFB)
id_val = (index <<8) + packet
# Set id.
setattr(self, "id%d_reg"%(out_ch), id_val)
# Set frequency.
setattr(self, "freq%d_reg"%(out_ch), cfg['f_int'])
# Set phase.
setattr(self, "phase%d_reg"%(out_ch), cfg['phase_int'])
#print("{}: f_int = {}, pfb_ch = {}, out_ch = {}, packet = {}, index = {}, id_val = {}".format(self.__class__.__name__, f_int, pfb_ch, out_ch, packet, index, id_val))
[docs]class AxisPFBReadoutV4(AxisPFBReadoutV3):
"""
AxisPFBReadoutV4 class.
This is identical to AxisPFBReadoutV3, but with 8 outputs instead of 4.
"""
bindto = ['user.org:user:axis_pfb_readout_v4:1.0',
'QICK:QICK:axis_pfb_readout_v4:1.0']
# Number of outputs.
NOUT = 8
[docs]class AbsDynReadout(AbsReadout, DummyIP):
"""tProc-controlled readout block.
This isn't a PYNQ driver, since the block has no registers for PYNQ control.
We still need this class to represent the block and its connectivity.
"""
# Bits of DDS.
B_DDS = 32
B_PHASE = 32
# Output mode selection is supported.
HAS_OUTSEL = True
# IP name, to be defined by subclass
IP_TYPE = ""
def __init__(self, fullpath):
# make a fake ip_dict that contains the info needed by QickIP
desc = {
"type": self.IP_TYPE,
"fullpath": fullpath
}
super().__init__(desc)
[docs]class AxisReadoutV3(AbsDynReadout):
"""tProc-controlled readout block.
This isn't a PYNQ driver, since the block has no registers for PYNQ control.
We still need this class to represent the block and its connectivity.
"""
# Downsampling ratio (RFDC samples per decimated readout sample)
DOWNSAMPLING = 4
IQ_OFFSET = -0.5
IP_TYPE = "axis_readout_v3:0.0"
def configure(self, rf):
super().configure(rf)
# there is a 2x1 resampler between the RFDC and readout, which doubles the effective fabric frequency.
self.cfg['f_fabric'] *= 2
[docs]class AxisDynReadoutV1(AbsDynReadout):
"""tProc-controlled readout block.
This isn't a PYNQ driver, since the block has no registers for PYNQ control.
We still need this class to represent the block and its connectivity.
"""
# Downsampling ratio (RFDC samples per decimated readout sample)
DOWNSAMPLING = 8
IQ_OFFSET = 0.0
IP_TYPE = "axis_dyn_readout_v1:0.0"
[docs]class AxisAvgBuffer(SocIP):
"""
AxisAvgBuffer class
Registers.
AVG_START_REG
* 0 : Averager Disabled.
* 1 : Averager Enabled (started by external trigger).
AVG_ADDR_REG : start address to write results.
AVG_LEN_REG : number of samples to be added.
AVG_DR_START_REG
* 0 : do not send any data.
* 1 : send data using m0_axis.
AVG_DR_ADDR_REG : start address to read data.
AVG_DR_LEN_REG : number of samples to be read.
BUF_START_REG
* 0 : Buffer Disabled.
* 1 : Buffer Enabled (started by external trigger).
BUF_ADDR_REG : start address to write results.
BUF_LEN_REG : number of samples to be buffered.
BUF_DR_START_REG
* 0 : do not send any data.
* 1 : send data using m1_axis.
BUF_DR_ADDR_REG : start address to read data.
BUF_DR_LEN_REG : number of samples to be read.
:param axi_dma_avg: dma block for average buffers
:type axi_dma_avg: str
:param switch_avg: switch block for average buffers
:type switch_avg: str
:param axi_dma_buf: dma block for raw buffers
:type axi_dma_buf: str
:param switch_buf: switch block for raw buffers
:type switch_buf: str
:param channel: readout channel selection
:type channel: int
"""
bindto = ['user.org:user:axis_avg_buffer:1.0',
'QICK:QICK:axis_avg_buffer:1.0']
EDGE_COUNTER = False
WEIGHTS = False
RO_PORT = 's_axis'
FIRST_OUT_SAMPLE_BUG_FIX = False # Bug is present in IP version <= 1.1
def __init__(self, description):
"""
Constructor method
"""
# Preallocate memory buffers for DMA transfers.
self.avg_buff = None
self.buf_buff = None
self.dma_avg = None
self.dma_buf = None
self.switch_avg = None
self.switch_buf = None
self.switch_ch = None
super().__init__(description)
def _init_config(self, description):
self.REGISTERS = {'avg_start_reg': 0,
'avg_addr_reg': 1,
'avg_len_reg': 2,
'avg_dr_start_reg': 3,
'avg_dr_addr_reg': 4,
'avg_dr_len_reg': 5,
'buf_start_reg': 6,
'buf_addr_reg': 7,
'buf_len_reg': 8,
'buf_dr_start_reg': 9,
'buf_dr_addr_reg': 10,
'buf_dr_len_reg': 11}
# Generics
self.B = int(description['parameters']['B'])
self.N_AVG = int(description['parameters']['N_AVG'])
self.N_BUF = int(description['parameters']['N_BUF'])
# Maximum number of samples
self.cfg['avg_maxlen'] = 2**self.N_AVG
self.cfg['buf_maxlen'] = 2**self.N_BUF
self.cfg['has_edge_counter'] = self.EDGE_COUNTER
self.cfg['has_weights'] = self.WEIGHTS
self.avg_buff = allocate(shape=self['avg_maxlen'], dtype=np.int64)
self.buf_buff = allocate(shape=self['buf_maxlen'], dtype=np.int32)
def _init_firmware(self):
# Default registers.
self.disable()
self._stop_transfer()
# print("%s: readout %s, switch %d, trigger %d, tProc port %d"%
# (self['fullpath'], self.readout['fullpath'], self.switch_ch, self.trigger_bit, self.tproc_ch))
[docs] def set_freq(self, f, gen_ch=0):
"""
Set the downconversion frequency on the readout that drvies this buffer.
:param f: frequency in MHz (before adding any DAC mixer frequency)
:type f: float
:param gen_ch: DAC channel (use None if you don't want to round to a valid DAC frequency)
:type gen_ch: int
"""
if isinstance(self.readout, AbsPFBReadout):
self.readout.set_freq(f, self.readoutport, gen_ch=gen_ch)
else:
self.readout.set_freq(f, gen_ch=gen_ch)
[docs] def enable(self, avg=True, buf=True):
"""
Enable both acculumated and decimated buffers
Parameters
----------
avg : bool
Enable the accumulated buffer
dec : bool
Enable the decimated buffer
"""
if avg: self.avg_start_reg = 1
if buf: self.buf_start_reg = 1
[docs] def disable(self):
"""
Disable both acculumated and decimated buffers
"""
self.avg_start_reg = 0
self.buf_start_reg = 0
def _start_transfer(self, mem_sel):
"""
Start a DMA transfer
Parameters
----------
mem_sel : bool
"avg" or "buf"
"""
if mem_sel == 'avg':
self.avg_dr_start_reg = 1
elif mem_sel == 'buf':
self.buf_dr_start_reg = 1
else:
raise RuntimeError("invalid mem_sel: %s"%(mem_sel))
def _stop_transfer(self):
"""
Stop all DMA transfers
"""
self.avg_dr_start_reg = 0
self.buf_dr_start_reg = 0
[docs] def config_avg(self, address=0, length=100):
"""
Configure accumulated data buffer
:param addr: Start address of first capture
:type addr: int
:param length: window size
:type length: int
"""
# Disable averaging.
self.disable()
# Set registers.
self.avg_addr_reg = address
self.avg_len_reg = length
[docs] def transfer_avg(self, address=0, length=100):
"""
Transfer data from accumulated buffer
:param addr: starting reading address
:type addr: int
:param length: number of samples
:type length: int
:return: I,Q pairs
:rtype: list
"""
if length >= self['avg_maxlen']:
raise RuntimeError("length=%d longer than %d" %
(length, self['avg_maxlen']))
# pad the transfer size to an even number (odd lengths seem to break the DMA)
transferlen = length + (length % 2)
# Route switch to channel.
if self.switch_avg is not None:
self.switch_avg.sel(slv=self.switch_ch)
if (not self.FIRST_OUT_SAMPLE_BUG_FIX):
# there is a bug which causes the first sample of a transfer to always be the sample at address 0
# we work around this by requesting an extra 2 samples at the beginning
address = (address-2) % self['avg_maxlen']
transferlen = transferlen + 2
# Set averager data reader address and length.
self.avg_dr_addr_reg = address
self.avg_dr_len_reg = transferlen
# Start send data mode.
self._start_transfer('avg')
# DMA data.
buff = self.avg_buff
# nbytes has to be a Python int (it gets passed to mmio.write, which requires int or bytes)
self.dma_avg.recvchannel.transfer(buff, nbytes=int(transferlen*8))
self.dma_avg.recvchannel.wait()
# Stop send data mode.
self._stop_transfer()
if self.dma_avg.recvchannel.transferred != transferlen*8:
raise RuntimeError("Requested %d samples but only got %d from DMA" % (
transferlen, self.dma_avg.recvchannel.transferred//8))
# Format:
# -> lower 32 bits: I value.
# -> higher 32 bits: Q value.
data = np.frombuffer(buff[:transferlen], dtype=np.int32).reshape((-1,2))
# realign returned data with true request
if (not self.FIRST_OUT_SAMPLE_BUG_FIX):
data = data[2:length+2]
else:
data = data[:length]
# data is a view into the data buffer, so copy it before returning
return data.copy()
[docs] def config_buf(self, address=0, length=100):
"""
Configure decimated data buffer
:param addr: Start address of first capture
:type addr: int
:param length: window size
:type length: int
"""
# Disable buffering.
self.disable()
if length >= self['buf_maxlen']:
raise RuntimeError("requested length=%d longer or equal to decimated buffer size=%d" %
(length, self['buf_maxlen']))
# Set registers.
self.buf_addr_reg = address
self.buf_len_reg = length
[docs] def transfer_buf(self, address=0, length=100):
"""
Transfer data from decimated buffer
:param addr: starting reading address
:type addr: int
:param length: number of samples
:type length: int
:return: I,Q pairs
:rtype: list
"""
if length >= self['buf_maxlen']:
raise RuntimeError("requested length=%d longer or equal to decimated buffer size=%d" %
(length, self['buf_maxlen']))
# pad the transfer size to an even number (odd lengths seem to break the DMA)
transferlen = length + (length % 2)
# Route switch to channel.
if self.switch_buf is not None:
self.switch_buf.sel(slv=self.switch_ch)
if (not self.FIRST_OUT_SAMPLE_BUG_FIX):
# there is a bug which causes the first sample of a transfer to always be the sample at address 0
# we work around this by requesting an extra 2 samples at the beginning
address = (address-2) % self['buf_maxlen']
transferlen = transferlen + 2
# Set buffer data reader address and length.
self.buf_dr_addr_reg = address
self.buf_dr_len_reg = transferlen
# Start send data mode.
self._start_transfer('buf')
# DMA data.
buff = self.buf_buff
# nbytes has to be a Python int (it gets passed to mmio.write, which requires int or bytes)
self.dma_buf.recvchannel.transfer(buff, nbytes=int(transferlen*4))
self.dma_buf.recvchannel.wait()
if self.dma_buf.recvchannel.transferred != transferlen*4:
raise RuntimeError("Requested %d samples but only got %d from DMA" % (
transferlen, self.dma_buf.recvchannel.transferred//4))
# Stop send data mode.
self._stop_transfer()
# Format:
# -> lower 16 bits: I value.
# -> higher 16 bits: Q value.
data = np.frombuffer(buff[:transferlen], dtype=np.int16).reshape((-1,2))
# realign returned data with true request
if (not self.FIRST_OUT_SAMPLE_BUG_FIX):
data = data[2:length+2]
else:
data = data[:length]
# data is a view into the data buffer, so copy it before returning
return data.copy()
[docs]class AxisAvgBufferV1pt1(AxisAvgBuffer):
bindto = ['user.org:user:axis_avg_buffer:1.1',
'QICK:QICK:axis_avg_buffer:1.1']
EDGE_COUNTER = True
def _init_config(self, description):
super()._init_config(description)
self.REGISTERS = {'avg_start_reg': 0,
'avg_addr_reg': 1,
'avg_len_reg': 2,
'avg_dr_start_reg': 3,
'avg_dr_addr_reg': 4,
'avg_dr_len_reg': 5,
'buf_start_reg': 6,
'buf_addr_reg': 7,
'buf_len_reg': 8,
'buf_dr_start_reg': 9,
'buf_dr_addr_reg': 10,
'buf_dr_len_reg': 11,
'avg_photon_mode_reg': 12,
'avg_h_threshold_reg': 13,
'avg_l_threshold_reg': 14}
def _init_firmware(self):
super()._init_firmware()
self.avg_photon_mode_reg = 0
[docs] def config_avg(
self, address=0, length=100,
edge_counting=False, high_threshold=1000, low_threshold=0):
"""
Configure average buffer data from average and buffering readout block
:param addr: Start address of first capture
:type addr: int
:param length: window size
:type length: int
"""
super().config_avg(address=address, length=length)
self.avg_photon_mode_reg = edge_counting
if edge_counting:
self.avg_h_threshold_reg = high_threshold
self.avg_l_threshold_reg = low_threshold
[docs]class AxisAvgBufferV1pt2(AxisAvgBufferV1pt1):
"""
Same as AxisAvgBufferV1pt1 but firmware has the first output sample bug fixed.
"""
bindto = ['user.org:user:axis_avg_buffer:1.2',
'QICK:QICK:axis_avg_buffer:1.2']
FIRST_OUT_SAMPLE_BUG_FIX = True # Bug is fixed in IP version >= 1.2
[docs]class AxisWeightedBuffer(AxisAvgBufferV1pt1):
bindto = ['user.org:user:axis_weighted_buffer:1.2',
'QICK:QICK:axis_weighted_buffer:1.2']
WEIGHTS = True
def __init__(self, description):
self.wgt_buff = None
self.dma_wgt = None
self.switch_wgt = None
self.switch_wgt_ch = None
super().__init__(description)
def _init_config(self, description):
super()._init_config(description)
self.REGISTERS = {'start_reg': 0,
'avg_addr_reg': 1,
'avg_len_reg': 2,
'dr_start_reg': 3,
'avg_dr_addr_reg': 4,
'avg_dr_len_reg': 5,
'buf_addr_reg': 6,
'buf_len_reg': 7,
'buf_dr_addr_reg': 8,
'buf_dr_len_reg': 9,
'avg_photon_mode_reg': 10,
'avg_h_threshold_reg': 11,
'avg_l_threshold_reg': 12,
'wgt_dw_addr_reg': 13,
}
self.N_WGT = int(description['parameters']['N_WGT'])
self.cfg['wgt_maxlen'] = 2**self.N_WGT
self.wgt_buff = allocate(shape=self['wgt_maxlen'], dtype=np.int32)
[docs] def enable(self, avg=True, buf=True):
"""
Enable both accumulated and decimated buffers
Parameters
----------
avg : bool
Enable the accumulated buffer
dec : bool
Enable the decimated buffer
"""
val = 0
if avg: val += 1
if buf: val += 2
self.start_reg = val
[docs] def disable(self):
"""
Disable both acculumated and decimated buffers
"""
self.start_reg = 0
def _start_transfer(self, mem_sel):
if mem_sel == 'avg':
self.dr_start_reg = 1
elif mem_sel == 'buf':
self.dr_start_reg = 2
elif mem_sel == 'wgt':
self.dr_start_reg = 4
else:
raise RuntimeError("invalid mem_sel: %s"%(mem_sel))
def _stop_transfer(self):
self.dr_start_reg = 0
[docs] def load_weights(self, data, addr=0):
"""
Load weights array
Parameters
----------
data : numpy.ndarray of int16
array of 16-bit (I, Q) values for weights
addr : int
starting address
"""
length = data.shape[0]
assert data.dtype==np.int16
# Check for max length.
if length > self['wgt_maxlen']:
raise RuntimeError("tried to load %d samples, which exceeds the buffer size (%d)." %
(length, self['wgt_maxlen']))
# Check for even transfer size.
#if length % 2 != 0:
# raise RuntimeError("Buffer transfer length must be even number.")
# Route switch to channel.
if self.switch_wgt is not None:
self.switch_wgt.sel(mst=self.switch_wgt_ch)
#print(self['fullpath'], xin.shape, addr, self.switch_ch)
# Pack the data into a single array; columns will be concatenated
# -> lower 16 bits: I value.
# -> higher 16 bits: Q value.
# Format and copy data.
np.copyto(self.wgt_buff[:length],
np.frombuffer(data, dtype=np.int32))
################
### Load I/Q ###
################
# Enable writes.
self.wgt_dw_addr_reg = addr
self._start_transfer('wgt')
# DMA data.
self.dma_wgt.sendchannel.transfer(self.wgt_buff, nbytes=int(length*4))
self.dma_wgt.sendchannel.wait()
# Disable writes.
self._stop_transfer()
[docs]class AxisWeightedBufferV1pt3(AxisWeightedBuffer):
"""
Same as AxisWeightedBuffer, but firmware has the first output sample bug fixed.
"""
bindto = ['QICK:QICK:axis_weighted_buffer:1.3']
FIRST_OUT_SAMPLE_BUG_FIX = True # Bug is fixed in IP version >= 1.3
[docs]class MrBufferEt(SocIP):
# Registers.
# DW_CAPTURE_REG
# * 0 : Capture disabled.
# * 1 : Capture enabled (capture started by external trigger).
#
# DR_START_REG
# * 0 : don't send.
# * 1 : start sending data.
#
# DW_CAPTURE_REG needs to be de-asserted and asserted again to allow a new capture.
# DR_START_REG needs to be de-assereted and asserted again to allow a new transfer.
#
bindto = ['user.org:user:mr_buffer_et:1.0',
'user.org:user:mr_buffer_et:1.1',
'QICK:QICK:mr_buffer_et:1.0',
'QICK:QICK:mr_buffer_et:1.1']
def __init__(self, description):
super().__init__(description)
# Switch for selecting input.
self.switch = None
# Map from avg_buf name to switch port.
self.buf2switch = {}
self.cfg['readouts'] = []
def _init_config(self, description):
# Generics
self.B = int(description['parameters']['B'])
self.N = int(description['parameters']['N'])
self.NM = int(description['parameters']['NM'])
self.REGISTERS = {'dw_capture_reg': 0, 'dr_start_reg': 1}
# Maximum number of samples
self.cfg['maxlen'] = 2**self.N * self.NM
if 'mr_buffer_et:1.0' in description['type']:
self.cfg['junk_len'] = 8 # Firmware Bug present and junk data samples are returned from buffer
else:
self.cfg['junk_len'] = 0 # Firmware Bug was fixed so no need for this value anymore
# Preallocate memory buffers for DMA transfers.
self.buff = allocate(shape=2*self['maxlen'], dtype=np.int16)
def _init_firmware(self):
# Default registers.
self.dw_capture_reg = 0
self.dr_start_reg = 0
def route(self, ch):
# Route switch to channel.
self.switch.sel(slv=ch)
def set_switch(self, bufname):
# if there's no switch, just check that the specified buffer is the one that's hardwired
if self.switch is None:
assert self.buf2switch[bufname]==0
else:
self.route(self.buf2switch[bufname])
def transfer(self, start=None):
if start is None: start = self['junk_len']
# Start send data mode.
self.dr_start_reg = 1
# DMA data.
self.dma.recvchannel.transfer(self.buff)
self.dma.recvchannel.wait()
# Stop send data mode.
self.dr_start_reg = 0
return np.copy(self.buff).reshape((-1,2))[start:]
def enable(self):
self.dw_capture_reg = 1
def disable(self):
self.dw_capture_reg = 0
[docs]class AxisBufferDdrV1(SocIP):
"""
The DDR4 buffer block is similar to the decimated buffer in the avg_buffer block, except that data is written to DDR4 memory instead of FPGA memory.
Typically multiple readouts will be connected to this buffer through a switch.
The driver assumes that input(s) to this buffer are also sent to avg_buffer blocks.
"""
# AXIS Buffer DDR V1 Registers.
bindto = ['user.org:user:axis_buffer_ddr_v1:1.0',
'QICK:QICK:axis_buffer_ddr_v1:1.0']
# Stream Input Port.
STREAM_IN_PORT = "s_axis"
JUNK_SAMPLES_BUG_FIX = False # Bug is present in IP version <= 1.0
def __init__(self, description):
# Initialize ip
super().__init__(description)
# DDR4 controller.
self.ddr4_mem = None
# DDR4 data array.
self.ddr4_array = None
# Switch for selecting input.
self.switch = None
# Map from avg_buf name to switch port.
self.buf2switch = {}
self.cfg['readouts'] = []
def _init_config(self, description):
# Generics.
self.TARGET_SLAVE_BASE_ADDR = int(description['parameters']['TARGET_SLAVE_BASE_ADDR'],0)
self.ID_WIDTH = int(description['parameters']['ID_WIDTH'])
self.DATA_WIDTH = int(description['parameters']['DATA_WIDTH']) # width of the AXI bus, in bits
self.BURST_SIZE = int(description['parameters']['BURST_SIZE']) + 1 # words per AXI burst
self.REGISTERS = {'rstart_reg' : 0,
'raddr_reg' : 1,
'rlength_reg': 2,
'wstart_reg' : 3,
'waddr_reg' : 4,
'wnburst_reg': 5
}
self.cfg['burst_len'] = self.DATA_WIDTH*self.BURST_SIZE//32
if self.JUNK_SAMPLES_BUG_FIX:
self.cfg['junk_len'] = 0
else:
self.cfg['junk_len'] = 50*self.DATA_WIDTH//32 + 1 # not clear where this 50 comes from, presumably some FIFO somewhere
def _init_firmware(self):
# Default registers.
self.rstart_reg = 0
self.raddr_reg = 0
self.rlength_reg = 10
self.wstart_reg = 0
self.waddr_reg = 0
self.wnburst_reg = 10
def rstop(self):
self.rstart_reg = 0
def rstart(self):
self.rstart_reg = 1
def wstop(self):
self.wstart_reg = 0
def wstart(self):
self.wstart_reg = 1
[docs] def wlen(self, len_=10):
"""
Set the number of bursts. Each burst is 256 IQ pairs.
"""
self.wnburst_reg = len_
def set_switch(self, bufname):
# if there's no switch, just check that the specified buffer is the one that's hardwired
if self.switch is None:
assert self.buf2switch[bufname]==0
else:
self.switch.sel(slv=self.buf2switch[bufname])
def clear_mem(self, length=None):
if length is None:
np.copyto(self.ddr4_array, 0)
else:
np.copyto(self.ddr4_array[:length], 0)
def get_mem(self, nt, start=None):
if start is None:
start = self['junk_len']
end = nt*self['burst_len']
else:
end = start + nt*self['burst_len']
length = end-start
# when we access memory-mapped data, the start and end need to be aligned to multiples of 64 bits.
# violations result in the Python interpreter crashing on SIGBUS/BUS_ADRALN
# this doesn't matter for all operations, but np.copy() definitely seems to care
# it seems that even if you slice out an address-aligned chunk of data and just print it, sometimes that will access it in an illegal way
# therefore we pad out the requested address block, copy the data, and trim
# this way, no special care needs to be taken with the returned array
buf_copy = self.ddr4_array[start - (start%2):end + (end%2)].copy()
return buf_copy[start%2:length + start%2].view(dtype=np.int16).reshape((-1,2))
def arm(self, nt, force_overwrite=False):
if nt > self['maxlen']//self['burst_len'] and not force_overwrite:
raise RuntimeError("the requested number of DDR4 transfers (nt) exceeds the memory size; the buffer will overwrite itself. You can disable this error message with force_overwrite=True.")
self.wlen(nt)
self.wstop()
self.wstart()
[docs]class AxisBufferDdrV1pt1(AxisBufferDdrV1):
"""
Same as AxisBufferDdrV1 but firmware has the junk samples bug fixed.
"""
# AXIS Buffer DDR V1 Registers.
bindto = ['QICK:QICK:axis_buffer_ddr:1.1']
JUNK_SAMPLES_BUG_FIX = True # Bug is fixed in IP version >= 1.1