implemented new parser: _run_parser_test_stream, activates via --parser_test

RFG_ADC_dataplotter.py

@@ -38,7 +38,7 @@ SWEEP_FREQ_MIN_GHZ = 3.3
 SWEEP_FREQ_MAX_GHZ = 14.3
 LOG_BASE = 10.0
 LOG_SCALER = 0.001  # int32 значения приходят в fixed-point лог-шкале с шагом 1e-3
-LOG_POSTSCALER = 1000
+LOG_POSTSCALER = 1
 LOG_EXP_LIMIT = 300.0  # запас до переполнения float64 при возведении LOG_BASE в степень
 C_M_S = 299_792_458.0
 # Порог для инверсии сырых данных: если среднее значение свипа ниже порога —
@@ -570,6 +570,7 @@ class SerialLineSource:
 
     def __init__(self, path: str, baud: int, timeout: float = 1.0):
         self._pyserial = try_open_pyserial(path, baud, timeout)
+        #self._pyserial = None
         self._fdreader = None
         self._using = "pyserial" if self._pyserial is not None else "raw"
         if self._pyserial is None:
@@ -669,6 +670,7 @@ class SweepReader(threading.Thread):
         bin_mode: bool = False,
         logscale: bool = False,
         parser_16_bit_x2: bool = False,
+        parser_test: bool = False,
     ):
         super().__init__(daemon=True)
         self._port_path = port_path
@@ -680,6 +682,7 @@ class SweepReader(threading.Thread):
         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._max_width: int = 0
         self._sweep_idx: int = 0
         self._last_sweep_ts: Optional[float] = None
@@ -1141,7 +1144,7 @@ class SweepReader(threading.Thread):
             #print(words)
             # Бинарный logscale-протокол (16-bit x2):
             #   старт свипа: 0xFFFF, 0xFFFF, 0xFFFF, (ch<<8)|0x0A
-            #   точка: step, avg1_lo16, avg2_lo16, 0x000A
+            #   точка: step, avg1_lo16, avg2_lo16, 0xFFFF
             while len(words) >= 4:
                 w0 = int(words[0])
                 w1 = int(words[1])
@@ -1175,13 +1178,13 @@ class SweepReader(threading.Thread):
                     continue
 
                 if syncing:
-                    if w0 == 0x000A:
+                    if w0 == 0xFFFF:
                         syncing = False
                         expected_step = None
                     words.popleft()
                     continue
 
-                if w3 != 0x000A:
+                if w3 != 0xFFFF:
                     syncing = True
                     continue
 
@@ -1220,6 +1223,133 @@ class SweepReader(threading.Thread):
             apply_inversion=False,
         )
 
