From e2d71662ff311519145e7d18615e7b91a41cb5ac Mon Sep 17 00:00:00 2001 From: jack Date: Mon, 6 Jul 2026 13:24:46 -0700 Subject: [PATCH] refactor: data_parser binary parsers use shared sensor_count_for() --- .../backend/controllers/device_controller.py | 2 +- src/pygui/serialcomm/data_parser.py | 40 +++++++++---------- tests/test_data_parser.py | 12 +++--- 3 files changed, 25 insertions(+), 29 deletions(-) diff --git a/src/pygui/backend/controllers/device_controller.py b/src/pygui/backend/controllers/device_controller.py index 248bbdb..3c1d209 100644 --- a/src/pygui/backend/controllers/device_controller.py +++ b/src/pygui/backend/controllers/device_controller.py @@ -24,7 +24,6 @@ from pygui.serialcomm.data_parser import ( remove_trailing_zeros, save_to_csv, ) -from pygui.serialcomm.device_service import DeviceService from pygui.serialcomm.serial_port import WaferInfo # import pygui.backend.wafer_map_item @@ -65,6 +64,7 @@ class DeviceController(QObject): super().__init__(parent) self._settings = settings self._data_dir = data_dir + from pygui.serialcomm.device_service import DeviceService self._service = DeviceService(settings) # Always start fresh — never restore connection status or logs from a prior session diff --git a/src/pygui/serialcomm/data_parser.py b/src/pygui/serialcomm/data_parser.py index db81964..3d37e79 100644 --- a/src/pygui/serialcomm/data_parser.py +++ b/src/pygui/serialcomm/data_parser.py @@ -12,13 +12,9 @@ import logging import os from typing import Optional -log = logging.getLogger(__name__) +from pygui.backend.wafer.family_spec import sensor_count_for -# Max DUTs (sensors) per row before overhead bytes are stripped -# P wafer: 244 valid readings per 256-block (12 overhead bytes) -# X wafer: 80 valid readings per 256-block (14 overhead bytes) -MAXDUT_P = 244 -MAXDUT_X = 80 +log = logging.getLogger(__name__) def csv_column_count(family_code: str) -> int: @@ -156,26 +152,27 @@ def _parse_p_binary(data_bytes: bytes) -> list[list[str]]: """Parse P-family (and A/B/C/D/E/F) binary data. Each block of 256 readings has 244 valid + 12 overhead. - Valid readings are chunked into rows of MAXDUT_P (244). + Valid readings are chunked into rows of the family's sensor count (244). """ + sensor_count = sensor_count_for("P") readings: list[str] = [] # Read 2 bytes at a time (UInt16 little-endian) - # Each 256-word block has MAXDUT_P valid readings (first N words), rest are overhead + # Each 256-word block has `sensor_count` valid readings (first N words), rest are overhead num_words = len(data_bytes) // 2 for i in range(num_words): value = int.from_bytes(data_bytes[i * 2 : i * 2 + 2], byteorder="little") - if i % 256 < MAXDUT_P: + if i % 256 < sensor_count: readings.append(f"{value:04X}") - # Chunk into rows of MAXDUT_P + # Chunk into rows of sensor_count result: list[list[str]] = [] idx = 0 - while idx + MAXDUT_P <= len(readings): - result.append(readings[idx : idx + MAXDUT_P]) - idx += MAXDUT_P + while idx + sensor_count <= len(readings): + result.append(readings[idx : idx + sensor_count]) + idx += sensor_count - log.info("Parsed P-family: %d rows × %d cols", len(result), MAXDUT_P) + log.info("Parsed P-family: %d rows × %d cols", len(result), sensor_count) return result @@ -183,24 +180,25 @@ def _parse_x_binary(data_bytes: bytes) -> list[list[str]]: """Parse X-family binary data. Each block of 256 readings has 80 valid + 14 overhead. - Valid readings are chunked into rows of MAXDUT_X (80). + Valid readings are chunked into rows of the family's sensor count (80). """ + sensor_count = sensor_count_for("X") readings: list[str] = [] num_words = len(data_bytes) // 2 for i in range(num_words): value = int.from_bytes(data_bytes[i * 2 : i * 2 + 2], byteorder="little") - if i % 256 < MAXDUT_X: + if i % 256 < sensor_count: readings.append(f"{value:04X}") - # Chunk into rows of MAXDUT_X + # Chunk into rows of sensor_count result: list[list[str]] = [] idx = 0 - while idx + MAXDUT_X <= len(readings): - result.append(readings[idx : idx + MAXDUT_X]) - idx += MAXDUT_X + while idx + sensor_count <= len(readings): + result.append(readings[idx : idx + sensor_count]) + idx += sensor_count - log.info("Parsed X-family: %d rows × %d cols", len(result), MAXDUT_X) + log.info("Parsed X-family: %d rows × %d cols", len(result), sensor_count) return result diff --git a/tests/test_data_parser.py b/tests/test_data_parser.py index 42909e9..c8336a7 100644 --- a/tests/test_data_parser.py +++ b/tests/test_data_parser.py @@ -8,8 +8,6 @@ from pygui.serialcomm.data_parser import ( remove_trailing_zeros, save_to_csv, _convert_hex_to_temp, - MAXDUT_P, - MAXDUT_X, ) # ── csv_column_count ────────────────────────────────────────────────────────── @@ -106,7 +104,7 @@ def _make_p_block(num_blocks: int = 1, value: int = 0x0100) -> bytes: data = bytearray() for _ in range(num_blocks): for i in range(256): - word = value if i < MAXDUT_P else 0 + word = value if i < 244 else 0 data += word.to_bytes(2, byteorder="little") return bytes(data) @@ -120,7 +118,7 @@ def _make_x_block(num_blocks: int = 1, value: int = 0x0100) -> bytes: data = bytearray() for _ in range(num_blocks): for i in range(256): - word = value if i < MAXDUT_X else 0 + word = value if i < 80 else 0 data += word.to_bytes(2, byteorder="little") return bytes(data) @@ -136,7 +134,7 @@ class TestParseBinaryData: data = _make_p_block(1) result = parse_binary_data(data, "P") assert result is not None - assert len(result[0]) == MAXDUT_P # 244 + assert len(result[0]) == 244 def test_p_family_two_blocks_returns_two_rows(self): data = _make_p_block(2) @@ -161,14 +159,14 @@ class TestParseBinaryData: data = _make_x_block(1) result = parse_binary_data(data, "X") assert result is not None - assert len(result[0]) == MAXDUT_X # 80 + assert len(result[0]) == 80 def test_a_family_uses_p_parser(self): # A/B/C/D/E/F all route to _parse_p_binary data = _make_p_block(1) result = parse_binary_data(data, "A") assert result is not None - assert len(result[0]) == MAXDUT_P + assert len(result[0]) == 244 def test_empty_bytes_returns_empty_list(self): result = parse_binary_data(b"", "P")