From cea4fb782e7c2f9883ed6a0e1798fc1a73365b13 Mon Sep 17 00:00:00 2001 From: jack Date: Mon, 15 Jun 2026 11:19:32 -0700 Subject: [PATCH] feat(wafer): introduce layout metadata modeling and shape resolution helpers - Define WaferLayout subclass to wrap sensor lists with shape and size attributes. - Add resolve_shape_and_size resolver to infer wafer shape/size from filename prefix or sensor count. - Update Sensor dataclass to support dynamic side alignments and coordinates offsets. - Expose waferShape and waferSize properties from SessionController for QML data-bindings. --- .../backend/controllers/session_controller.py | 128 ++++++++++++++++-- src/pygui/backend/data/csv_file_metadata.py | 7 +- src/pygui/backend/data/file_browser.py | 18 +-- src/pygui/backend/wafer/wafer_layouts.py | 98 +++++++++++++- src/pygui/backend/wafer/zwafer_models.py | 3 + 5 files changed, 217 insertions(+), 37 deletions(-) diff --git a/src/pygui/backend/controllers/session_controller.py b/src/pygui/backend/controllers/session_controller.py index 36425a1..549a2c4 100644 --- a/src/pygui/backend/controllers/session_controller.py +++ b/src/pygui/backend/controllers/session_controller.py @@ -70,7 +70,14 @@ class SessionController(QObject): def frameTotal(self) -> int: return self._player.total @Property(str, notify=stateChanged) - def state(self) -> str: return self._last.state if self._last else "idle" + def state(self) -> str: + if self._last: + return self._last.state + # IF we dont have data yet, but the reader is running -> Streaming + if self._mode == MODE_LIVE and self._reader is not None: + return "streaming" + + return "idle" @Property(bool, notify=recordingChanged) def recording(self) -> bool: return self._recorder.is_recording @@ -88,11 +95,30 @@ class SessionController(QObject): "value": round(v, 2), "band": band, "index": i}) return out - # ---- WaferMapItem bindings (split view of sensorDots) ---- @Property("QVariantList", notify=sensorsChanged) def sensorLayout(self) -> list: - """[{label, x, y}] for WaferMapItem.sensors.""" - return [{"label": s.label, "x": s.x, "y": s.y} for s in self._sensors] + """[{label, x, y, side, offset_x, offset_y}] for WaferMapItem.sensors.""" + return [ + { + "label": s.label, + "x": s.x, + "y": s.y, + "side": getattr(s, "side", "right"), + "offset_x": getattr(s, "offset_x", 0.0), + "offset_y": getattr(s, "offset_y", 0.0), + } + for s in self._sensors + ] + + @Property(str, notify=sensorsChanged) + def waferShape(self) -> str: + """Wafer shape: 'round' or 'square'.""" + return getattr(self._sensors, "shape", "round") + + @Property(float, notify=sensorsChanged) + def waferSize(self) -> float: + """Wafer size in mm.""" + return getattr(self._sensors, "size", 300.0) @Property("QVariantList", notify=frameUpdated) def sensorValues(self) -> list: @@ -155,7 +181,9 @@ class SessionController(QObject): # ---- review: file load + playback ---- @Slot(str) def loadFile(self, file_path: str) -> None: + from pathlib import Path from pygui.backend.data.data_records import is_official_csv, read_data_records, read_official_csv + from pygui.backend.wafer.wafer_layouts import WaferLayout, resolve_shape_and_size sensors: list[Sensor] = [] frames = [] @@ -180,7 +208,15 @@ class SessionController(QObject): log.warning("No sensors or data in %s", file_path) return - self._sensors = sensors + wafer_id = "" + if not is_official_csv(file_path): + wafer_id = data.serial if (data and data.serial) else "" + else: + stem = Path(file_path).stem + wafer_id = stem.split("-")[0] if "-" in stem else stem + + shape, size = resolve_shape_and_size(sensors, wafer_id) + self._sensors = WaferLayout(sensors, shape=shape, size=size) self._player.load(frames) self._model.reset() self._loaded_file = file_path @@ -256,28 +292,90 @@ class SessionController(QObject): self.stateChanged.emit() # ---- live: stream start/stop ---- - @Slot(str) - def startStream(self, port: str) -> None: + @Slot(str, str) + def startStream(self, port: str, family_code: str = "") -> None: from pygui.serialcomm.serial_port import SerialPort # transport open import serial as pyserial + from pygui.backend.wafer.wafer_layouts import load_layout_for_wafer_id + import time - def parse_line(raw: str, seq: int) -> Frame: - parts = [float(x) for x in raw.split(",")] - return Frame(seq=seq, time=parts[0], values=parts[1:]) + if family_code: + try: + self._sensors = load_layout_for_wafer_id(family_code) + self.sensorsChanged.emit() + except Exception as e: + log.warning("Could not load layout for %s: %s", family_code, e) + + # The new binary protocol sends payload of (sensorCount * 2) bytes + # Each sensor is a big-endian 16-bit value (Sign-Magnitude). + def parse_binary_frame(payload: bytes, seq: int) -> Frame: + values = [] + num_sensors = min(80, len(payload) // 2) + for i in range(num_sensors): + high_byte = payload[i * 2] + low_byte = payload[i * 2 + 1] + val16 = (high_byte << 8) | low_byte + + is_negative = (val16 & 0x8000) != 0 + integer_part = (val16 >> 7) & 0xFF + fractional_bits = val16 & 0x7F + fractional_part = fractional_bits / 128.0 + result = integer_part + fractional_part + if is_negative: + result = -result + values.append(result) + + # Create a mock timestamp or use elapsed time since start. + # We don't have a time value from the binary stream header right now. + t = seq * 0.05 # approx 20Hz if seq increments 1 per frame + return Frame(seq=seq, time=t, values=values) + + # Clear out any old data from the prev sessions + self._model.reset() + self._last = None + self._last_raw_frame = None + + try: + transport = pyserial.Serial(port, SerialPort.BAUDRATE, timeout=1) + except Exception as exc: + log.warning("Baud rate %d failed on %s, falling back to 115200: %s", SerialPort.BAUDRATE, port, exc) + transport = pyserial.Serial(port, 115200, timeout=1) + # Send 'D2' command padded to 512 bytes to start the stream + cmd = b"D2" + (b"F" * 510) + transport.write(cmd) + + def on_error(exc: Exception): + log.error("Live stream error: %s", exc) - transport = pyserial.Serial(port, SerialPort.BAUDRATE, timeout=1) self._reader = StreamReader( - transport, parse_line, - on_frame=lambda f: self._liveFrame.emit(f)) + transport, parse_binary_frame, + on_frame=lambda f: self._liveFrame.emit(f), + on_error=on_error, + family_code=family_code or "A") self._reader.start() - self._repaint_timer.start() # Q1: begin ~20 Hz repaints + self._repaint_timer.start() + + self.setMode("live") + self.stateChanged.emit() @Slot() def stopStream(self) -> None: self._repaint_timer.stop() - if self._reader is not None: + if self._reader: + transport = self._reader._transport self._reader.stop() self._reader = None + + # Send 'D2S' command padded to 512 bytes to stop the stream + if transport and transport.is_open: + try: + cmd = b"D2S" + (b"F" * 509) + transport.write(cmd) + transport.flush() + except Exception as exc: + log.error("Error sending stop command: %s", exc) + + self.stateChanged.emit() self.stopRecording() @Slot(object) diff --git a/src/pygui/backend/data/csv_file_metadata.py b/src/pygui/backend/data/csv_file_metadata.py index bcebb3d..6d48963 100644 --- a/src/pygui/backend/data/csv_file_metadata.py +++ b/src/pygui/backend/data/csv_file_metadata.py @@ -25,8 +25,8 @@ class CSVFileMetadata: try: return datetime.strptime(self.date, date_format) except (ValueError, TypeError): - # If format is wrong or date is None, return current time - return datetime.now() + # If format is wrong or date is None, return None + return None # ===== Formatting ===== @staticmethod @@ -48,4 +48,5 @@ class CSVFileMetadata: def string_date_format(self) -> str: """Helper to ensure the date is always a string for JSON.""" - return self.format_date(self.get_date()) + dt = self.get_date() + return self.format_date(dt) if dt else self.date diff --git a/src/pygui/backend/data/file_browser.py b/src/pygui/backend/data/file_browser.py index 45c20ad..817e425 100644 --- a/src/pygui/backend/data/file_browser.py +++ b/src/pygui/backend/data/file_browser.py @@ -193,21 +193,15 @@ class FileBrowser(QObject): # Fall back to CSV/header parsing if sidecar is malformed. pass + meta = self._metadata_from_filename(csv_path) parser_data, _ = self._parser.parse(str(csv_path)) if parser_data is not None: - wafer = parser_data.serial or "" - date_text = "" + if parser_data.serial: + meta.wafer = parser_data.serial if parser_data.date != datetime.min: - date_text = CSVFileMetadata.format_date(parser_data.date) - return CSVFileMetadata( - wafer=wafer, - date=date_text, - chamber="", - notes="", - filename=str(csv_path), - ) + meta.date = CSVFileMetadata.format_date(parser_data.date) - return self._metadata_from_filename(csv_path) + return meta def _metadata_from_filename(self, csv_path: Path) -> CSVFileMetadata: match = re.match( @@ -238,7 +232,7 @@ class FileBrowser(QObject): QStandardPaths.DocumentsLocation ) base_dir = Path(documents_dir) if documents_dir else (Path.home() / "Documents") - return base_dir / "isc_data" + return base_dir / "ISC_DATA" def _set_current_directory(self, directory: Path) -> None: normalized = Path(directory) diff --git a/src/pygui/backend/wafer/wafer_layouts.py b/src/pygui/backend/wafer/wafer_layouts.py index a54d936..6214d67 100644 --- a/src/pygui/backend/wafer/wafer_layouts.py +++ b/src/pygui/backend/wafer/wafer_layouts.py @@ -21,6 +21,45 @@ from pygui.backend.wafer.zwafer_models import Sensor _LAYOUTS_DIR = Path(__file__).parent.parent.parent / "assets" / "layouts" +class WaferLayout(list): + """Subclass of list that carries wafer shape and size metadata.""" + def __init__(self, sensors: list[Sensor], shape: str = "round", size: float = 300.0): + super().__init__(sensors) + self.shape = shape + self.size = size + + +def resolve_shape_and_size(sensors: list[Sensor], wafer_id: str = "") -> tuple[str, float]: + """Helper to determine the shape and size of a sensor layout. + + Tries matching by wafer_id prefix, falling back to the number of sensors. + """ + if wafer_id: + prefix = wafer_id[0].upper() + for path in _LAYOUTS_DIR.glob("*.yaml"): + try: + data = _load_yaml(path) + if prefix in data.get("wafers", []): + return data.get("shape", "round"), float(data.get("size", 300.0)) + except Exception: + continue + + # Fallback to sensor count mapping + n = len(sensors) + if n == 80: + return "square", 310.0 + elif n == 65: + return "round", 300.0 + elif n == 48: + return "round", 300.0 + elif n == 29: + return "round", 200.0 + elif n == 22: + return "round", 200.0 + + return "round", 300.0 + + def _family_name(raw_name: str) -> str: return raw_name.replace("wafer", "").strip("_") @@ -34,7 +73,7 @@ def available_families() -> list[str]: return [_family_name(_load_yaml(p)["name"]) for p in _LAYOUTS_DIR.glob("*.yaml")] -def load_layout(family: str) -> list[Sensor]: +def load_layout(family: str) -> WaferLayout: for path in _LAYOUTS_DIR.glob("*.yaml"): data = _load_yaml(path) if _family_name(data["name"]) == family: @@ -42,7 +81,7 @@ def load_layout(family: str) -> list[Sensor]: raise KeyError(f"Unknown wafer family: {family!r}") -def load_layout_for_wafer_id(wafer_id: str) -> list[Sensor]: +def load_layout_for_wafer_id(wafer_id: str) -> WaferLayout: """Match 'B00108' → bcdwafer by looking up the first char in each YAML's 'wafers' list.""" prefix = wafer_id[0].upper() if wafer_id else "" for path in _LAYOUTS_DIR.glob("*.yaml"): @@ -52,15 +91,17 @@ def load_layout_for_wafer_id(wafer_id: str) -> list[Sensor]: raise KeyError(f"No layout found for wafer ID prefix {prefix!r}") -def _to_sensors(data: dict) -> list[Sensor]: +def _to_sensors(data: dict) -> WaferLayout: xs: list[float] = data["X"] ys: list[float] = data["Y"] size: float = float(data["size"]) + shape: str = data.get("shape", "round") start_sn: int = data.get("start_sn", 1) reverse_x: bool = data.get("reverse_x", False) reverse_y: bool = data.get("reverse_y", False) x_orig: str = data.get("x_origin", "left") y_orig: str = data.get("y_origin", "bottom") + label_exceptions = data.get("label_exceptions", {}) or {} x_shift = {"left": size / 2, "right": -(size / 2), "center": 0.0}.get(x_orig, 0.0) y_shift = {"bottom": size / 2, "top": -(size / 2), "center": 0.0}.get(y_orig, 0.0) @@ -71,9 +112,52 @@ def _to_sensors(data: dict) -> list[Sensor]: x_mm = -x_mm if reverse_y: y_mm = -y_mm + + x_coord = x_mm - x_shift + y_coord = y_mm - y_shift + + # Determine label position defaults based on quadrants + sn = start_sn + i + side = "right" + offset_x = 0.0 + offset_y = 0.0 + + # Check explicit label exception from layout config + if sn in label_exceptions: + exc = label_exceptions[sn] + if exc: + side_key = list(exc.keys())[0] + val = exc[side_key] + side = side_key + if isinstance(val, list) and len(val) >= 2: + offset_x = float(val[0]) + offset_y = float(val[1]) + else: + # Smart default quadrant positioning to keep labels pointing inward + if x_coord > 30.0 and y_coord > 30.0: + side = "left" + offset_y = -0.5 + elif x_coord < -30.0 and y_coord > 30.0: + side = "right" + offset_y = -0.5 + elif x_coord < -30.0 and y_coord < -30.0: + side = "right" + offset_y = 0.5 + elif x_coord > 30.0 and y_coord < -30.0: + side = "left" + offset_y = 0.5 + elif x_coord > 15.0: + side = "left" + elif x_coord < -15.0: + side = "right" + sensors.append(Sensor( - label=str(start_sn + i), - x=x_mm - x_shift, - y=y_mm - y_shift, + label=str(sn), + x=x_coord, + y=y_coord, + side=side, + offset_x=offset_x, + offset_y=offset_y, )) - return sensors + return WaferLayout(sensors, shape=shape, size=size) + diff --git a/src/pygui/backend/wafer/zwafer_models.py b/src/pygui/backend/wafer/zwafer_models.py index 720df53..43e501a 100644 --- a/src/pygui/backend/wafer/zwafer_models.py +++ b/src/pygui/backend/wafer/zwafer_models.py @@ -9,6 +9,9 @@ class Sensor: label: str x: float y: float + side: str = "right" + offset_x: float = 0.0 + offset_y: float = 0.0 # ===== Data Row =====