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.
This commit is contained in:
Jack.Le
+30
@@ -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
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
@@ -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())
|
||||
@@ -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)
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
@@ -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}")
|
||||
Reference in New Issue
Block a user