From 428ba2a9e0a19a0af2879c856c110abeea7f44c1 Mon Sep 17 00:00:00 2001 From: awe Date: Mon, 17 Aug 2026 14:06:49 +0300 Subject: [PATCH] new mode --bin24 --- rfg_adc_plotter/cli.py | 13 ++ rfg_adc_plotter/gui/pyqtgraph_backend.py | 39 +++-- rfg_adc_plotter/io/sweep_parser_core.py | 146 ++++++++++++++++++ rfg_adc_plotter/io/sweep_reader.py | 7 + tests/test_processing.py | 21 +++ tests/test_sweep_parser_core.py | 180 +++++++++++++++++++++++ 6 files changed, 392 insertions(+), 14 deletions(-) diff --git a/rfg_adc_plotter/cli.py b/rfg_adc_plotter/cli.py index ebf877c..90f70a3 100644 --- a/rfg_adc_plotter/cli.py +++ b/rfg_adc_plotter/cli.py @@ -92,6 +92,19 @@ def build_parser() -> argparse.ArgumentParser: "Для 0x001A: code_i16 переводится в В, raw = V, FFT вход = exp(V)" ), ) + parser.add_argument( + "--bin24", + dest="bin24_mode", + action="store_true", + help=( + "Как --bin, но каждый отсчет I/Q — знаковый 24-битный (3 байта), little-endian. " + "Запись 10 байт: marker(u16), step(u16), ch1_i24(3б), ch2_i24(3б). " + "Маркеры те же: 0x000A (точка), 0x00A8 (secondary), " + "0x00A3/0x00A4 (DO1 LOW/HIGH tagged). " + "Математика как у --bin: сырая кривая = ch1^2+ch2^2, FFT вход = ch1+i*ch2. " + "Полная шкала для пересчета в В = 2^23-1." + ), + ) parser.add_argument( "--tty-range-v", type=float, diff --git a/rfg_adc_plotter/gui/pyqtgraph_backend.py b/rfg_adc_plotter/gui/pyqtgraph_backend.py index 277c33f..afc7b5c 100644 --- a/rfg_adc_plotter/gui/pyqtgraph_backend.py +++ b/rfg_adc_plotter/gui/pyqtgraph_backend.py @@ -75,6 +75,7 @@ DEFAULT_MAIN_WINDOW_HEIGHT = 680 MIN_MAIN_WINDOW_WIDTH = 640 MIN_MAIN_WINDOW_HEIGHT = 420 TTY_CODE_SCALE_DENOM = 32767.0 +TTY_CODE_SCALE_DENOM_24 = 8388607.0 TTY_RANGE_DEFAULT_V = 5.0 TTY_RANGE_MIN_V = 1e-6 TTY_RANGE_MAX_V = 1_000_000.0 @@ -640,13 +641,18 @@ def sanitize_tty_voltage_range(range_v: float, default: float = TTY_RANGE_DEFAUL return float(np.clip(abs(value), TTY_RANGE_MIN_V, TTY_RANGE_MAX_V)) -def convert_tty_i16_to_voltage(codes: np.ndarray, range_v: float) -> np.ndarray: - """Convert signed tty int16 code array to clipped voltage values in ``[-range_v, +range_v]``.""" +def convert_tty_i16_to_voltage( + codes: np.ndarray, range_v: float, *, denom: float = TTY_CODE_SCALE_DENOM +) -> np.ndarray: + """Convert signed tty code array to clipped voltage values in ``[-range_v, +range_v]``. + + ``denom`` is the code full-scale magnitude (2^15-1 for int16, 2^23-1 for int24). + """ code_arr = np.asarray(codes, dtype=np.float32).reshape(-1) if code_arr.size <= 0: return np.zeros((0,), dtype=np.float32) range_abs_v = sanitize_tty_voltage_range(range_v) - scale_v = range_abs_v / float(TTY_CODE_SCALE_DENOM) + scale_v = range_abs_v / float(denom) volt = code_arr * np.float32(scale_v) return np.clip(volt, -range_abs_v, range_abs_v).astype(np.float32, copy=False) @@ -656,9 +662,10 @@ def build_logdet_voltage_fft_input( range_v: float, *, exp_input_limit: float = LOGDET_EXP_INPUT_LIMIT, + denom: float = TTY_CODE_SCALE_DENOM, ) -> Tuple[np.ndarray, np.ndarray]: """Convert 1a00 log-detector codes to raw volts and a real FFT input ``exp(V)``.""" - volts = convert_tty_i16_to_voltage(codes, range_v) + volts = convert_tty_i16_to_voltage(codes, range_v, denom=denom) if volts.size <= 0: empty = np.zeros((0,), dtype=np.float32) return empty, empty @@ -889,15 +896,18 @@ def run_pyqtgraph(args) -> None: peak_calibrate_mode = bool(getattr(args, "calibrate", False)) peak_search_enabled = bool(getattr(args, "peak_search", False)) bin_mode = bool(getattr(args, "bin_mode", False)) + bin24_mode = bool(getattr(args, "bin24_mode", False)) tty_range_v = sanitize_tty_voltage_range(getattr(args, "tty_range_v", TTY_RANGE_DEFAULT_V)) + tty_code_denom = TTY_CODE_SCALE_DENOM_24 if bin24_mode else TTY_CODE_SCALE_DENOM complex_ascii_mode = bool(getattr(args, "parser_complex_ascii", False)) complex_sweep_mode = bool( bin_mode + or bin24_mode or complex_ascii_mode or getattr(args, "parser_16_bit_x2", False) or getattr(args, "parser_test", False) ) - bin_iq_power_mode = bool(bin_mode) + bin_iq_power_mode = bool(bin_mode or bin24_mode) if not sys.platform.startswith("win"): display_name = os.environ.get("DISPLAY") or os.environ.get("WAYLAND_DISPLAY") if not display_name: @@ -921,6 +931,7 @@ def run_pyqtgraph(args) -> None: stop_event, fancy=bool(args.fancy), bin_mode=bin_mode, + bin24_mode=bin24_mode, logscale=bool(args.logscale), parser_16_bit_x2=bool(args.parser_16_bit_x2), parser_test=bool(args.parser_test), @@ -1904,8 +1915,8 @@ def run_pyqtgraph(args) -> None: width = min(code_1_arr.size, code_2_arr.size) if width <= 0: return False - ch_1_v = convert_tty_i16_to_voltage(code_1_arr[:width], tty_range_v) - ch_2_v = convert_tty_i16_to_voltage(code_2_arr[:width], tty_range_v) + ch_1_v = convert_tty_i16_to_voltage(code_1_arr[:width], tty_range_v, denom=tty_code_denom) + ch_2_v = convert_tty_i16_to_voltage(code_2_arr[:width], tty_range_v, denom=tty_code_denom) runtime.full_do1_tagged_raw_low = None runtime.full_do1_tagged_raw_high = None runtime.full_do1_tagged_aux_low = None @@ -1936,10 +1947,10 @@ def run_pyqtgraph(args) -> None: if width <= 0: return False - low_ch_1_v = convert_tty_i16_to_voltage(low_code_1_arr[:width], tty_range_v) - low_ch_2_v = convert_tty_i16_to_voltage(low_code_2_arr[:width], tty_range_v) - high_ch_1_v = convert_tty_i16_to_voltage(high_code_1_arr[:width], tty_range_v) - high_ch_2_v = convert_tty_i16_to_voltage(high_code_2_arr[:width], tty_range_v) + low_ch_1_v = convert_tty_i16_to_voltage(low_code_1_arr[:width], tty_range_v, denom=tty_code_denom) + low_ch_2_v = convert_tty_i16_to_voltage(low_code_2_arr[:width], tty_range_v, denom=tty_code_denom) + high_ch_1_v = convert_tty_i16_to_voltage(high_code_1_arr[:width], tty_range_v, denom=tty_code_denom) + high_ch_2_v = convert_tty_i16_to_voltage(high_code_2_arr[:width], tty_range_v, denom=tty_code_denom) low_ch_1_v_f64 = low_ch_1_v.astype(np.float64, copy=False) low_ch_2_v_f64 = low_ch_2_v.astype(np.float64, copy=False) @@ -1964,7 +1975,7 @@ def run_pyqtgraph(args) -> None: code_arr = np.asarray(runtime.full_current_sweep_codes, dtype=np.float32).reshape(-1) if code_arr.size <= 0: return False - sweep_raw_v, fft_input = build_logdet_voltage_fft_input(code_arr, tty_range_v) + sweep_raw_v, fft_input = build_logdet_voltage_fft_input(code_arr, tty_range_v, denom=tty_code_denom) runtime.full_do1_tagged_raw_low = None runtime.full_do1_tagged_raw_high = None runtime.full_do1_tagged_aux_low = None @@ -3089,13 +3100,13 @@ def run_pyqtgraph(args) -> None: calibrate_freqs({"F": base_freqs, "I": sec_ch1})["I"], dtype=np.float32, ) - runtime.full_secondary_ch1 = convert_tty_i16_to_voltage(sec_ch1_calibrated, tty_range_v) + runtime.full_secondary_ch1 = convert_tty_i16_to_voltage(sec_ch1_calibrated, tty_range_v, denom=tty_code_denom) if sec_ch2 is not None: sec_ch2_calibrated = np.asarray( calibrate_freqs({"F": base_freqs, "I": sec_ch2})["I"], dtype=np.float32, ) - runtime.full_secondary_ch2 = convert_tty_i16_to_voltage(sec_ch2_calibrated, tty_range_v) + runtime.full_secondary_ch2 = convert_tty_i16_to_voltage(sec_ch2_calibrated, tty_range_v, denom=tty_code_denom) if runtime.full_secondary_ch1 is not None and runtime.full_secondary_ch2 is not None: w = min(runtime.full_secondary_ch1.size, runtime.full_secondary_ch2.size) v1 = runtime.full_secondary_ch1[:w] diff --git a/rfg_adc_plotter/io/sweep_parser_core.py b/rfg_adc_plotter/io/sweep_parser_core.py index 28a195b..6baf4e7 100644 --- a/rfg_adc_plotter/io/sweep_parser_core.py +++ b/rfg_adc_plotter/io/sweep_parser_core.py @@ -31,6 +31,10 @@ def u16_to_i16(value: int) -> int: return value - 0x1_0000 if (value & 0x8000) else value +def u24_to_i24(value: int) -> int: + return value - 0x100_0000 if (value & 0x80_0000) else value + + def log_value_to_linear(value: int) -> float: exponent = max(-LOG_EXP_LIMIT, min(LOG_EXP_LIMIT, float(value) * LOG_SCALER)) return float(LOG_BASE ** exponent) @@ -606,6 +610,148 @@ class LegacyBinaryParser: return events +class LegacyBinaryParser24(LegacyBinaryParser): + """Byte-resynchronizing parser for 10-byte tty records with 24-bit I/Q values. + + Same protocol as ``LegacyBinaryParser``'s tty family (markers 0x000A primary, + 0x00A8 secondary, 0x00A3/0x00A4 DO1 low/high tagged), but each channel value + is a signed 24-bit little-endian integer, so a record is 10 bytes: + ``marker(u16) | step(u16) | ch1_i24(3B) | ch2_i24(3B)``. The legacy 8-byte + (byte6==0x0A) and logdet (0x001A) record shapes do not exist here and are + omitted so int24 payload bytes cannot be mistaken for them. + """ + + RECORD_SIZE = 10 + + def __init__(self): + super().__init__(batch_events=False) + + @staticmethod + def _i24_at(buf: bytearray, offset: int) -> int: + u = int(buf[offset]) | (int(buf[offset + 1]) << 8) | (int(buf[offset + 2]) << 16) + return u24_to_i24(u) + + def _try_emit_tty_batch(self, events: List[ParserEvent], *, require_not_legacy: bool) -> bool: + # int24 has no native numpy dtype; the scalar per-record path handles it. + return False + + def _emit_tty_point24(self, events: List[ParserEvent], step: int, ch_1: int, ch_2: int) -> None: + self._prepare_bin_point(events, step=int(step), signal_kind="bin_iq") + ch_1 = int(ch_1) + ch_2 = int(ch_2) + 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_tty_tagged_point24( + self, + events: List[ParserEvent], + step: int, + ch_1: int, + ch_2: int, + do1_level: Do1Level, + ) -> None: + self._prepare_bin_point( + events, + step=int(step), + signal_kind="bin_iq_do1_tagged", + do1_level=do1_level, + ) + ch_1 = int(ch_1) + ch_2 = int(ch_2) + 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_do1_tagged", + do1_level=do1_level, + ) + ) + + def _emit_secondary_point24(self, events: List[ParserEvent], step: int, ch_1: int, ch_2: int) -> None: + self._mode = "bin" + self._current_signal_kind = self._current_signal_kind or "bin_iq" + ch_1 = int(ch_1) + ch_2 = int(ch_2) + events.append( + PointEvent( + ch=0, + x=int(step), + y=0.0, + aux=(float(ch_1), float(ch_2)), + signal_kind="bin_iq", + is_secondary=True, + ) + ) + + def feed(self, data: bytes) -> List[ParserEvent]: + if data: + self._buf += data + events: List[ParserEvent] = [] + while len(self._buf) >= self.RECORD_SIZE: + w0 = self._u16_at(self._buf, 0) + w1 = self._u16_at(self._buf, 2) + + is_tty_start = w0 == 0x000A and all(b == 0xFF for b in self._buf[2:10]) + is_tty_point = w0 == 0x000A and w1 != 0xFFFF + is_tty_tagged_low_point = w0 == 0x00A3 and w1 != 0xFFFF + is_tty_tagged_high_point = w0 == 0x00A4 and w1 != 0xFFFF + is_secondary_point = w0 == 0x00A8 and w1 != 0xFFFF + + if is_tty_start: + self._emit_tty_start(events) + del self._buf[:10] + continue + if is_tty_point: + self._emit_tty_point24( + events, + step=int(w1), + ch_1=self._i24_at(self._buf, 4), + ch_2=self._i24_at(self._buf, 7), + ) + del self._buf[:10] + continue + if is_tty_tagged_low_point: + self._emit_tty_tagged_point24( + events, + step=int(w1), + ch_1=self._i24_at(self._buf, 4), + ch_2=self._i24_at(self._buf, 7), + do1_level="low", + ) + del self._buf[:10] + continue + if is_tty_tagged_high_point: + self._emit_tty_tagged_point24( + events, + step=int(w1), + ch_1=self._i24_at(self._buf, 4), + ch_2=self._i24_at(self._buf, 7), + do1_level="high", + ) + del self._buf[:10] + continue + if is_secondary_point: + self._emit_secondary_point24( + events, + step=int(w1), + ch_1=self._i24_at(self._buf, 4), + ch_2=self._i24_at(self._buf, 7), + ) + del self._buf[:10] + continue + del self._buf[:1] + return events + + class LogScaleBinaryParser32: """Byte-resynchronizing parser for 32-bit logscale pair records.""" diff --git a/rfg_adc_plotter/io/sweep_reader.py b/rfg_adc_plotter/io/sweep_reader.py index bedca8e..d301e97 100644 --- a/rfg_adc_plotter/io/sweep_reader.py +++ b/rfg_adc_plotter/io/sweep_reader.py @@ -12,6 +12,7 @@ from rfg_adc_plotter.io.sweep_parser_core import ( AsciiSweepParser, ComplexAsciiSweepParser, LegacyBinaryParser, + LegacyBinaryParser24, LogScale16BitX2BinaryParser, LogScaleBinaryParser32, ParserTestStreamParser, @@ -113,6 +114,7 @@ class SweepReader(threading.Thread): stop_event: threading.Event, fancy: bool = False, bin_mode: bool = False, + bin24_mode: bool = False, logscale: bool = False, parser_16_bit_x2: bool = False, parser_test: bool = False, @@ -125,6 +127,7 @@ class SweepReader(threading.Thread): self._stop_event = stop_event self._fancy = bool(fancy) self._bin_mode = bool(bin_mode) + self._bin24_mode = bool(bin24_mode) self._logscale = bool(logscale) self._parser_16_bit_x2 = bool(parser_16_bit_x2) self._parser_test = bool(parser_test) @@ -143,6 +146,8 @@ class SweepReader(threading.Thread): return "parser_16_bit_x2" if self._logscale: return "logscale_32" + if self._bin24_mode: + return "legacy_10byte_i24" if self._bin_mode: return "legacy_8byte" return "ascii" @@ -156,6 +161,8 @@ class SweepReader(threading.Thread): return LogScale16BitX2BinaryParser(), SweepAssembler(fancy=self._fancy, apply_inversion=False) if self._logscale: return LogScaleBinaryParser32(), SweepAssembler(fancy=self._fancy, apply_inversion=False) + if self._bin24_mode: + return LegacyBinaryParser24(), SweepAssembler(fancy=self._fancy, apply_inversion=True) if self._bin_mode: return LegacyBinaryParser(batch_events=True), SweepAssembler(fancy=self._fancy, apply_inversion=True) return AsciiSweepParser(), SweepAssembler(fancy=self._fancy, apply_inversion=True) diff --git a/tests/test_processing.py b/tests/test_processing.py index 6bf77d1..20b7725 100644 --- a/tests/test_processing.py +++ b/tests/test_processing.py @@ -13,6 +13,7 @@ from rfg_adc_plotter.gui.pyqtgraph_backend import ( build_logdet_voltage_fft_input, build_main_window_layout, coalesce_packets_for_ui, + TTY_CODE_SCALE_DENOM_24, compute_background_subtracted_bscan_levels, compute_aux_phase_curve, compute_do1_tagged_aggregate, @@ -87,6 +88,26 @@ class ProcessingTests(unittest.TestCase): self.assertTrue(np.all(volts >= -5.0)) self.assertTrue(np.all(volts <= 5.0)) + def test_convert_tty_i16_to_voltage_int24_denominator_maps_full_scale(self): + full_scale = float(TTY_CODE_SCALE_DENOM_24) + codes = np.asarray([-full_scale - 1.0, 0.0, full_scale], dtype=np.float32) + volts = convert_tty_i16_to_voltage(codes, 5.0, denom=TTY_CODE_SCALE_DENOM_24) + + self.assertAlmostEqual(float(volts[0]), -5.0, places=4) + self.assertAlmostEqual(float(volts[1]), 0.0, places=6) + self.assertAlmostEqual(float(volts[2]), 5.0, places=4) + + # A quarter-scale 24-bit code must yield ~1.25 V with the int24 denom, but the + # default int16 denom over-scales it ~256x and clips to the ±5 V range. This + # guards that --bin24 uses the wider full-scale rather than the int16 one. + quarter = np.asarray([full_scale / 4.0], dtype=np.float32) + self.assertAlmostEqual( + float(convert_tty_i16_to_voltage(quarter, 5.0, denom=TTY_CODE_SCALE_DENOM_24)[0]), + 1.25, + places=4, + ) + self.assertAlmostEqual(float(convert_tty_i16_to_voltage(quarter, 5.0)[0]), 5.0, places=6) + 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) diff --git a/tests/test_sweep_parser_core.py b/tests/test_sweep_parser_core.py index 212c7d0..59dd16f 100644 --- a/tests/test_sweep_parser_core.py +++ b/tests/test_sweep_parser_core.py @@ -10,6 +10,7 @@ from rfg_adc_plotter.io.sweep_parser_core import ( BatchPointEvent, ComplexAsciiSweepParser, LegacyBinaryParser, + LegacyBinaryParser24, LogScale16BitX2BinaryParser, LogScaleBinaryParser32, ParserTestStreamParser, @@ -17,6 +18,7 @@ from rfg_adc_plotter.io.sweep_parser_core import ( StartEvent, SweepAssembler, log_pair_to_sweep, + u24_to_i24, ) @@ -120,6 +122,35 @@ def _pack_logdet_point(step: int, value: int) -> bytes: ) +def _i24le(value: int) -> bytes: + v = int(value) & 0xFFFFFF + return bytes((v & 0xFF, (v >> 8) & 0xFF, (v >> 16) & 0xFF)) + + +def _pack_tty24_start() -> bytes: + return _u16le(0x000A) + b"\xff" * 8 + + +def _pack_tty24_point(step: int, ch1: int, ch2: int) -> bytes: + return _u16le(0x000A) + _u16le(step) + _i24le(ch1) + _i24le(ch2) + + +def _pack_tty24_tagged_point(marker_word0: int, step: int, ch1: int, ch2: int) -> bytes: + return _u16le(marker_word0) + _u16le(step) + _i24le(ch1) + _i24le(ch2) + + +def _pack_tty24_tagged_low_point(step: int, ch1: int, ch2: int) -> bytes: + return _pack_tty24_tagged_point(0x00A3, step, ch1, ch2) + + +def _pack_tty24_tagged_high_point(step: int, ch1: int, ch2: int) -> bytes: + return _pack_tty24_tagged_point(0x00A4, step, ch1, ch2) + + +def _pack_tty24_secondary_point(step: int, ch1: int, ch2: int) -> bytes: + return _u16le(0x00A8) + _u16le(step) + _i24le(ch1) + _i24le(ch2) + + class SweepParserCoreTests(unittest.TestCase): def test_ascii_parser_emits_start_and_points(self): parser = AsciiSweepParser() @@ -697,5 +728,154 @@ class SweepParserCoreTests(unittest.TestCase): self.assertNotIn("_secondary_payload", info) +class LegacyBinaryParser24Tests(unittest.TestCase): + def test_u24_to_i24_sign_extension_boundaries(self): + self.assertEqual(u24_to_i24(0x000000), 0) + self.assertEqual(u24_to_i24(0x000001), 1) + self.assertEqual(u24_to_i24(0x7FFFFF), 8388607) + self.assertEqual(u24_to_i24(0x800000), -8388608) + self.assertEqual(u24_to_i24(0xFFFFFF), -1) + + def test_accepts_tty24_ch1_ch2_stream(self): + parser = LegacyBinaryParser24() + stream = b"".join( + [ + _pack_tty24_start(), + _pack_tty24_point(1, 100, 90), + _pack_tty24_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.assertEqual(events[2].x, 2) + self.assertEqual(events[2].y, 23425.0) + self.assertEqual(events[2].aux, (120.0, 95.0)) + + def test_decodes_negative_int24_values(self): + parser = LegacyBinaryParser24() + stream = _pack_tty24_start() + _pack_tty24_point(1, -8388608, 8388607) + + events = parser.feed(stream) + + point = events[1] + self.assertEqual(point.aux, (-8388608.0, 8388607.0)) + self.assertEqual(point.y, float(8388608 ** 2 + 8388607 ** 2)) + + def test_never_emits_batch_event(self): + parser = LegacyBinaryParser24() + stream = _pack_tty24_start() + b"".join( + _pack_tty24_point(i, i * 10, -i * 5) for i in range(1, 6) + ) + + events = parser.feed(stream) + + self.assertFalse(any(isinstance(e, BatchPointEvent) for e in events)) + self.assertFalse(parser._try_emit_tty_batch(events, require_not_legacy=False)) + + def test_resynchronizes_after_garbage(self): + parser = LegacyBinaryParser24() + stream = b"\x11\x22\x33" + _pack_tty24_start() + _pack_tty24_point(5, 7, -3) + + events = parser.feed(stream) + + points = [e for e in events if isinstance(e, PointEvent)] + self.assertEqual(len(points), 1) + self.assertEqual(points[0].x, 5) + self.assertEqual(points[0].aux, (7.0, -3.0)) + + def test_detects_new_sweep_on_step_reset(self): + parser = LegacyBinaryParser24() + stream = _pack_tty24_start() + b"".join( + [ + _pack_tty24_point(5, 1, 1), + _pack_tty24_point(6, 2, 2), + _pack_tty24_point(1, 3, 3), + ] + ) + + events = parser.feed(stream) + + start_count = sum(1 for e in events if isinstance(e, StartEvent)) + self.assertEqual(start_count, 2) + + def test_do1_tagged_routing(self): + parser = LegacyBinaryParser24() + stream = b"".join( + [ + _pack_tty24_tagged_low_point(1, 10, 20), + _pack_tty24_tagged_high_point(1, 30, 40), + ] + ) + + events = parser.feed(stream) + points = [e for e in events if isinstance(e, PointEvent)] + + self.assertEqual(points[0].signal_kind, "bin_iq_do1_tagged") + self.assertEqual(points[0].do1_level, "low") + self.assertEqual(points[0].aux, (10.0, 20.0)) + self.assertEqual(points[1].do1_level, "high") + self.assertEqual(points[1].aux, (30.0, 40.0)) + + def test_secondary_point(self): + parser = LegacyBinaryParser24() + stream = _pack_tty24_start() + _pack_tty24_secondary_point(3, -51, -36) + + events = parser.feed(stream) + secondary = [e for e in events if isinstance(e, PointEvent) and e.is_secondary] + + self.assertEqual(len(secondary), 1) + self.assertEqual(secondary[0].x, 3) + self.assertEqual(secondary[0].y, 0.0) + self.assertEqual(secondary[0].aux, (-51.0, -36.0)) + + def test_assembles_sweep_packet(self): + parser = LegacyBinaryParser24() + assembler = SweepAssembler(fancy=False, apply_inversion=True) + stream = _pack_tty24_start() + b"".join( + _pack_tty24_point(i, i * 10, -i * 5) for i in range(1, 20) + ) + + packets = [] + for event in parser.feed(stream): + packet = assembler.consume(event) + if packet is not None: + packets.append(packet) + final = assembler.finalize_current() + if final is not None: + packets.append(final) + + self.assertTrue(packets) + sweep, info, _aux = packets[-1] + self.assertEqual(info["signal_kind"], "bin_iq") + self.assertTrue(np.isfinite(info["mean"])) + + +class SweepReaderBin24Tests(unittest.TestCase): + def test_build_parser_selects_int24_parser(self): + from queue import Queue + import threading + + from rfg_adc_plotter.io.sweep_reader import SweepReader + + reader = SweepReader( + "unused", + 115200, + Queue(), + threading.Event(), + bin24_mode=True, + ) + parser, _assembler = reader._build_parser() + self.assertIsInstance(parser, LegacyBinaryParser24) + self.assertEqual(reader._resolve_parser_mode_label(), "legacy_10byte_i24") + + if __name__ == "__main__": unittest.main()