+    def _run_parser_test_stream(self, chunk_reader: SerialChunkReader):
+        """Тестовый парсер 16-bit x2:
+        - разделитель точки: одиночный 0xFFFF
+        - маркер старта свипа: серия 0xFFFF (>=2), затем (ch<<8)|0x0A
+        """
+        xs: list[int] = []
+        ys: list[float] = []
+        avg_1_vals: list[int] = []
+        avg_2_vals: list[int] = []
+        cur_channel: Optional[int] = None
+        cur_channels: set[int] = set()
+        expected_step: Optional[int] = None
+        in_sweep = False
+        point_buf: list[int] = []
+        ffff_run = 0
+        local_resync = False
+
+        def _finalize_sweep():
+            self._finalize_current(
+                xs,
+                ys,
+                cur_channels,
+                raw_curves=(avg_1_vals, avg_2_vals),
+                apply_inversion=False,
+            )
+            xs.clear()
+            ys.clear()
+            avg_1_vals.clear()
+            avg_2_vals.clear()
+            cur_channels.clear()
+
+        def _consume_point() -> bool:
+            nonlocal expected_step
+            if len(point_buf) != 3:
+                return False
+            step = int(point_buf[0])
+            if step <= 0:
+                return False
+            if expected_step is not None and step < expected_step:
+                return False
+            if cur_channel is not None:
+                cur_channels.add(cur_channel)
+            avg_1 = self._u16_to_i16(int(point_buf[1]))
+            avg_2 = self._u16_to_i16(int(point_buf[2]))
+            xs.append(step)
+            avg_1_vals.append(avg_1)
+            avg_2_vals.append(avg_2)
+            ys.append(_log_pair_to_sweep(avg_1, avg_2))
+            expected_step = step + 1
+            return True
+
+        buf = bytearray()
+        buf_pos = 0
+        while not self._stop.is_set():
+            data = chunk_reader.read_available()
+            if data:
+                buf += data
+            else:
+                time.sleep(0.000005)
+                continue
+
+            # Обрабатываем поток строго по одному слову (16 бит) за шаг.
+            while (buf_pos + 1) < len(buf):
+                w = int(buf[buf_pos]) | (int(buf[buf_pos + 1]) << 8)
+                buf_pos += 2
+
+                if w == 0xFFFF:
+                    ffff_run += 1
+                    continue
+
+                if ffff_run > 0:
+                    bad_point_on_this_delim = False
+                    if in_sweep and point_buf and (not local_resync):
+                        if not _consume_point():
+                            local_resync = True
+                            bad_point_on_this_delim = True
+                        point_buf.clear()
+
+                    if ffff_run >= 2:
+                        if (w & 0x00FF) == 0x000A:
+                            _finalize_sweep()
+                            cur_channel = (w >> 8) & 0x00FF
+                            cur_channels.add(cur_channel)
+                            in_sweep = True
+                            expected_step = 1
+                            local_resync = False
+                            point_buf.clear()
+                            ffff_run = 0
+                            continue
+                        # Некорректная серия FFFF: остаёмся в текущем свипе, ждём
+                        # следующего одиночного разделителя точки.
+                        if in_sweep:
+                            local_resync = True
+                        ffff_run = 0
+                        continue
+
+                    # ffff_run == 1: это просто разделитель точки
+                    if local_resync and (not bad_point_on_this_delim):
+                        # Нашли следующий одиночный разделитель: выходим из локального resync.
+                        local_resync = False
+                        point_buf.clear()
+                    ffff_run = 0
+
+                if in_sweep and (not local_resync):
+                    point_buf.append(w)
+                    if len(point_buf) > 3:
+                        point_buf.clear()
+                        local_resync = True
+
+            # Периодически уплотняем буфер, сохраняя возможный хвост из 1 байта.
+            if buf_pos >= 262144:
+                del buf[:buf_pos]
+                buf_pos = 0
+            # Аварийный лимит на остаток неразобранных данных.
+            if (len(buf) - buf_pos) > 1_000_000:
+                tail = buf[buf_pos:]
+                if len(tail) > 262144:
+                    tail = tail[-262144:]
+                buf = bytearray(tail)
+                buf_pos = 0
+
+        # Попробуем завершить последнюю точку, если поток закончился ровно на разделителе.
+        if in_sweep and (not local_resync) and ffff_run == 1 and point_buf:
+            _consume_point()
+            point_buf.clear()
+        _finalize_sweep()
+
     def run(self):
         try:
             self._src = SerialLineSource(self._port_path, self._baud, timeout=1.0)
@@ -1230,7 +1360,9 @@ class SweepReader(threading.Thread):
 
         try:
             chunk_reader = SerialChunkReader(self._src)
-            if self._parser_16_bit_x2:
+            if self._parser_test:
+                self._run_parser_test_stream(chunk_reader)
+            elif self._parser_16_bit_x2:
                 self._run_logscale_16_bit_x2_binary_stream(chunk_reader)
             elif self._logscale:
                 self._run_logscale_binary_stream(chunk_reader)
@@ -1325,7 +1457,15 @@ def main():
         action="store_true",
         help=(
             "Бинарный logscale-протокол c парой int16 (avg_1, avg_2): "
-            "старт 0xFFFF,0xFFFF,0xFFFF,(CH<<8)|0x0A; точка step,avg1_lo16,avg2_lo16,0x000A"
+            "старт 0xFFFF,0xFFFF,0xFFFF,(CH<<8)|0x0A; точка step,avg1_lo16,avg2_lo16,0xFFFF"
+        ),
+    )
+    parser.add_argument(
+        "--parser_test",
+        action="store_true",
+        help=(
+            "Тестовый парсер для формата 16-bit x2: "
+            "одиночный 0xFFFF завершает точку, серия 0xFFFF начинает новый свип"
         ),
     )
     parser.add_argument(
@@ -1374,6 +1514,7 @@ def main():
         bin_mode=bool(args.bin_mode),
         logscale=bool(args.logscale),
         parser_16_bit_x2=bool(args.parser_16_bit_x2),
+        parser_test=bool(args.parser_test),
     )
     reader.start()
 
@@ -2103,6 +2244,7 @@ def run_pyqtgraph(args):
         bin_mode=bool(args.bin_mode),
         logscale=bool(args.logscale),
         parser_16_bit_x2=bool(args.parser_16_bit_x2),
+        parser_test=bool(args.parser_test),
     )
     reader.start()