feat(export): batch wafer-map PNG export from file list
- BatchExport worker + BatchExportController (threaded, busy flag) - SourcePanel: Export Maps button → FolderDialog → batch export with progress/summary banner (dismissible) - StreamControlPanel: name single export after loaded CSV + timestamp
This commit is contained in:
@@ -0,0 +1,62 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import threading
|
||||
|
||||
from PySide6.QtCore import Property, QObject, Qt, Signal, Slot
|
||||
|
||||
from pygui.backend.utils import slot_error_boundary
|
||||
from pygui.backend.wafer.batch_export import BatchExportResult, export_batch
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class BatchExportController(QObject):
|
||||
busyChanged = Signal()
|
||||
exportFinished = Signal(str)
|
||||
|
||||
_batchFinished = Signal(list)
|
||||
|
||||
def __init__(self, parent: QObject | None = None) -> None:
|
||||
super().__init__(parent)
|
||||
self._busy = False
|
||||
self._batchFinished.connect(self._handle_batch_finished, Qt.ConnectionType.QueuedConnection)
|
||||
|
||||
@Property(bool, notify=busyChanged)
|
||||
def busy(self) -> bool:
|
||||
return self._busy
|
||||
|
||||
@Slot("QVariantList", str)
|
||||
def start(self, file_paths: list, output_dir: str) -> None:
|
||||
if self._busy:
|
||||
return
|
||||
self._busy = True
|
||||
self.busyChanged.emit()
|
||||
threading.Thread(
|
||||
target=self._worker, args=(list(file_paths), output_dir), daemon=True
|
||||
).start()
|
||||
|
||||
def _worker(self, file_paths: list[str], output_dir: str) -> None:
|
||||
try:
|
||||
results = export_batch(file_paths, output_dir)
|
||||
except Exception as exc:
|
||||
log.exception("Batch export worker crashed: %s", exc)
|
||||
results = []
|
||||
self._batchFinished.emit(results)
|
||||
|
||||
@Slot(list)
|
||||
@slot_error_boundary
|
||||
def _handle_batch_finished(self, results: list[BatchExportResult]) -> None:
|
||||
try:
|
||||
total = len(results)
|
||||
succeeded = sum(1 for r in results if r.success)
|
||||
failed = total - succeeded
|
||||
summary = (
|
||||
f"Exported {succeeded}/{total} — {failed} failed"
|
||||
if failed
|
||||
else f"Exported {succeeded}/{total}"
|
||||
)
|
||||
self.exportFinished.emit(summary)
|
||||
finally:
|
||||
self._busy = False
|
||||
self.busyChanged.emit()
|
||||
@@ -0,0 +1,68 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
from pygui.backend.data.data_records import (
|
||||
is_official_csv,
|
||||
read_data_records,
|
||||
read_official_csv,
|
||||
)
|
||||
from pygui.backend.models.frame_player import frames_from_wafer_data
|
||||
from pygui.backend.visualization.wafer_map_item import WaferMapItem
|
||||
from pygui.backend.wafer.zwafer_parser import ZWaferParser
|
||||
|
||||
_EXPORT_SIZE = 600
|
||||
|
||||
|
||||
@dataclass
|
||||
class BatchExportResult:
|
||||
file_path: str
|
||||
output_path: str
|
||||
success: bool
|
||||
error: str = ""
|
||||
|
||||
|
||||
def export_batch(
|
||||
file_paths: list[str], output_dir: str, timestamp: str | None = None
|
||||
) -> list[BatchExportResult]:
|
||||
"""Render the last frame of each file to a PNG in output_dir.
|
||||
|
||||
Skips files that fail to parse or render, continuing with the rest.
|
||||
"""
|
||||
ts = timestamp or datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
results = []
|
||||
for file_path in file_paths:
|
||||
results.append(_export_one(file_path, output_dir, ts))
|
||||
return results
|
||||
|
||||
|
||||
def _export_one(file_path: str, output_dir: str, ts: str) -> BatchExportResult:
|
||||
base_name = Path(file_path).stem
|
||||
output_path = str(Path(output_dir) / f"{base_name}_{ts}.png")
|
||||
try:
|
||||
if is_official_csv(file_path):
|
||||
sensors, records = read_official_csv(file_path)
|
||||
else:
|
||||
data, _ = ZWaferParser().parse(file_path)
|
||||
if data is None or not data.sensors:
|
||||
return BatchExportResult(file_path, output_path, False, "Invalid layout or missing sensors")
|
||||
sensors = data.sensors
|
||||
records = read_data_records(file_path)
|
||||
|
||||
frames = frames_from_wafer_data(None, records)
|
||||
if not sensors or not frames:
|
||||
return BatchExportResult(file_path, output_path, False, "No sensors or frames found")
|
||||
|
||||
item = WaferMapItem()
|
||||
item.setWidth(_EXPORT_SIZE)
|
||||
item.setHeight(_EXPORT_SIZE)
|
||||
item.sensors = [{"label": s.label, "x": s.x, "y": s.y} for s in sensors] # type: ignore[method-assign]
|
||||
item.values = frames[-1].values # type: ignore[method-assign]
|
||||
|
||||
if not item.export_image(output_path):
|
||||
return BatchExportResult(file_path, output_path, False, "Export render failed")
|
||||
return BatchExportResult(file_path, output_path, True)
|
||||
except Exception as exc:
|
||||
return BatchExportResult(file_path, output_path, False, str(exc))
|
||||
Reference in New Issue
Block a user