I'm not sure why exp creates such a bad tone

charon_vna/vna.py

@@ -1,7 +1,7 @@
 # %% imports
 import copy
 from pathlib import Path
-from typing import Any, Dict, Optional
+from typing import Any, Dict, Optional, Tuple
 
 import adi
 import iio
@@ -19,16 +19,22 @@ dir_ = Path(__file__).parent
 
 # %% connection
 class Charon:
+    FREQUENCY_OFFSET = 1e6
+
     def __init__(
         self,
         uri: str,
         frequency: npt.ArrayLike = np.linspace(1e9, 2e9, 3),
+        ports: Tuple[int] = (1,),
     ):
+        self.ports = ports
+        self.frequency = frequency
+
         # everything RF
         self.sdr = adi.ad9361(uri=uri)
         for attr, expected in [
             ("adi,2rx-2tx-mode-enable", True),
-            ("adi,gpo-manual-mode-enable", True),
+            # ("adi,gpo-manual-mode-enable", True),
         ]:
             # available configuration options:
             # https://wiki.analog.com/resources/tools-software/linux-drivers/iio-transceiver/ad9361-customization
@@ -40,8 +46,8 @@ class Charon:
                 # TODO: it might be possible to change this on the fly.
                 # I think we'll actually just fail in __init__ of ad9361 if 2rx-2tx is wrong
 
-        self.sdr.rx_lo = int(frequency[0])
+        self.sdr.rx_lo = int(self.frequency[0])
-        self.sdr.tx_lo = int(frequency[0])
+        self.sdr.tx_lo = int(self.frequency[0])
         self.sdr.sample_rate = 30e6
         self.sdr.rx_rf_bandwidth = int(4e6)
         self.sdr.tx_rf_bandwidth = int(4e6)
@@ -63,7 +69,7 @@ class Charon:
         # 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
         self.ctrl.reg_write(0x26, 0x90)  # bit 7: AuxDAC Manual, bit 4: GPO Manual
-        self._set_gpo(0)
+        self._set_gpo(self.ports[0] - 1)
         # TODO: init AuxDAC
 
     def get_config(self) -> Dict[str, Any]:
@@ -102,60 +108,155 @@ class Charon:
             # tx_gain = pout.coords["tx_gain"][tx_gain_idx]
         self.sdr.tx_hardwaregain_chan0 = float(tx_gain)
 
-    def set_output(self, frequency: float, power: float, offset_frequency: float = 1e6):
+    def generate_tone(self, f: float, N: int = 1024, fs: Optional[float] = None):
-        # TODO: switch to DDS in Pluto
+        if fs is None:
-        def generate_tone(f: float, N: int = 1024, fs: Optional[float] = None):
+            fs = self.sdr.sample_rate
-            if fs is None:
+        fs = int(fs)
-                fs = sdr.sample_rate
+        fc = int(f / (fs / N)) * (fs / N)
-            fs = int(fs)
+        ts = 1 / float(fs)
-            fc = int(f / (fs / N)) * (fs / N)
+        t = np.arange(0, N * ts, ts)
-            ts = 1 / float(fs)
+        i = np.cos(2 * np.pi * t * fc) * 2**14
-            t = np.arange(0, N * ts, ts)
+        q = np.sin(2 * np.pi * t * fc) * 2**14
-            i = np.cos(2 * np.pi * t * fc) * 2**14
+        iq = i + 1j * q
-            q = np.sin(2 * np.pi * t * fc) * 2**14
-            iq = i + 1j * q
 
-            return iq
+        return iq
+
+    def set_output(self, frequency: float, power: float):
+        # TODO: switch to DDS in Pluto
 
         self.sdr.tx_destroy_buffer()
         self.set_output_power(power)
-        self.sdr.tx_lo = int(frequency - offset_frequency)
+        self.sdr.tx_lo = int(frequency - self.FREQUENCY_OFFSET)
-        offset_frequency = frequency - self.sdr.tx_lo
         self.sdr.tx_cyclic_buffer = True
-        self.sdr.tx(generate_tone(offset_frequency))
+        self.sdr.tx(self.generate_tone(self.FREQUENCY_OFFSET))
 
