ampl parser

README.md

@@ -109,13 +109,21 @@ Legacy binary:
 .venv/bin/python -m rfg_adc_plotter.main /dev/ttyACM0 --bin
 ```
 
+`--bin` понимает mixed 8-байтный поток:
+- `0x000A,step,ch1_i16,ch2_i16` для CH1/CH2 из `kamil_adc` (`tty:/tmp/ttyADC_data`)
+- `0x001A,step,data_i16,0x0000` для логарифмического детектора
+
+Для `0x000A` сырая кривая строится как `ch1^2 + ch2^2`, а FFT рассчитывается от комплексного сигнала `ch1 + i*ch2`.
+Для `0x001A` signed `data_i16` сначала переводится в В, затем raw отображается как `V`, а FFT рассчитывается от `exp(V)`.
+Параметр `--tty-range-v` применяется к обоим типам `--bin`-данных.
+
 Logscale binary с парой `int32`:
 
 ```bash
 .venv/bin/python -m rfg_adc_plotter.main /dev/ttyACM0 --logscale
 ```
 
-Logscale binary `16-bit x2`:
+Complex binary `16-bit x2`:
 
 ```bash
 .venv/bin/python -m rfg_adc_plotter.main /dev/ttyACM0 --parser_16_bit_x2
@@ -127,6 +135,12 @@ Logscale binary `16-bit x2`:
 .venv/bin/python -m rfg_adc_plotter.main /dev/ttyACM0 --parser_test
 ```
 
+Комплексный ASCII-поток `step real imag`:
+
+```bash
+.venv/bin/python -m rfg_adc_plotter.main /dev/ttyACM0 --parser_complex_ascii
+```
+
 ## Локальная проверка через replay_pty
 
 Если есть лог-файл захвата, его можно воспроизвести как виртуальный последовательный порт.
@@ -164,6 +178,12 @@ ssh 192.148.0.148 'ls -l /dev/ttyACM0'
 
 Если на удаленной машине есть доступ к потоку, удобнее сохранять его в файл и уже этот файл гонять локально через `replay_pty.py`.
 
+Для локального `tty`-потока из `kamil_adc` используйте:
+
+```bash
+.venv/bin/python -m rfg_adc_plotter.main /tmp/ttyADC_data --bin
+```
+
 ## Проверка и тесты
 
 Синтаксическая проверка:

rfg_adc_plotter/cli.py

@@ -55,6 +55,11 @@ def build_parser() -> argparse.ArgumentParser:
         default="pg",
         help="Совместимый флаг. Поддерживаются только auto и pg; mpl удален.",
     )
+    parser.add_argument(
+        "--opengl",
+        action="store_true",
+        help="Включить OpenGL-ускорение для PyQtGraph. По умолчанию используется CPU-отрисовка.",
+    )
     parser.add_argument(
         "--norm-type",
         choices=["projector", "simple"],
@@ -66,14 +71,26 @@ def build_parser() -> argparse.ArgumentParser:
         dest="bin_mode",
         action="store_true",
         help=(
-            "Бинарный протокол: старт свипа 0xFFFF,0xFFFF,0xFFFF,(CH<<8)|0x0A; "
-            "точки step,uint32(hi16,lo16),0x000A"
+            "8-байтный бинарный протокол: либо legacy старт "
+            "0xFFFF,0xFFFF,0xFFFF,(CH<<8)|0x0A и точки step,uint32(hi16,lo16),0x000A, "
+            "либо mixed поток 0x000A,step,ch1_i16,ch2_i16 и 0x001A,step,data_i16,0x0000. "
+            "Для 0x000A: после парсинга int16 переводятся в В, "
+            "сырая кривая = ch1^2+ch2^2 (В^2), FFT вход = ch1+i*ch2 (В). "
+            "Для 0x001A: code_i16 переводится в В, raw = V, FFT вход = exp(V)"
+        ),
+    )
+    parser.add_argument(
+        "--tty-range-v",
+        type=float,
+        default=5.0,
+        help=(
+            "Полный диапазон для пересчета tty int16 в напряжение ±V "
+            "(для --bin 0x000A CH1/CH2 и 0x001A log-detector, по умолчанию 5.0)"
         ),
     )
     parser.add_argument(
         "--logscale",
         action="store_true",
-        default=True,
         help=(
             "Новый бинарный протокол: точка несет пару int32 (avg_1, avg_2), "
             "а свип считается как |10**(avg_1*0.001) - 10**(avg_2*0.001)|"
@@ -83,18 +100,26 @@ def build_parser() -> argparse.ArgumentParser:
         "--parser_16_bit_x2",
         action="store_true",
         help=(
-            "Бинарный logscale-протокол c парой int16 (avg_1, avg_2): "
-            "старт 0xFFFF,0xFFFF,0xFFFF,(CH<<8)|0x0A; точка step,avg1_lo16,avg2_lo16,0xFFFF"
+            "Бинарный complex-протокол c парой int16 (Re, Im): "
+            "старт 0xFFFF,0xFFFF,0xFFFF,(CH<<8)|0x0A; точка step,re_lo16,im_lo16,0xFFFF"
         ),
     )
     parser.add_argument(
         "--parser_test",
         action="store_true",
         help=(
-            "Тестовый парсер для формата 16-bit x2: "
+            "Тестовый парсер для complex-формата 16-bit x2: "
             "одиночный 0xFFFF завершает точку, серия 0xFFFF начинает новый свип"
         ),
     )
+    parser.add_argument(
+        "--parser_complex_ascii",
+        action="store_true",
+        help=(
+            "ASCII-поток из трех чисел на строку: step real imag. "
+            "Новый свип определяется по сбросу/повтору step, FFT строится по комплексным данным"
+        ),
+    )
     parser.add_argument(
         "--calibrate",
         action="store_true",

rfg_adc_plotter/constants.py

@@ -1,11 +1,11 @@
 """Shared constants for sweep parsing and visualization."""
 
 WF_WIDTH = 1000
-FFT_LEN = 1024
+FFT_LEN = 2048
 BACKGROUND_MEDIAN_SWEEPS = 64
 
 SWEEP_FREQ_MIN_GHZ = 3.3
-SWEEP_FREQ_MAX_GHZ = 14.3
+SWEEP_FREQ_MAX_GHZ = 6.3
 
 LOG_BASE = 10.0
 LOG_SCALER = 0.001

rfg_adc_plotter/io/sweep_parser_core.py

@@ -10,7 +10,15 @@ from typing import List, Optional, Sequence, Set
 import numpy as np
 
 from rfg_adc_plotter.constants import DATA_INVERSION_THRESHOLD, LOG_BASE, LOG_EXP_LIMIT, LOG_POSTSCALER, LOG_SCALER
-from rfg_adc_plotter.types import ParserEvent, PointEvent, StartEvent, SweepAuxCurves, SweepInfo, SweepPacket
+from rfg_adc_plotter.types import (
+    ParserEvent,
+    PointEvent,
+    SignalKind,
+    StartEvent,
+    SweepAuxCurves,
+    SweepInfo,
+    SweepPacket,
+)
 
 
 def u32_to_i32(value: int) -> int:
@@ -32,6 +40,13 @@ def log_pair_to_sweep(avg_1: int, avg_2: int) -> float:
     return abs(value_1 - value_2) * LOG_POSTSCALER
 
 
+def tty_ch_pair_to_sweep(ch_1: int, ch_2: int) -> float:
+    """Reduce a raw CH1/CH2 TTY point to power-like scalar ``ch1^2 + ch2^2``."""
+    ch_1_i = int(ch_1)
+    ch_2_i = int(ch_2)
+    return float((ch_1_i * ch_1_i) + (ch_2_i * ch_2_i))
+
+
 class AsciiSweepParser:
     """Incremental parser for ASCII sweep streams."""
 
@@ -82,16 +97,144 @@ class AsciiSweepParser:
         return events
 
 
-class LegacyBinaryParser:
-    """Byte-resynchronizing parser for legacy 8-byte binary records."""
+class ComplexAsciiSweepParser:
+    """Incremental parser for ASCII ``step real imag`` streams."""
 
     def __init__(self):
         self._buf = bytearray()
+        self._last_step: Optional[int] = None
+        self._seen_points = False
+
+    def feed(self, data: bytes) -> List[ParserEvent]:
+        if data:
+            self._buf += data
+        events: List[ParserEvent] = []
+        while True:
+            nl = self._buf.find(b"\n")
+            if nl == -1:
+                break
+            line = bytes(self._buf[:nl])
+            del self._buf[: nl + 1]
+            if line.endswith(b"\r"):
+                line = line[:-1]
+            if not line:
+                continue
+
+            if line.lower().startswith(b"sweep_start"):
+                self._last_step = None
+                self._seen_points = False
+                events.append(StartEvent())
+                continue
+
+            parts = line.split()
+            if len(parts) < 3:
+                continue
+            try:
+                step = int(parts[0], 10)
+                real = float(parts[1])
+                imag = float(parts[2])
+            except Exception:
+                continue
+            if step < 0 or (not math.isfinite(real)) or (not math.isfinite(imag)):
+                continue
+
+            if self._seen_points and self._last_step is not None and step <= self._last_step:
+                events.append(StartEvent())
+            self._seen_points = True
+            self._last_step = step
+            events.append(
+                PointEvent(
+                    ch=0,
+                    x=step,
+                    y=float(abs(complex(real, imag))),
+                    aux=(float(real), float(imag)),
+                )
+            )
+        return events
+
+
+class LegacyBinaryParser:
+    """Byte-resynchronizing parser for supported 8-byte binary record formats."""
+
+    def __init__(self):
+        self._buf = bytearray()
+        self._last_step: Optional[int] = None
+        self._seen_points = False
+        self._mode: Optional[str] = None
+        self._current_signal_kind: Optional[SignalKind] = None
 
     @staticmethod
     def _u16_at(buf: bytearray, offset: int) -> int:
         return int(buf[offset]) | (int(buf[offset + 1]) << 8)
 
+    def _emit_legacy_start(self, events: List[ParserEvent], ch: int) -> None:
+        self._mode = "legacy"
+        self._last_step = None
+        self._seen_points = False
+        self._current_signal_kind = None
+        events.append(StartEvent(ch=int(ch)))
+
+    def _emit_bin_start(self, events: List[ParserEvent], signal_kind: SignalKind) -> None:
+        self._mode = "bin"
+        self._last_step = None
+        self._seen_points = False
+        self._current_signal_kind = signal_kind
+        events.append(StartEvent(ch=0, signal_kind=signal_kind))
+
+    def _emit_tty_start(self, events: List[ParserEvent]) -> None:
+        self._emit_bin_start(events, signal_kind="bin_iq")
+
+    def _emit_legacy_point(self, events: List[ParserEvent], step: int, value_word_hi: int, value_word_lo: int, ch: int) -> None:
+        self._mode = "legacy"
+        self._current_signal_kind = None
+        if self._seen_points and self._last_step is not None and step <= self._last_step:
+            events.append(StartEvent(ch=int(ch)))
+        self._seen_points = True
+        self._last_step = int(step)
+        value = u32_to_i32((int(value_word_hi) << 16) | int(value_word_lo))
+        events.append(PointEvent(ch=int(ch), x=int(step), y=float(value)))
+
+    def _prepare_bin_point(self, events: List[ParserEvent], step: int, signal_kind: SignalKind) -> None:
+        self._mode = "bin"
+        if self._current_signal_kind != signal_kind:
+            if self._seen_points:
+                events.append(StartEvent(ch=0, signal_kind=signal_kind))
+            self._last_step = None
+            self._seen_points = False
+            self._current_signal_kind = signal_kind
+        if self._seen_points and self._last_step is not None and step <= self._last_step:
+            events.append(StartEvent(ch=0, signal_kind=signal_kind))
+            self._last_step = None
+            self._seen_points = False
+        self._seen_points = True
+        self._last_step = int(step)
+
+    def _emit_tty_point(self, events: List[ParserEvent], step: int, ch_1_word: int, ch_2_word: int) -> None:
+        self._prepare_bin_point(events, step=int(step), signal_kind="bin_iq")
+        ch_1 = u16_to_i16(int(ch_1_word))
+        ch_2 = u16_to_i16(int(ch_2_word))
+        events.append(
+            PointEvent(
+                ch=0,
+                x=int(step),
+                y=tty_ch_pair_to_sweep(ch_1, ch_2),
+                aux=(float(ch_1), float(ch_2)),
+                signal_kind="bin_iq",
+            )
+        )
+
+    def _emit_logdet_point(self, events: List[ParserEvent], step: int, value_word: int) -> None:
+        self._prepare_bin_point(events, step=int(step), signal_kind="bin_logdet")
+        value = u16_to_i16(int(value_word))
+        events.append(
+            PointEvent(
+                ch=0,
+                x=int(step),
+                y=float(value),
+                signal_kind="bin_logdet",
+            )
+        )
+
     def feed(self, data: bytes) -> List[ParserEvent]:
         if data:
             self._buf += data
@@ -100,16 +243,83 @@ class LegacyBinaryParser:
             w0 = self._u16_at(self._buf, 0)
             w1 = self._u16_at(self._buf, 2)
             w2 = self._u16_at(self._buf, 4)
-            if w0 == 0xFFFF and w1 == 0xFFFF and w2 == 0xFFFF and self._buf[6] == 0x0A:
-                events.append(StartEvent(ch=int(self._buf[7])))
+            w3 = self._u16_at(self._buf, 6)
+
+            is_legacy_start = (w0 == 0xFFFF and w1 == 0xFFFF and w2 == 0xFFFF and self._buf[6] == 0x0A)
+            is_tty_start = (w0 == 0x000A and w1 == 0xFFFF and w2 == 0xFFFF and w3 == 0xFFFF)
+            is_legacy_point = (self._buf[6] == 0x0A and w0 != 0xFFFF)
+            is_tty_point = (w0 == 0x000A and w1 != 0xFFFF)
+            is_logdet_point = (w0 == 0x001A and w3 == 0x0000)
+
+            if is_legacy_start:
+                self._emit_legacy_start(events, ch=int(self._buf[7]))
                 del self._buf[:8]
                 continue
-            if self._buf[6] == 0x0A:
-                ch = int(self._buf[7])
-                value = u32_to_i32((w1 << 16) | w2)
-                events.append(PointEvent(ch=ch, x=int(w0), y=float(value)))
+
+            if is_tty_start:
+                self._emit_tty_start(events)
                 del self._buf[:8]
                 continue
+
+            if is_logdet_point:
+                self._emit_logdet_point(events, step=int(w1), value_word=int(w2))
+                del self._buf[:8]
+                continue
+
+            if self._mode == "legacy":
+                if is_legacy_point:
+                    self._emit_legacy_point(
+                        events,
+                        step=int(w0),
+                        value_word_hi=int(w1),
+                        value_word_lo=int(w2),
+                        ch=int(self._buf[7]),
+                    )
+                    del self._buf[:8]
+                    continue
+                if is_tty_point and (not is_legacy_point):
+                    self._emit_tty_point(events, step=int(w1), ch_1_word=int(w2), ch_2_word=int(w3))
+                    del self._buf[:8]
+                    continue
+                del self._buf[:1]
+                continue
+
+            if self._mode == "bin":
+                if is_tty_point:
+                    self._emit_tty_point(events, step=int(w1), ch_1_word=int(w2), ch_2_word=int(w3))
+                    del self._buf[:8]
+                    continue
+                if is_legacy_point and (not is_tty_point):
+                    self._emit_legacy_point(
+                        events,
+                        step=int(w0),
+                        value_word_hi=int(w1),
+                        value_word_lo=int(w2),
+                        ch=int(self._buf[7]),
+                    )
+                    del self._buf[:8]
+                    continue
+                del self._buf[:1]
+                continue
+
+            # Mode is still unknown. Accept only unambiguous point shapes to avoid
+            # jumping between tty and legacy interpretations on coincidental bytes.
+            if is_tty_point and (not is_legacy_point):
+                self._emit_tty_point(events, step=int(w1), ch_1_word=int(w2), ch_2_word=int(w3))
+                del self._buf[:8]
+                continue
+
+            if is_legacy_point and (not is_tty_point):
+                self._emit_legacy_point(
+                    events,
+                    step=int(w0),
+                    value_word_hi=int(w1),
+                    value_word_lo=int(w2),
+                    ch=int(self._buf[7]),
+                )
+                del self._buf[:8]
+                continue
+
             del self._buf[:1]
         return events
 
@@ -119,6 +329,8 @@ class LogScaleBinaryParser32:
 
     def __init__(self):
         self._buf = bytearray()
+        self._last_step: Optional[int] = None
+        self._seen_points = False
 
     @staticmethod
     def _u16_at(buf: bytearray, offset: int) -> int:
@@ -131,11 +343,17 @@ class LogScaleBinaryParser32:
         while len(self._buf) >= 12:
             words = [self._u16_at(self._buf, idx * 2) for idx in range(6)]
             if words[0:5] == [0xFFFF] * 5 and (words[5] & 0x00FF) == 0x000A:
+                self._last_step = None
+                self._seen_points = False
                 events.append(StartEvent(ch=int((words[5] >> 8) & 0x00FF)))
                 del self._buf[:12]
                 continue
             if (words[5] & 0x00FF) == 0x000A and words[0] != 0xFFFF:
                 ch = int((words[5] >> 8) & 0x00FF)
+                if self._seen_points and self._last_step is not None and words[0] <= self._last_step:
+                    events.append(StartEvent(ch=ch))
+                self._seen_points = True
+                self._last_step = int(words[0])
                 avg_1 = u32_to_i32((words[1] << 16) | words[2])
                 avg_2 = u32_to_i32((words[3] << 16) | words[4])
                 events.append(
@@ -158,6 +376,8 @@ class LogScale16BitX2BinaryParser:
     def __init__(self):
         self._buf = bytearray()
         self._current_channel = 0
+        self._last_step: Optional[int] = None
+        self._seen_points = False
 
     @staticmethod
     def _u16_at(buf: bytearray, offset: int) -> int:
@@ -171,18 +391,24 @@ class LogScale16BitX2BinaryParser:
             words = [self._u16_at(self._buf, idx * 2) for idx in range(4)]
             if words[0:3] == [0xFFFF, 0xFFFF, 0xFFFF] and (words[3] & 0x00FF) == 0x000A:
                 self._current_channel = int((words[3] >> 8) & 0x00FF)
+                self._last_step = None
+                self._seen_points = False
                 events.append(StartEvent(ch=self._current_channel))
                 del self._buf[:8]
                 continue
             if words[3] == 0xFFFF and words[0] != 0xFFFF:
-                avg_1 = u16_to_i16(words[1])
-                avg_2 = u16_to_i16(words[2])
+                if self._seen_points and self._last_step is not None and words[0] <= self._last_step:
+                    events.append(StartEvent(ch=self._current_channel))
+                self._seen_points = True
+                self._last_step = int(words[0])
+                real = u16_to_i16(words[1])
+                imag = u16_to_i16(words[2])
                 events.append(
                     PointEvent(
                         ch=self._current_channel,
                         x=int(words[0]),
-                        y=log_pair_to_sweep(avg_1, avg_2),
-                        aux=(float(avg_1), float(avg_2)),
+                        y=float(abs(complex(real, imag))),
+                        aux=(float(real), float(imag)),
                     )
                 )
                 del self._buf[:8]
@@ -212,14 +438,14 @@ class ParserTestStreamParser:
             return None
         if self._expected_step is not None and step < self._expected_step:
             return None
-        avg_1 = u16_to_i16(int(self._point_buf[1]))
-        avg_2 = u16_to_i16(int(self._point_buf[2]))
+        real = u16_to_i16(int(self._point_buf[1]))
+        imag = u16_to_i16(int(self._point_buf[2]))
         self._expected_step = step + 1
         return PointEvent(
             ch=self._current_channel,
             x=step,
-            y=log_pair_to_sweep(avg_1, avg_2),
-            aux=(float(avg_1), float(avg_2)),
+            y=float(abs(complex(real, imag))),
+            aux=(float(real), float(imag)),
         )
 
     def feed(self, data: bytes) -> List[ParserEvent]:
@@ -299,6 +525,7 @@ class SweepAssembler:
         self._aux_1: list[float] = []
         self._aux_2: list[float] = []
         self._cur_channel: Optional[int] = None
+        self._cur_signal_kind: Optional[SignalKind] = None
         self._cur_channels: set[int] = set()
 
     def _reset_current(self) -> None:
@@ -307,6 +534,7 @@ class SweepAssembler:
         self._aux_1.clear()
         self._aux_2.clear()
         self._cur_channel = None
+        self._cur_signal_kind = None
         self._cur_channels.clear()
 
     def _scatter(self, xs: Sequence[int], values: Sequence[float], width: int) -> np.ndarray:
@@ -345,18 +573,35 @@ class SweepAssembler:
             self._reset_current()
             if event.ch is not None:
                 self._cur_channel = int(event.ch)
-                self._cur_channels.add(int(event.ch))
+            self._cur_signal_kind = event.signal_kind
             return packet
 
+        point_ch = int(event.ch)
+        point_signal_kind = event.signal_kind
+        packet: Optional[SweepPacket] = None
         if self._cur_channel is None:
-            self._cur_channel = int(event.ch)
-        self._cur_channels.add(int(event.ch))
+            self._cur_channel = point_ch
+        elif point_ch != self._cur_channel:
+            if self._xs:
+                # Never mix channels in a single sweep packet: otherwise
+                # identical step indexes can overwrite each other.
+                packet = self.finalize_current()
+                self._reset_current()
+            self._cur_channel = point_ch
+        if self._cur_signal_kind != point_signal_kind:
+            if self._xs:
+                packet = self.finalize_current()
+                self._reset_current()
+                self._cur_channel = point_ch
+            self._cur_signal_kind = point_signal_kind
+
+        self._cur_channels.add(point_ch)
         self._xs.append(int(event.x))
         self._ys.append(float(event.y))
         if event.aux is not None:
             self._aux_1.append(float(event.aux[0]))
             self._aux_2.append(float(event.aux[1]))
-        return None
+        return packet
 
     def finalize_current(self) -> Optional[SweepPacket]:
         if not self._xs:
@@ -417,6 +662,7 @@ class SweepAssembler:
             "sweep": self._sweep_idx,
             "ch": ch_primary,
             "chs": ch_list,
+            "signal_kind": self._cur_signal_kind,
             "n_valid": n_valid,
             "min": vmin,
             "max": vmax,

rfg_adc_plotter/io/sweep_reader.py

@@ -10,13 +10,96 @@ from queue import Full, Queue
 from rfg_adc_plotter.io.serial_source import SerialChunkReader, SerialLineSource
 from rfg_adc_plotter.io.sweep_parser_core import (
     AsciiSweepParser,
+    ComplexAsciiSweepParser,
     LegacyBinaryParser,
     LogScale16BitX2BinaryParser,
     LogScaleBinaryParser32,
     ParserTestStreamParser,
     SweepAssembler,
 )
-from rfg_adc_plotter.types import SweepPacket
+from rfg_adc_plotter.types import ParserEvent, PointEvent, StartEvent, SweepPacket
+
+_PARSER_16_BIT_X2_PROBE_BYTES = 64 * 1024
+_LEGACY_STREAM_MIN_RECORDS = 32
+_LEGACY_STREAM_MIN_MATCH_RATIO = 0.95
+_TTY_STREAM_MIN_MATCH_RATIO = 0.60
+_DEBUG_FRAME_LOG_EVERY = 10
+_NO_INPUT_WARN_INTERVAL_S = 5.0
+_NO_PACKET_WARN_INTERVAL_S = 5.0
+_NO_PACKET_HINT_AFTER_S = 10.0
+
+
+def _u16le_at(data: bytes, offset: int) -> int:
+    return int(data[offset]) | (int(data[offset + 1]) << 8)
+
+
+def _looks_like_legacy_8byte_stream(data: bytes) -> bool:
+    """Heuristically detect supported 8-byte binary streams on an arbitrary byte offset."""
+    buf = bytes(data)
+    for offset in range(8):
+        blocks = (len(buf) - offset) // 8
+        if blocks < _LEGACY_STREAM_MIN_RECORDS:
+            continue
+        min_matches = max(_LEGACY_STREAM_MIN_RECORDS, int(blocks * _LEGACY_STREAM_MIN_MATCH_RATIO))
+        matched_steps_legacy: list[int] = []
+        matched_steps_tty: list[int] = []
+        matched_steps_logdet: list[int] = []
+        for block_idx in range(blocks):
+            base = offset + (block_idx * 8)
+            if (_u16le_at(buf, base + 6) & 0x00FF) != 0x000A:
+                w0 = _u16le_at(buf, base)
+                w1 = _u16le_at(buf, base + 2)
+                w3 = _u16le_at(buf, base + 6)
+                if w0 == 0x000A and w1 != 0xFFFF:
+                    matched_steps_tty.append(w1)
+                elif w0 == 0x001A and w3 == 0x0000:
+                    matched_steps_logdet.append(w1)
+                continue
+            matched_steps_legacy.append(_u16le_at(buf, base))
+
+        if len(matched_steps_legacy) >= min_matches:
+            monotonic_or_reset = 0
+            for prev_step, next_step in zip(matched_steps_legacy, matched_steps_legacy[1:]):
+                if next_step == (prev_step + 1) or next_step <= prev_step:
+                    monotonic_or_reset += 1
+            if monotonic_or_reset >= max(4, len(matched_steps_legacy) - 4):
+                return True
+
+        tty_min_matches = max(_LEGACY_STREAM_MIN_RECORDS, int(blocks * _TTY_STREAM_MIN_MATCH_RATIO))
+        if len(matched_steps_tty) >= tty_min_matches:
+            monotonic_or_reset = 0
+            for prev_step, next_step in zip(matched_steps_tty, matched_steps_tty[1:]):
+                if next_step == (prev_step + 1) or next_step <= 2:
+                    monotonic_or_reset += 1
+            if monotonic_or_reset >= max(4, len(matched_steps_tty) - 4):
+                return True
+
+        if len(matched_steps_logdet) >= tty_min_matches:
+            monotonic_or_reset = 0
+            for prev_step, next_step in zip(matched_steps_logdet, matched_steps_logdet[1:]):
+                if next_step == (prev_step + 1) or next_step <= 2:
+                    monotonic_or_reset += 1
+            if monotonic_or_reset >= max(4, len(matched_steps_logdet) - 4):
+                return True
+
+    return False
+
+
+def _is_valid_parser_16_bit_x2_probe(events: list[ParserEvent]) -> bool:
+    """Accept only plausible complex streams and ignore resync noise."""
+    point_steps: list[int] = []
+    for event in events:
+        if isinstance(event, PointEvent):
+            point_steps.append(int(event.x))
+
+    if len(point_steps) < 3:
+        return False
+
+    monotonic_or_small_reset = 0
+    for prev_step, next_step in zip(point_steps, point_steps[1:]):
+        if next_step == (prev_step + 1) or next_step <= 2:
+            monotonic_or_small_reset += 1
+    return monotonic_or_small_reset >= max(2, len(point_steps) - 3)
 
 
 class SweepReader(threading.Thread):
@@ -33,20 +116,40 @@ class SweepReader(threading.Thread):
         logscale: bool = False,
         parser_16_bit_x2: bool = False,
         parser_test: bool = False,
+        parser_complex_ascii: bool = False,
     ):
         super().__init__(daemon=True)
         self._port_path = port_path
         self._baud = int(baud)
         self._queue = out_queue
-        self._stop = stop_event
+        self._stop_event = stop_event
         self._fancy = bool(fancy)
         self._bin_mode = bool(bin_mode)
         self._logscale = bool(logscale)
         self._parser_16_bit_x2 = bool(parser_16_bit_x2)
         self._parser_test = bool(parser_test)
+        self._parser_complex_ascii = bool(parser_complex_ascii)
         self._src: SerialLineSource | None = None
+        self._frames_read = 0
+        self._frames_dropped = 0
+        self._started_at = time.perf_counter()
+
+    def _resolve_parser_mode_label(self) -> str:
+        if self._parser_complex_ascii:
+            return "complex_ascii"
+        if self._parser_test:
+            return "parser_test_16x2"
+        if self._parser_16_bit_x2:
+            return "parser_16_bit_x2"
+        if self._logscale:
+            return "logscale_32"
+        if self._bin_mode:
+            return "legacy_8byte"
+        return "ascii"
 
     def _build_parser(self):
+        if self._parser_complex_ascii:
+            return ComplexAsciiSweepParser(), SweepAssembler(fancy=self._fancy, apply_inversion=False)
         if self._parser_test:
             return ParserTestStreamParser(), SweepAssembler(fancy=self._fancy, apply_inversion=False)
         if self._parser_16_bit_x2:
@@ -57,40 +160,216 @@ class SweepReader(threading.Thread):
             return LegacyBinaryParser(), SweepAssembler(fancy=self._fancy, apply_inversion=True)
         return AsciiSweepParser(), SweepAssembler(fancy=self._fancy, apply_inversion=True)
 
+    @staticmethod
+    def _consume_events(assembler: SweepAssembler, events) -> list[SweepPacket]:
+        packets: list[SweepPacket] = []
+        for event in events:
+            packet = assembler.consume(event)
+            if packet is not None:
+                packets.append(packet)
+        return packets
+
+    def _probe_parser_16_bit_x2(self, chunk_reader: SerialChunkReader):
+        parser = LogScale16BitX2BinaryParser()
+        probe_buf = bytearray()
+        probe_events: list[ParserEvent] = []
+        probe_started_at = time.perf_counter()
+
+        while not self._stop_event.is_set() and len(probe_buf) < _PARSER_16_BIT_X2_PROBE_BYTES:
+            data = chunk_reader.read_available()
+            if not data:
+                time.sleep(0.0005)
+                continue
+            probe_buf += data
+            probe_events.extend(parser.feed(data))
+            if _is_valid_parser_16_bit_x2_probe(probe_events):
+                assembler = SweepAssembler(fancy=self._fancy, apply_inversion=False)
+                probe_packets = self._consume_events(assembler, probe_events)
+                n_points = int(sum(1 for event in probe_events if isinstance(event, PointEvent)))
+                n_starts = int(sum(1 for event in probe_events if isinstance(event, StartEvent)))
+                probe_ms = (time.perf_counter() - probe_started_at) * 1000.0
+                sys.stderr.write(
+                    "[info] parser_16_bit_x2 probe: bytes:%d events:%d points:%d starts:%d parser:16x2 elapsed_ms:%.1f\n"
+                    % (
+                        len(probe_buf),
+                        len(probe_events),
+                        n_points,
+                        n_starts,
+                        probe_ms,
+                    )
+                )
+                return parser, assembler, probe_packets
+
+        probe_looks_legacy = bool(probe_buf) and _looks_like_legacy_8byte_stream(bytes(probe_buf))
+        n_points = int(sum(1 for event in probe_events if isinstance(event, PointEvent)))
+        n_starts = int(sum(1 for event in probe_events if isinstance(event, StartEvent)))
+        probe_ms = (time.perf_counter() - probe_started_at) * 1000.0
+        if probe_looks_legacy:
+            sys.stderr.write(
+                "[info] parser_16_bit_x2 probe: bytes:%d events:%d points:%d starts:%d parser:legacy(fallback) elapsed_ms:%.1f\n"
+                % (
+                    len(probe_buf),
+                    len(probe_events),
+                    n_points,
+                    n_starts,
+                    probe_ms,
+                )
+            )
+            sys.stderr.write("[info] parser_16_bit_x2: fallback -> legacy\n")
+            parser = LegacyBinaryParser()
+            assembler = SweepAssembler(fancy=self._fancy, apply_inversion=True)
+            probe_packets = self._consume_events(assembler, parser.feed(bytes(probe_buf)))
+            return parser, assembler, probe_packets
+
+        sys.stderr.write(
+            "[warn] parser_16_bit_x2 probe inconclusive: bytes:%d events:%d points:%d starts:%d parser:16x2 elapsed_ms:%.1f\n"
+            % (
+                len(probe_buf),
+                len(probe_events),
+                n_points,
+                n_starts,
+                probe_ms,
+            )
+        )
+        sys.stderr.write(
+            "[hint] parser_16_bit_x2: if source is 8-byte tty CH1/CH2 stream (0x000A,step,ch1,ch2), try --bin\n"
+        )
+        assembler = SweepAssembler(fancy=self._fancy, apply_inversion=False)
+        return parser, assembler, []
+
     def _enqueue(self, packet: SweepPacket) -> None:
+        dropped = False
         try:
             self._queue.put_nowait(packet)
         except Full:
             try:
                 _ = self._queue.get_nowait()
+                dropped = True
             except Exception:
                 pass
             try:
                 self._queue.put_nowait(packet)
             except Exception:
                 pass
+        if dropped:
+            self._frames_dropped += 1
+
+        self._frames_read += 1
+        if self._frames_read % _DEBUG_FRAME_LOG_EVERY == 0:
+            sweep, info, _aux = packet
+            try:
+                queue_size = self._queue.qsize()
+            except Exception:
+                queue_size = -1
+            elapsed_s = max(time.perf_counter() - self._started_at, 1e-9)
+            frames_per_sec = float(self._frames_read) / elapsed_s
+            sweep_idx = info.get("sweep") if isinstance(info, dict) else None
+            channel = info.get("ch") if isinstance(info, dict) else None
+            sys.stderr.write(
+                "[debug] reader frames:%d rate:%.2f/s last_sweep:%s ch:%s width:%d queue:%d dropped:%d\n"
+                % (
+                    self._frames_read,
+                    frames_per_sec,
+                    str(sweep_idx),
+                    str(channel),
+                    int(getattr(sweep, "size", 0)),
+                    int(queue_size),
+                    self._frames_dropped,
+                )
+            )
 
     def run(self) -> None:
         try:
             self._src = SerialLineSource(self._port_path, self._baud, timeout=1.0)
+            queue_cap = int(getattr(self._queue, "maxsize", -1))
             sys.stderr.write(f"[info] Открыл порт {self._port_path} ({self._src._using})\n")
+            sys.stderr.write(
+                "[info] reader start: parser:%s fancy:%d queue_max:%d source:%s\n"
+                % (
+                    self._resolve_parser_mode_label(),
+                    int(self._fancy),
+                    queue_cap,
+                    getattr(self._src, "_using", "unknown"),
+                )
+            )
         except Exception as exc:
             sys.stderr.write(f"[error] {exc}\n")
             return
 
-        parser, assembler = self._build_parser()
-
         try:
             chunk_reader = SerialChunkReader(self._src)
-            while not self._stop.is_set():
+            if self._parser_16_bit_x2:
+                parser, assembler, pending_packets = self._probe_parser_16_bit_x2(chunk_reader)
+            else:
+                parser, assembler = self._build_parser()
+                pending_packets = []
+
+            for packet in pending_packets:
+                self._enqueue(packet)
+
+            loop_started_at = time.perf_counter()
+            last_input_at = loop_started_at
+            last_packet_at = loop_started_at if self._frames_read > 0 else loop_started_at
+            last_no_input_warn_at = loop_started_at
+            last_no_packet_warn_at = loop_started_at
+            parser_hint_emitted = False
+
+            while not self._stop_event.is_set():
                 data = chunk_reader.read_available()
+                now_s = time.perf_counter()
                 if not data:
+                    input_idle_s = now_s - last_input_at
+                    if (
+                        input_idle_s >= _NO_INPUT_WARN_INTERVAL_S
+                        and (now_s - last_no_input_warn_at) >= _NO_INPUT_WARN_INTERVAL_S
+                    ):
+                        sys.stderr.write(
+                            "[warn] reader no input bytes for %.1fs on %s (parser:%s)\n"
+                            % (
+                                input_idle_s,
+                                self._port_path,
+                                self._resolve_parser_mode_label(),
+                            )
+                        )
+                        last_no_input_warn_at = now_s
+
+                    packets_idle_s = now_s - last_packet_at
+                    if (
+                        packets_idle_s >= _NO_PACKET_WARN_INTERVAL_S
+                        and (now_s - last_no_packet_warn_at) >= _NO_PACKET_WARN_INTERVAL_S
+                    ):
+                        try:
+                            queue_size = self._queue.qsize()
+                        except Exception:
+                            queue_size = -1
+                        sys.stderr.write(
+                            "[warn] reader no sweep packets for %.1fs (input_idle:%.1fs queue:%d parser:%s)\n"
+                            % (
+                                packets_idle_s,
+                                input_idle_s,
+                                int(queue_size),
+                                self._resolve_parser_mode_label(),
+                            )
+                        )
+                        last_no_packet_warn_at = now_s
+                        if (
+                            self._parser_16_bit_x2
+                            and (not parser_hint_emitted)
+                            and (now_s - self._started_at) >= _NO_PACKET_HINT_AFTER_S
+                        ):
+                            sys.stderr.write(
+                                "[hint] parser_16_bit_x2 still has no sweeps; if source is tty CH1/CH2, rerun with --bin\n"
+                            )
+                            parser_hint_emitted = True
                     time.sleep(0.0005)
                     continue
-                for event in parser.feed(data):
-                    packet = assembler.consume(event)
-                    if packet is not None:
-                        self._enqueue(packet)
+
+                last_input_at = now_s
+                packets = self._consume_events(assembler, parser.feed(data))
+                if packets:
+                    last_packet_at = now_s
+                for packet in packets:
+                    self._enqueue(packet)
             packet = assembler.finalize_current()
             if packet is not None:
                 self._enqueue(packet)

rfg_adc_plotter/processing/__init__.py

@@ -8,16 +8,20 @@ from rfg_adc_plotter.processing.background import (
 )
 from rfg_adc_plotter.processing.calibration import (
     build_calib_envelope,
+    build_complex_calibration_curve,
     calibrate_freqs,
     get_calibration_base,
     get_calibration_coeffs,
     load_calib_envelope,
+    load_complex_calibration,
     recalculate_calibration_c,
     save_calib_envelope,
+    save_complex_calibration,
     set_calibration_base_value,
 )
 from rfg_adc_plotter.processing.fft import (
     compute_distance_axis,
+    compute_fft_complex_row,
     compute_fft_mag_row,
     compute_fft_row,
     fft_mag_to_db,
@@ -29,6 +33,8 @@ from rfg_adc_plotter.processing.formatting import (
 )
 from rfg_adc_plotter.processing.normalization import (
     build_calib_envelopes,
+    fit_complex_calibration_to_width,
+    normalize_by_complex_calibration,
     normalize_by_envelope,
     normalize_by_calib,
 )
@@ -41,9 +47,11 @@ from rfg_adc_plotter.processing.peaks import (
 __all__ = [
     "build_calib_envelopes",
     "build_calib_envelope",
+    "build_complex_calibration_curve",
     "calibrate_freqs",
     "compute_auto_ylim",
     "compute_distance_axis",
+    "compute_fft_complex_row",
     "compute_fft_mag_row",
     "compute_fft_row",
     "fft_mag_to_db",
@@ -53,13 +61,17 @@ __all__ = [
     "get_calibration_base",
     "get_calibration_coeffs",
     "load_calib_envelope",
+    "load_complex_calibration",
     "load_fft_background",
+    "fit_complex_calibration_to_width",
+    "normalize_by_complex_calibration",
     "normalize_by_envelope",
     "normalize_by_calib",
     "parse_spec_clip",
     "recalculate_calibration_c",
     "rolling_median_ref",
     "save_calib_envelope",
+    "save_complex_calibration",
     "save_fft_background",
     "set_calibration_base_value",
     "subtract_fft_background",

rfg_adc_plotter/processing/calibration.py

@@ -65,14 +65,23 @@ def set_calibration_base_value(index: int, value: float) -> np.ndarray:
 def calibrate_freqs(sweep: Mapping[str, Any]) -> SweepData:
     """Return a sweep copy with calibrated and resampled frequency axis."""
     freqs = np.asarray(sweep["F"], dtype=np.float64).copy()
-    values = np.asarray(sweep["I"], dtype=np.float64).copy()
+    values_in = np.asarray(sweep["I"]).reshape(-1)
+    values = np.asarray(
+        values_in,
+        dtype=np.complex128 if np.iscomplexobj(values_in) else np.float64,
+    ).copy()
     coeffs = np.asarray(CALIBRATION_C, dtype=np.float64)
     if freqs.size > 0:
         freqs = coeffs[0] + coeffs[1] * freqs + coeffs[2] * (freqs * freqs)
 
     if freqs.size >= 2:
         freqs_cal = np.linspace(float(freqs[0]), float(freqs[-1]), freqs.size, dtype=np.float64)
-        values_cal = np.interp(freqs_cal, freqs, values).astype(np.float64)
+        if np.iscomplexobj(values):
+            values_real = np.interp(freqs_cal, freqs, values.real.astype(np.float64, copy=False))
+            values_imag = np.interp(freqs_cal, freqs, values.imag.astype(np.float64, copy=False))
+            values_cal = (values_real + (1j * values_imag)).astype(np.complex64)
+        else:
+            values_cal = np.interp(freqs_cal, freqs, values).astype(np.float64)
     else:
         freqs_cal = freqs.copy()
         values_cal = values.copy()
@@ -92,6 +101,17 @@ def build_calib_envelope(sweep: np.ndarray) -> np.ndarray:
     return np.asarray(upper, dtype=np.float32)
 
 
+def build_complex_calibration_curve(ch1: np.ndarray, ch2: np.ndarray) -> np.ndarray:
+    """Build a complex calibration curve as ``ch1 + 1j*ch2``."""
+    ch1_arr = np.asarray(ch1, dtype=np.float32).reshape(-1)
+    ch2_arr = np.asarray(ch2, dtype=np.float32).reshape(-1)
+    width = min(ch1_arr.size, ch2_arr.size)
+    if width <= 0:
+        raise ValueError("Complex calibration source is empty")
+    curve = ch1_arr[:width].astype(np.complex64) + (1j * ch2_arr[:width].astype(np.complex64))
+    return validate_complex_calibration_curve(curve)
+
+
 def validate_calib_envelope(envelope: np.ndarray) -> np.ndarray:
     """Validate a saved calibration envelope payload."""
     values = np.asarray(envelope, dtype=np.float32).reshape(-1)
@@ -102,6 +122,16 @@ def validate_calib_envelope(envelope: np.ndarray) -> np.ndarray:
     return values
 
 
+def validate_complex_calibration_curve(curve: np.ndarray) -> np.ndarray:
+    """Validate a saved complex calibration payload."""
+    values = np.asarray(curve).reshape(-1)
+    if values.size == 0:
+        raise ValueError("Complex calibration curve is empty")
+    if not np.issubdtype(values.dtype, np.number):
+        raise ValueError("Complex calibration curve must be numeric")
+    return np.asarray(values, dtype=np.complex64)
+
+
 def _normalize_calib_path(path: str | Path) -> Path:
     out = Path(path).expanduser()
     if out.suffix.lower() != ".npy":
@@ -122,3 +152,18 @@ def load_calib_envelope(path: str | Path) -> np.ndarray:
     normalized_path = _normalize_calib_path(path)
     loaded = np.load(normalized_path, allow_pickle=False)
     return validate_calib_envelope(loaded)
+
+
+def save_complex_calibration(path: str | Path, curve: np.ndarray) -> str:
+    """Persist a complex calibration curve as a .npy file and return the final path."""
+    normalized_path = _normalize_calib_path(path)
+    values = validate_complex_calibration_curve(curve)
+    np.save(normalized_path, values.astype(np.complex64, copy=False))
+    return str(normalized_path)
+
+
+def load_complex_calibration(path: str | Path) -> np.ndarray:
+    """Load and validate a complex calibration curve from a .npy file."""
+    normalized_path = _normalize_calib_path(path)
+    loaded = np.load(normalized_path, allow_pickle=False)
+    return validate_complex_calibration_curve(loaded)

rfg_adc_plotter/processing/fft.py

@@ -24,6 +24,185 @@ def _finite_freq_bounds(freqs: Optional[np.ndarray]) -> Optional[Tuple[float, fl
     return f_min, f_max
 
 
+def _coerce_sweep_array(sweep: np.ndarray) -> np.ndarray:
+    values = np.asarray(sweep).reshape(-1)
+    if np.iscomplexobj(values):
+        return np.asarray(values, dtype=np.complex64)
+    return np.asarray(values, dtype=np.float32)
+
+
+def _interp_signal(x_uniform: np.ndarray, x_known: np.ndarray, y_known: np.ndarray) -> np.ndarray:
+    if np.iscomplexobj(y_known):
+        real = np.interp(x_uniform, x_known, np.asarray(y_known.real, dtype=np.float64))
+        imag = np.interp(x_uniform, x_known, np.asarray(y_known.imag, dtype=np.float64))
+        return (real + (1j * imag)).astype(np.complex64)
+    return np.interp(x_uniform, x_known, np.asarray(y_known, dtype=np.float64)).astype(np.float32)
+
+
+def _fit_complex_bins(values: np.ndarray, bins: int) -> np.ndarray:
+    arr = np.asarray(values, dtype=np.complex64).reshape(-1)
+    if bins <= 0:
+        return np.zeros((0,), dtype=np.complex64)
+    if arr.size == bins:
+        return arr
+    out = np.full((bins,), np.nan + 0j, dtype=np.complex64)
+    take = min(arr.size, bins)
+    out[:take] = arr[:take]
+    return out
+
+
+def _extract_positive_exact_band(
+    sweep: np.ndarray,
+    freqs: Optional[np.ndarray],
+) -> Optional[Tuple[np.ndarray, np.ndarray, float, float]]:
+    """Return sorted positive band data and exact-grid parameters."""
+    if freqs is None:
+        return None
+
+    sweep_arr = _coerce_sweep_array(sweep)
+    freq_arr = np.asarray(freqs, dtype=np.float64).reshape(-1)
+    take = min(int(sweep_arr.size), int(freq_arr.size))
+    if take <= 1:
+        return None
+
+    sweep_seg = sweep_arr[:take]
+    freq_seg = freq_arr[:take]
+    valid = np.isfinite(freq_seg) & np.isfinite(sweep_seg) & (freq_seg > 0.0)
+    if int(np.count_nonzero(valid)) < 2:
+        return None
+
+    freq_band = np.asarray(freq_seg[valid], dtype=np.float64)
+    sweep_band = np.asarray(sweep_seg[valid])
+    order = np.argsort(freq_band, kind="mergesort")
+    freq_band = freq_band[order]
+    sweep_band = sweep_band[order]
+
+    n_band = int(freq_band.size)
+    if n_band <= 1:
+        return None
+
+    f_min = float(freq_band[0])
+    f_max = float(freq_band[-1])
+    if (not np.isfinite(f_min)) or (not np.isfinite(f_max)) or f_max <= f_min:
+        return None
+
+    df_ghz = float((f_max - f_min) / max(1, n_band - 1))
+    if (not np.isfinite(df_ghz)) or df_ghz <= 0.0:
+        return None
+
+    return freq_band, sweep_band, f_max, df_ghz
+
+
+def _positive_exact_shift_size(f_max: float, df_ghz: float) -> int:
+    if (not np.isfinite(f_max)) or (not np.isfinite(df_ghz)) or f_max <= 0.0 or df_ghz <= 0.0:
+        return 0
+    return int(np.arange(-f_max, f_max + (0.5 * df_ghz), df_ghz, dtype=np.float64).size)
+
+
+def _resolve_positive_exact_band_size(
+    f_min: float,
+    f_max: float,
+    n_band: int,
+    max_shift_len: Optional[int],
+) -> int:
+    if n_band <= 2:
+        return max(2, int(n_band))
+    if max_shift_len is None:
+        return int(n_band)
+
+    limit = int(max_shift_len)
+    if limit <= 1:
+        return max(2, int(n_band))
+
+    span = float(f_max - f_min)
+    if (not np.isfinite(span)) or span <= 0.0:
+        return int(n_band)
+
+    df_current = float(span / max(1, int(n_band) - 1))
+    if _positive_exact_shift_size(f_max, df_current) <= limit:
+        return int(n_band)
+
+    denom = max(2.0 * f_max, 1e-12)
+    approx = int(np.floor(1.0 + ((float(limit - 1) * span) / denom)))
+    target = min(int(n_band), max(2, approx))
+    while target > 2:
+        df_try = float(span / max(1, target - 1))
+        if _positive_exact_shift_size(f_max, df_try) <= limit:
+            break
+        target -= 1
+    return max(2, target)
+
+
+def _normalize_positive_exact_band(
+    freq_band: np.ndarray,
+    sweep_band: np.ndarray,
+    *,
+    max_shift_len: Optional[int] = None,
+) -> Optional[Tuple[np.ndarray, np.ndarray, float, float]]:
+    freq_arr = np.asarray(freq_band, dtype=np.float64).reshape(-1)
+    sweep_arr = np.asarray(sweep_band).reshape(-1)
+    width = min(int(freq_arr.size), int(sweep_arr.size))
+    if width <= 1:
+        return None
+
+    freq_arr = freq_arr[:width]
+    sweep_arr = sweep_arr[:width]
+    f_min = float(freq_arr[0])
+    f_max = float(freq_arr[-1])
+    if (not np.isfinite(f_min)) or (not np.isfinite(f_max)) or f_max <= f_min:
+        return None
+
+    target_band = _resolve_positive_exact_band_size(f_min, f_max, int(freq_arr.size), max_shift_len)
+    if target_band < int(freq_arr.size):
+        target_freqs = np.linspace(f_min, f_max, target_band, dtype=np.float64)
+        target_sweep = _interp_signal(target_freqs, freq_arr, sweep_arr)
+        freq_arr = target_freqs
+        sweep_arr = np.asarray(target_sweep).reshape(-1)
+
+    n_band = int(freq_arr.size)
+    if n_band <= 1:
+        return None
+
+    df_ghz = float((f_max - f_min) / max(1, n_band - 1))
+    if (not np.isfinite(df_ghz)) or df_ghz <= 0.0:
+        return None
+
+    return freq_arr, sweep_arr, f_max, df_ghz
+
+
+def _resolve_positive_only_exact_geometry(
+    freqs: Optional[np.ndarray],
+    *,
+    max_shift_len: Optional[int] = None,
+) -> Optional[Tuple[int, float]]:
+    """Return (N_shift, df_hz) for the exact centered positive-only mode."""
+    if freqs is None:
+        return None
+
+    freq_arr = np.asarray(freqs, dtype=np.float64).reshape(-1)
+    finite = np.asarray(freq_arr[np.isfinite(freq_arr) & (freq_arr > 0.0)], dtype=np.float64)
+    if finite.size < 2:
+        return None
+
+    finite.sort(kind="mergesort")
+    f_min = float(finite[0])
+    f_max = float(finite[-1])
+    if (not np.isfinite(f_min)) or (not np.isfinite(f_max)) or f_max <= f_min:
+        return None
+
+    n_band = int(finite.size)
+    target_band = _resolve_positive_exact_band_size(f_min, f_max, n_band, max_shift_len)
+    n_band = max(2, min(n_band, target_band))
+    df_ghz = float((f_max - f_min) / max(1, n_band - 1))
+    if (not np.isfinite(df_ghz)) or df_ghz <= 0.0:
+        return None
+
+    n_shift = _positive_exact_shift_size(f_max, df_ghz)
+    if n_shift <= 1:
+        return None
+    return int(n_shift), float(df_ghz * 1e9)
+
+
 def prepare_fft_segment(
     sweep: np.ndarray,
     freqs: Optional[np.ndarray],
@@ -34,8 +213,10 @@ def prepare_fft_segment(
     if take_fft <= 0:
         return None
 
-    sweep_seg = np.asarray(sweep[:take_fft], dtype=np.float32)
-    fallback = np.nan_to_num(sweep_seg, nan=0.0).astype(np.float32, copy=False)
+    sweep_arr = _coerce_sweep_array(sweep)
+    sweep_seg = sweep_arr[:take_fft]
+    fallback_dtype = np.complex64 if np.iscomplexobj(sweep_seg) else np.float32
+    fallback = np.nan_to_num(sweep_seg, nan=0.0).astype(fallback_dtype, copy=False)
     if freqs is None:
         return fallback, take_fft
 
@@ -59,7 +240,7 @@ def prepare_fft_segment(
         return fallback, take_fft
 
     x_uniform = np.linspace(float(x_unique[0]), float(x_unique[-1]), take_fft, dtype=np.float64)
-    resampled = np.interp(x_uniform, x_unique, y_unique).astype(np.float32)
+    resampled = _interp_signal(x_uniform, x_unique, y_unique)
     return resampled, take_fft
 
 
@@ -94,18 +275,20 @@ def build_symmetric_ifft_spectrum(
 
     fft_seg, take_fft = prepared
     if take_fft != band_len:
-        fft_seg = np.asarray(fft_seg[:band_len], dtype=np.float32)
+        fft_dtype = np.complex64 if np.iscomplexobj(fft_seg) else np.float32
+        fft_seg = np.asarray(fft_seg[:band_len], dtype=fft_dtype)
         if fft_seg.size < band_len:
-            padded = np.zeros((band_len,), dtype=np.float32)
+            padded = np.zeros((band_len,), dtype=fft_dtype)
             padded[: fft_seg.size] = fft_seg
             fft_seg = padded
 
     window = np.hanning(band_len).astype(np.float32)
-    band = np.nan_to_num(fft_seg, nan=0.0).astype(np.float32, copy=False) * window
+    band_dtype = np.complex64 if np.iscomplexobj(fft_seg) else np.float32
+    band = np.nan_to_num(fft_seg, nan=0.0).astype(band_dtype, copy=False) * window
 
-    spectrum = np.zeros((int(fft_len),), dtype=np.float32)
+    spectrum = np.zeros((int(fft_len),), dtype=band_dtype)
     spectrum[pos_idx] = band
-    spectrum[neg_idx] = band[::-1]
+    spectrum[neg_idx] = np.conj(band[::-1]) if np.iscomplexobj(band) else band[::-1]
     return spectrum
 
 
@@ -137,20 +320,56 @@ def build_positive_only_centered_ifft_spectrum(
 
     fft_seg, take_fft = prepared
     if take_fft != band_len:
-        fft_seg = np.asarray(fft_seg[:band_len], dtype=np.float32)
+        fft_dtype = np.complex64 if np.iscomplexobj(fft_seg) else np.float32
+        fft_seg = np.asarray(fft_seg[:band_len], dtype=fft_dtype)
         if fft_seg.size < band_len:
-            padded = np.zeros((band_len,), dtype=np.float32)
+            padded = np.zeros((band_len,), dtype=fft_dtype)
             padded[: fft_seg.size] = fft_seg
             fft_seg = padded
 
     window = np.hanning(band_len).astype(np.float32)
-    band = np.nan_to_num(fft_seg, nan=0.0).astype(np.float32, copy=False) * window
+    band_dtype = np.complex64 if np.iscomplexobj(fft_seg) else np.float32
+    band = np.nan_to_num(fft_seg, nan=0.0).astype(band_dtype, copy=False) * window
 
-    spectrum = np.zeros((int(fft_len),), dtype=np.float32)
+    spectrum = np.zeros((int(fft_len),), dtype=band_dtype)
     spectrum[pos_idx] = band
     return spectrum
 
 
+def build_positive_only_exact_centered_ifft_spectrum(
+    sweep: np.ndarray,
+    freqs: Optional[np.ndarray],
+    *,
+    max_shift_len: Optional[int] = None,
+) -> Optional[np.ndarray]:
+    """Build centered spectrum exactly as zeros[-f_max..+f_min) + measured positive band."""
+    prepared = _extract_positive_exact_band(sweep, freqs)
+    if prepared is None:
+        return None
+
+    freq_band, sweep_band, _f_max, _df_ghz = prepared
+    normalized = _normalize_positive_exact_band(
+        freq_band,
+        sweep_band,
+        max_shift_len=max_shift_len,
+    )
+    if normalized is None:
+        return None
+
+    freq_band, sweep_band, f_max, df_ghz = normalized
+    f_shift = np.arange(-f_max, f_max + (0.5 * df_ghz), df_ghz, dtype=np.float64)
+    if f_shift.size <= 1:
+        return None
+
+    band_dtype = np.complex64 if np.iscomplexobj(sweep_band) else np.float32
+    band = np.nan_to_num(np.asarray(sweep_band, dtype=band_dtype), nan=0.0)
+    spectrum = np.zeros((int(f_shift.size),), dtype=band_dtype)
+    idx = np.round((freq_band - f_shift[0]) / df_ghz).astype(np.int64)
+    idx = np.clip(idx, 0, spectrum.size - 1)
+    spectrum[idx] = band
+    return spectrum
+
+
 def fft_mag_to_db(mag: np.ndarray) -> np.ndarray:
     """Convert magnitude to dB with safe zero handling."""
     mag_arr = np.asarray(mag, dtype=np.float32)
@@ -158,24 +377,21 @@ def fft_mag_to_db(mag: np.ndarray) -> np.ndarray:
     return (20.0 * np.log10(safe_mag + 1e-9)).astype(np.float32, copy=False)
 
 
-def _compute_fft_mag_row_direct(
+def _compute_fft_complex_row_direct(
     sweep: np.ndarray,
     freqs: Optional[np.ndarray],
     bins: int,
 ) -> np.ndarray:
     prepared = prepare_fft_segment(sweep, freqs, fft_len=FFT_LEN)
     if prepared is None:
-        return np.full((bins,), np.nan, dtype=np.float32)
+        return np.full((bins,), np.nan + 0j, dtype=np.complex64)
 
     fft_seg, take_fft = prepared
-    fft_in = np.zeros((FFT_LEN,), dtype=np.float32)
+    fft_in = np.zeros((FFT_LEN,), dtype=np.complex64)
     window = np.hanning(take_fft).astype(np.float32)
-    fft_in[:take_fft] = fft_seg * window
+    fft_in[:take_fft] = np.asarray(fft_seg, dtype=np.complex64) * window
     spec = np.fft.ifft(fft_in)
-    mag = np.abs(spec).astype(np.float32)
-    if mag.shape[0] != bins:
-        mag = mag[:bins]
-    return mag
+    return _fit_complex_bins(spec, bins)
 
 
 def _normalize_fft_mode(mode: str | None, symmetric: Optional[bool]) -> str:
@@ -188,9 +404,44 @@ def _normalize_fft_mode(mode: str | None, symmetric: Optional[bool]) -> str:
         return "symmetric"
     if normalized in {"positive_only", "positive-centered", "positive_centered", "zero_left"}:
         return "positive_only"
+    if normalized in {"positive_only_exact", "positive-centered-exact", "positive_centered_exact", "zero_left_exact"}:
+        return "positive_only_exact"
     raise ValueError(f"Unsupported FFT mode: {mode!r}")
 
 
+def compute_fft_complex_row(
+    sweep: np.ndarray,
+    freqs: Optional[np.ndarray],
+    bins: int,
+    *,
+    mode: str = "symmetric",
+    symmetric: Optional[bool] = None,
+) -> np.ndarray:
+    """Compute a complex FFT/IFFT row on the distance axis."""
+    if bins <= 0:
+        return np.zeros((0,), dtype=np.complex64)
+
+    fft_mode = _normalize_fft_mode(mode, symmetric)
+    if fft_mode == "direct":
+        return _compute_fft_complex_row_direct(sweep, freqs, bins)
+
+    if fft_mode == "positive_only":
+        spectrum_centered = build_positive_only_centered_ifft_spectrum(sweep, freqs, fft_len=FFT_LEN)
+    elif fft_mode == "positive_only_exact":
+        spectrum_centered = build_positive_only_exact_centered_ifft_spectrum(
+            sweep,
+            freqs,
+            max_shift_len=bins,
+        )
+    else:
+        spectrum_centered = build_symmetric_ifft_spectrum(sweep, freqs, fft_len=FFT_LEN)
+    if spectrum_centered is None:
+        return np.full((bins,), np.nan + 0j, dtype=np.complex64)
+
+    spec = np.fft.ifft(np.fft.ifftshift(np.asarray(spectrum_centered, dtype=np.complex64)))
+    return _fit_complex_bins(spec, bins)
+
+
 def compute_fft_mag_row(
     sweep: np.ndarray,
     freqs: Optional[np.ndarray],
@@ -200,25 +451,8 @@ def compute_fft_mag_row(
     symmetric: Optional[bool] = None,
 ) -> np.ndarray:
     """Compute a linear FFT magnitude row."""
-    if bins <= 0:
-        return np.zeros((0,), dtype=np.float32)
-
-    fft_mode = _normalize_fft_mode(mode, symmetric)
-    if fft_mode == "direct":
-        return _compute_fft_mag_row_direct(sweep, freqs, bins)
-
-    if fft_mode == "positive_only":
-        spectrum_centered = build_positive_only_centered_ifft_spectrum(sweep, freqs, fft_len=FFT_LEN)
-    else:
-        spectrum_centered = build_symmetric_ifft_spectrum(sweep, freqs, fft_len=FFT_LEN)
-    if spectrum_centered is None:
-        return np.full((bins,), np.nan, dtype=np.float32)
-
-    spec = np.fft.ifft(np.fft.ifftshift(spectrum_centered))
-    mag = np.abs(spec).astype(np.float32)
-    if mag.shape[0] != bins:
-        mag = mag[:bins]
-    return mag
+    complex_row = compute_fft_complex_row(sweep, freqs, bins, mode=mode, symmetric=symmetric)
+    return np.abs(complex_row).astype(np.float32, copy=False)
 
 
 def compute_fft_row(
@@ -244,6 +478,16 @@ def compute_distance_axis(
     if bins <= 0:
         return np.zeros((0,), dtype=np.float64)
     fft_mode = _normalize_fft_mode(mode, symmetric)
+    if fft_mode == "positive_only_exact":
+        geometry = _resolve_positive_only_exact_geometry(freqs, max_shift_len=bins)
+        if geometry is None:
+            return np.arange(bins, dtype=np.float64)
+        n_shift, df_hz = geometry
+        if (not np.isfinite(df_hz)) or df_hz <= 0.0 or n_shift <= 0:
+            return np.arange(bins, dtype=np.float64)
+        step_m = C_M_S / (2.0 * float(n_shift) * df_hz)
+        return np.arange(bins, dtype=np.float64) * step_m
+
     if fft_mode in {"symmetric", "positive_only"}:
         bounds = _finite_freq_bounds(freqs)
         if bounds is None:

rfg_adc_plotter/processing/normalization.py

@@ -148,20 +148,77 @@ def resample_envelope(envelope: np.ndarray, width: int) -> np.ndarray:
     return np.interp(x_dst, x_src[finite], values[finite]).astype(np.float32)
 
 
+def fit_complex_calibration_to_width(calib: np.ndarray, width: int) -> np.ndarray:
+    """Fit a complex calibration curve to the signal width via trim/pad with ones."""
+    target_width = int(width)
+    if target_width <= 0:
+        return np.zeros((0,), dtype=np.complex64)
+
+    values = np.asarray(calib, dtype=np.complex64).reshape(-1)
+    if values.size <= 0:
+        return np.ones((target_width,), dtype=np.complex64)
+    if values.size == target_width:
+        return values.astype(np.complex64, copy=True)
+    if values.size > target_width:
+        return np.asarray(values[:target_width], dtype=np.complex64)
+
+    out = np.ones((target_width,), dtype=np.complex64)
+    out[: values.size] = values
+    return out
+
+
+def normalize_by_complex_calibration(
+    signal: np.ndarray,
+    calib: np.ndarray,
+    eps: float = 1e-9,
+) -> np.ndarray:
+    """Normalize complex signal by a complex calibration curve with zero protection."""
+    sig_arr = np.asarray(signal, dtype=np.complex64).reshape(-1)
+    if sig_arr.size <= 0:
+        return sig_arr.copy()
+
+    calib_fit = fit_complex_calibration_to_width(calib, sig_arr.size)
+    eps_abs = max(abs(float(eps)), 1e-12)
+    denom = np.asarray(calib_fit, dtype=np.complex64).copy()
+    safe_denom = (
+        np.isfinite(denom.real)
+        & np.isfinite(denom.imag)
+        & (np.abs(denom) >= eps_abs)
+    )
+    if np.any(~safe_denom):
+        denom[~safe_denom] = np.complex64(1.0 + 0.0j)
+
+    out = np.full(sig_arr.shape, np.nan + 0j, dtype=np.complex64)
+    valid_sig = np.isfinite(sig_arr.real) & np.isfinite(sig_arr.imag)
+    if np.any(valid_sig):
+        with np.errstate(divide="ignore", invalid="ignore"):
+            out[valid_sig] = sig_arr[valid_sig] / denom[valid_sig]
+
+    out_real = np.nan_to_num(out.real, nan=np.nan, posinf=np.nan, neginf=np.nan)
+    out_imag = np.nan_to_num(out.imag, nan=np.nan, posinf=np.nan, neginf=np.nan)
+    return (out_real + (1j * out_imag)).astype(np.complex64, copy=False)
+
+
 def normalize_by_envelope(raw: np.ndarray, envelope: np.ndarray) -> np.ndarray:
     """Normalize a sweep by an envelope with safe resampling and zero protection."""
-    raw_arr = np.asarray(raw, dtype=np.float32).reshape(-1)
+    raw_in = np.asarray(raw).reshape(-1)
+    raw_dtype = np.complex64 if np.iscomplexobj(raw_in) else np.float32
+    raw_arr = np.asarray(raw_in, dtype=raw_dtype).reshape(-1)
     if raw_arr.size == 0:
         return raw_arr.copy()
 
     env = resample_envelope(envelope, raw_arr.size)
-    out = np.full_like(raw_arr, np.nan, dtype=np.float32)
+    out = np.full(raw_arr.shape, np.nan + 0j if np.iscomplexobj(raw_arr) else np.nan, dtype=raw_dtype)
     den_eps = np.float32(1e-9)
     valid = np.isfinite(raw_arr) & np.isfinite(env)
     if np.any(valid):
         with np.errstate(divide="ignore", invalid="ignore"):
             denom = env[valid] + np.where(env[valid] >= 0.0, den_eps, -den_eps)
             out[valid] = raw_arr[valid] / denom
+    if np.iscomplexobj(out):
+        out_real = np.nan_to_num(out.real, nan=np.nan, posinf=np.nan, neginf=np.nan)
+        out_imag = np.nan_to_num(out.imag, nan=np.nan, posinf=np.nan, neginf=np.nan)
+        return (out_real + (1j * out_imag)).astype(np.complex64, copy=False)
     return np.nan_to_num(out, nan=np.nan, posinf=np.nan, neginf=np.nan)
 
 

rfg_adc_plotter/state/ring_buffer.py

@@ -23,6 +23,7 @@ class RingBuffer:
         self.ring: Optional[np.ndarray] = None
         self.ring_time: Optional[np.ndarray] = None
         self.ring_fft: Optional[np.ndarray] = None
+        self.ring_fft_input: Optional[np.ndarray] = None
         self.x_shared: Optional[np.ndarray] = None
         self.distance_axis: Optional[np.ndarray] = None
         self.last_fft_mag: Optional[np.ndarray] = None
@@ -30,6 +31,9 @@ class RingBuffer:
         self.last_freqs: Optional[np.ndarray] = None
         self.y_min_fft: Optional[float] = None
         self.y_max_fft: Optional[float] = None
+        self.last_push_valid_points = 0
+        self.last_push_fft_valid = False
+        self.last_push_axis_valid = False
 
     @property
     def is_ready(self) -> bool:
@@ -46,6 +50,7 @@ class RingBuffer:
         self.ring = None
         self.ring_time = None
         self.ring_fft = None
+        self.ring_fft_input = None
         self.x_shared = None
         self.distance_axis = None
         self.last_fft_mag = None
@@ -53,6 +58,38 @@ class RingBuffer:
         self.last_freqs = None
         self.y_min_fft = None
         self.y_max_fft = None
+        self.last_push_valid_points = 0
+        self.last_push_fft_valid = False
+        self.last_push_axis_valid = False
+
+    def _promote_fft_cache(self, fft_mag: np.ndarray) -> bool:
+        fft_mag_arr = np.asarray(fft_mag, dtype=np.float32).reshape(-1)
+        if fft_mag_arr.size <= 0:
+            self.last_push_fft_valid = False
+            return False
+        fft_db = fft_mag_to_db(fft_mag_arr)
+        finite_db = fft_db[np.isfinite(fft_db)]
+        if finite_db.size <= 0:
+            self.last_push_fft_valid = False
+            return False
+
+        self.last_fft_mag = fft_mag_arr.copy()
+        self.last_fft_db = fft_db
+        fr_min = float(np.min(finite_db))
+        fr_max = float(np.max(finite_db))
+        self.y_min_fft = fr_min if self.y_min_fft is None else min(self.y_min_fft, fr_min)
+        self.y_max_fft = fr_max if self.y_max_fft is None else max(self.y_max_fft, fr_max)
+        self.last_push_fft_valid = True
+        return True
+
+    def _promote_distance_axis(self, axis: np.ndarray) -> bool:
+        axis_arr = np.asarray(axis, dtype=np.float64).reshape(-1)
+        if axis_arr.size <= 0 or not np.all(np.isfinite(axis_arr)):
+            self.last_push_axis_valid = False
+            return False
+        self.distance_axis = axis_arr.copy()
+        self.last_push_axis_valid = True
+        return True
 
     def ensure_init(self, sweep_width: int) -> bool:
         """Allocate or resize buffers. Returns True when geometry changed."""
@@ -63,13 +100,18 @@ class RingBuffer:
             self.ring = np.full((self.max_sweeps, self.width), np.nan, dtype=np.float32)
             self.ring_time = np.full((self.max_sweeps,), np.nan, dtype=np.float64)
             self.ring_fft = np.full((self.max_sweeps, self.fft_bins), np.nan, dtype=np.float32)
+            self.ring_fft_input = np.full((self.max_sweeps, self.width), np.nan + 0j, dtype=np.complex64)
             self.head = 0
             changed = True
         elif target_width != self.width:
             new_ring = np.full((self.max_sweeps, target_width), np.nan, dtype=np.float32)
+            new_fft_input = np.full((self.max_sweeps, target_width), np.nan + 0j, dtype=np.complex64)
             take = min(self.width, target_width)
             new_ring[:, :take] = self.ring[:, :take]
+            if self.ring_fft_input is not None:
+                new_fft_input[:, :take] = self.ring_fft_input[:, :take]
             self.ring = new_ring
+            self.ring_fft_input = new_fft_input
             self.width = target_width
             changed = True
 
@@ -92,7 +134,9 @@ class RingBuffer:
             normalized_mode = "symmetric"
         if normalized_mode in {"positive-centered", "positive_centered", "zero_left"}:
             normalized_mode = "positive_only"
-        if normalized_mode not in {"direct", "symmetric", "positive_only"}:
+        if normalized_mode in {"positive-centered-exact", "positive_centered_exact", "zero_left_exact"}:
+            normalized_mode = "positive_only_exact"
+        if normalized_mode not in {"direct", "symmetric", "positive_only", "positive_only_exact"}:
             raise ValueError(f"Unsupported FFT mode: {mode!r}")
         if normalized_mode == self.fft_mode:
             return False
@@ -100,35 +144,44 @@ class RingBuffer:
         self.fft_mode = normalized_mode
         self.y_min_fft = None
         self.y_max_fft = None
+        self.last_push_fft_valid = False
+        self.last_push_axis_valid = False
 
         if self.ring is None or self.ring_fft is None:
             return True
 
         self.ring_fft.fill(np.nan)
         for row_idx in range(self.ring.shape[0]):
-            sweep_row = self.ring[row_idx]
-            if not np.any(np.isfinite(sweep_row)):
+            fft_source_row = self.ring_fft_input[row_idx] if self.ring_fft_input is not None else self.ring[row_idx]
+            if not np.any(np.isfinite(fft_source_row)):
                 continue
+            finite_idx = np.flatnonzero(np.isfinite(fft_source_row))
+            if finite_idx.size <= 0:
+                continue
+            row_width = int(finite_idx[-1]) + 1
+            fft_source = fft_source_row[:row_width]
+            freqs = self.last_freqs[:row_width] if self.last_freqs is not None and self.last_freqs.size >= row_width else self.last_freqs
             fft_mag = compute_fft_mag_row(
-                sweep_row,
-                self.last_freqs,
+                fft_source,
+                freqs,
                 self.fft_bins,
                 mode=self.fft_mode,
             )
             self.ring_fft[row_idx, :] = fft_mag
 
         if self.last_freqs is not None:
-            self.distance_axis = compute_distance_axis(
-                self.last_freqs,
-                self.fft_bins,
-                mode=self.fft_mode,
+            self._promote_distance_axis(
+                compute_distance_axis(
+                    self.last_freqs,
+                    self.fft_bins,
+                    mode=self.fft_mode,
+                )
             )
 
         last_idx = (self.head - 1) % self.max_sweeps
         if self.ring_fft.shape[0] > 0:
             last_fft = self.ring_fft[last_idx]
-            self.last_fft_mag = np.asarray(last_fft, dtype=np.float32).copy()
-            self.last_fft_db = fft_mag_to_db(last_fft)
+            self._promote_fft_cache(last_fft)
             finite = self.ring_fft[np.isfinite(self.ring_fft)]
             if finite.size > 0:
                 finite_db = fft_mag_to_db(finite.astype(np.float32, copy=False))
@@ -140,34 +193,39 @@ class RingBuffer:
         """Backward-compatible wrapper for the old two-state FFT switch."""
         return self.set_fft_mode("symmetric" if enabled else "direct")
 
-    def push(self, sweep: np.ndarray, freqs: Optional[np.ndarray] = None) -> None:
+    def push(
+        self,
+        sweep: np.ndarray,
+        freqs: Optional[np.ndarray] = None,
+        *,
+        fft_input: Optional[np.ndarray] = None,
+    ) -> None:
         """Push a processed sweep and refresh raw/FFT buffers."""
         if sweep is None or sweep.size == 0:
             return
         self.ensure_init(int(sweep.size))
-        if self.ring is None or self.ring_time is None or self.ring_fft is None:
+        if self.ring is None or self.ring_time is None or self.ring_fft is None or self.ring_fft_input is None:
             return
 
         row = np.full((self.width,), np.nan, dtype=np.float32)
         take = min(self.width, int(sweep.size))
         row[:take] = np.asarray(sweep[:take], dtype=np.float32)
+        self.last_push_valid_points = int(np.count_nonzero(np.isfinite(row[:take])))
         self.ring[self.head, :] = row
         self.ring_time[self.head] = time.time()
         if freqs is not None:
             self.last_freqs = np.asarray(freqs, dtype=np.float64).copy()
 
-        fft_mag = compute_fft_mag_row(sweep, freqs, self.fft_bins, mode=self.fft_mode)
+        fft_source = np.asarray(fft_input if fft_input is not None else sweep).reshape(-1)
+        fft_row = np.full((self.width,), np.nan + 0j, dtype=np.complex64)
+        fft_take = min(self.width, int(fft_source.size))
+        fft_row[:fft_take] = np.asarray(fft_source[:fft_take], dtype=np.complex64)
+        self.ring_fft_input[self.head, :] = fft_row
+
+        fft_mag = compute_fft_mag_row(fft_source, freqs, self.fft_bins, mode=self.fft_mode)
         self.ring_fft[self.head, :] = fft_mag
-        self.last_fft_mag = np.asarray(fft_mag, dtype=np.float32).copy()
-        self.last_fft_db = fft_mag_to_db(fft_mag)
-
-        if self.last_fft_db.size > 0:
-            fr_min = float(np.nanmin(self.last_fft_db))
-            fr_max = float(np.nanmax(self.last_fft_db))
-            self.y_min_fft = fr_min if self.y_min_fft is None else min(self.y_min_fft, fr_min)
-            self.y_max_fft = fr_max if self.y_max_fft is None else max(self.y_max_fft, fr_max)
-
-        self.distance_axis = compute_distance_axis(freqs, self.fft_bins, mode=self.fft_mode)
+        self._promote_fft_cache(fft_mag)
+        self._promote_distance_axis(compute_distance_axis(freqs, self.fft_bins, mode=self.fft_mode))
         self.head = (self.head + 1) % self.max_sweeps
 
     def get_display_raw(self) -> np.ndarray:
@@ -176,6 +234,21 @@ class RingBuffer:
         base = self.ring if self.head == 0 else np.roll(self.ring, -self.head, axis=0)
         return base.T
 
+    def get_display_raw_decimated(self, max_points: int) -> np.ndarray:
+        """Return a display-oriented raw waterfall with optional frequency decimation."""
+        if self.ring is None:
+            return np.zeros((1, 1), dtype=np.float32)
+
+        limit = int(max_points)
+        if limit <= 0 or self.width <= limit:
+            return self.get_display_raw()
+
+        row_order = np.arange(self.ring.shape[0], dtype=np.int64)
+        if self.head:
+            row_order = np.roll(row_order, -self.head)
+        col_idx = np.linspace(0, self.width - 1, limit, dtype=np.int64)
+        return self.ring[np.ix_(row_order, col_idx)].T
+
     def get_display_fft_linear(self) -> np.ndarray:
         if self.ring_fft is None:
             return np.zeros((1, 1), dtype=np.float32)

rfg_adc_plotter/state/runtime_state.py

@@ -20,10 +20,16 @@ class RuntimeState:
     range_max_ghz: float = 0.0
     full_current_freqs: Optional[np.ndarray] = None
     full_current_sweep_raw: Optional[np.ndarray] = None
+    full_current_sweep_codes: Optional[np.ndarray] = None
+    full_current_fft_source: Optional[np.ndarray] = None
     full_current_aux_curves: SweepAuxCurves = None
+    full_current_aux_curves_codes: SweepAuxCurves = None
     current_freqs: Optional[np.ndarray] = None
     current_distances: Optional[np.ndarray] = None
     current_sweep_raw: Optional[np.ndarray] = None
+    current_fft_source: Optional[np.ndarray] = None
+    current_fft_input: Optional[np.ndarray] = None
+    current_fft_complex: Optional[np.ndarray] = None
     current_aux_curves: SweepAuxCurves = None
     current_sweep_norm: Optional[np.ndarray] = None
     current_fft_mag: Optional[np.ndarray] = None
@@ -31,6 +37,8 @@ class RuntimeState:
     last_calib_sweep: Optional[np.ndarray] = None
     calib_envelope: Optional[np.ndarray] = None
     calib_file_path: Optional[str] = None
+    complex_calib_curve: Optional[np.ndarray] = None
+    complex_calib_file_path: Optional[str] = None
     background_buffer: BackgroundMedianBuffer = field(
         default_factory=lambda: BackgroundMedianBuffer(BACKGROUND_MEDIAN_SWEEPS)
     )

rfg_adc_plotter/types.py

@@ -3,12 +3,13 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
-from typing import Any, Dict, Optional, Tuple, TypeAlias, Union
+from typing import Any, Dict, Literal, Optional, Tuple, TypeAlias, Union
 
 import numpy as np
 
 
 Number = Union[int, float]
+SignalKind = Literal["bin_iq", "bin_logdet"]
 SweepInfo = Dict[str, Any]
 SweepData = Dict[str, np.ndarray]
 SweepAuxCurves = Optional[Tuple[np.ndarray, np.ndarray]]
@@ -18,6 +19,7 @@ SweepPacket = Tuple[np.ndarray, SweepInfo, SweepAuxCurves]
 @dataclass(frozen=True)
 class StartEvent:
     ch: Optional[int] = None
+    signal_kind: Optional[SignalKind] = None
 
 
 @dataclass(frozen=True)
@@ -26,6 +28,7 @@ class PointEvent:
     x: int
     y: float
     aux: Optional[Tuple[float, float]] = None
+    signal_kind: Optional[SignalKind] = None
 
 
 ParserEvent: TypeAlias = Union[StartEvent, PointEvent]

tests/test_cli.py

@@ -5,6 +5,8 @@ import sys
 import unittest
 from pathlib import Path
 
+from rfg_adc_plotter.cli import build_parser
+
 
 ROOT = Path(__file__).resolve().parents[1]
 
@@ -20,6 +22,17 @@ def _run(*args: str) -> subprocess.CompletedProcess[str]:
 
 
 class CliTests(unittest.TestCase):
+    def test_logscale_and_opengl_are_opt_in(self):
+        args = build_parser().parse_args(["/dev/null"])
+        self.assertFalse(args.logscale)
+        self.assertFalse(args.opengl)
+        self.assertAlmostEqual(float(args.tty_range_v), 5.0, places=6)
+
+        args_log = build_parser().parse_args(["/dev/null", "--logscale", "--opengl", "--tty-range-v", "2.5"])
+        self.assertTrue(args_log.logscale)
+        self.assertTrue(args_log.opengl)
+        self.assertAlmostEqual(float(args_log.tty_range_v), 2.5, places=6)
+
     def test_wrapper_help_works(self):
         proc = _run("RFG_ADC_dataplotter.py", "--help")
         self.assertEqual(proc.returncode, 0)
@@ -31,6 +44,8 @@ class CliTests(unittest.TestCase):
         self.assertEqual(proc.returncode, 0)
         self.assertIn("usage:", proc.stdout)
         self.assertIn("--parser_16_bit_x2", proc.stdout)
+        self.assertIn("--parser_complex_ascii", proc.stdout)
+        self.assertIn("--opengl", proc.stdout)
 
     def test_backend_mpl_reports_removal(self):
         proc = _run("-m", "rfg_adc_plotter.main", "/dev/null", "--backend", "mpl")

tests/test_processing.py

@@ -5,19 +5,37 @@ import tempfile
 import numpy as np
 import unittest
 
-from rfg_adc_plotter.constants import FFT_LEN, SWEEP_FREQ_MAX_GHZ, SWEEP_FREQ_MIN_GHZ
+from rfg_adc_plotter.constants import C_M_S, FFT_LEN, SWEEP_FREQ_MAX_GHZ, SWEEP_FREQ_MIN_GHZ
 from rfg_adc_plotter.gui.pyqtgraph_backend import (
+    apply_distance_cut_to_axis,
     apply_working_range,
     apply_working_range_to_aux_curves,
+    build_logdet_voltage_fft_input,
+    build_main_window_layout,
+    coalesce_packets_for_ui,
     compute_background_subtracted_bscan_levels,
+    compute_aux_phase_curve,
+    convert_tty_i16_to_voltage,
+    decimate_curve_for_display,
+    resolve_axis_bounds,
+    resolve_heavy_refresh_stride,
+    resolve_initial_window_size,
+    resolve_distance_cut_start,
+    sanitize_curve_data_for_display,
+    sanitize_image_for_display,
+    set_image_rect_if_ready,
+    resolve_visible_fft_curves,
     resolve_visible_aux_curves,
 )
 from rfg_adc_plotter.processing.calibration import (
     build_calib_envelope,
+    build_complex_calibration_curve,
     calibrate_freqs,
     load_calib_envelope,
+    load_complex_calibration,
     recalculate_calibration_c,
     save_calib_envelope,
+    save_complex_calibration,
 )
 from rfg_adc_plotter.processing.background import (
     load_fft_background,
@@ -25,16 +43,20 @@ from rfg_adc_plotter.processing.background import (
     subtract_fft_background,
 )
 from rfg_adc_plotter.processing.fft import (
+    build_positive_only_exact_centered_ifft_spectrum,
     build_positive_only_centered_ifft_spectrum,
     build_symmetric_ifft_spectrum,
     compute_distance_axis,
+    compute_fft_complex_row,
     compute_fft_mag_row,
     compute_fft_row,
     fft_mag_to_db,
 )
 from rfg_adc_plotter.processing.normalization import (
     build_calib_envelopes,
+    fit_complex_calibration_to_width,
     normalize_by_calib,
+    normalize_by_complex_calibration,
     normalize_by_envelope,
     resample_envelope,
 )
@@ -42,6 +64,40 @@ from rfg_adc_plotter.processing.peaks import find_peak_width_markers, find_top_p
 
 
 class ProcessingTests(unittest.TestCase):
+    def test_convert_tty_i16_to_voltage_maps_and_clips_full_range(self):
+        codes = np.asarray([-32768.0, -16384.0, 0.0, 16384.0, 32767.0], dtype=np.float32)
+        volts = convert_tty_i16_to_voltage(codes, 5.0)
+
+        self.assertEqual(volts.shape, codes.shape)
+        self.assertAlmostEqual(float(volts[0]), -5.0, places=6)
+        self.assertAlmostEqual(float(volts[2]), 0.0, places=6)
+        self.assertAlmostEqual(float(volts[-1]), 5.0, places=6)
+        self.assertTrue(np.all(volts >= -5.0))
+        self.assertTrue(np.all(volts <= 5.0))
+
+    def test_build_logdet_voltage_fft_input_converts_codes_and_exponentiates(self):
+        codes = np.asarray([-32768.0, 0.0, 32767.0], dtype=np.float32)
+        volts, fft_input = build_logdet_voltage_fft_input(codes, 5.0)
+
+        self.assertEqual(volts.shape, codes.shape)
+        self.assertEqual(fft_input.shape, codes.shape)
+        self.assertAlmostEqual(float(volts[0]), -5.0, places=6)
+        self.assertAlmostEqual(float(volts[1]), 0.0, places=6)
+        self.assertAlmostEqual(float(volts[2]), 5.0, places=6)
+        self.assertTrue(np.allclose(fft_input, np.exp(volts.astype(np.float32))))
+
+    def test_build_logdet_voltage_fft_input_clips_exp_argument_and_respects_range(self):
+        codes = np.asarray([32767.0], dtype=np.float32)
+        volts_5, fft_5 = build_logdet_voltage_fft_input(codes, 5.0, exp_input_limit=2.0)
+        volts_10, fft_10 = build_logdet_voltage_fft_input(codes, 10.0, exp_input_limit=2.0)
+
+        self.assertAlmostEqual(float(volts_5[0]), 5.0, places=6)
+        self.assertAlmostEqual(float(volts_10[0]), 10.0, places=6)
+        self.assertAlmostEqual(float(fft_5[0]), float(np.exp(np.float32(2.0))), places=5)
+        self.assertAlmostEqual(float(fft_10[0]), float(np.exp(np.float32(2.0))), places=5)
+        self.assertTrue(np.isfinite(fft_5[0]))
+        self.assertTrue(np.isfinite(fft_10[0]))
+
     def test_recalculate_calibration_preserves_requested_edges(self):
         coeffs = recalculate_calibration_c(np.asarray([0.0, 1.0, 0.025], dtype=np.float64), 3.3, 14.3)
         y0 = coeffs[0] + coeffs[1] * 3.3 + coeffs[2] * (3.3 ** 2)
@@ -56,6 +112,18 @@ class ProcessingTests(unittest.TestCase):
         self.assertEqual(calibrated["I"].shape, (32,))
         self.assertTrue(np.all(np.diff(calibrated["F"]) >= 0.0))
 
+    def test_calibrate_freqs_keeps_complex_payload(self):
+        sweep = {
+            "F": np.linspace(3.3, 14.3, 32),
+            "I": np.exp(1j * np.linspace(0.0, np.pi, 32)).astype(np.complex64),
+        }
+        calibrated = calibrate_freqs(sweep)
+
+        self.assertEqual(calibrated["F"].shape, (32,))
+        self.assertEqual(calibrated["I"].shape, (32,))
+        self.assertTrue(np.iscomplexobj(calibrated["I"]))
+        self.assertTrue(np.all(np.isfinite(calibrated["I"])))
+
     def test_normalizers_and_envelopes_return_finite_ranges(self):
         calib = (np.sin(np.linspace(0.0, 4.0 * np.pi, 64)) * 5.0).astype(np.float32)
         raw = calib * 0.75
@@ -105,6 +173,15 @@ class ProcessingTests(unittest.TestCase):
         self.assertAlmostEqual(float(normalized[1]), 2.0, places=5)
         self.assertAlmostEqual(float(normalized[2]), -3.0, places=5)
 
+    def test_normalize_by_envelope_supports_complex_input(self):
+        raw = np.asarray([1.0 + 1.0j, 2.0 - 2.0j], dtype=np.complex64)
+        envelope = np.asarray([1.0, 2.0], dtype=np.float32)
+        normalized = normalize_by_envelope(raw, envelope)
+
+        self.assertTrue(np.iscomplexobj(normalized))
+        self.assertTrue(np.all(np.isfinite(normalized)))
+        self.assertTrue(np.allclose(normalized, np.asarray([1.0 + 1.0j, 1.0 - 1.0j], dtype=np.complex64)))
+
     def test_load_calib_envelope_rejects_empty_payload(self):
         with tempfile.TemporaryDirectory() as tmp_dir:
             path = os.path.join(tmp_dir, "empty.npy")
@@ -112,6 +189,46 @@ class ProcessingTests(unittest.TestCase):
             with self.assertRaises(ValueError):
                 load_calib_envelope(path)
 
+    def test_complex_calibration_curve_roundtrip(self):
+        ch1 = np.asarray([1.0, 2.0, 3.0], dtype=np.float32)
+        ch2 = np.asarray([0.5, -1.0, 4.0], dtype=np.float32)
+        curve = build_complex_calibration_curve(ch1, ch2)
+        expected = np.asarray([1.0 + 0.5j, 2.0 - 1.0j, 3.0 + 4.0j], dtype=np.complex64)
+
+        self.assertTrue(np.iscomplexobj(curve))
+        self.assertTrue(np.allclose(curve, expected))
+
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            path = os.path.join(tmp_dir, "complex_calibration")
+            saved_path = save_complex_calibration(path, curve)
+            loaded = load_complex_calibration(saved_path)
+            self.assertTrue(saved_path.endswith(".npy"))
+            self.assertEqual(loaded.dtype, np.complex64)
+            self.assertTrue(np.allclose(loaded, expected))
+
+    def test_fit_complex_calibration_to_width_pads_or_trims(self):
+        calib = np.asarray([1.0 + 1.0j, 2.0 + 2.0j], dtype=np.complex64)
+        padded = fit_complex_calibration_to_width(calib, 4)
+        trimmed = fit_complex_calibration_to_width(
+            np.asarray([1.0 + 1.0j, 2.0 + 2.0j, 3.0 + 3.0j], dtype=np.complex64),
+            2,
+        )
+
+        self.assertEqual(padded.shape, (4,))
+        self.assertTrue(np.allclose(padded, np.asarray([1.0 + 1.0j, 2.0 + 2.0j, 1.0 + 0.0j, 1.0 + 0.0j], dtype=np.complex64)))
+        self.assertEqual(trimmed.shape, (2,))
+        self.assertTrue(np.allclose(trimmed, np.asarray([1.0 + 1.0j, 2.0 + 2.0j], dtype=np.complex64)))
+
+    def test_normalize_by_complex_calibration_handles_zero_and_length_mismatch(self):
+        signal = np.asarray([2.0 + 2.0j, 4.0 + 0.0j, 3.0 + 3.0j], dtype=np.complex64)
+        calib = np.asarray([1.0 + 1.0j, 0.0 + 0.0j], dtype=np.complex64)
+        normalized = normalize_by_complex_calibration(signal, calib)
+        expected = np.asarray([2.0 + 0.0j, 4.0 + 0.0j, 3.0 + 3.0j], dtype=np.complex64)
+
+        self.assertTrue(np.iscomplexobj(normalized))
+        self.assertTrue(np.all(np.isfinite(normalized)))
+        self.assertTrue(np.allclose(normalized, expected))
+
     def test_fft_background_roundtrip_and_rejects_non_1d_payload(self):
         background = np.asarray([0.5, 1.5, 2.5], dtype=np.float32)
         with tempfile.TemporaryDirectory() as tmp_dir:
@@ -182,6 +299,167 @@ class ProcessingTests(unittest.TestCase):
         self.assertTrue(np.allclose(visible[0], aux[0]))
         self.assertTrue(np.allclose(visible[1], aux[1]))
 
+    def test_compute_aux_phase_curve_returns_atan2_of_aux_channels(self):
+        aux = (
+            np.asarray([1.0, 1.0, -1.0, 0.0], dtype=np.float32),
+            np.asarray([0.0, 1.0, 1.0, 1.0], dtype=np.float32),
+        )
+
+        phase = compute_aux_phase_curve(aux)
+
+        self.assertIsNotNone(phase)
+        expected = np.asarray([0.0, np.pi / 4.0, 3.0 * np.pi / 4.0, np.pi / 2.0], dtype=np.float32)
+        self.assertEqual(phase.shape, expected.shape)
+        self.assertTrue(np.allclose(phase, expected, atol=1e-6))
+
+    def test_decimate_curve_for_display_preserves_small_series(self):
+        xs = np.linspace(3.3, 14.3, 64, dtype=np.float64)
+        ys = np.linspace(-1.0, 1.0, 64, dtype=np.float32)
+
+        decimated_x, decimated_y = decimate_curve_for_display(xs, ys, max_points=128)
+
+        self.assertTrue(np.allclose(decimated_x, xs))
+        self.assertTrue(np.allclose(decimated_y, ys))
+
+    def test_decimate_curve_for_display_limits_points_and_keeps_endpoints(self):
+        xs = np.linspace(3.3, 14.3, 10000, dtype=np.float64)
+        ys = np.sin(np.linspace(0.0, 12.0 * np.pi, 10000)).astype(np.float32)
+
+        decimated_x, decimated_y = decimate_curve_for_display(xs, ys, max_points=512)
+
+        self.assertLessEqual(decimated_x.size, 512)
+        self.assertEqual(decimated_x.shape, decimated_y.shape)
+        self.assertAlmostEqual(float(decimated_x[0]), float(xs[0]), places=12)
+        self.assertAlmostEqual(float(decimated_x[-1]), float(xs[-1]), places=12)
+        self.assertAlmostEqual(float(decimated_y[0]), float(ys[0]), places=6)
+        self.assertAlmostEqual(float(decimated_y[-1]), float(ys[-1]), places=6)
+
+    def test_coalesce_packets_for_ui_keeps_newest_packets(self):
+        packets = [
+            (np.asarray([float(idx)], dtype=np.float32), {"sweep": idx}, None)
+            for idx in range(6)
+        ]
+
+        kept, skipped = coalesce_packets_for_ui(packets, max_packets=2)
+
+        self.assertEqual(skipped, 4)
+        self.assertEqual(len(kept), 2)
+        self.assertEqual(int(kept[0][1]["sweep"]), 4)
+        self.assertEqual(int(kept[1][1]["sweep"]), 5)
+
+    def test_coalesce_packets_for_ui_never_returns_empty_for_non_empty_input(self):
+        packets = [
+            (np.asarray([1.0], dtype=np.float32), {"sweep": 1}, None),
+        ]
+
+        kept, skipped = coalesce_packets_for_ui(packets, max_packets=0)
+
+        self.assertEqual(skipped, 0)
+        self.assertEqual(len(kept), 1)
+        self.assertEqual(int(kept[0][1]["sweep"]), 1)
+
+    def test_coalesce_packets_for_ui_switches_to_latest_only_on_large_backlog(self):
+        packets = [
+            (np.asarray([float(idx)], dtype=np.float32), {"sweep": idx}, None)
+            for idx in range(40)
+        ]
+
+        kept, skipped = coalesce_packets_for_ui(packets, max_packets=8, backlog_packets=40)
+
+        self.assertEqual(skipped, 39)
+        self.assertEqual(len(kept), 1)
+        self.assertEqual(int(kept[0][1]["sweep"]), 39)
+
+    def test_resolve_heavy_refresh_stride_increases_with_backlog(self):
+        self.assertEqual(resolve_heavy_refresh_stride(0, max_packets=8), 1)
+        self.assertEqual(resolve_heavy_refresh_stride(20, max_packets=8), 2)
+        self.assertEqual(resolve_heavy_refresh_stride(40, max_packets=8), 4)
+
+    def test_sanitize_curve_data_for_display_rejects_fully_nonfinite_series(self):
+        xs, ys = sanitize_curve_data_for_display(
+            np.asarray([np.nan, np.nan], dtype=np.float64),
+            np.asarray([np.nan, np.nan], dtype=np.float32),
+        )
+
+        self.assertEqual(xs.shape, (0,))
+        self.assertEqual(ys.shape, (0,))
+
+    def test_sanitize_image_for_display_rejects_fully_nonfinite_frame(self):
+        data = sanitize_image_for_display(np.full((4, 4), np.nan, dtype=np.float32))
+
+        self.assertIsNone(data)
+
+    def test_set_image_rect_if_ready_skips_uninitialized_image(self):
+        class _DummyImageItem:
+            def __init__(self):
+                self.calls = 0
+
+            def width(self):
+                return None
+
+            def height(self):
+                return None
+
+            def setRect(self, *_args):
+                self.calls += 1
+
+        image_item = _DummyImageItem()
+        applied = set_image_rect_if_ready(image_item, 0.0, 0.0, 10.0, 1.0)
+
+        self.assertFalse(applied)
+        self.assertEqual(image_item.calls, 0)
+
+    def test_resolve_axis_bounds_rejects_nonfinite_ranges(self):
+        bounds = resolve_axis_bounds(np.asarray([np.nan, np.inf], dtype=np.float64))
+
+        self.assertIsNone(bounds)
+
+    def test_resolve_distance_cut_start_interpolates_with_percent(self):
+        axis = np.asarray([0.0, 1.0, 2.0, 3.0], dtype=np.float64)
+        cut_start = resolve_distance_cut_start(axis, 50.0)
+
+        self.assertIsNotNone(cut_start)
+        self.assertAlmostEqual(float(cut_start), 1.5, places=6)
+
+    def test_apply_distance_cut_to_axis_keeps_farthest_point_for_extreme_cut(self):
+        axis = np.asarray([0.0, 1.0, 2.0, 3.0], dtype=np.float64)
+        cut_axis, keep_mask = apply_distance_cut_to_axis(axis, 10.0)
+
+        self.assertEqual(cut_axis.shape, (1,))
+        self.assertEqual(keep_mask.shape, axis.shape)
+        self.assertTrue(bool(keep_mask[-1]))
+        self.assertAlmostEqual(float(cut_axis[0]), 3.0, places=6)
+
+    def test_resolve_initial_window_size_stays_within_small_screen(self):
+        width, height = resolve_initial_window_size(800, 480)
+
+        self.assertLessEqual(width, 800)
+        self.assertLessEqual(height, 480)
+        self.assertGreaterEqual(width, 640)
+        self.assertGreaterEqual(height, 420)
+
+    def test_build_main_window_layout_uses_splitter_and_scroll_area(self):
+        os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
+        try:
+            from PyQt5 import QtCore, QtWidgets
+        except Exception as exc:  # pragma: no cover - environment-dependent
+            self.skipTest(f"Qt unavailable: {exc}")
+
+        app = QtWidgets.QApplication.instance() or QtWidgets.QApplication([])
+        main_window = QtWidgets.QWidget()
+        try:
+            _layout, splitter, _plot_layout, settings_widget, settings_layout, settings_scroll = build_main_window_layout(
+                QtCore,
+                QtWidgets,
+                main_window,
+            )
+            self.assertIsInstance(splitter, QtWidgets.QSplitter)
+            self.assertIsInstance(settings_scroll, QtWidgets.QScrollArea)
+            self.assertIs(settings_scroll.widget(), settings_widget)
+            self.assertIsInstance(settings_layout, QtWidgets.QVBoxLayout)
+        finally:
+            main_window.close()
+
     def test_background_subtracted_bscan_levels_ignore_zero_floor(self):
         disp_fft_lin = np.zeros((4, 8), dtype=np.float32)
         disp_fft_lin[1, 2:6] = np.asarray([0.05, 0.1, 0.5, 2.0], dtype=np.float32)
@@ -247,6 +525,147 @@ class ProcessingTests(unittest.TestCase):
         self.assertTrue(np.allclose(spectrum[zero_mask], 0.0))
         self.assertTrue(np.any(np.abs(spectrum[pos_idx]) > 0.0))
 
+    def test_positive_only_exact_spectrum_uses_direct_index_insertion_without_window(self):
+        sweep = np.asarray([1.0, 2.0, 3.0], dtype=np.float32)
+        freqs = np.asarray([4.0, 5.0, 6.0], dtype=np.float64)
+        spectrum = build_positive_only_exact_centered_ifft_spectrum(sweep, freqs)
+
+        self.assertIsNotNone(spectrum)
+        df = (6.0 - 4.0) / 2.0
+        f_shift = np.arange(-6.0, 6.0 + (0.5 * df), df, dtype=np.float64)
+        idx = np.round((freqs - f_shift[0]) / df).astype(np.int64)
+        zero_mask = (f_shift > -6.0) & (f_shift < 4.0)
+
+        self.assertEqual(int(spectrum.size), int(f_shift.size))
+        self.assertTrue(np.allclose(spectrum[zero_mask], 0.0))
+        self.assertTrue(np.allclose(spectrum[idx], sweep))
+
+    def test_complex_symmetric_ifft_spectrum_uses_conjugate_mirror(self):
+        sweep = np.exp(1j * np.linspace(0.0, np.pi, 128)).astype(np.complex64)
+        freqs = np.linspace(4.0, 10.0, 128, dtype=np.float64)
+        spectrum = build_symmetric_ifft_spectrum(sweep, freqs, fft_len=FFT_LEN)
+
+        self.assertIsNotNone(spectrum)
+        freq_axis = np.linspace(-10.0, 10.0, FFT_LEN, dtype=np.float64)
+        neg_idx_all = np.flatnonzero(freq_axis <= (-4.0))
+        pos_idx_all = np.flatnonzero(freq_axis >= 4.0)
+        band_len = int(min(neg_idx_all.size, pos_idx_all.size))
+        neg_idx = neg_idx_all[:band_len]
+        pos_idx = pos_idx_all[-band_len:]
+
+        self.assertTrue(np.iscomplexobj(spectrum))
+        self.assertTrue(np.allclose(spectrum[neg_idx], np.conj(spectrum[pos_idx][::-1])))
+
+    def test_compute_fft_helpers_accept_complex_input(self):
+        sweep = np.exp(1j * np.linspace(0.0, 2.0 * np.pi, 128)).astype(np.complex64)
+        freqs = np.linspace(3.3, 14.3, 128, dtype=np.float64)
+        complex_row = compute_fft_complex_row(sweep, freqs, 513, mode="positive_only")
+        mag = compute_fft_mag_row(sweep, freqs, 513, mode="positive_only")
+        row = compute_fft_row(sweep, freqs, 513, mode="positive_only")
+
+        self.assertEqual(complex_row.shape, (513,))
+        self.assertTrue(np.iscomplexobj(complex_row))
+        self.assertEqual(mag.shape, (513,))
+        self.assertEqual(row.shape, (513,))
+        self.assertTrue(np.allclose(mag, np.abs(complex_row), equal_nan=True))
+        self.assertTrue(np.any(np.isfinite(mag)))
+        self.assertTrue(np.any(np.isfinite(row)))
+
+    def test_compute_fft_complex_row_positive_only_exact_matches_manual_ifftshift_ifft(self):
+        sweep = np.asarray([1.0 + 1.0j, 2.0 + 0.0j, 3.0 - 1.0j], dtype=np.complex64)
+        freqs = np.asarray([4.0, 5.0, 6.0], dtype=np.float64)
+        bins = 16
+        row = compute_fft_complex_row(sweep, freqs, bins, mode="positive_only_exact")
+
+        df = (6.0 - 4.0) / 2.0
+        f_shift = np.arange(-6.0, 6.0 + (0.5 * df), df, dtype=np.float64)
+        manual_shift = np.zeros((f_shift.size,), dtype=np.complex64)
+        idx = np.round((freqs - f_shift[0]) / df).astype(np.int64)
+        manual_shift[idx] = sweep
+        manual_ifft = np.fft.ifft(np.fft.ifftshift(manual_shift))
+        expected = np.full((bins,), np.nan + 0j, dtype=np.complex64)
+        expected[: manual_ifft.size] = np.asarray(manual_ifft, dtype=np.complex64)
+
+        self.assertEqual(row.shape, (bins,))
+        self.assertTrue(np.allclose(row, expected, equal_nan=True))
+
+    def test_positive_only_exact_distance_axis_uses_exact_grid_geometry(self):
+        freqs = np.asarray([4.0, 5.0, 6.0], dtype=np.float64)
+        bins = 8
+        axis = compute_distance_axis(freqs, bins, mode="positive_only_exact")
+
+        # With a small bins budget the exact-mode grid is downsampled so
+        # internal IFFT length does not exceed visible bins.
+        df_hz = 2e9
+        n_shift = int(np.arange(-6.0, 6.0 + 1.0, 2.0, dtype=np.float64).size)
+        expected_step = C_M_S / (2.0 * n_shift * df_hz)
+        expected = np.arange(bins, dtype=np.float64) * expected_step
+
+        self.assertEqual(axis.shape, (bins,))
+        self.assertTrue(np.allclose(axis, expected))
+
+    def test_positive_only_exact_mode_remains_stable_when_input_points_double(self):
+        bins = FFT_LEN // 2 + 1
+        tau_s = 45e-9
+
+        freqs_400 = np.linspace(3.3, 14.3, 400, dtype=np.float64)
+        freqs_800 = np.linspace(3.3, 14.3, 800, dtype=np.float64)
+        sweep_400 = np.exp(-1j * 2.0 * np.pi * freqs_400 * 1e9 * tau_s).astype(np.complex64)
+        sweep_800 = np.exp(-1j * 2.0 * np.pi * freqs_800 * 1e9 * tau_s).astype(np.complex64)
+
+        mag_400 = compute_fft_mag_row(sweep_400, freqs_400, bins, mode="positive_only_exact")
+        mag_800 = compute_fft_mag_row(sweep_800, freqs_800, bins, mode="positive_only_exact")
+
+        self.assertEqual(mag_400.shape, mag_800.shape)
+        finite = np.isfinite(mag_400) & np.isfinite(mag_800)
+        self.assertGreater(int(np.count_nonzero(finite)), int(0.95 * bins))
+
+        idx_400 = int(np.nanargmax(mag_400))
+        idx_800 = int(np.nanargmax(mag_800))
+        peak_400 = float(np.nanmax(mag_400))
+        peak_800 = float(np.nanmax(mag_800))
+
+        self.assertLess(abs(idx_400 - idx_800), 64)
+        self.assertGreater(idx_400, 8)
+        self.assertGreater(idx_800, 8)
+        self.assertLess(idx_400, bins - 8)
+        self.assertLess(idx_800, bins - 8)
+        self.assertGreater(peak_400, 0.05)
+        self.assertGreater(peak_800, 0.05)
+
+    def test_resolve_visible_fft_curves_handles_complex_mode(self):
+        complex_row = np.asarray([1.0 + 2.0j, -3.0 + 4.0j], dtype=np.complex64)
+        mag = np.abs(complex_row).astype(np.float32)
+
+        abs_curve, real_curve, imag_curve = resolve_visible_fft_curves(
+            complex_row,
+            mag,
+            complex_mode=True,
+            show_abs=True,
+            show_real=False,
+            show_imag=True,
+        )
+
+        self.assertTrue(np.allclose(abs_curve, mag))
+        self.assertIsNone(real_curve)
+        self.assertTrue(np.allclose(imag_curve, np.asarray([2.0, 4.0], dtype=np.float32)))
+
+    def test_resolve_visible_fft_curves_preserves_legacy_abs_mode(self):
+        mag = np.asarray([1.0, 2.0, 3.0], dtype=np.float32)
+
+        abs_curve, real_curve, imag_curve = resolve_visible_fft_curves(
+            None,
+            mag,
+            complex_mode=False,
+            show_abs=True,
+            show_real=True,
+            show_imag=True,
+        )
+
+        self.assertTrue(np.allclose(abs_curve, mag))
+        self.assertIsNone(real_curve)
+        self.assertIsNone(imag_curve)
+
     def test_symmetric_distance_axis_uses_windowed_frequency_bounds(self):
         freqs = np.linspace(4.0, 10.0, 128, dtype=np.float64)
         axis = compute_distance_axis(freqs, 513, mode="symmetric")

tests/test_ring_buffer.py

@@ -2,7 +2,10 @@ from __future__ import annotations
 
 import numpy as np
 import unittest
+import warnings
+from unittest.mock import patch
 
+from rfg_adc_plotter.processing.fft import compute_fft_mag_row
 from rfg_adc_plotter.state.ring_buffer import RingBuffer
 
 
@@ -40,6 +43,22 @@ class RingBufferTests(unittest.TestCase):
         self.assertIsNotNone(ring.last_fft_db)
         self.assertEqual(ring.last_fft_db.shape, (ring.fft_bins,))
 
+    def test_ring_buffer_can_return_decimated_display_raw(self):
+        ring = RingBuffer(max_sweeps=3)
+        sweep_a = np.linspace(0.0, 1.0, 4096, dtype=np.float32)
+        sweep_b = np.linspace(1.0, 2.0, 4096, dtype=np.float32)
+        sweep_c = np.linspace(2.0, 3.0, 4096, dtype=np.float32)
+        freqs = np.linspace(3.3, 14.3, 4096, dtype=np.float64)
+        ring.push(sweep_a, freqs)
+        ring.push(sweep_b, freqs)
+        ring.push(sweep_c, freqs)
+
+        raw = ring.get_display_raw_decimated(256)
+
+        self.assertEqual(raw.shape, (256, 3))
+        self.assertAlmostEqual(float(raw[0, -1]), float(sweep_c[0]), places=6)
+        self.assertAlmostEqual(float(raw[-1, -1]), float(sweep_c[-1]), places=6)
+
     def test_ring_buffer_can_switch_fft_mode_and_rebuild_fft_rows(self):
         ring = RingBuffer(max_sweeps=2)
         sweep = np.linspace(0.0, 1.0, 64, dtype=np.float32)
@@ -72,6 +91,34 @@ class RingBufferTests(unittest.TestCase):
         self.assertEqual(ring.last_fft_db.shape, (ring.fft_bins,))
         self.assertIsNotNone(ring.distance_axis)
 
+    def test_ring_buffer_can_switch_to_positive_only_exact_fft_mode(self):
+        ring = RingBuffer(max_sweeps=2)
+        sweep = np.linspace(0.0, 1.0, 64, dtype=np.float32)
+        freqs = np.linspace(3.3, 14.3, 64, dtype=np.float64)
+        ring.push(sweep, freqs)
+
+        changed = ring.set_fft_mode("positive_only_exact")
+
+        self.assertTrue(changed)
+        self.assertEqual(ring.fft_mode, "positive_only_exact")
+        self.assertIsNotNone(ring.last_fft_db)
+        self.assertEqual(ring.last_fft_db.shape, (ring.fft_bins,))
+        self.assertIsNotNone(ring.distance_axis)
+
+    def test_ring_buffer_rebuilds_fft_from_complex_input(self):
+        ring = RingBuffer(max_sweeps=2)
+        freqs = np.linspace(3.3, 14.3, 64, dtype=np.float64)
+        complex_input = np.exp(1j * np.linspace(0.0, 2.0 * np.pi, 64)).astype(np.complex64)
+        display_sweep = np.abs(complex_input).astype(np.float32)
+        ring.push(display_sweep, freqs, fft_input=complex_input)
+
+        ring.set_fft_mode("direct")
+
+        expected = compute_fft_mag_row(complex_input, freqs, ring.fft_bins, mode="direct")
+        self.assertTrue(np.allclose(ring.get_last_fft_linear(), expected))
+        self.assertFalse(np.iscomplexobj(ring.get_display_fft_linear()))
+        self.assertTrue(np.allclose(ring.get_display_raw()[: display_sweep.size, -1], display_sweep))
+
     def test_ring_buffer_reset_clears_cached_history(self):
         ring = RingBuffer(max_sweeps=2)
         ring.push(np.linspace(0.0, 1.0, 64, dtype=np.float32), np.linspace(4.0, 10.0, 64))
@@ -85,6 +132,45 @@ class RingBufferTests(unittest.TestCase):
         self.assertEqual(ring.width, 0)
         self.assertEqual(ring.head, 0)
 
+    def test_ring_buffer_push_ignores_all_nan_fft_without_runtime_warning(self):
+        ring = RingBuffer(max_sweeps=2)
+        freqs = np.linspace(3.3, 14.3, 64, dtype=np.float64)
+        ring.push(np.linspace(0.0, 1.0, 64, dtype=np.float32), freqs)
+        fft_before = ring.last_fft_db.copy()
+        y_min_before = ring.y_min_fft
+        y_max_before = ring.y_max_fft
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("error", RuntimeWarning)
+            with patch(
+                "rfg_adc_plotter.state.ring_buffer.compute_fft_mag_row",
+                return_value=np.full((ring.fft_bins,), np.nan, dtype=np.float32),
+            ):
+                ring.push(np.linspace(1.0, 2.0, 64, dtype=np.float32), freqs)
+
+        self.assertFalse(ring.last_push_fft_valid)
+        self.assertTrue(np.allclose(ring.last_fft_db, fft_before))
+        self.assertEqual(ring.y_min_fft, y_min_before)
+        self.assertEqual(ring.y_max_fft, y_max_before)
+
+    def test_ring_buffer_set_fft_mode_ignores_all_nan_rebuild_without_runtime_warning(self):
+        ring = RingBuffer(max_sweeps=2)
+        freqs = np.linspace(3.3, 14.3, 64, dtype=np.float64)
+        ring.push(np.linspace(0.0, 1.0, 64, dtype=np.float32), freqs)
+        fft_before = ring.last_fft_db.copy()
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("error", RuntimeWarning)
+            with patch(
+                "rfg_adc_plotter.state.ring_buffer.compute_fft_mag_row",
+                return_value=np.full((ring.fft_bins,), np.nan, dtype=np.float32),
+            ):
+                ring.set_fft_mode("direct")
+
+        self.assertFalse(ring.last_push_fft_valid)
+        self.assertTrue(np.allclose(ring.last_fft_db, fft_before))
+        self.assertEqual(ring.fft_mode, "direct")
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_sweep_parser_core.py

@@ -5,6 +5,7 @@ import unittest
 
 from rfg_adc_plotter.io.sweep_parser_core import (
     AsciiSweepParser,
+    ComplexAsciiSweepParser,
     LegacyBinaryParser,
     LogScale16BitX2BinaryParser,
     LogScaleBinaryParser32,
@@ -71,6 +72,32 @@ def _pack_log16_point(step: int, avg1: int, avg2: int) -> bytes:
     )
 
 
+def _pack_tty_start() -> bytes:
+    return b"".join([_u16le(0x000A), _u16le(0xFFFF), _u16le(0xFFFF), _u16le(0xFFFF)])
+
+
+def _pack_tty_point(step: int, ch1: int, ch2: int) -> bytes:
+    return b"".join(
+        [
+            _u16le(0x000A),
+            _u16le(step),
+            _u16le(ch1),
+            _u16le(ch2),
+        ]
+    )
+
+
+def _pack_logdet_point(step: int, value: int) -> bytes:
+    return b"".join(
+        [
+            _u16le(0x001A),
+            _u16le(step),
+            _u16le(value),
+            _u16le(0x0000),
+        ]
+    )
+
+
 class SweepParserCoreTests(unittest.TestCase):
     def test_ascii_parser_emits_start_and_points(self):
         parser = AsciiSweepParser()
@@ -96,6 +123,161 @@ class SweepParserCoreTests(unittest.TestCase):
         self.assertEqual(events[1].x, 1)
         self.assertEqual(events[1].y, -2.0)
 
+    def test_legacy_binary_parser_detects_new_sweep_on_step_reset(self):
+        parser = LegacyBinaryParser()
+        stream = b"".join(
+            [
+                _pack_legacy_point(3, 1, -2),
+                _pack_legacy_point(3, 2, -3),
+                _pack_legacy_point(3, 1, -4),
+            ]
+        )
+        events = parser.feed(stream)
+        self.assertIsInstance(events[0], PointEvent)
+        self.assertIsInstance(events[1], PointEvent)
+        self.assertIsInstance(events[2], StartEvent)
+        self.assertEqual(events[2].ch, 3)
+        self.assertIsInstance(events[3], PointEvent)
+        self.assertEqual(events[3].x, 1)
+        self.assertEqual(events[3].y, -4.0)
+
+    def test_legacy_binary_parser_accepts_tty_ch1_ch2_stream(self):
+        parser = LegacyBinaryParser()
+        stream = b"".join(
+            [
+                _pack_tty_start(),
+                _pack_tty_point(1, 100, 90),
+                _pack_tty_point(2, 120, 95),
+            ]
+        )
+
+        events = parser.feed(stream)
+
+        self.assertIsInstance(events[0], StartEvent)
+        self.assertEqual(events[0].ch, 0)
+        self.assertIsInstance(events[1], PointEvent)
+        self.assertEqual(events[1].x, 1)
+        self.assertEqual(events[1].y, 18100.0)
+        self.assertEqual(events[1].aux, (100.0, 90.0))
+        self.assertEqual(events[1].signal_kind, "bin_iq")
+        self.assertIsInstance(events[2], PointEvent)
+        self.assertEqual(events[2].x, 2)
+        self.assertEqual(events[2].y, 23425.0)
+        self.assertEqual(events[2].aux, (120.0, 95.0))
+        self.assertEqual(events[2].signal_kind, "bin_iq")
+
+    def test_legacy_binary_parser_detects_new_tty_sweep_on_step_reset(self):
+        parser = LegacyBinaryParser()
+        stream = b"".join(
+            [
+                _pack_tty_start(),
+                _pack_tty_point(1, 100, 90),
+                _pack_tty_point(2, 110, 95),
+                _pack_tty_point(1, 120, 80),
+            ]
+        )
+
+        events = parser.feed(stream)
+
+        self.assertIsInstance(events[0], StartEvent)
+        self.assertIsInstance(events[1], PointEvent)
+        self.assertIsInstance(events[2], PointEvent)
+        self.assertIsInstance(events[3], StartEvent)
+        self.assertEqual(events[3].ch, 0)
+        self.assertIsInstance(events[4], PointEvent)
+        self.assertEqual(events[4].x, 1)
+        self.assertEqual(events[4].aux, (120.0, 80.0))
+        self.assertEqual(events[4].signal_kind, "bin_iq")
+
+    def test_legacy_binary_parser_tty_mode_does_not_flip_to_legacy_on_ch2_low_byte_0x0a(self):
+        parser = LegacyBinaryParser()
+        stream = b"".join(
+            [
+                _pack_tty_start(),
+                _pack_tty_point(1, 100, 0x040A),  # low byte is 0x0A: used to be misparsed as legacy
+                _pack_tty_point(2, 120, 0x0410),
+            ]
+        )
+
+        events = parser.feed(stream)
+
+        self.assertEqual(len(events), 3)
+        self.assertIsInstance(events[0], StartEvent)
+        self.assertEqual(events[0].ch, 0)
+
+        self.assertIsInstance(events[1], PointEvent)
+        self.assertEqual(events[1].ch, 0)
+        self.assertEqual(events[1].x, 1)
+        self.assertEqual(events[1].aux, (100.0, 1034.0))
+        self.assertEqual(events[1].y, 1079156.0)
+
+        self.assertIsInstance(events[2], PointEvent)
+        self.assertEqual(events[2].ch, 0)
+        self.assertEqual(events[2].x, 2)
+        self.assertEqual(events[2].aux, (120.0, 1040.0))
+        self.assertEqual(events[2].y, 1096000.0)
+
+    def test_legacy_binary_parser_accepts_logdet_stream(self):
+        parser = LegacyBinaryParser()
+        stream = b"".join(
+            [
+                _pack_logdet_point(1, 0x0F77),
+                _pack_logdet_point(2, 0xF234),
+            ]
+        )
+
+        events = parser.feed(stream)
+
+        self.assertEqual(len(events), 2)
+        self.assertIsInstance(events[0], PointEvent)
+        self.assertEqual(events[0].x, 1)
+        self.assertEqual(events[0].y, 3959.0)
+        self.assertIsNone(events[0].aux)
+        self.assertEqual(events[0].signal_kind, "bin_logdet")
+        self.assertIsInstance(events[1], PointEvent)
+        self.assertEqual(events[1].x, 2)
+        self.assertEqual(events[1].y, -3532.0)
+        self.assertEqual(events[1].signal_kind, "bin_logdet")
+
+    def test_legacy_binary_parser_splits_packet_on_bin_signal_kind_change(self):
+        parser = LegacyBinaryParser()
+        stream = b"".join(
+            [
+                _pack_tty_start(),
+                _pack_tty_point(1, 100, 90),
+                _pack_tty_point(2, 110, 95),
+                _pack_logdet_point(3, 0x0F77),
+            ]
+        )
+
+        events = parser.feed(stream)
+
+        self.assertIsInstance(events[0], StartEvent)
+        self.assertEqual(events[0].signal_kind, "bin_iq")
+        self.assertIsInstance(events[1], PointEvent)
+        self.assertEqual(events[1].signal_kind, "bin_iq")
+        self.assertIsInstance(events[2], PointEvent)
+        self.assertEqual(events[2].signal_kind, "bin_iq")
+        self.assertIsInstance(events[3], StartEvent)
+        self.assertEqual(events[3].signal_kind, "bin_logdet")
+        self.assertIsInstance(events[4], PointEvent)
+        self.assertEqual(events[4].x, 3)
+        self.assertEqual(events[4].signal_kind, "bin_logdet")
+
+    def test_complex_ascii_parser_detects_new_sweep_on_step_reset(self):
+        parser = ComplexAsciiSweepParser()
+        events = parser.feed(b"0 3 4\n1 5 12\n0 8 15\n")
+
+        self.assertIsInstance(events[0], PointEvent)
+        self.assertEqual(events[0].x, 0)
+        self.assertEqual(events[0].y, 5.0)
+        self.assertEqual(events[0].aux, (3.0, 4.0))
+        self.assertIsInstance(events[1], PointEvent)
+        self.assertEqual(events[1].y, 13.0)
+        self.assertIsInstance(events[2], StartEvent)
+        self.assertIsInstance(events[3], PointEvent)
+        self.assertEqual(events[3].aux, (8.0, 15.0))
+
     def test_logscale_32_parser_keeps_channel_and_aux_values(self):
         parser = LogScaleBinaryParser32()
         stream = _pack_log_start(5) + _pack_log_point(7, 1500, 700, ch=5)
@@ -108,6 +290,24 @@ class SweepParserCoreTests(unittest.TestCase):
         self.assertAlmostEqual(events[1].y, log_pair_to_sweep(1500, 700), places=6)
         self.assertEqual(events[1].aux, (1500.0, 700.0))
 
+    def test_logscale_32_parser_detects_new_sweep_on_step_reset(self):
+        parser = LogScaleBinaryParser32()
+        stream = b"".join(
+            [
+                _pack_log_point(1, 1500, 700, ch=5),
+                _pack_log_point(2, 1400, 650, ch=5),
+                _pack_log_point(1, 1300, 600, ch=5),
+            ]
+        )
+        events = parser.feed(stream)
+        self.assertIsInstance(events[0], PointEvent)
+        self.assertIsInstance(events[1], PointEvent)
+        self.assertIsInstance(events[2], StartEvent)
+        self.assertEqual(events[2].ch, 5)
+        self.assertIsInstance(events[3], PointEvent)
+        self.assertEqual(events[3].x, 1)
+        self.assertAlmostEqual(events[3].y, log_pair_to_sweep(1300, 600), places=6)
+
     def test_log_pair_to_sweep_is_order_independent(self):
         self.assertAlmostEqual(log_pair_to_sweep(1500, 700), log_pair_to_sweep(700, 1500), places=6)
 
@@ -119,8 +319,29 @@ class SweepParserCoreTests(unittest.TestCase):
         self.assertEqual(events[0].ch, 2)
         self.assertIsInstance(events[1], PointEvent)
         self.assertEqual(events[1].ch, 2)
+        self.assertAlmostEqual(events[1].y, math.hypot(100.0, 90.0), places=6)
         self.assertEqual(events[1].aux, (100.0, 90.0))
 
+    def test_logscale_16bit_parser_detects_new_sweep_on_step_reset(self):
+        parser = LogScale16BitX2BinaryParser()
+        stream = b"".join(
+            [
+                _pack_log16_start(2),
+                _pack_log16_point(1, 100, 90),
+                _pack_log16_point(2, 110, 95),
+                _pack_log16_point(1, 120, 80),
+            ]
+        )
+        events = parser.feed(stream)
+        self.assertIsInstance(events[0], StartEvent)
+        self.assertIsInstance(events[1], PointEvent)
+        self.assertIsInstance(events[2], PointEvent)
+        self.assertIsInstance(events[3], StartEvent)
+        self.assertEqual(events[3].ch, 2)
+        self.assertIsInstance(events[4], PointEvent)
+        self.assertEqual(events[4].x, 1)
+        self.assertAlmostEqual(events[4].y, math.hypot(120.0, 80.0), places=6)
+
     def test_parser_test_stream_parser_recovers_point_after_single_separator(self):
         parser = ParserTestStreamParser()
         stream = b"".join(
@@ -140,20 +361,56 @@ class SweepParserCoreTests(unittest.TestCase):
         self.assertIsInstance(events[1], PointEvent)
         self.assertEqual(events[1].ch, 4)
         self.assertEqual(events[1].x, 1)
-        self.assertTrue(math.isfinite(events[1].y))
+        self.assertAlmostEqual(events[1].y, math.hypot(100.0, 90.0), places=6)
+        self.assertEqual(events[1].aux, (100.0, 90.0))
 
     def test_sweep_assembler_builds_aux_curves_without_inversion(self):
         assembler = SweepAssembler(fancy=False, apply_inversion=False)
-        self.assertIsNone(assembler.consume(StartEvent(ch=1)))
-        assembler.consume(PointEvent(ch=1, x=1, y=10.0, aux=(100.0, 90.0)))
-        assembler.consume(PointEvent(ch=1, x=2, y=20.0, aux=(110.0, 95.0)))
+        self.assertIsNone(assembler.consume(StartEvent(ch=1, signal_kind="bin_iq")))
+        assembler.consume(PointEvent(ch=1, x=1, y=10.0, aux=(100.0, 90.0), signal_kind="bin_iq"))
+        assembler.consume(PointEvent(ch=1, x=2, y=20.0, aux=(110.0, 95.0), signal_kind="bin_iq"))
         sweep, info, aux = assembler.finalize_current()
         self.assertEqual(sweep.shape[0], 3)
         self.assertEqual(info["ch"], 1)
+        self.assertEqual(info["signal_kind"], "bin_iq")
         self.assertIsNotNone(aux)
         self.assertEqual(aux[0][1], 100.0)
         self.assertEqual(aux[1][2], 95.0)
 
+    def test_sweep_assembler_splits_packet_on_channel_switch(self):
+        assembler = SweepAssembler(fancy=False, apply_inversion=False)
+        self.assertIsNone(assembler.consume(PointEvent(ch=1, x=1, y=10.0)))
+        packet = assembler.consume(PointEvent(ch=2, x=1, y=20.0))
+        self.assertIsNotNone(packet)
+
+        sweep_1, info_1, aux_1 = packet
+        self.assertIsNone(aux_1)
+        self.assertEqual(info_1["ch"], 1)
+        self.assertEqual(info_1["chs"], [1])
+        self.assertAlmostEqual(float(sweep_1[1]), 10.0, places=6)
+
+        sweep_2, info_2, aux_2 = assembler.finalize_current()
+        self.assertIsNone(aux_2)
+        self.assertEqual(info_2["ch"], 2)
+        self.assertEqual(info_2["chs"], [2])
+        self.assertAlmostEqual(float(sweep_2[1]), 20.0, places=6)
+
+    def test_sweep_assembler_splits_packet_on_signal_kind_switch(self):
+        assembler = SweepAssembler(fancy=False, apply_inversion=False)
+        self.assertIsNone(assembler.consume(PointEvent(ch=0, x=1, y=10.0, signal_kind="bin_iq")))
+        packet = assembler.consume(PointEvent(ch=0, x=1, y=20.0, signal_kind="bin_logdet"))
+        self.assertIsNotNone(packet)
+
+        sweep_1, info_1, aux_1 = packet
+        self.assertIsNone(aux_1)
+        self.assertEqual(info_1["signal_kind"], "bin_iq")
+        self.assertAlmostEqual(float(sweep_1[1]), 10.0, places=6)
+
+        sweep_2, info_2, aux_2 = assembler.finalize_current()
+        self.assertIsNone(aux_2)
+        self.assertEqual(info_2["signal_kind"], "bin_logdet")
+        self.assertAlmostEqual(float(sweep_2[1]), 20.0, places=6)
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_sweep_reader.py

@@ -0,0 +1,262 @@
+from __future__ import annotations
+
+import contextlib
+import io
+import threading
+import time
+import unittest
+from queue import Queue
+from unittest.mock import patch
+
+from rfg_adc_plotter.io import sweep_reader as sweep_reader_module
+from rfg_adc_plotter.io.sweep_reader import SweepReader, _PARSER_16_BIT_X2_PROBE_BYTES
+
+
+def _u16le(word: int) -> bytes:
+    value = int(word) & 0xFFFF
+    return bytes((value & 0xFF, (value >> 8) & 0xFF))
+
+
+def _pack_legacy_point(ch: int, step: int, value_i32: int) -> bytes:
+    value = int(value_i32) & 0xFFFF_FFFF
+    return b"".join(
+        [
+            _u16le(step),
+            _u16le((value >> 16) & 0xFFFF),
+            _u16le(value & 0xFFFF),
+            bytes((0x0A, int(ch) & 0xFF)),
+        ]
+    )
+
+
+def _pack_log16_start(ch: int) -> bytes:
+    return b"\xff\xff" * 3 + bytes((0x0A, int(ch) & 0xFF))
+
+
+def _pack_log16_point(step: int, real: int, imag: int) -> bytes:
+    return b"".join(
+        [
+            _u16le(step),
+            _u16le(real),
+            _u16le(imag),
+            _u16le(0xFFFF),
+        ]
+    )
+
+
+def _pack_tty_start() -> bytes:
+    return b"".join(
+        [
+            _u16le(0x000A),
+            _u16le(0xFFFF),
+            _u16le(0xFFFF),
+            _u16le(0xFFFF),
+        ]
+    )
+
+
+def _pack_tty_point(step: int, ch1: int, ch2: int) -> bytes:
+    return b"".join(
+        [
+            _u16le(0x000A),
+            _u16le(step),
+            _u16le(ch1),
+            _u16le(ch2),
+        ]
+    )
+
+
+def _pack_logdet_point(step: int, value: int) -> bytes:
+    return b"".join(
+        [
+            _u16le(0x001A),
+            _u16le(step),
+            _u16le(value),
+            _u16le(0x0000),
+        ]
+    )
+
+
+def _chunk_bytes(data: bytes, size: int = 4096) -> list[bytes]:
+    return [data[idx : idx + size] for idx in range(0, len(data), size)]
+
+
+class _FakeSerialLineSource:
+    def __init__(self, path: str, baud: int, timeout: float = 1.0):
+        self.path = path
+        self.baud = baud
+        self.timeout = timeout
+        self._using = "fake"
+
+    def close(self) -> None:
+        pass
+
+
+class _FakeChunkReader:
+    payload_chunks: list[bytes] = []
+
+    def __init__(self, src):
+        self._src = src
+        self._chunks = list(type(self).payload_chunks)
+
+    def read_available(self) -> bytes:
+        if self._chunks:
+            return self._chunks.pop(0)
+        return b""
+
+
+class SweepReaderTests(unittest.TestCase):
+    def _start_reader(self, payload: bytes, **reader_kwargs):
+        queue: Queue = Queue()
+        stop_event = threading.Event()
+        stderr = io.StringIO()
+        _FakeChunkReader.payload_chunks = _chunk_bytes(payload)
+        reader = SweepReader(
+            "/tmp/fake-tty",
+            115200,
+            queue,
+            stop_event,
+            **reader_kwargs,
+        )
+        stack = contextlib.ExitStack()
+        stack.enter_context(patch.object(sweep_reader_module, "SerialLineSource", _FakeSerialLineSource))
+        stack.enter_context(patch.object(sweep_reader_module, "SerialChunkReader", _FakeChunkReader))
+        stack.enter_context(contextlib.redirect_stderr(stderr))
+        reader.start()
+        return stack, reader, queue, stop_event, stderr
+
+    def test_parser_16_bit_x2_falls_back_to_legacy_stream(self):
+        payload = bytearray()
+        while len(payload) < (_PARSER_16_BIT_X2_PROBE_BYTES + 24):
+            payload += _pack_legacy_point(3, 1, -2)
+            payload += _pack_legacy_point(3, 2, -3)
+            payload += _pack_legacy_point(3, 1, -4)
+
+        stack, reader, queue, stop_event, stderr = self._start_reader(bytes(payload), parser_16_bit_x2=True)
+        try:
+            sweep, info, aux = queue.get(timeout=2.0)
+            self.assertEqual(info["ch"], 3)
+            self.assertIsNone(aux)
+            self.assertGreaterEqual(sweep.shape[0], 3)
+            self.assertIn("fallback -> legacy", stderr.getvalue())
+        finally:
+            stop_event.set()
+            reader.join(timeout=1.0)
+            stack.close()
+
+    def test_parser_16_bit_x2_falls_back_to_tty_ch1_ch2_stream(self):
+        payload = bytearray()
+        while len(payload) < (_PARSER_16_BIT_X2_PROBE_BYTES + 24):
+            payload += _pack_tty_start()
+            payload += _pack_tty_point(1, 100, 90)
+            payload += _pack_tty_point(2, 120, 95)
+            payload += _pack_tty_point(1, 80, 70)
+
+        stack, reader, queue, stop_event, stderr = self._start_reader(bytes(payload), parser_16_bit_x2=True)
+        try:
+            sweep, info, aux = queue.get(timeout=2.0)
+            self.assertEqual(info["ch"], 0)
+            self.assertIsNotNone(aux)
+            self.assertGreaterEqual(sweep.shape[0], 3)
+            self.assertAlmostEqual(float(sweep[1]), 18100.0, places=6)
+            self.assertAlmostEqual(float(sweep[2]), 23425.0, places=6)
+            self.assertIn("fallback -> legacy", stderr.getvalue())
+        finally:
+            stop_event.set()
+            reader.join(timeout=1.0)
+            stack.close()
+
+    def test_parser_16_bit_x2_keeps_true_complex_stream(self):
+        payload = b"".join(
+            [
+                _pack_log16_start(2),
+                _pack_log16_point(1, 3, 4),
+                _pack_log16_point(2, 5, 12),
+                _pack_log16_point(1, 8, 15),
+            ]
+        )
+
+        stack, reader, queue, stop_event, stderr = self._start_reader(payload, parser_16_bit_x2=True)
+        try:
+            sweep, info, aux = queue.get(timeout=1.0)
+            self.assertEqual(info["ch"], 2)
+            self.assertIsNotNone(aux)
+            self.assertAlmostEqual(float(sweep[1]), 5.0, places=6)
+            self.assertAlmostEqual(float(sweep[2]), 13.0, places=6)
+            self.assertNotIn("fallback -> legacy", stderr.getvalue())
+        finally:
+            stop_event.set()
+            reader.join(timeout=1.0)
+            stack.close()
+
+    def test_parser_16_bit_x2_falls_back_to_logdet_1a00_stream(self):
+        payload = bytearray()
+        while len(payload) < (_PARSER_16_BIT_X2_PROBE_BYTES + 24):
+            payload += _pack_logdet_point(1, 0x0F77)
+            payload += _pack_logdet_point(2, 0x0FCB)
+            payload += _pack_logdet_point(1, 0x0F88)
+
+        stack, reader, queue, stop_event, stderr = self._start_reader(bytes(payload), parser_16_bit_x2=True)
+        try:
+            sweep, info, aux = queue.get(timeout=2.0)
+            self.assertEqual(info["signal_kind"], "bin_logdet")
+            self.assertIsNone(aux)
+            self.assertGreaterEqual(sweep.shape[0], 3)
+            self.assertAlmostEqual(float(sweep[1]), 3959.0, places=6)
+            self.assertIn("fallback -> legacy", stderr.getvalue())
+        finally:
+            stop_event.set()
+            reader.join(timeout=1.0)
+            stack.close()
+
+    def test_parser_16_bit_x2_probe_inconclusive_logs_hint(self):
+        payload = b"\x00" * (_PARSER_16_BIT_X2_PROBE_BYTES + 128)
+
+        stack, reader, queue, stop_event, stderr = self._start_reader(payload, parser_16_bit_x2=True)
+        try:
+            deadline = time.time() + 1.5
+            logs = ""
+            while time.time() < deadline:
+                logs = stderr.getvalue()
+                if "probe inconclusive" in logs:
+                    break
+                time.sleep(0.02)
+            self.assertTrue(queue.empty())
+            self.assertIn("probe inconclusive", logs)
+            self.assertIn("try --bin", logs)
+        finally:
+            stop_event.set()
+            reader.join(timeout=1.0)
+            stack.close()
+
+    def test_reader_logs_no_input_warning_when_source_is_idle(self):
+        with patch.object(sweep_reader_module, "_NO_INPUT_WARN_INTERVAL_S", 0.02), patch.object(
+            sweep_reader_module, "_NO_PACKET_WARN_INTERVAL_S", 0.02
+        ):
+            stack, reader, _queue, stop_event, stderr = self._start_reader(b"", parser_16_bit_x2=False)
+            try:
+                time.sleep(0.12)
+                logs = stderr.getvalue()
+                self.assertIn("no input bytes", logs)
+                self.assertIn("no sweep packets", logs)
+            finally:
+                stop_event.set()
+                reader.join(timeout=1.0)
+                stack.close()
+
+    def test_reader_join_does_not_raise_when_stopped(self):
+        stack, reader, _queue, stop_event, _stderr = self._start_reader(b"", parser_16_bit_x2=True)
+        try:
+            time.sleep(0.01)
+            stop_event.set()
+            reader.join(timeout=1.0)
+            self.assertFalse(reader.is_alive())
+        finally:
+            stop_event.set()
+            if reader.is_alive():
+                reader.join(timeout=1.0)
+            stack.close()
+
+
+if __name__ == "__main__":
+    unittest.main()