ChTheo/RFG_stm32_ADC_receiver_GUI
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

new mode --bin24

Browse Source
This commit is contained in:
awe
2026-08-17 14:06:49 +03:00
parent 2cf7543bbe
commit 428ba2a9e0
6 changed files with 392 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rfg_adc_plotter/cli.py
+13
View File
@@ -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,
rfg_adc_plotter/gui/pyqtgraph_backend.py
+25 -14
View File
@@ -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]
rfg_adc_plotter/io/sweep_parser_core.py
+146
View File
@@ -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."""
rfg_adc_plotter/io/sweep_reader.py
+7
View File
@@ -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)
tests/test_processing.py
+21
View File
@@ -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)
tests/test_sweep_parser_core.py
+180
View File
@@ -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()