Receiver Phase Calibration on the AIR-T¶
Introduction¶
Achieving phase alignment across transceiver channels is essential for a variety of RF applications, including but not limited to: direction finding (DF), angle of arrival (AoA), and beamforming. This tutorial aims to describe how to perform phase correction across the receive channels of your AIR-T. For a refresher on the theory involved in this process, please refer to the Theory of Operation.
NOTE: The processes described in this tutorial are for narrow-band (~100MHz bandwidth) RX phase calibration only. Each calibration is valid only for your current configured Rx LO frequency (baseband frequency is set to 0Hz in this example).
For the purpose of this demonstration, we are going to talk about an internal (AIR-T system) calibration, and an external calbration (external to AIR-T). The general architecture of these paths is described in the following section.
Internal Path¶
The internal calibration path, to be used for system calibration of your AIR-T, is part of the internal circuitry of the radio. Specifically, when the internal calibration path is active, the following is true, with a graphical architectural overview below:
- Instead of transmitting through the RF front-end SMA connectors, the AD9371 transmits data internal to the AIR-T via an RF switch.
- The data is then routed through an RF switching network across all receive channels on the AIR-T, back into the AD9371.
- Due to the nature of this design, the power of the transmitted signal along this internal path is divided equally among all receive channels (4 channels for AIR7311, 2 channels for AIR7310).
flowchart LR
%% Define Shapes
subgraph RFIC_TX[<b>AD9371 RFIC</b>]
TRANSMIT[<b>TX</b>]
end
subgraph RFIC_RX[<b>AD9371 RFIC</b>]
RECEIVE[<b>RX</b>]
end
subgraph TX_CHAN[<b>TX Path</b>]
TXOUT1[<b>RF Channel</b>]
VATT21[<b>RF Switch</b>]
end
subgraph RX_CHAN[<b>RX Path</b>]
RXIN[<b>RF Switch</b>]
RXCHAN[<b>RF Channel</b>]
end
subgraph INT[<b>Calibration Path</b>]
AD1C[<b>Switch Network</b>]
end
%% Define Connections
TRANSMIT --> TXOUT1 --> VATT21 --> AD1C --> RXIN -->RXCHAN --> RECEIVEExternal Path¶
The external calibration path setup is application-dependent, but for the purposes of this demonstration, a 1:4 power splitter is assumed to be used from TX0 to RX0, RX1, RX2, and RX3 of an AIR7311, ideally via phase-matched RF cables. In this case, the internal calibration path is not active, and the following is true, with a graphical architectural overview below:
- As in normal operation, the AD9371 transmits data through the TX RF channel, and out of the RF front-end TX SMA connector. The same RF switch gates the SMA connector as the internal calibration path; it is simply configured differently.
- The data is then externally routed through an RF SMA cable, into a 1:4 power splitter, and finally, through four RF SMA cables (one on each output port of the power splitter) back into the RF front-end RX SMA connectors. The data then makes its back through the RX RF channel and into the AD9371.
- This external calibration is intended to be applied on top of the internal system calibration, as described later in Step 3 of the tutorial.
flowchart LR
%% Define Shapes
subgraph RFIC_TX[<b>AD9371 RFIC</b>]
TRANSMIT[<b>TX</b>]
end
subgraph RFIC_RX[<b>AD9371 RFIC</b>]
RECEIVE[<b>RX</b>]
end
subgraph TX_CHAN[<b>TX Path</b>]
TXOUT[<b>RF Channel</b>]
VATT2[<b>RF Switch</b>]
end
subgraph RX_CHAN[<b>RX Path</b>]
RXOUT[<b>RF Channel</b>]
RXSWITCH[<b>RF Switch</b>]
end
subgraph EXT[<b>External Network</b>]
ADC["<b>TX<br>SMA</b>"]
CABLE["<b>Cable Network<br>(Application Specific)</b>"]
DAC["<b>RX<br>SMA</b>"]
end
%% Define Connections
TRANSMIT --> TXOUT --> VATT2 --> ADC --> CABLE --> DAC --> RXSWITCH --> RXOUT --> RECEIVEOverview¶
This Python tutorial will provide the basic tools needed to ensure that the receive channels on your AIR-T are aligned in phase. The general flow of this process is as follows:
- Step 1: Create a phase calibration class through which to initialize the radio and execute phase calibration
- Step 2: Perform phase calibration on received data
- Step 3: Use open RX phase aligned data streams for your specific application
Requirements¶
- AirStack 2.4.0+
- Connectorized Power Splitter(s) (Application Dependent)
- RF Cables (x1 for Transmit, x2-4 for Receive)
Step 1¶
Create Phase Calibration Class¶
Create a PhaseCal class in Python that, upon initialization, will:
- Ensure the version of AirStack Core running on your AIR-T supports receiver phase calibration.
- Store your SoapySDR device object
device, chosen transmit sourcetx_cal_srcfor calibration, as well as your transmit streamtx_streamand/or receiver stream(s)rx_streamas member variables. Note that therx_streamobject can be accessed by the user after calibration is run, as long as thePhaseCalobject remains in scope.
Setup Streams¶
The PhaseCal class constructor will setup the transmit and receive stream(s) for you, given user-provided:
freq: Default1400e6; Center Frequency (Hz)master_clk: Default125e6; Master Clock Rate (Hz)chan: Default(0,1,2,3); Receive Channelsext-gain: Default-15.0; External Calibration Gain (dB) (Only relevant when using external calibration path)
For the purposes of this demonstration, the default parameters listed above will be used.
Step 2¶
Phase Correlation¶
Once run() is called by the PhaseCal object, the calibration path is configured as one of the following:
internal: Default; AIR-T internal system calibrationexternal: External to AIR-T; Application specific cable networkboth: Performs internal system calibration first, then performs'external'calibration on top
Once the chosen path(s) are configured, the receive channels can be categorized into test channels (chan[1:]) and a reference channel (chan[0]).
In estimate_pair_delay_phase(), the reference and test channel data is represented by sig_ref_zm and sig_test_zm, respectively, where the _zm suffix indicates that the mean has been subtracted in order to remove any DC bias.
The full cross-correlation of the received samples is computed by np.correlate(sig_test_zm, sig_ref_zm, mode='full'). If a test channel is delayed by delay_seconds seconds relative to the reference channel, then the two signals align best when the test signal is shifted back by that delay.
This implementation limits the peak delay search to \(\pm100~ns\) with lag_mask, then stores this peak lag as delay_samples.
Using the SOAPY_SDR_HAS_TIME flag when receiving data as shown in the Complete Code Listing below should ensure time-alignment across receive channels, resulting in delay_samples = 0
NOTE: if delay_samples is non-zero, the current channel pair is not time-aligned. In this case, please do not proceed unless you intend on accounting for this delay, as time-alignment is required in order for the phase calibration to be accurate.
Once time-alignment between receive channels is confirmed, the code slices sig_ref and sig_test so only their overlapping, time-aligned samples remain. The residual phase offset (radians) is then estimated with the dot product np.angle(np.vdot(ref_aligned, test_aligned)). This dot product is used to calculate the phase offset between the chosen test and reference channel, which is stored in phase_deg as degrees.
def estimate_pair_delay_phase(sig_ref: np.ndarray, sig_test: np.ndarray, samp_rate: float):
"""Estimate relative delay and phase between two complex signals.
Takes in the reference samples, test samples, and the sample rate of the
signal. Returns a dictionary containing the computed delay (in samples and
seconds) and the computed phase (in degrees).
"""
max_lag_samples = int(round(100e-9 * samp_rate))
sig_ref_zm = sig_ref.astype(np.complex128) - np.mean(sig_ref)
sig_test_zm = sig_test.astype(np.complex128) - np.mean(sig_test)
corr_full = scipy.signal.correlate(sig_test_zm, sig_ref_zm, method='fft')
lag_idxs_full = np.arange(-(len(sig_ref) - 1), len(sig_test))
lag_mask = np.abs(lag_idxs_full) <= max_lag_samples
corr = corr_full[lag_mask]
lag_idxs = lag_idxs_full[lag_mask]
peak_idx = int(np.argmax(np.abs(corr)))
delay_samples = int(lag_idxs[peak_idx])
delay_seconds = delay_samples / samp_rate
if delay_samples >= 0:
ref_aligned = sig_ref[:len(sig_ref) - delay_samples]
test_aligned = sig_test[delay_samples:]
else:
shift = -delay_samples
ref_aligned = sig_ref[shift:]
test_aligned = sig_test[:len(sig_test) - shift]
n_overlap = min(len(ref_aligned), len(test_aligned))
ref_aligned = ref_aligned[:n_overlap]
test_aligned = test_aligned[:n_overlap]
if n_overlap == 0:
phase_deg = np.nan
else:
phase_est = np.vdot(ref_aligned, test_aligned)
phase_deg = float(np.degrees(np.angle(phase_est)))
return {
'delay_samples': delay_samples,
'delay_seconds': delay_seconds,
'phase_deg': phase_deg,
}
def analyze_channel_pairs(chans: list[int], buffs: list[np.ndarray], samp_rate: float):
"""Compute delay/phase results for every received channel pair.
Returns a list where each element is a dictionary containing the reference
channel, the test channel, and the parameters computed by
estimate_pair_delay_phase().
"""
pair_results = []
for idx_ref, idx_test in combinations(range(len(chans)), 2):
result = estimate_pair_delay_phase(buffs[idx_ref], buffs[idx_test], samp_rate)
result['chan_ref'] = chans[idx_ref]
result['chan_test'] = chans[idx_test]
pair_results.append(result)
return pair_results
Phasor Generation¶
The previous step estimates phase offsets between pairs of channels, stored as degrees in phase_deg = result['phase_deg'] for each pair. These offsets are then converted to radians via np.radians(phase_deg).
This leaves one unknown phase per remaining channel. We can represent this relationship using a system of linear equations, where each row in the lhs matrix represents each valid ref and test channel pair.
Since there are more pair measurements than unknown channel phases, the equations are likely to overdetermined. np.linalg.lstsq() finds the set of channel phases that best fits all pairwise measurements in the least-squares sense.
The returned calibration values are complex unit phasors, represented as np.complex128(np.exp(-1j * channel_phases[chan])) for each test channel chan, where the channel_phases are in radians. The reference channel receives the identity phasor np.complex128(np.exp(-1j * 0)), which is equal to 1.
def compute_channel_phase_rotations(chans: list[int], pair_results: list[dict]):
"""Solve for one complex phase rotation per channel from pairwise
measurements.
Takes in the data structure produced by analyze_channel_pairs() and returns
the computed phase rotations as a dictionary where the key is the
corresponding reference channel.
"""
ref_chan = chans[0]
chan_to_col = {chan: idx for idx, chan in enumerate(chans[1:])}
num_unknowns = len(chans) - 1
if num_unknowns == 0:
return {ref_chan: np.complex128(1.0 + 0.0j)}
rows = []
rhs = []
for result in pair_results:
phase_deg = result["phase_deg"]
if not np.isfinite(phase_deg):
continue
m_ij = np.exp(1j * np.radians(phase_deg))
row = np.zeros(num_unknowns, dtype=np.complex128)
b = 0.0j
test_chan = result["chan_test"]
ref_pair_chan = result["chan_ref"]
if test_chan == ref_chan:
# 1 - m_ij * r_j ~= 0
row[chan_to_col[ref_pair_chan]] = -m_ij
b = -1.0
elif ref_pair_chan == ref_chan:
# r_i - m_ij * 1 ~= 0
row[chan_to_col[test_chan]] = 1.0
b = m_ij
else:
# r_i - m_ij * r_j ~= 0
row[chan_to_col[test_chan]] = 1.0
row[chan_to_col[ref_pair_chan]] = -m_ij
rows.append(row)
rhs.append(b)
if not rows:
return {chan: np.complex128(1.0 + 0.0j) for chan in chans}
A = np.vstack(rows)
b = np.asarray(rhs, dtype=np.complex128)
sol, _, _, _ = np.linalg.lstsq(A, b, rcond=None)
channel_rotations = {ref_chan: np.complex128(1.0 + 0.0j)}
for chan, r in zip(chans[1:], sol):
# Project back onto the unit circle
mag = np.abs(r)
channel_rotations[chan] = (
np.conj(np.complex128(r / mag)) if mag > 0 else np.complex128(1.0 + 0.0j)
)
return channel_rotations
Phasor Application¶
The per-channel correction phasors, denoted delta_phasor, computed in the previous section are then programmed into the receive datapaths for each test channel. This correction gets applied on top of the phasor correction existing_phasor currently stored on that receive datapath, simply by multiplying the two values together: phasor = existing_phasor * delta_phasor.
This makes calibration cumulative: each calibration pass adjusts the phasor already programmed for that channel rather than replacing it with only the latest estimate.
The real and imaginary parts are then written independently using the Soapy Settings API, via arguments cal_phasor_real and cal_phasor_imag, respectively. Internally, the radio uses those two values as a complex scalar along the corresponding receive datapath.
Once programmed, these phasors remain applied to each respective receive datapath for the life of the current SoapySDR device object, including later streams opened from that same object. Recreating the device object starts a new device-control lifetime, so applications that need calibrated streams should run or restore calibration after constructing their SoapySDR device.
def _apply_channel_phase_rotations(self, phase_rotations: dict[np.complex128]):
"""Accumulate and program per-channel RX calibration phasor settings.
Takes in the computed per-channel phase rotations and returns the
applied phasors as a per-channel dictionary. See below for contents of each
index.
"""
applied_phasors = {}
for chan in self._rx_cal_chans:
delta_phasor = phase_rotations.get(chan, np.complex128(1.0 + 0.0j))
existing_real = float(self._device.readSetting(SOAPY_SDR_RX, chan, "cal_phasor_real"))
existing_imag = float(self._device.readSetting(SOAPY_SDR_RX, chan, "cal_phasor_imag"))
existing_phasor = np.complex128(existing_real + 1j * existing_imag)
phasor = existing_phasor * delta_phasor
phasor_mag = np.abs(phasor)
if phasor_mag > 0:
phasor /= phasor_mag
else:
phasor = np.complex128(1.0 + 0.0j)
self._device.writeSetting(SOAPY_SDR_RX, chan, "cal_phasor_real", phasor.real)
self._device.writeSetting(SOAPY_SDR_RX, chan, "cal_phasor_imag", phasor.imag)
applied_phasors[chan] = {
'delta_phasor': delta_phasor,
'phasor': phasor,
'delta_phase_deg': float(np.degrees(np.angle(delta_phasor))),
'phase_deg': float(np.degrees(np.angle(phasor))),
}
return applied_phasors
Step 3¶
Phase-Aligned Streams Usage¶
Within the context of your current PhaseCal() object, you now have the ability to access the rx_stream used in this example as required by your specific application. The correction phasors programmed in each receive datapath will remain applied for the lifetime of your SoapySDR device object, denoted as sdr in this example.
Command Line Usage¶
Below is example command line usage of the included test script phase_calibration.py using different receive channel configurations on the AIR-T and performing an AIR-T internal system calibration, followed immediately by a cumulative external calibration through your external application specific cable network:
Two RX Channel Calibration:
./phase_calibration.py --channels 0 1 --calibration both
Four RX Channel Calibration (default; channel specification here is for emphasis only):
./phase_calibration.py --channels 0 1 2 3 --calibration both
For a complete listing of all available arguments, issue the following command:
./phase_calibration.py --help
Expected output:
usage: phase_calibration.py [-h] [--freq FREQ] [--master-clock MASTER_CLOCK] [--captures CAPTURES] [--delay DELAY]
[--samples SAMPLES] [--channels CHANNELS [CHANNELS ...]] [--tx-channel TX_CHANNEL]
[--calibration {internal,external,both}] [--ext-gain EXT_GAIN]
Run RX Phase Calibration on an AIR-T Device.
options:
-h, --help show this help message and exit
--freq FREQ RX carrier frequency in Hz. (default: 1400000000.0)
--master-clock MASTER_CLOCK
Master clock rate in Hz. (default: 125000000.0)
--captures CAPTURES number of calibration captures. (default: 5)
--delay DELAY delay between captures in seconds. (default: 1.0)
--samples SAMPLES samples per capture. (default: 4096)
--channels CHANNELS [CHANNELS ...]
space-separated RX channels. (default: (0, 1, 2, 3))
--tx-channel TX_CHANNEL
TX channel used as the calibration source. (default: 0)
--calibration {internal,external,both}
calibration path to run. (default: internal)
--ext-gain EXT_GAIN external calibration gain in dB.
NOTE: only relevant when --calibration is external or both.
Ignored for internal calibration. (default: -15.0)
Some troubleshooting tips:
- The default
capturesvalue of5may be too small to get an accurate phase correction for your specific application. In this case, try increasing the number of captures to10or more. - The default
delayvalue of1second may be too short in order to get accurate consecutive captures for your specific application. In this case, try increasing the delay between captures to be multiple seconds long.
Complete Code Listing¶
This code example is Open Source, made available under the terms of the BSD 3-Clause License. It may be found in the AirStack Examples repository on GitHub and is reproduced below.
#!/usr/bin/env python3
"""
PhaseCal - RX Phase calibration utility for AIR-T devices.
Estimates and applies phase corrections across multiple receive channels
to enable coherent combining.
"""
import argparse
import sys
import time
import warnings
from itertools import combinations
import numpy as np
import scipy.signal
import SoapySDR
from SoapySDR import SOAPY_SDR_RX, SOAPY_SDR_TX, SOAPY_SDR_CF32, SOAPY_SDR_HAS_TIME, errToStr
# Helper functions for various DSP computations
def analyze_channel_pairs(chans: list[int], buffs: list[np.ndarray], samp_rate: float):
"""Compute delay/phase results for every received channel pair.
Returns a list where each element is a dictionary containing the reference
channel, the test channel, and the parameters computed by
estimate_pair_delay_phase().
"""
pair_results = []
for idx_ref, idx_test in combinations(range(len(chans)), 2):
result = estimate_pair_delay_phase(buffs[idx_ref], buffs[idx_test], samp_rate)
result['chan_ref'] = chans[idx_ref]
result['chan_test'] = chans[idx_test]
pair_results.append(result)
return pair_results
def estimate_pair_delay_phase(sig_ref: np.ndarray, sig_test: np.ndarray, samp_rate: float):
"""Estimate relative delay and phase between two complex signals.
Takes in the reference samples, test samples, and the sample rate of the
signal. Returns a dictionary containing the computed delay (in samples and
seconds) and the computed phase (in degrees).
"""
max_lag_samples = int(round(100e-9 * samp_rate))
sig_ref_zm = sig_ref.astype(np.complex128) - np.mean(sig_ref)
sig_test_zm = sig_test.astype(np.complex128) - np.mean(sig_test)
corr_full = scipy.signal.correlate(sig_test_zm, sig_ref_zm, method='fft')
lag_idxs_full = np.arange(-(len(sig_ref) - 1), len(sig_test))
lag_mask = np.abs(lag_idxs_full) <= max_lag_samples
corr = corr_full[lag_mask]
lag_idxs = lag_idxs_full[lag_mask]
peak_idx = int(np.argmax(np.abs(corr)))
delay_samples = int(lag_idxs[peak_idx])
delay_seconds = delay_samples / samp_rate
if delay_samples >= 0:
ref_aligned = sig_ref[:len(sig_ref) - delay_samples]
test_aligned = sig_test[delay_samples:]
else:
shift = -delay_samples
ref_aligned = sig_ref[shift:]
test_aligned = sig_test[:len(sig_test) - shift]
n_overlap = min(len(ref_aligned), len(test_aligned))
ref_aligned = ref_aligned[:n_overlap]
test_aligned = test_aligned[:n_overlap]
if n_overlap == 0:
phase_deg = np.nan
else:
phase_est = np.vdot(ref_aligned, test_aligned)
phase_deg = float(np.degrees(np.angle(phase_est)))
return {
'delay_samples': delay_samples,
'delay_seconds': delay_seconds,
'phase_deg': phase_deg,
}
def compute_channel_phase_rotations(chans: list[int], pair_results: list[dict]):
"""Solve for one complex phase rotation per channel from pairwise
measurements.
Takes in the data structure produced by analyze_channel_pairs() and returns
the computed phase rotations as a dictionary where the key is the
corresponding reference channel.
"""
ref_chan = chans[0]
chan_to_col = {chan: idx for idx, chan in enumerate(chans[1:])}
num_unknowns = len(chans) - 1
if num_unknowns == 0:
return {ref_chan: np.complex128(1.0 + 0.0j)}
rows = []
rhs = []
for result in pair_results:
phase_deg = result["phase_deg"]
if not np.isfinite(phase_deg):
continue
m_ij = np.exp(1j * np.radians(phase_deg))
row = np.zeros(num_unknowns, dtype=np.complex128)
b = 0.0j
test_chan = result["chan_test"]
ref_pair_chan = result["chan_ref"]
if test_chan == ref_chan:
# 1 - m_ij * r_j ~= 0
row[chan_to_col[ref_pair_chan]] = -m_ij
b = -1.0
elif ref_pair_chan == ref_chan:
# r_i - m_ij * 1 ~= 0
row[chan_to_col[test_chan]] = 1.0
b = m_ij
else:
# r_i - m_ij * r_j ~= 0
row[chan_to_col[test_chan]] = 1.0
row[chan_to_col[ref_pair_chan]] = -m_ij
rows.append(row)
rhs.append(b)
if not rows:
return {chan: np.complex128(1.0 + 0.0j) for chan in chans}
A = np.vstack(rows)
b = np.asarray(rhs, dtype=np.complex128)
sol, _, _, _ = np.linalg.lstsq(A, b, rcond=None)
channel_rotations = {ref_chan: np.complex128(1.0 + 0.0j)}
for chan, r in zip(chans[1:], sol):
# Project back onto the unit circle
mag = np.abs(r)
channel_rotations[chan] = (
np.conj(np.complex128(r / mag)) if mag > 0 else np.complex128(1.0 + 0.0j)
)
return channel_rotations
class PhaseCal:
def __init__(self, device: SoapySDR.Device, tx_cal_src: int,
rx_cal_chans: list[int], master_clk_rate: float, freq: float):
self._tx_stream = None
self.rx_stream = None
self.path_results = {"internal": [], "external": []}
global_settings = device.getSettingInfo()
info = next((x for x in global_settings if x.key == "phase_cal:mode"), None)
if info is None:
raise ValueError("Device does not support phase calibration!")
self._device = device
if self._device.getMasterClockRate() != master_clk_rate:
# Note that it is much more efficient to set the master clock rate when
# creating a new SDR device, so hopefully this warning is never printed.
warnings.warn("SDR master clock rate not set previously, applying now...")
self._device.setMasterClockRate(master_clk_rate)
# Setup Rx
if not isinstance(rx_cal_chans, list):
rx_cal_chans = list(rx_cal_chans)
self._rx_cal_chans = rx_cal_chans
for chan in self._rx_cal_chans:
self._device.setSampleRate(SOAPY_SDR_RX, chan, master_clk_rate / 2)
self._device.setFrequency(SOAPY_SDR_RX, chan, freq)
self._device.setGainMode(SOAPY_SDR_RX, chan, False) # AGC off
self.rx_stream = self._device.setupStream(SOAPY_SDR_RX, SOAPY_SDR_CF32, self._rx_cal_chans)
print(f'\nReceiving on Channel(s): {self._rx_cal_chans}\n')
# Setup Tx
tx_fs = master_clk_rate
bb_freq = 0
tx_lo_freq = freq - bb_freq
self._tx_cal_src = tx_cal_src
self._device.setSampleRate(SOAPY_SDR_TX, self._tx_cal_src, tx_fs)
self._device.setFrequency(SOAPY_SDR_TX, self._tx_cal_src, tx_lo_freq)
self._tx_stream = self._device.setupStream(SOAPY_SDR_TX, SOAPY_SDR_CF32, [self._tx_cal_src])
# Activate noise generator on FPGA for calibration signal.
self._device.activateStream(self._tx_stream)
self._device.writeSetting("dds:frequency", bb_freq)
print('Enabling noise generator for calibration...')
self._device.writeSetting("dds:mode", "Noise")
print(f'\nTransmitting on Channel: {self._tx_cal_src}\n')
def __del__(self):
self._device.writeSetting("dds:mode", "Off")
if self._tx_stream is not None:
self._device.closeStream(self._tx_stream)
self._tx_stream = None
if self.rx_stream is not None:
self._device.closeStream(self.rx_stream)
self.rx_stream = None
def run(self, cal_type: str = "internal", num_captures: int = 5,
delay: float = 1.0, buff_len: int = 4096, ext_gain: float = -15.0):
"""Configures the device for calibration and runs the calibration."""
cal_type = cal_type.lower()
if cal_type not in ('internal', 'external', 'both'):
raise ValueError(f"Unsupported calibration type: {cal_type}")
if cal_type in ('internal', 'both'):
print("Performing Internal Calibration...")
if self._rx_cal_chans == [0, 1] and self._tx_cal_src in self._rx_cal_chans:
phase_cal_mode = 'Slot A'
elif self._rx_cal_chans == [2, 3] and self._tx_cal_src in self._rx_cal_chans:
phase_cal_mode = 'Slot B'
else:
phase_cal_mode = 'Four Channel'
if cal_type == 'both':
self._device.writeSetting('phase_cal:mode', 'Off')
self._device.setGain(SOAPY_SDR_TX, self._tx_cal_src, 0.0)
for chan in self._rx_cal_chans:
self._device.setGain(SOAPY_SDR_RX, chan, 0.0)
self._device.writeSetting('phase_cal:mode', phase_cal_mode)
self._device.writeSetting('phase_cal:src_chan', self._tx_cal_src)
phasors_internal = self._run_cal_path("Internal",
num_captures=num_captures,
delay=delay,
buff_len=buff_len)
if cal_type in ('external', 'both'):
print("Performing External Calibration...")
self._device.writeSetting('phase_cal:mode', "Off")
self._device.setGain(SOAPY_SDR_TX, self._tx_cal_src, -15.0)
for chan in self._rx_cal_chans:
self._device.setGain(SOAPY_SDR_RX, chan, -15.0)
phasors_external = self._run_cal_path("External",
num_captures=num_captures,
delay=delay,
buff_len=buff_len,
ext_gain=ext_gain)
def _run_cal_path(self, path_name: str = "Internal", num_captures: int = 5,
delay: float = 1.0, buff_len: int = 4096, ext_gain: float = -15.0):
"""Runs the requested calibration path and applies the phasor settings.
Returns the results of each capture as a list, where each index contains
applied phasor settings (i.e., the data structure returned from
apply_channel_phase_rotations()) from a specific internal or external run.
"""
samp_rate = self._device.getSampleRate(SOAPY_SDR_RX, self._rx_cal_chans[0])
start_time_ns = int(time.time() * 1e9)
self._device.setHardwareTime(start_time_ns, "now")
repeat_time_ns = int(delay * 1e9)
rx_time_ns = start_time_ns + int(1e9)
for capture in range(num_captures):
rx_buff = [np.zeros(buff_len, np.complex64) for _ in self._rx_cal_chans]
self._device.activateStream(self.rx_stream,
flags=SOAPY_SDR_HAS_TIME,
timeNs=rx_time_ns)
now = self._device.getHardwareTime("now")
if now > rx_time_ns:
print(f"capture: {capture} is late, skipping")
print(f"now - rx_time_ns = {now - rx_time_ns} ns")
self._device.deactivateStream(self.rx_stream)
rx_time_ns += repeat_time_ns
continue
rc = self._device.readStream(self.rx_stream, rx_buff, buff_len,
timeoutUs=int((rx_time_ns - now) / 1000) + 100000)
if rc.ret != buff_len:
print('capture: {}, RX Error {}: {}'.format(capture, rc.ret,
errToStr(rc.ret)))
self._device.deactivateStream(self.rx_stream)
break
self._device.deactivateStream(self.rx_stream)
pair_results = analyze_channel_pairs(self._rx_cal_chans, rx_buff, samp_rate)
if capture == num_captures - 1:
print('Results:')
for result in pair_results:
print(
f" CH{result['chan_test']} vs CH{result['chan_ref']} "
f"delay={result['delay_samples']} samples "
f"({result['delay_seconds'] * 1e9:.1f} ns), "
f"phase={result['phase_deg']:.2f} deg, "
)
print()
phase_rotations = compute_channel_phase_rotations(self._rx_cal_chans, pair_results)
phasor_results = self._apply_channel_phase_rotations(phase_rotations)
self.path_results[path_name.lower()].append(phasor_results)
rx_time_ns += repeat_time_ns
return self.path_results[path_name.lower()]
def _apply_channel_phase_rotations(self, phase_rotations: dict[np.complex128]):
"""Accumulate and program per-channel RX calibration phasor settings.
Takes in the computed per-channel phase rotations and returns the
applied phasors as a per-channel dictionary. See below for contents of each
index.
"""
applied_phasors = {}
for chan in self._rx_cal_chans:
delta_phasor = phase_rotations.get(chan, np.complex128(1.0 + 0.0j))
existing_real = float(self._device.readSetting(SOAPY_SDR_RX, chan, "cal_phasor_real"))
existing_imag = float(self._device.readSetting(SOAPY_SDR_RX, chan, "cal_phasor_imag"))
existing_phasor = np.complex128(existing_real + 1j * existing_imag)
phasor = existing_phasor * delta_phasor
phasor_mag = np.abs(phasor)
if phasor_mag > 0:
phasor /= phasor_mag
else:
phasor = np.complex128(1.0 + 0.0j)
self._device.writeSetting(SOAPY_SDR_RX, chan, "cal_phasor_real", phasor.real)
self._device.writeSetting(SOAPY_SDR_RX, chan, "cal_phasor_imag", phasor.imag)
applied_phasors[chan] = {
'delta_phasor': delta_phasor,
'phasor': phasor,
'delta_phase_deg': float(np.degrees(np.angle(delta_phasor))),
'phase_deg': float(np.degrees(np.angle(phasor))),
}
return applied_phasors
class DefaultsRawTextHelpFormatter(argparse.ArgumentDefaultsHelpFormatter,
argparse.RawTextHelpFormatter):
"""Preserve help-text newlines while showing argument default values."""
def parse_command_line_arguments(argv=None):
parser = argparse.ArgumentParser(description="Run RX Phase Calibration on an AIR-T Device.",
formatter_class=DefaultsRawTextHelpFormatter)
argument_definitions = (
('--freq', dict(type=float, default=1400e6, help='RX carrier frequency in Hz.')),
('--master-clock', dict(type=float, default=125e6, help='Master clock rate in Hz.')),
('--captures', dict(type=int, default=5, help='number of calibration captures.')),
('--delay', dict(type=float, default=1.0, help='delay between captures in seconds.')),
('--samples', dict(type=int, default=4096, help='samples per capture.')),
('--channels', dict(type=int, action='store', default=(0, 1, 2, 3),
nargs='+', help='space-separated RX channels.')),
('--tx-channel', dict(type=int, default=0,
help='TX channel used as the calibration source.')),
('--calibration', dict(choices=('internal', 'external', 'both'),
default='internal', help='calibration path to run.')),
('--ext-gain', dict(type=float, default=-15.0,
help=('external calibration gain in dB.\n'
'NOTE: only relevant when --calibration is external or both.\n'
'Ignored for internal calibration.'))),
)
for opt, kwargs in argument_definitions:
parser.add_argument(opt, **kwargs)
args = parser.parse_args(argv)
command_line = argv if argv is not None else sys.argv[1:]
ext_gain_specified = any(
argument == '--ext-gain' or argument.startswith('--ext-gain=')
for argument in command_line
)
if args.calibration == 'internal' and ext_gain_specified:
warnings.warn('--ext-gain is ignored when --calibration is internal.',
UserWarning, stacklevel=2)
return args
def print_configuration(args: argparse.ArgumentParser):
print("\nRadio Configuration:\n")
for label, value in (
('RX Channels', args.channels),
('TX Calibration Source', f'CH{args.tx_channel}'),
('Frequency', f'{args.freq / 1e6:.3f} MHz'),
('Master Clock Rate', f'{args.master_clock / 1e6:.6f} MHz'),
('Captures', args.captures),
('Delay', f'{args.delay:g} s'),
('Samples per Capture', args.samples),
('Calibration Path(s)', args.calibration)
):
print(f' {label}: {value}')
print()
def main():
sys.stdout.reconfigure(line_buffering=True)
args = parse_command_line_arguments()
print_configuration(args)
dev_args = dict(master_clock_rate=str(args.master_clock),
time_src="internal",
lock_timeout=str(10))
sdr = SoapySDR.Device(dev_args)
cal = PhaseCal(sdr, args.tx_channel, args.channels, args.master_clock, args.freq)
cal.run(args.calibration, args.captures, args.delay, args.samples, args.ext_gain)
# Note that after calibration is run, cal.rx_stream can be used to read
# samples from the calibrated channels. The calibration will be valid so
# long as the cal object is in scope and not garbage collected.
if __name__ == '__main__':
main()