Compare commits
5 Commits
958d1f96d1
...
v0.1.1
Author | SHA1 | Date | |
---|---|---|---|
78d0034e34 | |||
a56b2e30e2 | |||
ec010af947 | |||
7e0df9e643 | |||
a20217967f |
10
.github/workflows/python_publish.yml
vendored
10
.github/workflows/python_publish.yml
vendored
@ -54,13 +54,7 @@ jobs:
|
|||||||
with:
|
with:
|
||||||
name: python-package-distributions
|
name: python-package-distributions
|
||||||
path: dist/
|
path: dist/
|
||||||
- name: Set up Python
|
|
||||||
uses: actions/setup-python@v5
|
|
||||||
with:
|
|
||||||
python-version: "3.x"
|
|
||||||
- name: Publish distribution 📦 to PyPI
|
- name: Publish distribution 📦 to PyPI
|
||||||
uses: pypa/gh-action-pypi-publish@release/v1
|
uses: pypa/gh-action-pypi-publish@0ab0b79471669eb3a4d647e625009c62f9f3b241
|
||||||
with:
|
with:
|
||||||
password: ${{ secrets.PYPI_API_TOKEN }}
|
password: ${{ secrets.PYPI_API_TOKEN }}
|
||||||
verbose: true
|
|
||||||
print-hash: true
|
|
@ -1,7 +1,7 @@
|
|||||||
# %% imports
|
# %% imports
|
||||||
import copy
|
import copy
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Any, Dict, Optional, Tuple
|
from typing import Any, Dict, Tuple
|
||||||
|
|
||||||
import adi
|
import adi
|
||||||
import iio
|
import iio
|
||||||
@ -18,6 +18,20 @@ dir_ = Path(__file__).parent
|
|||||||
|
|
||||||
|
|
||||||
# %% connection
|
# %% connection
|
||||||
|
|
||||||
|
|
||||||
|
def generate_tone(f: float, fs: float, N: int = 1024, scale: int = 2**14):
|
||||||
|
fs = int(fs)
|
||||||
|
fc = int(f / (fs / N)) * (fs / N)
|
||||||
|
ts = 1 / float(fs)
|
||||||
|
t = np.arange(0, N * ts, ts)
|
||||||
|
i = np.cos(2 * np.pi * t * fc) * scale
|
||||||
|
q = np.sin(2 * np.pi * t * fc) * scale
|
||||||
|
iq = i + 1j * q
|
||||||
|
|
||||||
|
return iq
|
||||||
|
|
||||||
|
|
||||||
class Charon:
|
class Charon:
|
||||||
FREQUENCY_OFFSET = 1e6
|
FREQUENCY_OFFSET = 1e6
|
||||||
|
|
||||||
@ -58,19 +72,19 @@ class Charon:
|
|||||||
self.sdr.loopback = 0
|
self.sdr.loopback = 0
|
||||||
self.sdr.gain_control_mode_chan0 = "manual"
|
self.sdr.gain_control_mode_chan0 = "manual"
|
||||||
self.sdr.gain_control_mode_chan1 = "manual"
|
self.sdr.gain_control_mode_chan1 = "manual"
|
||||||
self.sdr.rx_hardwaregain_chan0 = 40
|
self.sdr.rx_hardwaregain_chan0 = 10
|
||||||
self.sdr.rx_hardwaregain_chan1 = 40
|
self.sdr.rx_hardwaregain_chan1 = 10
|
||||||
self.sdr.tx_hardwaregain_chan0 = -10
|
self.sdr.tx_hardwaregain_chan0 = -10
|
||||||
|
|
||||||
# switch control
|
# # switch control
|
||||||
ctx = iio.Context(uri)
|
# ctx = iio.Context(uri)
|
||||||
self.ctrl = ctx.find_device("ad9361-phy")
|
# self.ctrl = ctx.find_device("ad9361-phy")
|
||||||
# raw ad9361 register accesss:
|
# # raw ad9361 register accesss:
|
||||||
# https://ez.analog.com/linux-software-drivers/f/q-a/120853/control-fmcomms3-s-gpo-with-python
|
# # https://ez.analog.com/linux-software-drivers/f/q-a/120853/control-fmcomms3-s-gpo-with-python
|
||||||
# https://www.analog.com/media/cn/technical-documentation/user-guides/ad9364_register_map_reference_manual_ug-672.pdf
|
# # https://www.analog.com/media/cn/technical-documentation/user-guides/ad9364_register_map_reference_manual_ug-672.pdf
|
||||||
self.ctrl.reg_write(0x26, 0x90) # bit 7: AuxDAC Manual, bit 4: GPO Manual
|
# self.ctrl.reg_write(0x26, 0x90) # bit 7: AuxDAC Manual, bit 4: GPO Manual
|
||||||
self._set_gpo(self.ports[0] - 1)
|
# self._set_gpo(self.ports[0] - 1)
|
||||||
# TODO: init AuxDAC
|
# # TODO: init AuxDAC
|
||||||
|
|
||||||
def get_config(self) -> Dict[str, Any]:
|
def get_config(self) -> Dict[str, Any]:
|
||||||
config = dict()
|
config = dict()
|
||||||
@ -94,13 +108,24 @@ class Charon:
|
|||||||
|
|
||||||
return config
|
return config
|
||||||
|
|
||||||
|
def _get_gpo(self) -> int:
|
||||||
|
return (self.ctrl.reg_read(0x27) >> 4) & 0x0F
|
||||||
|
|
||||||
def _set_gpo(self, value: int) -> None:
|
def _set_gpo(self, value: int) -> None:
|
||||||
self.ctrl.reg_write(0x27, (value & 0x0F) << 4) # bits 7-4: GPO3-0
|
self.ctrl.reg_write(0x27, (value & 0x0F) << 4) # bits 7-4: GPO3-0
|
||||||
|
|
||||||
def set_output_power(self, power: float):
|
def set_output_power(self, power: float):
|
||||||
if power == -5:
|
# FIXME: this is a hack because I don't want to go through re-calibration
|
||||||
# FIXME: this is a hack because I don't want to go through re-calibration
|
if power == 5:
|
||||||
tx_gain = -8
|
tx_gain = -1
|
||||||
|
elif power == 0:
|
||||||
|
tx_gain = -7
|
||||||
|
elif power == -5:
|
||||||
|
tx_gain = -12
|
||||||
|
elif power == -10:
|
||||||
|
tx_gain = -17
|
||||||
|
elif power == -15:
|
||||||
|
tx_gain = -22
|
||||||
else:
|
else:
|
||||||
raise NotImplementedError()
|
raise NotImplementedError()
|
||||||
# # TODO: correct over frequency
|
# # TODO: correct over frequency
|
||||||
@ -108,19 +133,6 @@ class Charon:
|
|||||||
# tx_gain = pout.coords["tx_gain"][tx_gain_idx]
|
# tx_gain = pout.coords["tx_gain"][tx_gain_idx]
|
||||||
self.sdr.tx_hardwaregain_chan0 = float(tx_gain)
|
self.sdr.tx_hardwaregain_chan0 = float(tx_gain)
|
||||||
|
|
||||||
def generate_tone(self, f: float, N: int = 1024, fs: Optional[float] = None):
|
|
||||||
if fs is None:
|
|
||||||
fs = self.sdr.sample_rate
|
|
||||||
fs = int(fs)
|
|
||||||
fc = int(f / (fs / N)) * (fs / N)
|
|
||||||
ts = 1 / float(fs)
|
|
||||||
t = np.arange(0, N * ts, ts)
|
|
||||||
i = np.cos(2 * np.pi * t * fc) * 2**14
|
|
||||||
q = np.sin(2 * np.pi * t * fc) * 2**14
|
|
||||||
iq = i + 1j * q
|
|
||||||
|
|
||||||
return iq
|
|
||||||
|
|
||||||
def set_output(self, frequency: float, power: float):
|
def set_output(self, frequency: float, power: float):
|
||||||
# TODO: switch to DDS in Pluto
|
# TODO: switch to DDS in Pluto
|
||||||
|
|
||||||
@ -128,73 +140,109 @@ class Charon:
|
|||||||
self.set_output_power(power)
|
self.set_output_power(power)
|
||||||
self.sdr.tx_lo = int(frequency - self.FREQUENCY_OFFSET)
|
self.sdr.tx_lo = int(frequency - self.FREQUENCY_OFFSET)
|
||||||
self.sdr.tx_cyclic_buffer = True
|
self.sdr.tx_cyclic_buffer = True
|
||||||
self.sdr.tx(self.generate_tone(self.FREQUENCY_OFFSET))
|
# self.sdr.tx(generate_tone(f=self.FREQUENCY_OFFSET, fs=self.sdr.sample_rate))
|
||||||
|
self.sdr.dds_single_tone(self.FREQUENCY_OFFSET, scale=0.9, channel=0)
|
||||||
|
|
||||||
def _rx(self, count: int = 1) -> npt.ArrayLike:
|
def _rx(self, count: int = 1, fc: float | None = None) -> npt.ArrayLike:
|
||||||
if count < 1:
|
if count < 1:
|
||||||
raise ValueError
|
raise ValueError
|
||||||
|
|
||||||
self.sdr.rx_destroy_buffer()
|
self.sdr.rx_destroy_buffer()
|
||||||
|
if fc is not None:
|
||||||
|
self.sdr.rx_lo = int(fc)
|
||||||
|
self.sdr.rx_enabled_channels = [0, 1]
|
||||||
|
self.sdr.gain_control_mode_chan0 = "manual"
|
||||||
|
self.sdr.gain_control_mode_chan1 = "manual"
|
||||||
|
self.sdr.rx_hardwaregain_chan0 = 30
|
||||||
|
self.sdr.rx_hardwaregain_chan1 = 30
|
||||||
return np.concatenate([np.array(self.sdr.rx()) for _ in range(count)], axis=-1)
|
return np.concatenate([np.array(self.sdr.rx()) for _ in range(count)], axis=-1)
|
||||||
|
|
||||||
|
def get_b_over_a(self, frequency: float):
|
||||||
|
self.set_output(frequency=frequency, power=-5)
|
||||||
|
|
||||||
|
data = self._rx(1, fc=frequency - self.FREQUENCY_OFFSET)
|
||||||
|
ddc_tone = generate_tone(f=-self.FREQUENCY_OFFSET, fs=self.sdr.sample_rate, scale=1)
|
||||||
|
ddc_data = data * ddc_tone
|
||||||
|
|
||||||
|
ddc_rel = ddc_data[1] / ddc_data[0]
|
||||||
|
|
||||||
|
# plt.figure()
|
||||||
|
# plt.plot(
|
||||||
|
# np.fft.fftshift(np.fft.fftfreq(ddc_data.shape[-1], 1 / self.sdr.sample_rate)),
|
||||||
|
# np.abs(np.fft.fftshift(np.fft.fft(ddc_data, axis=-1))).T,
|
||||||
|
# )
|
||||||
|
# plt.show()
|
||||||
|
|
||||||
|
# TODO: calculate sos only once
|
||||||
|
n, wn = signal.buttord(
|
||||||
|
wp=0.3 * sdr.FREQUENCY_OFFSET,
|
||||||
|
ws=0.8 * sdr.FREQUENCY_OFFSET,
|
||||||
|
gpass=1,
|
||||||
|
gstop=40,
|
||||||
|
analog=False,
|
||||||
|
fs=self.sdr.sample_rate,
|
||||||
|
)
|
||||||
|
sos = signal.butter(n, wn, "lowpass", analog=False, output="sos", fs=self.sdr.sample_rate)
|
||||||
|
# TODO: figure out why filt sucks. Introduces SO much phase noise (out to several MHz)
|
||||||
|
filt_data = signal.sosfiltfilt(sos, ddc_data, axis=-1)
|
||||||
|
|
||||||
|
filt_rel = filt_data[1] / filt_data[0]
|
||||||
|
|
||||||
|
return np.mean(data[1] / data[0])
|
||||||
|
|
||||||
|
def sweep_b_over_a(self):
|
||||||
|
s = xr.DataArray(
|
||||||
|
np.zeros(
|
||||||
|
len(self.frequency),
|
||||||
|
dtype=np.complex128,
|
||||||
|
),
|
||||||
|
dims=["frequency"],
|
||||||
|
coords=dict(
|
||||||
|
frequency=self.frequency,
|
||||||
|
),
|
||||||
|
)
|
||||||
|
for frequency in self.frequency:
|
||||||
|
s.loc[dict(frequency=frequency)] = self.get_b_over_a(frequency=frequency)
|
||||||
|
return s
|
||||||
|
|
||||||
|
def vna_capture(self, frequency: npt.ArrayLike):
|
||||||
|
s = xr.DataArray(
|
||||||
|
np.empty(len(frequency), dtype=np.complex128),
|
||||||
|
dims=["frequency"],
|
||||||
|
coords=dict(
|
||||||
|
frequency=frequency,
|
||||||
|
),
|
||||||
|
)
|
||||||
|
for freq in s.frequency.data:
|
||||||
|
self.set_output(frequency=freq, power=-5)
|
||||||
|
self.sdr.rx_destroy_buffer()
|
||||||
|
self.sdr.rx_lo = int(freq)
|
||||||
|
self.sdr.rx_enabled_channels = [0, 1]
|
||||||
|
self.sdr.gain_control_mode_chan0 = "manual"
|
||||||
|
self.sdr.gain_control_mode_chan1 = "manual"
|
||||||
|
self.sdr.rx_hardwaregain_chan0 = 40
|
||||||
|
self.sdr.rx_hardwaregain_chan1 = 40
|
||||||
|
rx = self.sdr.rx()
|
||||||
|
s.loc[dict(frequency=freq)] = np.mean(rx[1] / rx[0])
|
||||||
|
|
||||||
|
return s
|
||||||
|
|
||||||
|
|
||||||
# %%
|
# %%
|
||||||
sdr = Charon("ip:192.168.3.1")
|
sdr = Charon("ip:192.168.3.1", frequency=np.linspace(1e9, 1.1e9, 11))
|
||||||
|
|
||||||
# %% initialization
|
# %% initialization
|
||||||
config = sdr.get_config()
|
config = sdr.get_config()
|
||||||
|
# print(sdr.ctrl.debug_attrs["adi,rx-rf-port-input-select"].value)
|
||||||
|
# print(sdr.ctrl.debug_attrs["adi,tx-rf-port-input-select"].value)
|
||||||
config
|
config
|
||||||
|
|
||||||
# %% generate tone
|
# %% generate tone
|
||||||
sdr.set_output(frequency=1e9 + sdr.FREQUENCY_OFFSET, power=-5)
|
fc = 1e9
|
||||||
|
sdr.set_output(frequency=fc + sdr.FREQUENCY_OFFSET, power=-5)
|
||||||
|
|
||||||
# %% capture data
|
# %% capture data
|
||||||
data = sdr._rx(1)
|
data = sdr._rx(1, fc=fc)
|
||||||
|
|
||||||
# %%
|
|
||||||
fig, axs = plt.subplots(2, 2, sharex=True, tight_layout=True)
|
|
||||||
# ddc_tone = np.exp(
|
|
||||||
# -1j * 2 * np.pi * (sdr.FREQUENCY_OFFSET / sdr.sdr.sample_rate) * np.arange(data.shape[-1]), dtype=np.complex128
|
|
||||||
# )
|
|
||||||
ddc_tone = sdr.generate_tone(-sdr.FREQUENCY_OFFSET) * 2**-14
|
|
||||||
ddc_data = data * ddc_tone
|
|
||||||
axs[0][0].plot(np.real(ddc_data).T, label="DDC")
|
|
||||||
axs[1][0].plot(np.imag(ddc_data).T, label="DDC")
|
|
||||||
ddc_rel = ddc_data[1] / ddc_data[0]
|
|
||||||
axs[0][1].plot(np.real(ddc_rel).T, label="DDC")
|
|
||||||
axs[1][1].plot(np.imag(ddc_rel).T, label="DDC")
|
|
||||||
n, wn = signal.buttord(
|
|
||||||
wp=0.3 * sdr.FREQUENCY_OFFSET,
|
|
||||||
ws=0.8 * sdr.FREQUENCY_OFFSET,
|
|
||||||
gpass=1,
|
|
||||||
gstop=40,
|
|
||||||
analog=False,
|
|
||||||
fs=sdr.sdr.sample_rate,
|
|
||||||
)
|
|
||||||
sos = signal.butter(n, wn, "lowpass", analog=False, output="sos", fs=sdr.sdr.sample_rate)
|
|
||||||
filt_data = signal.sosfiltfilt(sos, ddc_data, axis=-1)
|
|
||||||
axs[0][0].plot(np.real(filt_data).T, label="FILT")
|
|
||||||
axs[1][0].plot(np.imag(filt_data).T, label="FILT")
|
|
||||||
filt_rel = filt_data[1] / filt_data[0]
|
|
||||||
axs[0][1].plot(np.real(filt_rel).T, label="FILT")
|
|
||||||
axs[1][1].plot(np.imag(filt_rel).T, label="FILT")
|
|
||||||
s = np.mean(filt_rel)
|
|
||||||
axs[0][1].axhline(np.real(s), color="k")
|
|
||||||
axs[1][1].axhline(np.imag(s), color="k")
|
|
||||||
|
|
||||||
axs[0][0].grid(True)
|
|
||||||
axs[1][0].grid(True)
|
|
||||||
axs[0][1].grid(True)
|
|
||||||
axs[1][1].grid(True)
|
|
||||||
|
|
||||||
axs[0][0].legend(loc="lower right")
|
|
||||||
axs[1][0].legend(loc="lower right")
|
|
||||||
axs[0][1].legend(loc="lower right")
|
|
||||||
axs[1][1].legend(loc="lower right")
|
|
||||||
|
|
||||||
axs[0][0].set_ylabel("Real")
|
|
||||||
axs[1][0].set_ylabel("Imag")
|
|
||||||
axs[0][0].set_title("By Channel")
|
|
||||||
axs[0][1].set_title("Relative")
|
|
||||||
|
|
||||||
# %% Plot in time
|
# %% Plot in time
|
||||||
fig, axs = plt.subplots(2, 1, sharex=True, tight_layout=True)
|
fig, axs = plt.subplots(2, 1, sharex=True, tight_layout=True)
|
||||||
@ -221,35 +269,16 @@ fig.show()
|
|||||||
|
|
||||||
# %% Plot in frequency
|
# %% Plot in frequency
|
||||||
f = np.fft.fftfreq(data.shape[-1], 1 / sdr.sdr.sample_rate)
|
f = np.fft.fftfreq(data.shape[-1], 1 / sdr.sdr.sample_rate)
|
||||||
RX_BITS = 10
|
RX_BITS = 12 # for each of i, q (including sign bit)
|
||||||
Pxx_den = np.fft.fft(data, axis=-1) / (len(data) * 2 ** (2 * RX_BITS))
|
fft_data = np.fft.fft(data, axis=-1, norm="forward") / (2 ** (RX_BITS - 1))
|
||||||
Pxx_den_ddc = np.fft.fft(ddc_data, axis=-1) / (len(ddc_data) * 2 ** (2 * RX_BITS))
|
|
||||||
Pxx_den_filt = np.fft.fft(filt_data, axis=-1) / (len(filt_data) * 2 ** (2 * RX_BITS))
|
|
||||||
fft_ddc_tone = np.fft.fft(ddc_tone, axis=-1) / (len(ddc_tone))
|
|
||||||
plt.figure()
|
plt.figure()
|
||||||
for cc, chan in enumerate(sdr.sdr.rx_enabled_channels):
|
for cc, chan in enumerate(sdr.sdr.rx_enabled_channels):
|
||||||
# plt.plot(
|
|
||||||
# np.fft.fftshift(f),
|
|
||||||
# db20(np.fft.fftshift(Pxx_den[cc])),
|
|
||||||
# label=f"Channel {chan}",
|
|
||||||
# )
|
|
||||||
plt.plot(
|
plt.plot(
|
||||||
np.fft.fftshift(f),
|
np.fft.fftshift(f),
|
||||||
db20(np.fft.fftshift(Pxx_den_ddc[cc])),
|
db20(np.fft.fftshift(fft_data[cc])),
|
||||||
label=f"Channel {chan}",
|
label=f"Channel {chan}",
|
||||||
)
|
)
|
||||||
plt.plot(
|
|
||||||
np.fft.fftshift(f),
|
|
||||||
db20(np.fft.fftshift(Pxx_den_filt[cc])),
|
|
||||||
label=f"Channel {chan}",
|
|
||||||
)
|
|
||||||
# plt.plot(
|
|
||||||
# np.fft.fftshift(f),
|
|
||||||
# db20(np.fft.fftshift(fft_ddc_tone)),
|
|
||||||
# label="DDC Tone",
|
|
||||||
# )
|
|
||||||
plt.legend()
|
plt.legend()
|
||||||
# plt.ylim(1e-7, 1e2)
|
|
||||||
plt.ylim(-100, 0)
|
plt.ylim(-100, 0)
|
||||||
plt.xlabel("Frequency [Hz]")
|
plt.xlabel("Frequency [Hz]")
|
||||||
plt.ylabel("Power [dBfs]")
|
plt.ylabel("Power [dBfs]")
|
||||||
@ -260,31 +289,7 @@ plt.show()
|
|||||||
|
|
||||||
|
|
||||||
# %%
|
# %%
|
||||||
def vna_capture(frequency: npt.ArrayLike):
|
s = sdr.vna_capture(frequency=np.linspace(70e6, 200e6, 101))
|
||||||
s = xr.DataArray(
|
|
||||||
np.empty(len(frequency), dtype=np.complex128),
|
|
||||||
dims=["frequency"],
|
|
||||||
coords=dict(
|
|
||||||
frequency=frequency,
|
|
||||||
),
|
|
||||||
)
|
|
||||||
for freq in s.frequency.data:
|
|
||||||
set_output(frequency=freq, power=-5)
|
|
||||||
sdr.rx_destroy_buffer()
|
|
||||||
sdr.rx_lo = int(freq)
|
|
||||||
sdr.rx_enabled_channels = [0, 1]
|
|
||||||
sdr.gain_control_mode_chan0 = "manual"
|
|
||||||
sdr.gain_control_mode_chan1 = "manual"
|
|
||||||
sdr.rx_hardwaregain_chan0 = 40
|
|
||||||
sdr.rx_hardwaregain_chan1 = 40
|
|
||||||
rx = sdr.rx()
|
|
||||||
s.loc[dict(frequency=freq)] = np.mean(rx[1] / rx[0])
|
|
||||||
|
|
||||||
return s
|
|
||||||
|
|
||||||
|
|
||||||
# %%
|
|
||||||
s = vna_capture(frequency=np.linspace(70e6, 200e6, 101))
|
|
||||||
|
|
||||||
# %% Plot Logmag
|
# %% Plot Logmag
|
||||||
fig, axs = plt.subplots(2, 1, sharex=True, tight_layout=True)
|
fig, axs = plt.subplots(2, 1, sharex=True, tight_layout=True)
|
||||||
@ -322,11 +327,11 @@ plt.show()
|
|||||||
cal_frequency = np.linspace(70e6, 600e6, 101)
|
cal_frequency = np.linspace(70e6, 600e6, 101)
|
||||||
ideal_cal_frequency = rf.Frequency(np.min(cal_frequency), np.max(cal_frequency), len(cal_frequency))
|
ideal_cal_frequency = rf.Frequency(np.min(cal_frequency), np.max(cal_frequency), len(cal_frequency))
|
||||||
input("Connect SHORT and press ENTER...")
|
input("Connect SHORT and press ENTER...")
|
||||||
short = vna_capture(frequency=cal_frequency)
|
short = sdr.vna_capture(frequency=cal_frequency)
|
||||||
input("Connect OPEN and press ENTER...")
|
input("Connect OPEN and press ENTER...")
|
||||||
open = vna_capture(frequency=cal_frequency)
|
open = sdr.vna_capture(frequency=cal_frequency)
|
||||||
input("Connect LOAD and press ENTER...")
|
input("Connect LOAD and press ENTER...")
|
||||||
load = vna_capture(frequency=cal_frequency)
|
load = sdr.vna_capture(frequency=cal_frequency)
|
||||||
|
|
||||||
short_net = s2net(short)
|
short_net = s2net(short)
|
||||||
open_net = s2net(open)
|
open_net = s2net(open)
|
||||||
@ -340,7 +345,7 @@ calibration = rf.calibration.OnePort(
|
|||||||
|
|
||||||
|
|
||||||
# %%
|
# %%
|
||||||
s = vna_capture(frequency=cal_frequency)
|
s = sdr.vna_capture(frequency=cal_frequency)
|
||||||
|
|
||||||
# %%
|
# %%
|
||||||
s_calibrated = calibration.apply_cal(s2net(s))
|
s_calibrated = calibration.apply_cal(s2net(s))
|
||||||
|
Reference in New Issue
Block a user