Jack.Le
f24dd89e8e
- Updated requirements.txt to specify minimum versions for PySide6, pyqtgraph, and numpy. - Modified CSVFileMetadata class to include a new master_type attribute for improved metadata representation. - Enhanced FileBrowser class to handle master_type in metadata saving and retrieval. - Added a new method in FileBrowser for parsing CSV metadata, improving error handling and data extraction. - Adjusted LocalSettingsModel to format code and change default reverseZWafer setting to False.
361 lines
12 KiB
Python
361 lines
12 KiB
Python
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": False,
|
|
"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
|