jack
294 lines
10 KiB
Python
294 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
from PySide6.QtCore import Property, QObject, QStandardPaths, Signal, Slot
|
|
from PySide6.QtWidgets import QFileDialog, QMessageBox
|
|
|
|
from pygui.backend.data.constants import DEFAULT_DATA_DIR_NAME
|
|
from pygui.backend.data.csv_file_metadata import CSVFileMetadata
|
|
from pygui.backend.wafer.zwafer_parser import ZWaferParser
|
|
|
|
|
|
# ===== File Browser Model =====
|
|
class FileBrowser(QObject):
|
|
# ===== Change Signals =====
|
|
filesChanged = Signal()
|
|
currentDirectoryChanged = Signal()
|
|
|
|
# ===== Lifecycle =====
|
|
def __init__(self, parent: QObject | None = None) -> None:
|
|
super().__init__(parent)
|
|
self._parser = ZWaferParser()
|
|
self._files: list[dict[str, object]] = []
|
|
self._current_directory = self._resolve_default_directory()
|
|
self._refresh_files(show_empty_message=False)
|
|
|
|
# ===== Exposed Properties =====
|
|
@Property("QVariantList", notify=filesChanged)
|
|
def files(self) -> list[dict[str, object]]:
|
|
return [dict(row) for row in self._files]
|
|
|
|
@Property(str, notify=currentDirectoryChanged)
|
|
def currentDirectory(self) -> str:
|
|
return str(self._current_directory)
|
|
|
|
# ===== Directory Selection =====
|
|
@Slot()
|
|
def chooseDirectory(self) -> None:
|
|
selected_dir = QFileDialog.getExistingDirectory(
|
|
None,
|
|
"Select CSV Folder",
|
|
self.currentDirectory,
|
|
)
|
|
if not selected_dir:
|
|
return
|
|
|
|
self._set_current_directory(Path(selected_dir))
|
|
self._refresh_files(show_empty_message=True)
|
|
|
|
@Slot()
|
|
def refreshFiles(self) -> None:
|
|
self._refresh_files(show_empty_message=False)
|
|
|
|
# ===== Metadata Persistence =====
|
|
@Slot(str, str, str, str, str, bool, str)
|
|
def saveMetadata(
|
|
self,
|
|
file_path: str,
|
|
wafer: str,
|
|
date: str,
|
|
chamber: str,
|
|
notes: str,
|
|
selected: bool,
|
|
master_type: str = "",
|
|
) -> None:
|
|
csv_path = Path(file_path)
|
|
if not csv_path.exists():
|
|
self._show_error(f"CSV file was not found:\n{file_path}")
|
|
return
|
|
|
|
row = self._find_row(file_path)
|
|
if row is None:
|
|
self._show_error("The selected CSV row is no longer available.")
|
|
return
|
|
|
|
row["wafer"] = wafer
|
|
row["date"] = date
|
|
row["chamber"] = chamber
|
|
row["notes"] = notes
|
|
row["selected"] = selected
|
|
row["masterType"] = master_type
|
|
|
|
payload = {
|
|
"wafer": wafer,
|
|
"date": date,
|
|
"chamber": chamber,
|
|
"notes": notes,
|
|
"masterType": master_type,
|
|
}
|
|
|
|
try:
|
|
sidecar_path = self._sidecar_path(csv_path)
|
|
sidecar_path.write_text(json.dumps(payload, indent=4), encoding="utf-8")
|
|
except Exception as exc: # pragma: no cover - defensive UI path
|
|
self._show_error(f"Failed to save metadata:\n{exc}")
|
|
return
|
|
|
|
self.filesChanged.emit()
|
|
self._show_info(f"Saved metadata for:\n{csv_path.name}")
|
|
|
|
# ===== Internal Data Loading =====
|
|
def _refresh_files(self, show_empty_message: bool) -> None:
|
|
selected_state = {
|
|
str(row.get("fileName", "")): bool(row.get("selected", False))
|
|
for row in self._files
|
|
}
|
|
master_state = {
|
|
str(row.get("fileName", "")): row.get("masterType", "") or ""
|
|
for row in self._files
|
|
}
|
|
|
|
self._files = []
|
|
|
|
if not self._current_directory.exists():
|
|
self.filesChanged.emit()
|
|
return
|
|
|
|
for csv_path in sorted(self._current_directory.glob("*.csv")):
|
|
try:
|
|
metadata = self._load_metadata(csv_path)
|
|
wafer_type = metadata.get_wafer_type().upper() or (csv_path.stem[0].upper() if csv_path.stem else "")
|
|
date_str = metadata.string_date_format()
|
|
serial = metadata.wafer[1:] if len(metadata.wafer) > 1 else ""
|
|
if not serial:
|
|
stem = csv_path.stem
|
|
serial = stem[1:] if len(stem) > 1 else ""
|
|
if "-" in serial:
|
|
serial = serial.split("-")[0]
|
|
if "_" in serial:
|
|
serial = serial.split("_")[0]
|
|
time_str = date_str[11:] if len(date_str) > 10 else ""
|
|
self._files.append(
|
|
{
|
|
"selected": selected_state.get(str(csv_path), False),
|
|
"baseName": csv_path.stem,
|
|
"wafer": metadata.wafer,
|
|
"waferType": wafer_type,
|
|
"serialNumber": serial,
|
|
"date": date_str,
|
|
"timeStr": time_str,
|
|
"chamber": metadata.chamber,
|
|
"notes": metadata.notes,
|
|
"masterType": master_state.get(str(csv_path), "") or metadata.master_type,
|
|
"fileName": str(csv_path),
|
|
"highlight": wafer_type in {"A", "B", "C"},
|
|
}
|
|
)
|
|
except Exception:
|
|
stem = csv_path.stem
|
|
serial_fallback = stem[1:] if len(stem) > 1 else ""
|
|
if "-" in serial_fallback:
|
|
serial_fallback = serial_fallback.split("-")[0]
|
|
if "_" in serial_fallback:
|
|
serial_fallback = serial_fallback.split("_")[0]
|
|
self._files.append(
|
|
{
|
|
"selected": selected_state.get(str(csv_path), False),
|
|
"baseName": stem,
|
|
"wafer": stem,
|
|
"waferType": stem[0].upper() if stem else "",
|
|
"serialNumber": serial_fallback,
|
|
"date": "",
|
|
"timeStr": "",
|
|
"chamber": "",
|
|
"notes": "Unable to parse metadata",
|
|
"masterType": master_state.get(str(csv_path), ""),
|
|
"fileName": str(csv_path),
|
|
"highlight": False,
|
|
}
|
|
)
|
|
|
|
self.filesChanged.emit()
|
|
|
|
if show_empty_message and not self._files:
|
|
self._show_info("No CSV files were found in the selected folder.")
|
|
|
|
def _load_metadata(self, csv_path: Path) -> CSVFileMetadata:
|
|
sidecar_path = self._sidecar_path(csv_path)
|
|
if sidecar_path.exists():
|
|
try:
|
|
payload = json.loads(sidecar_path.read_text(encoding="utf-8"))
|
|
return CSVFileMetadata(
|
|
wafer=str(payload.get("wafer", "")),
|
|
date=str(payload.get("date", "")),
|
|
chamber=str(payload.get("chamber", "")),
|
|
notes=str(payload.get("notes", "")),
|
|
master_type=str(payload.get("masterType", "")),
|
|
filename=str(csv_path),
|
|
)
|
|
except Exception:
|
|
# Fall back to CSV/header parsing if sidecar is malformed.
|
|
pass
|
|
|
|
meta = self._metadata_from_filename(csv_path)
|
|
parser_data, _ = self._parser.parse(str(csv_path))
|
|
if parser_data is not None:
|
|
if parser_data.serial:
|
|
meta.wafer = parser_data.serial
|
|
if parser_data.date != datetime.min:
|
|
meta.date = CSVFileMetadata.format_date(parser_data.date)
|
|
|
|
return meta
|
|
|
|
def _metadata_from_filename(self, csv_path: Path) -> CSVFileMetadata:
|
|
match = re.match(
|
|
r"^(?P<wafer>[^-]+)-(?P<date>\d{8})(?:_(?P<time>\d{6}))?$",
|
|
csv_path.stem,
|
|
)
|
|
if not match:
|
|
return CSVFileMetadata(filename=str(csv_path))
|
|
|
|
date_text = ""
|
|
try:
|
|
date_part = match.group("date")
|
|
time_part = match.group("time") or "000000"
|
|
parsed = datetime.strptime(date_part + time_part, "%Y%m%d%H%M%S")
|
|
date_text = CSVFileMetadata.format_date(parsed)
|
|
except ValueError:
|
|
date_text = ""
|
|
|
|
return CSVFileMetadata(
|
|
wafer=match.group("wafer"),
|
|
date=date_text,
|
|
filename=str(csv_path),
|
|
)
|
|
|
|
# ===== Internal Helpers =====
|
|
def _resolve_default_directory(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 _set_current_directory(self, directory: Path) -> None:
|
|
normalized = Path(directory)
|
|
if normalized == self._current_directory:
|
|
return
|
|
self._current_directory = normalized
|
|
self.currentDirectoryChanged.emit()
|
|
|
|
def _find_row(self, file_path: str) -> dict[str, object] | None:
|
|
for row in self._files:
|
|
if str(row.get("fileName", "")) == file_path:
|
|
return row
|
|
return None
|
|
|
|
def _sidecar_path(self, csv_path: Path) -> Path:
|
|
return csv_path.with_suffix(f"{csv_path.suffix}.json")
|
|
|
|
def _show_error(self, message: str) -> None:
|
|
QMessageBox.critical(None, "iSenseCloud", message)
|
|
|
|
@Slot(str)
|
|
def showInfo(self, message: str) -> None:
|
|
self._show_info(message)
|
|
|
|
@Slot(str, result=dict)
|
|
def parseCsvMetadata(self, file_path: str) -> dict:
|
|
from pathlib import Path as _Path
|
|
|
|
csv_path = _Path(file_path)
|
|
if not csv_path.exists():
|
|
return {"success": False, "error": "File not found"}
|
|
|
|
parser_data, _ = self._parser.parse(file_path)
|
|
if parser_data is None:
|
|
return {"success": False, "error": "Failed to parse CSV"}
|
|
|
|
wafer = parser_data.serial or ""
|
|
date_text = ""
|
|
if parser_data.date != datetime.min:
|
|
date_text = CSVFileMetadata.format_date(parser_data.date)
|
|
columns = len(parser_data.csv_headers) if parser_data.csv_headers else 0
|
|
family_code = wafer[0].upper() if wafer else ""
|
|
|
|
return {
|
|
"success": True,
|
|
"wafer": wafer,
|
|
"date": date_text,
|
|
"columns": columns,
|
|
"familyCode": family_code,
|
|
}
|
|
|
|
def _show_info(self, message: str) -> None:
|
|
QMessageBox.information(None, "iSenseCloud", message)
|
|
|
|
|
|
# Backward-compatible alias while the QML dialog is still named SelectFileDialog.
|
|
SelectFileDialogModel = FileBrowser
|