-    def _rx(self) -> npt.ArrayLike:
+    def _rx(self, count: int = 1) -> npt.ArrayLike:
+        if count < 1:
+            raise ValueError
         self.sdr.rx_destroy_buffer()
-        return np.array(self.sdr.rx())
+        return np.concatenate([np.array(self.sdr.rx()) for _ in range(count)], axis=-1)
 
 
 # %%
 sdr = Charon("ip:192.168.3.1")
 
-
 # %% initialization
 config = sdr.get_config()
 config
 
+# %% generate tone
+sdr.set_output(frequency=1e9 + sdr.FREQUENCY_OFFSET, power=-5)
+
+# %% capture data
+data = sdr._rx(1)
+
+# %%
+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,
+)
+b, a = signal.butter(n, wn, "lowpass", analog=False, output="ba", fs=sdr.sdr.sample_rate)
+filt_data = signal.lfilter(b, a, 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
-data = sdr._rx()
 fig, axs = plt.subplots(2, 1, sharex=True, tight_layout=True)
 axs[0].plot(np.real(data).T)
 axs[1].plot(np.imag(data).T)
 axs[0].set_ylabel("Real")
 axs[1].set_ylabel("Imag")
-axs[-1].set_xlabel("Time")
+axs[0].grid(True)
+axs[1].grid(True)
+axs[-1].set_xlabel("Sample")
+axs[-1].set_xlim(0, data.shape[-1])
+fig.show()
+
+# %%
+fig, ax = plt.subplots(1, 1, tight_layout=True)
+ax.plot(np.real(data).T, np.imag(data).T)
+ax.grid(True)
+ax.set_aspect("equal")
+ax.set_xlabel("Real")
+ax.set_ylabel("Imag")
+ax.set_xlim(np.array([-1, 1]) * (2 ** (12 - 1) - 1))
+ax.set_ylim(ax.get_xlim())
 fig.show()
 
 # %% Plot in frequency
-f, Pxx_den = signal.periodogram(data, sdr.sample_rate, axis=-1, return_onesided=False)
+f, Pxx_den = signal.periodogram(data, sdr.sdr.sample_rate, axis=-1, return_onesided=False)
+f_ddc, Pxx_den_ddc = signal.periodogram(ddc_data, sdr.sdr.sample_rate, axis=-1, return_onesided=False)
+f_filt, Pxx_den_filt = signal.periodogram(filt_data, sdr.sdr.sample_rate, axis=-1, return_onesided=False)
+f = np.fft.fftfreq(data.shape[-1], 1 / sdr.sdr.sample_rate)
+Pxx_den = np.fft.fft(data, axis=-1)
+Pxx_den_ddc = np.fft.fft(ddc_data, axis=-1)
+Pxx_den_filt = np.fft.fft(filt_data, axis=-1)
+fft_ddc_tone = np.fft.fft(ddc_tone, axis=-1)
 plt.figure()
-for cc, chan in enumerate(sdr.rx_enabled_channels):
+for cc, chan in enumerate(sdr.sdr.rx_enabled_channels):
-    plt.semilogy(f, Pxx_den[cc], label=f"Channel {chan}")
+    plt.plot(
+        np.fft.fftshift(f),
+        db20(np.fft.fftshift(Pxx_den[cc])),
+        label=f"Channel {chan}",
+    )
+    plt.plot(
+        np.fft.fftshift(f),
+        db20(np.fft.fftshift(Pxx_den_ddc[cc])),
+        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.ylim([1e-7, 1e2])
+# plt.ylim(1e-7, 1e2)
-plt.xlabel("frequency [Hz]")
+plt.ylim(0)
-plt.ylabel("PSD [V**2/Hz]")
+plt.xlabel("Frequency [Hz]")
+plt.ylabel("PSD [$V^2/Hz$]")
+plt.title(f"Fc = {sdr.sdr.rx_lo / 1e9} GHz")
+plt.gca().xaxis.set_major_formatter(EngFormatter())
 plt.grid(True)
 plt.show()