jack
333 lines
12 KiB
Python
333 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
from copy import deepcopy
|
|
from pathlib import Path
|
|
from typing import Any, Callable
|
|
|
|
from PySide6.QtCore import Property, QDateTime, QObject, Signal, Slot
|
|
|
|
from pygui.backend.data.local_settings import LocalSettings
|
|
|
|
MASTER_FAMILIES = ("A", "B", "C", "D", "E", "F", "P", "X", "Z")
|
|
|
|
|
|
class LocalSettingsModel(QObject):
|
|
# Draft attrs are seeded via setattr() in __init__ from _new_defaults();
|
|
# declared here so mypy knows their types.
|
|
_chamber_id: str
|
|
_reverse_z_wafer: bool
|
|
_debug_mode: bool
|
|
_wafer_read_timeout: int
|
|
_wafer_detect_timeout: int
|
|
_wafer_retries: int
|
|
_masters: dict[str, str]
|
|
|
|
chamberIdChanged = Signal()
|
|
reverseZWaferChanged = Signal()
|
|
debugModeChanged = Signal()
|
|
waferReadTimeoutChanged = Signal()
|
|
waferDetectTimeoutChanged = Signal()
|
|
waferRetriesChanged = Signal()
|
|
mastersChanged = Signal()
|
|
isDirtyChanged = Signal()
|
|
isValidChanged = Signal()
|
|
validationMessageChanged = Signal()
|
|
saveStatusChanged = Signal()
|
|
lastSavedAtChanged = Signal()
|
|
|
|
# Draft attr (this class) <-> LocalSettings attr (the shared, persisted
|
|
# instance also held by DeviceController) <-> coercer applied on both
|
|
# load and revert/reset. One table drives load/save/revert/reset instead
|
|
# of four hand-written field-by-field copies.
|
|
_FIELD_MAP: tuple[tuple[str, str, Callable[[Any], Any]], ...] = (
|
|
("_chamber_id", "chamber_id", str),
|
|
("_reverse_z_wafer", "reverse_z_wafer", bool),
|
|
("_debug_mode", "debug", bool),
|
|
("_wafer_read_timeout", "wafer_read_timeout", int),
|
|
("_wafer_detect_timeout", "wafer_detect_timeout", int),
|
|
("_wafer_retries", "wafer_read_retries", int),
|
|
)
|
|
|
|
def __init__(
|
|
self,
|
|
settings: LocalSettings,
|
|
data_dir: Path | str,
|
|
parent: QObject | None = None,
|
|
) -> None:
|
|
super().__init__(parent)
|
|
# Shared with DeviceController — saving here must mutate this same
|
|
# instance, not a throwaway copy, or the device layer keeps using
|
|
# stale config until restart.
|
|
self._settings = settings
|
|
self._data_dir = Path(data_dir)
|
|
self._defaults = self._new_defaults()
|
|
|
|
for draft_attr, default_value in self._defaults.items():
|
|
setattr(self, draft_attr, deepcopy(default_value))
|
|
|
|
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 _new_defaults(self) -> dict[str, Any]:
|
|
return {
|
|
"_chamber_id": "2",
|
|
"_reverse_z_wafer": False,
|
|
"_debug_mode": False,
|
|
"_wafer_read_timeout": 120000,
|
|
"_wafer_detect_timeout": 5000,
|
|
"_wafer_retries": 8,
|
|
"_masters": {family: "" for family in MASTER_FAMILIES},
|
|
}
|
|
|
|
def _snapshot(self) -> dict[str, Any]:
|
|
snap = {attr: getattr(self, attr) for attr, _, _ in self._FIELD_MAP}
|
|
snap["_masters"] = deepcopy(self._masters)
|
|
return snap
|
|
|
|
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 _normalized_masters(self, source: dict[str, str]) -> dict[str, str]:
|
|
masters = {family: "" for family in MASTER_FAMILIES}
|
|
for family, value in source.items():
|
|
normalized = str(family).strip().upper()
|
|
if normalized in masters:
|
|
masters[normalized] = str(value).strip()
|
|
return masters
|
|
|
|
@Property(str, notify=chamberIdChanged)
|
|
def chamberId(self) -> str:
|
|
return self._chamber_id
|
|
|
|
@chamberId.setter # type: ignore[no-redef]
|
|
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 # type: ignore[no-redef]
|
|
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 # type: ignore[no-redef]
|
|
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 # type: ignore[no-redef]
|
|
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 # type: ignore[no-redef]
|
|
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 # type: ignore[no-redef]
|
|
def waferRetries(self, value: int) -> None:
|
|
if self._wafer_retries == value:
|
|
return
|
|
self._wafer_retries = int(value)
|
|
self.waferRetriesChanged.emit()
|
|
self._recompute_derived()
|
|
|
|
@Property(dict, 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:
|
|
"""Seed draft fields from the shared LocalSettings instance."""
|
|
for draft_attr, settings_attr, coerce in self._FIELD_MAP:
|
|
setattr(self, draft_attr, coerce(getattr(self._settings, settings_attr)))
|
|
self._chamber_id = self._chamber_id.strip() or str(self._defaults["_chamber_id"])
|
|
self._masters = self._normalized_masters(self._settings.master)
|
|
|
|
self._saved_snapshot = self._snapshot()
|
|
self._emit_all_changed()
|
|
self._recompute_derived()
|
|
self._set_save_status("Loaded settings")
|
|
|
|
@Slot()
|
|
def saveSettings(self) -> None:
|
|
"""Write draft fields onto the shared LocalSettings instance and persist it."""
|
|
self._recompute_derived()
|
|
if not self._is_valid:
|
|
self._set_save_status("error: invalid settings")
|
|
return
|
|
|
|
try:
|
|
for draft_attr, settings_attr, _coerce in self._FIELD_MAP:
|
|
setattr(self._settings, settings_attr, getattr(self, draft_attr))
|
|
self._settings.master = deepcopy(self._masters)
|
|
LocalSettings.save_settings(str(self._data_dir), self._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
|
|
for draft_attr, _settings_attr, coerce in self._FIELD_MAP:
|
|
setattr(self, draft_attr, coerce(snap[draft_attr]))
|
|
self._masters = deepcopy(snap["_masters"])
|
|
|
|
self._emit_all_changed()
|
|
self._recompute_derived()
|
|
self._set_save_status("Reverted unsaved changes")
|
|
|
|
@Slot()
|
|
def resetDefaults(self) -> None:
|
|
for draft_attr, _settings_attr, coerce in self._FIELD_MAP:
|
|
setattr(self, draft_attr, coerce(self._defaults[draft_attr]))
|
|
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, "")
|