From e43bf258e3cc6e140326c5ee5c37ae5c18afb91c Mon Sep 17 00:00:00 2001 From: "Jack.Le" Date: Thu, 23 Apr 2026 11:40:47 -0700 Subject: [PATCH] Add initial backend structure with models, utilities, and settings management - Introduced .gitignore to exclude common Python artifacts and IDE files. - Added contour models for contour generation. - Implemented cryptographic utilities for data encryption and decryption. - Created CSV metadata model for handling wafer data. - Developed time-indexed data segment model. - Established local settings model for application configuration. - Added Z-wafer data models and parser for CSV file handling. - Implemented Marching Squares algorithm for contour generation. --- .gitignore | 30 +++ backend/contour_models.py | 26 +++ backend/crypto_helper.py | 90 ++++++++ backend/csv_file_metadata.py | 49 +++++ backend/data_segment.py | 55 +++++ backend/local_settings.py | 93 +++++++++ backend/local_settings_model.py | 356 ++++++++++++++++++++++++++++++++ backend/marching_squares.py | 139 +++++++++++++ backend/zwafer_models.py | 28 +++ backend/zwafer_parser.py | 131 ++++++++++++ 10 files changed, 997 insertions(+) create mode 100644 .gitignore create mode 100644 backend/contour_models.py create mode 100644 backend/crypto_helper.py create mode 100644 backend/csv_file_metadata.py create mode 100644 backend/data_segment.py create mode 100644 backend/local_settings.py create mode 100644 backend/local_settings_model.py create mode 100644 backend/marching_squares.py create mode 100644 backend/zwafer_models.py create mode 100644 backend/zwafer_parser.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..843dba4 --- /dev/null +++ b/.gitignore @@ -0,0 +1,30 @@ +# Python +__pycache__/ +*.pyc +*.pyo +*.pyd +*.so +*.a + +# Virtual environment +.venv/ +venv/ +env/ + +# IDE +.vscode/ +.idea/ +*.swp +*.swo +*~ + +# OS +.DS_Store +Thumbs.db + +# Qt / QML cache +Qt/ +qrc_*.cpp +moc_*.cpp +*.qmlcache +*.qmldir.cache diff --git a/backend/contour_models.py b/backend/contour_models.py new file mode 100644 index 0000000..f91b712 --- /dev/null +++ b/backend/contour_models.py @@ -0,0 +1,26 @@ +from dataclasses import dataclass, field +from typing import List + + +# ===== Single Contour Segment ===== +@dataclass +class ContourSegment: + start_x: float + start_y: float + end_x: float + end_y: float + + @property + def start(self): + return (self.start_x, self.start_y) + + @property + def end(self): + return (self.end_x, self.end_y) + + +# ===== Contour Line ===== +@dataclass +class ContourLine: + level: float + segments: List[ContourSegment] = field(default_factory=list) diff --git a/backend/crypto_helper.py b/backend/crypto_helper.py new file mode 100644 index 0000000..3a97b35 --- /dev/null +++ b/backend/crypto_helper.py @@ -0,0 +1,90 @@ +import secrets +from pathlib import Path +from typing import Union + +from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes +from cryptography.hazmat.backends import default_backend + + +# ===== Crypto Utilities ===== +class CryptoHelper: + """Cryptographic utilities for wafer data encryption/decryption.""" + + # ===== Hex/Byte Conversion ===== + @staticmethod + def hex_string_to_bytearray(hex_str: str) -> bytes: + """Converts a hex string to bytes (e.g., '48656C6C6F' → b'Hello').""" + if len(hex_str) % 2 != 0: + raise ValueError("Hex string must have even length") + try: + return bytes.fromhex(hex_str) + except ValueError as e: + raise ValueError(f"Invalid hex string: {e}") + + @staticmethod + def bytearray_to_hex_string(data: Union[bytes, bytearray]) -> str: + """Converts bytes to lowercase hex string.""" + return data.hex() + + # ===== Random Data ===== + @staticmethod + def generate_random_bytes(length: int) -> bytes: + """Generates cryptographically secure random bytes.""" + if length < 0: + raise ValueError("Length must be non-negative") + return secrets.token_bytes(length) + + # ===== AES Decryption ===== + @staticmethod + def decrypt_aes(data: bytes, key: bytes, iv: bytes) -> bytes: + """ + Decrypts AES-128-CBC data with PKCS7 padding. + + Args: + data: Encrypted ciphertext (must be multiple of 16 bytes) + key: 16-byte AES key + iv: 16-byte IV + + Returns: + Decrypted plaintext + + Raises: + ValueError: If inputs are invalid + cryptography.exceptions.InvalidSignature: On decryption failure + """ + if len(key) != 16: + raise ValueError("Key must be 16 bytes (AES-128)") + if len(iv) != 16: + raise ValueError("IV must be 16 bytes") + if len(data) == 0: + return b"" + if len(data) % 16 != 0: + raise ValueError("Ciphertext length must be multiple of 16 bytes") + + cipher = Cipher(algorithms.AES(key), modes.CBC(iv), backend=default_backend()) + decrypt_ctx = cipher.decryptor() + plaintext_padded = decrypt_ctx.update(data) + decrypt_ctx.finalize() + + # Remove PKCS7 padding + pad_len = plaintext_padded[-1] + if pad_len > 16 or pad_len == 0: + raise ValueError("Invalid padding") + return plaintext_padded[:-pad_len] + + # ===== Binary File I/O ===== + @staticmethod + def save_to_binary_file(file_path: Union[str, Path], data: bytes) -> None: + """Writes bytes to a binary file.""" + Path(file_path).write_bytes(data) + + @staticmethod + def read_from_binary_file(file_path: Union[str, Path]) -> bytes: + """Reads bytes from a binary file.""" + return Path(file_path).read_bytes() + + # ===== Convenience Helpers ===== + @classmethod + def read_hex_string_from_binary_file(cls, file_path: Union[str, Path]) -> str: + """Reads binary file and returns hex string.""" + data = cls.read_from_binary_file(file_path) + return cls.bytearray_to_hex_string(data) diff --git a/backend/csv_file_metadata.py b/backend/csv_file_metadata.py new file mode 100644 index 0000000..2b2c1c0 --- /dev/null +++ b/backend/csv_file_metadata.py @@ -0,0 +1,49 @@ +from datetime import datetime + + +# ===== CSV Metadata Model ===== +class CSVFileMetadata: + # ===== Lifecycle ===== + def __init__(self, wafer="", date="", chamber="", notes="", filename="", columns=0): + self.wafer = wafer + self.date = date + self.chamber = chamber + self.notes = notes + + self.filename = filename + self.columns = columns + + # ===== Accessors ===== + def get_wafer_type(self) -> str: + """Return the first character of the wafer string, or empty if none""" + return self.wafer[0] if self.wafer else "" + + def get_date(self) -> datetime: + """Parsifes the date string. Returns current time if parsing fails""" + date_format = "%Y-%m-%d %H:%M:%S" + 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() + + # ===== Formatting ===== + @staticmethod + def format_date(dt: datetime) -> str: + """Formats a datetime object into the standard ISO-like string.""" + if not isinstance(dt, datetime): + return "" + return dt.strftime("%Y-%m-%d %H:%M:%S") + + # ===== Serialization ===== + def to_dict(self) -> dict: + return { + "wafer": self.wafer, + "date": self.string_date_format(), # Ensure we save as string + "chamber": self.chamber, + "notes": self.notes, + } + + def string_date_format(self) -> str: + """Helper to ensure the date is always a string for JSON.""" + return self.format_date(self.get_date()) diff --git a/backend/data_segment.py b/backend/data_segment.py new file mode 100644 index 0000000..4a92c19 --- /dev/null +++ b/backend/data_segment.py @@ -0,0 +1,55 @@ +from datetime import datetime, timedelta + + +# ===== Time-Indexed Data Segment ===== +class DataSegment: + # ===== Lifecycle ===== + def __init__( + self, + full_data: list[float], + start_time: datetime, + start_index: int, + end_index: int, + ): + if not isinstance(start_index, int): + raise TypeError("start_index must be an integer") + if not isinstance(end_index, int): + raise TypeError("end_index must be an integer") + """ + full_data: The complete list of sensor readings. + start_time: The timestamp of the very first reading in full_data. + start_index: The offset (in seconds) from start_time to the beginning of this segment. + end_index: The offset (in seconds) from start_time to the end of this segment. + """ + self.full_data = full_data + self._base_start_time = start_time + self._start_index = start_index + self._end_index = end_index + self.chamber = "" + self.notes = "" + + # ===== Index Properties ===== + @property + def start_index(self) -> int: + return self._start_index + + @start_index.setter + def start_index(self, value: int): + self._start_index = value + + @property + def end_index(self) -> int: + return self._end_index + + @end_index.setter + def end_index(self, value: int): + self._end_index = value + + # ===== Derived Time Properties ===== + @property + def start_time(self) -> datetime: + return self._base_start_time + timedelta(seconds=self.start_index) + + @property + def end_time(self) -> datetime: + return self._base_start_time + timedelta(seconds=self.end_index) diff --git a/backend/local_settings.py b/backend/local_settings.py new file mode 100644 index 0000000..d89be31 --- /dev/null +++ b/backend/local_settings.py @@ -0,0 +1,93 @@ +import json +import os +from pathlib import Path + + +# ===== Settings Persistence Model ===== +class LocalSettings: + # ===== Defaults ===== + def __init__(self): + self.chamber_id = "" + self.reverse_z_wafer = False + self.master = {} # Dict[str, str] + self.wafer_data_size = {} # Dict[str, int] + self.debug = False + self.wafer_read_retries = 8 + # Timeout in msec for reading from the wafer (default 2 min) + self.wafer_read_timeout = 120000 + self.wafer_detect_timeout = 5000 + self.split_threshold = 40.0 + + # ===== File Path Helpers ===== + @classmethod + def _settings_path(cls, directory: str) -> Path: + return Path(directory) / "settings.json" + + # ===== Load/Save ===== + @classmethod + def read_settings(cls, directory: str) -> "LocalSettings": + """Read settings from settings.json in the provided directory.""" + settings = cls() + path = cls._settings_path(directory) + + if not path.exists(): + return settings + + try: + with open(path, "r", encoding="utf-8") as f: + data = json.load(f) + for key, value in data.items(): + if hasattr(settings, key): + setattr(settings, key, value) + except Exception as e: + print(f"Error reading setting file: {e}") + + return settings + + @classmethod + def save_settings(cls, directory: str, settings: "LocalSettings") -> None: + """Save the current settings instance to settings.json.""" + path = cls._settings_path(directory) + path.parent.mkdir(parents=True, exist_ok=True) + data = settings.__dict__.copy() + + try: + with open(path, "w", encoding="utf-8") as f: + json.dump(data, f, indent=4) + except Exception as e: + print(f"Error saving settings: {e}") + + # ===== Master File Helpers ===== + @classmethod + def get_master(cls, directory: str, wtype: str) -> str: + """Return the full path to a master file if it exists.""" + settings = cls.read_settings(directory) + if wtype not in settings.master: + return "" + + file_path = Path(directory) / settings.master[wtype] + if file_path.exists(): + return str(file_path) + return "" + + @classmethod + def set_master(cls, directory: str, wtype: str, filename: str) -> None: + """Store a master filename for a wafer type and save settings.""" + settings = cls.read_settings(directory) + settings.master[wtype] = os.path.basename(filename) + cls.save_settings(directory, settings) + + # ===== Wafer Sizing ===== + def get_wafer_data_size(self, wtype: str) -> int: + """Return the expected transfer size in bytes for a wafer type.""" + if not wtype: + return 393216 + + if wtype in self.wafer_data_size: + return self.wafer_data_size[wtype] + + if wtype == "P": + return 393216 + if wtype == "X": + return 1310720 + return 393216 diff --git a/backend/local_settings_model.py b/backend/local_settings_model.py new file mode 100644 index 0000000..f58ca7f --- /dev/null +++ b/backend/local_settings_model.py @@ -0,0 +1,356 @@ +from __future__ import annotations + +from copy import deepcopy +from pathlib import Path +from typing import Any + +from PySide6.QtCore import QObject, Property, QDateTime, QStandardPaths, Signal, Slot + +from backend.local_settings import LocalSettings + + +MASTER_FAMILIES = ("A", "B", "C", "D", "E", "F", "P", "X", "Z") + + +class LocalSettingsModel(QObject): + chamberIdChanged = Signal() + reverseZWaferChanged = Signal() + debugModeChanged = Signal() + waferReadTimeoutChanged = Signal() + waferDetectTimeoutChanged = Signal() + waferRetriesChanged = Signal() + mastersChanged = Signal() + isDirtyChanged = Signal() + isValidChanged = Signal() + validationMessageChanged = Signal() + saveStatusChanged = Signal() + lastSavedAtChanged = Signal() + + def __init__(self, parent: QObject | None = None) -> None: + super().__init__(parent) + self._data_dir = self._resolve_data_dir() + self._defaults = self._new_defaults() + + self._chamber_id = self._defaults["chamberId"] + self._reverse_z_wafer = self._defaults["reverseZWafer"] + self._debug_mode = self._defaults["debugMode"] + self._wafer_read_timeout = self._defaults["waferReadTimeout"] + self._wafer_detect_timeout = self._defaults["waferDetectTimeout"] + self._wafer_retries = self._defaults["waferRetries"] + self._masters = deepcopy(self._defaults["masters"]) + + self._is_dirty = False + self._is_valid = True + self._validation_message = "" + self._save_status = "Ready" + self._last_saved_at = "" + self._saved_snapshot = self._snapshot() + self._recompute_derived() + + def _resolve_data_dir(self) -> Path: + documents_dir = QStandardPaths.writableLocation(QStandardPaths.DocumentsLocation) + base_dir = Path(documents_dir) if documents_dir else (Path.home() / "Documents") + return base_dir / "isc_data" + + def _new_defaults(self) -> dict[str, Any]: + return { + "chamberId": "2", + "reverseZWafer": True, + "debugMode": False, + "waferReadTimeout": 120000, + "waferDetectTimeout": 5000, + "waferRetries": 8, + "masters": {family: "" for family in MASTER_FAMILIES}, + } + + def _snapshot(self) -> dict[str, Any]: + return { + "chamberId": self._chamber_id, + "reverseZWafer": self._reverse_z_wafer, + "debugMode": self._debug_mode, + "waferReadTimeout": self._wafer_read_timeout, + "waferDetectTimeout": self._wafer_detect_timeout, + "waferRetries": self._wafer_retries, + "masters": deepcopy(self._masters), + } + + def _set_is_dirty(self, value: bool) -> None: + if self._is_dirty != value: + self._is_dirty = value + self.isDirtyChanged.emit() + + def _set_is_valid(self, value: bool) -> None: + if self._is_valid != value: + self._is_valid = value + self.isValidChanged.emit() + + def _set_validation_message(self, value: str) -> None: + if self._validation_message != value: + self._validation_message = value + self.validationMessageChanged.emit() + + def _set_save_status(self, value: str) -> None: + if self._save_status != value: + self._save_status = value + self.saveStatusChanged.emit() + + def _set_last_saved_at(self, value: str) -> None: + if self._last_saved_at != value: + self._last_saved_at = value + self.lastSavedAtChanged.emit() + + def _validate(self) -> str: + if not self._chamber_id.strip(): + return "Chamber ID is required." + if self._wafer_retries < 0 or self._wafer_retries > 10: + return "Retries must be between 0 and 10." + if self._wafer_read_timeout < 100 or self._wafer_read_timeout > 300000: + return "Read timeout must be between 100 and 300000 ms." + if self._wafer_detect_timeout < 100 or self._wafer_detect_timeout > 300000: + return "Detect timeout must be between 100 and 300000 ms." + + for family, path_text in self._masters.items(): + cleaned = path_text.strip() + if not cleaned: + continue + path = Path(cleaned) + if path.suffix.lower() != ".csv": + return f"Master {family} must point to a CSV file." + if not path.exists(): + return f"Master {family} file was not found." + return "" + + def _recompute_derived(self) -> None: + message = self._validate() + self._set_is_valid(message == "") + self._set_validation_message(message) + self._set_is_dirty(self._snapshot() != self._saved_snapshot) + + def _emit_all_changed(self) -> None: + self.chamberIdChanged.emit() + self.reverseZWaferChanged.emit() + self.debugModeChanged.emit() + self.waferReadTimeoutChanged.emit() + self.waferDetectTimeoutChanged.emit() + self.waferRetriesChanged.emit() + self.mastersChanged.emit() + + def _to_local_settings(self) -> LocalSettings: + settings = LocalSettings() + settings.chamber_id = self._chamber_id + settings.reverse_z_wafer = self._reverse_z_wafer + settings.debug = self._debug_mode + settings.wafer_read_timeout = self._wafer_read_timeout + settings.wafer_detect_timeout = self._wafer_detect_timeout + settings.wafer_read_retries = self._wafer_retries + settings.master = deepcopy(self._masters) + return settings + + @Property(str, notify=chamberIdChanged) + def chamberId(self) -> str: + return self._chamber_id + + @chamberId.setter + def chamberId(self, value: str) -> None: + next_value = str(value).strip() + if self._chamber_id == next_value: + return + self._chamber_id = next_value + self.chamberIdChanged.emit() + self._recompute_derived() + + @Property(bool, notify=reverseZWaferChanged) + def reverseZWafer(self) -> bool: + return self._reverse_z_wafer + + @reverseZWafer.setter + def reverseZWafer(self, value: bool) -> None: + if self._reverse_z_wafer == value: + return + self._reverse_z_wafer = bool(value) + self.reverseZWaferChanged.emit() + self._recompute_derived() + + @Property(bool, notify=debugModeChanged) + def debugMode(self) -> bool: + return self._debug_mode + + @debugMode.setter + def debugMode(self, value: bool) -> None: + if self._debug_mode == value: + return + self._debug_mode = bool(value) + self.debugModeChanged.emit() + self._recompute_derived() + + @Property(int, notify=waferReadTimeoutChanged) + def waferReadTimeout(self) -> int: + return self._wafer_read_timeout + + @waferReadTimeout.setter + def waferReadTimeout(self, value: int) -> None: + if self._wafer_read_timeout == value: + return + self._wafer_read_timeout = int(value) + self.waferReadTimeoutChanged.emit() + self._recompute_derived() + + @Property(int, notify=waferDetectTimeoutChanged) + def waferDetectTimeout(self) -> int: + return self._wafer_detect_timeout + + @waferDetectTimeout.setter + def waferDetectTimeout(self, value: int) -> None: + if self._wafer_detect_timeout == value: + return + self._wafer_detect_timeout = int(value) + self.waferDetectTimeoutChanged.emit() + self._recompute_derived() + + @Property(int, notify=waferRetriesChanged) + def waferRetries(self) -> int: + return self._wafer_retries + + @waferRetries.setter + def waferRetries(self, value: int) -> None: + if self._wafer_retries == value: + return + self._wafer_retries = int(value) + self.waferRetriesChanged.emit() + self._recompute_derived() + + @Property("QVariantMap", notify=mastersChanged) + def masters(self) -> dict[str, str]: + return dict(self._masters) + + @Property(bool, notify=isDirtyChanged) + def isDirty(self) -> bool: + return self._is_dirty + + @Property(bool, notify=isValidChanged) + def isValid(self) -> bool: + return self._is_valid + + @Property(str, notify=validationMessageChanged) + def validationMessage(self) -> str: + return self._validation_message + + @Property(str, notify=saveStatusChanged) + def saveStatus(self) -> str: + return self._save_status + + @Property(str, notify=lastSavedAtChanged) + def lastSavedAt(self) -> str: + return self._last_saved_at + + @Property(str, notify=saveStatusChanged) + def settingsFilePath(self) -> str: + return str(self._data_dir / "settings.json") + + @Slot() + def loadSettings(self) -> None: + loaded = LocalSettings.read_settings(str(self._data_dir)) + self._chamber_id = str(loaded.chamber_id).strip() or str(self._defaults["chamberId"]) + self._reverse_z_wafer = bool(loaded.reverse_z_wafer) + self._debug_mode = bool(loaded.debug) + self._wafer_read_timeout = int(loaded.wafer_read_timeout) + self._wafer_detect_timeout = int(loaded.wafer_detect_timeout) + self._wafer_retries = int(loaded.wafer_read_retries) + + masters = {family: "" for family in MASTER_FAMILIES} + for family, value in loaded.master.items(): + normalized = str(family).strip().upper() + if normalized in masters: + masters[normalized] = str(value).strip() + self._masters = masters + + self._saved_snapshot = self._snapshot() + self._emit_all_changed() + self._recompute_derived() + self._set_save_status("Loaded settings") + + @Slot() + def saveSettings(self) -> None: + self._recompute_derived() + if not self._is_valid: + self._set_save_status("error: invalid settings") + return + + try: + LocalSettings.save_settings(str(self._data_dir), self._to_local_settings()) + self._saved_snapshot = self._snapshot() + self._recompute_derived() + self._set_save_status("Saved") + timestamp = QDateTime.currentDateTime().toString("yyyy-MM-dd HH:mm:ss") + self._set_last_saved_at(timestamp) + except Exception as exc: # pragma: no cover - defensive + self._set_save_status(f"error: {exc}") + + @Slot() + def revertChanges(self) -> None: + snap = self._saved_snapshot + self._chamber_id = str(snap["chamberId"]) + self._reverse_z_wafer = bool(snap["reverseZWafer"]) + self._debug_mode = bool(snap["debugMode"]) + self._wafer_read_timeout = int(snap["waferReadTimeout"]) + self._wafer_detect_timeout = int(snap["waferDetectTimeout"]) + self._wafer_retries = int(snap["waferRetries"]) + self._masters = deepcopy(snap["masters"]) + + self._emit_all_changed() + self._recompute_derived() + self._set_save_status("Reverted unsaved changes") + + @Slot() + def resetDefaults(self) -> None: + self._chamber_id = str(self._defaults["chamberId"]) + self._reverse_z_wafer = bool(self._defaults["reverseZWafer"]) + self._debug_mode = bool(self._defaults["debugMode"]) + self._wafer_read_timeout = int(self._defaults["waferReadTimeout"]) + self._wafer_detect_timeout = int(self._defaults["waferDetectTimeout"]) + self._wafer_retries = int(self._defaults["waferRetries"]) + self._masters = deepcopy(self._defaults["masters"]) + + self._emit_all_changed() + self._recompute_derived() + self._set_save_status("Defaults restored (not saved)") + + @Slot(str, str) + def setMaster(self, family: str, file_path: str) -> None: + normalized = family.strip().upper() + if normalized not in self._masters: + return + next_path = file_path.strip() + if self._masters[normalized] == next_path: + return + self._masters[normalized] = next_path + self.mastersChanged.emit() + self._recompute_derived() + + @Slot(str) + def clearMaster(self, family: str) -> None: + self.setMaster(family, "") + + @Slot(str) + def setChamberId(self, value: str) -> None: + self.chamberId = value + + @Slot(bool) + def setReverseZWafer(self, value: bool) -> None: + self.reverseZWafer = value + + @Slot(bool) + def setDebugMode(self, value: bool) -> None: + self.debugMode = value + + @Slot(int) + def setWaferReadTimeout(self, value: int) -> None: + self.waferReadTimeout = value + + @Slot(int) + def setWaferDetectTimeout(self, value: int) -> None: + self.waferDetectTimeout = value + + @Slot(int) + def setWaferRetries(self, value: int) -> None: + self.waferRetries = value diff --git a/backend/marching_squares.py b/backend/marching_squares.py new file mode 100644 index 0000000..7f1787d --- /dev/null +++ b/backend/marching_squares.py @@ -0,0 +1,139 @@ +from typing import List, Tuple, Optional + +import numpy as np + +from backend.contour_models import ContourLine, ContourSegment + + +# ===== Contour Generation ===== +class MarchingSquares: + + # ===== Public API ===== + @staticmethod + def generate_contours(grid: np.ndarray, levels: List[float]) -> List[ContourLine]: + """ + Generate contour lines for a 2D grid at specified levels. + + Args: + grid: 2D numpy array (shape: [width, height]) + levels: List of contour levels to compute + + Returns: + List of ContourLine objects + """ + if grid.size == 0: + return [] + + width, height = grid.shape[0], grid.shape[1] + contours = [] + + for level in levels: + contour = ContourLine(level=level) + + # Iterate over each cell (x, y) in the grid + for y in range(height - 1): + for x in range(width - 1): + v0 = float(grid[x, y]) # top-left + v1 = float(grid[x + 1, y]) # top-right + v2 = float(grid[x + 1, y + 1]) # bottom-right + v3 = float(grid[x, y + 1]) # bottom-left + + if any(np.isnan([v0, v1, v2, v3])): + continue # Skip cells with NaN values + + state = ( + (1 if v0 > level else 0) + | (2 if v1 > level else 0) + | (4 if v2 > level else 0) + | (8 if v3 > level else 0) + ) + + seg = MarchingSquares._get_segment( + x, y, v0, v1, v2, v3, level, state + ) + if seg is not None: + contour.segments.append(seg) + + contours.append(contour) + + return contours + + # ===== Geometry Helpers ===== + @staticmethod + def _lerp( + x1: float, + y1: float, + x2: float, + y2: float, + val1: float, + val2: float, + level: float, + ) -> Tuple[float, float]: + """Linear interpolation between (x1,y1) and (x2,y2).""" + if val2 == val1: + return (x1 + x2) / 2, (y1 + y2) / 2 + t = (level - val1) / (val2 - val1) + px = x1 + t * (x2 - x1) + py = y1 + t * (y2 - y1) + return px, py + + # ===== Segment Lookup ===== + @staticmethod + def _get_segment( + x: int, + y: int, + v0: float, + v1: float, + v2: float, + v3: float, + level: float, + state: int, + ) -> Optional[ContourSegment]: + """Return a ContourSegment for the given cell and state.""" + if state in (10,): # Ambiguous case — skip + return None + + # Map C# states to Python logic + if state == 1 or state == 14: + start = MarchingSquares._lerp(x, y, x + 1, y, v0, v1, level) + end = MarchingSquares._lerp(x, y, x, y + 1, v0, v3, level) + elif state == 2 or state == 13: + start = MarchingSquares._lerp(x + 1, y, x + 1, y + 1, v1, v2, level) + end = MarchingSquares._lerp(x, y, x + 1, y, v0, v1, level) + elif state == 3 or state == 12: + start = MarchingSquares._lerp(x, y, x, y + 1, v0, v3, level) + end = MarchingSquares._lerp(x + 1, y, x + 1, y + 1, v1, v2, level) + elif state == 4 or state == 11: + start = MarchingSquares._lerp(x + 1, y, x + 1, y + 1, v1, v2, level) + end = MarchingSquares._lerp(x + 1, y + 1, x, y + 1, v2, v3, level) + elif state == 5: + start = MarchingSquares._lerp(x, y, x, y + 1, v0, v3, level) + end = MarchingSquares._lerp(x + 1, y, x + 1, y + 1, v1, v2, level) + elif state == 6 or state == 9: + start = MarchingSquares._lerp(x, y, x + 1, y, v0, v1, level) + end = MarchingSquares._lerp(x + 1, y + 1, x, y + 1, v2, v3, level) + elif state == 7 or state == 8: + start = MarchingSquares._lerp(x, y, x, y + 1, v0, v3, level) + end = MarchingSquares._lerp(x + 1, y + 1, x, y + 1, v2, v3, level) + else: + return None + + return ContourSegment( + start_x=start[0], start_y=start[1], end_x=end[0], end_y=end[1] + ) + + # ===== Color Mapping ===== + @staticmethod + def color_from_level( + value: float, min_val: float, max_val: float + ) -> Tuple[int, int, int]: + """Return (R, G, B) tuple for a value between min and max.""" + range_val = max_val - min_val + if range_val == 0: + t = 0.5 + else: + t = max(0.0, min(1.0, (value - min_val) / range_val)) + + r = int(255 * t) + b = int(255 * (1 - t)) + return (r, 0, b) diff --git a/backend/zwafer_models.py b/backend/zwafer_models.py new file mode 100644 index 0000000..720df53 --- /dev/null +++ b/backend/zwafer_models.py @@ -0,0 +1,28 @@ +from dataclasses import dataclass, field +from datetime import datetime +from typing import Dict, List, Optional + + +# ===== Sensor Geometry ===== +@dataclass +class Sensor: + label: str + x: float + y: float + + +# ===== Data Row ===== +@dataclass +class DataRecord: + time: float + values: List[float] = field(default_factory=list) + + +# ===== Parsed Z-Wafer Container ===== +@dataclass +class ZWaferData: + date: datetime = field(default_factory=lambda: datetime.min) + serial: Optional[str] = None + header: Dict[str, str] = field(default_factory=dict) + sensors: List[Sensor] = field(default_factory=list) + csv_headers: List[str] = field(default_factory=list) diff --git a/backend/zwafer_parser.py b/backend/zwafer_parser.py new file mode 100644 index 0000000..41e6cf5 --- /dev/null +++ b/backend/zwafer_parser.py @@ -0,0 +1,131 @@ +from pathlib import Path +from typing import Tuple, Optional +from datetime import datetime, timedelta + +from backend.zwafer_models import ZWaferData, Sensor + + +# ===== Z-Wafer CSV Parser ===== +class ZWaferParser: + """Parses Z-wafer CSV files (header + sensor layout + data rows).""" + + # ===== Public Parse API ===== + def parse(self, file_path: str) -> Tuple[Optional[ZWaferData], Optional[Path]]: + """ + Parse a Z-wafer file. + + Returns: + (ZWaferData, Path) on success + (None, None) on error (e.g., file not found) + """ + try: + path = Path(file_path) + if not path.exists(): + raise FileNotFoundError(f"File not found: {file_path}") + + with path.open("r", encoding="utf-8") as f: + return self._process_header(f), path + except (ValueError, KeyError): + raise + except Exception: + return None, None + + # ===== Header Parsing ===== + def _process_header(self, file_obj) -> ZWaferData: + """Parse header and sensor layout from open file object.""" + wafer_data = ZWaferData() + labels: Optional[list] = None + x_coords: Optional[list] = None + y_coords: Optional[list] = None + + for line in file_obj: + # Strip trailing comma + whitespace + line = line.rstrip().rstrip(",").strip() + if not line or line.replace(",", "").strip() == "": + continue + + parts = [p.strip() for p in line.split(",")] + first_part = parts[0].lower() + + # Detect end of header section + if first_part == "data": + wafer_data.csv_headers = labels or [] + self._build_sensor_layout(wafer_data, labels, x_coords, y_coords) + return wafer_data + + # Detect switch from metadata to sensor layout + # Parse metadata (key=value pairs) + if not self._parse_header_line(wafer_data, parts): + # If not metadata, it's part of sensor layout + label = parts[0].lower() + values = parts[1:] + + if label == "label": + labels = values + elif label == "x (mm)": + x_coords = values + elif label == "y (mm)": + y_coords = values + + return wafer_data # Incomplete file — return partial data + + # ===== Metadata Parsing ===== + def _parse_header_line(self, wafer_data: ZWaferData, parts: list) -> bool: + """Parse key=value pairs from header line. Returns True if handled.""" + non_empty_parts = [p for p in parts if p] + if not non_empty_parts: + return False + + found_kv = False + for part in non_empty_parts: + eq_idx = part.find("=") + if eq_idx < 0: + continue + + key = part[:eq_idx].strip() + value = part[eq_idx + 1 :].strip('=" ') + found_kv = True + + wafer_data.header[key] = value + + # Extract special fields + if key.lower() == "acquisition date": + try: + from datetime import datetime as dt + + wafer_data.date = dt.strptime(value, "%m/%d/%Y") + except ValueError: + wafer_data.date = datetime.min # Fallback on parse error + elif key.lower() == "wafer id": + wafer_data.serial = value + + if found_kv and wafer_data.date == datetime.min: + wafer_data.date = datetime.min + + return found_kv + + # ===== Sensor Layout Parsing ===== + def _build_sensor_layout( + self, + wafer_data: ZWaferData, + labels: Optional[list], + x_coords: Optional[list], + y_coords: Optional[list], + ) -> None: + """Build sensor list from layout arrays.""" + if not all([labels, x_coords, y_coords]): + raise ValueError("Sensor layout section is incomplete or missing.") + + if len(labels) != len(x_coords) or len(labels) != len(y_coords): + raise ValueError( + f"Mismatched sensor columns: labels={len(labels)}, " + f"x={len(x_coords)}, y={len(y_coords)}" + ) + + for i in range(len(labels)): + try: + wafer_data.sensors.append( + Sensor(label=labels[i], x=float(x_coords[i]), y=float(y_coords[i])) + ) + except ValueError as e: + raise ValueError(f"Invalid coordinate at index {i}: {e}")