refactor(backend): clean up package re-exports, settings model properties, and unused signals
This commit is contained in:
@@ -2,17 +2,26 @@ from __future__ import annotations
|
||||
|
||||
from copy import deepcopy
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from typing import Any, Callable
|
||||
|
||||
from PySide6.QtCore import Property, QDateTime, QObject, QStandardPaths, Signal, Slot
|
||||
from PySide6.QtCore import Property, QDateTime, QObject, Signal, Slot
|
||||
|
||||
from pygui.backend.data.constants import DEFAULT_DATA_DIR_NAME
|
||||
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()
|
||||
@@ -26,18 +35,35 @@ class LocalSettingsModel(QObject):
|
||||
saveStatusChanged = Signal()
|
||||
lastSavedAtChanged = Signal()
|
||||
|
||||
def __init__(self, parent: QObject | None = None) -> None:
|
||||
# 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)
|
||||
self._data_dir = self._resolve_data_dir()
|
||||
# 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()
|
||||
|
||||
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"])
|
||||
for draft_attr, default_value in self._defaults.items():
|
||||
setattr(self, draft_attr, deepcopy(default_value))
|
||||
|
||||
self._is_dirty = False
|
||||
self._is_valid = True
|
||||
@@ -47,34 +73,21 @@ class LocalSettingsModel(QObject):
|
||||
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 / DEFAULT_DATA_DIR_NAME
|
||||
|
||||
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},
|
||||
"_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]:
|
||||
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),
|
||||
}
|
||||
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:
|
||||
@@ -137,22 +150,19 @@ class LocalSettingsModel(QObject):
|
||||
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
|
||||
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
|
||||
@chamberId.setter # type: ignore[no-redef]
|
||||
def chamberId(self, value: str) -> None:
|
||||
next_value = str(value).strip()
|
||||
if self._chamber_id == next_value:
|
||||
@@ -165,7 +175,7 @@ class LocalSettingsModel(QObject):
|
||||
def reverseZWafer(self) -> bool:
|
||||
return self._reverse_z_wafer
|
||||
|
||||
@reverseZWafer.setter
|
||||
@reverseZWafer.setter # type: ignore[no-redef]
|
||||
def reverseZWafer(self, value: bool) -> None:
|
||||
if self._reverse_z_wafer == value:
|
||||
return
|
||||
@@ -177,7 +187,7 @@ class LocalSettingsModel(QObject):
|
||||
def debugMode(self) -> bool:
|
||||
return self._debug_mode
|
||||
|
||||
@debugMode.setter
|
||||
@debugMode.setter # type: ignore[no-redef]
|
||||
def debugMode(self, value: bool) -> None:
|
||||
if self._debug_mode == value:
|
||||
return
|
||||
@@ -189,7 +199,7 @@ class LocalSettingsModel(QObject):
|
||||
def waferReadTimeout(self) -> int:
|
||||
return self._wafer_read_timeout
|
||||
|
||||
@waferReadTimeout.setter
|
||||
@waferReadTimeout.setter # type: ignore[no-redef]
|
||||
def waferReadTimeout(self, value: int) -> None:
|
||||
if self._wafer_read_timeout == value:
|
||||
return
|
||||
@@ -201,7 +211,7 @@ class LocalSettingsModel(QObject):
|
||||
def waferDetectTimeout(self) -> int:
|
||||
return self._wafer_detect_timeout
|
||||
|
||||
@waferDetectTimeout.setter
|
||||
@waferDetectTimeout.setter # type: ignore[no-redef]
|
||||
def waferDetectTimeout(self, value: int) -> None:
|
||||
if self._wafer_detect_timeout == value:
|
||||
return
|
||||
@@ -213,7 +223,7 @@ class LocalSettingsModel(QObject):
|
||||
def waferRetries(self) -> int:
|
||||
return self._wafer_retries
|
||||
|
||||
@waferRetries.setter
|
||||
@waferRetries.setter # type: ignore[no-redef]
|
||||
def waferRetries(self, value: int) -> None:
|
||||
if self._wafer_retries == value:
|
||||
return
|
||||
@@ -251,22 +261,11 @@ class LocalSettingsModel(QObject):
|
||||
|
||||
@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
|
||||
"""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()
|
||||
@@ -275,13 +274,18 @@ class LocalSettingsModel(QObject):
|
||||
|
||||
@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:
|
||||
LocalSettings.save_settings(str(self._data_dir), self._to_local_settings())
|
||||
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")
|
||||
@@ -293,13 +297,9 @@ class LocalSettingsModel(QObject):
|
||||
@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"])
|
||||
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()
|
||||
@@ -307,13 +307,9 @@ class LocalSettingsModel(QObject):
|
||||
|
||||
@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"])
|
||||
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()
|
||||
|
||||
Reference in New Issue
Block a user