Your Name
02327a7a21
- AlignmentThresholds dataclass + VisionEngine.__init__ for injection - alignment_state computed per snapshot (ready/aligning/fault) - MonitorPlot circle/vector/legend driven by state color - Status dot + label moved into POSITION VISUALIZATION panel header - CsvSessionSource + DemoSource pluggable DataSource layer - Synthetic DATA/ready/ and DATA/aligning/ sessions for visual testing - Fix CAM 3 label placement (below circle, not above)
104 lines
4.3 KiB
Python
104 lines
4.3 KiB
Python
"""Pluggable data sources for the wafer monitor.
|
|
|
|
A DataSource produces a MonitorSnapshot on demand. Today: CSV session + synthetic
|
|
demo. Future: live MQTT from NanoBerry. App.py only talks to this interface.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from typing import Protocol
|
|
|
|
from .Calibration import BaselineCalibration
|
|
from .CsvSource import find_session_files, load_camera_records_from_csv
|
|
from .VisionEngine import MonitorSnapshot, VisionEngine
|
|
|
|
|
|
class DataSource(Protocol):
|
|
def latest_snapshot(self, baseline: BaselineCalibration) -> MonitorSnapshot: ...
|
|
|
|
def description(self) -> str: ...
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# P2.2 — CsvSessionSource for replaying a selected session.
|
|
#
|
|
# Resolves sibling files once from the seed path using P1.1, then re-reads
|
|
# those files on every latest_snapshot() call. If discovery returns no files,
|
|
# raises FileNotFoundError so a bad operator selection is visible immediately.
|
|
# ---------------------------------------------------------------------------
|
|
class CsvSessionSource:
|
|
"""Reads a captured session from a directory of CSV files.
|
|
|
|
The operator selects the session folder (e.g. ``sessions/0540/``); this
|
|
source discovers all cameras inside it automatically. Re-reads files on
|
|
every ``latest_snapshot()`` call so "Reload Current Session" always reflects
|
|
the latest on-disk state without cache invalidation.
|
|
"""
|
|
|
|
def __init__(self, folder: Path, engine: VisionEngine) -> None:
|
|
self.folder = Path(folder)
|
|
self.engine = engine
|
|
# Discover once at construction so missing-folder errors are immediate.
|
|
paths = find_session_files(self.folder)
|
|
if not paths:
|
|
raise FileNotFoundError(f"No NanoBerry CSV files found in {self.folder}")
|
|
self._cam_count = len(paths)
|
|
|
|
def latest_snapshot(self, baseline: BaselineCalibration) -> MonitorSnapshot:
|
|
# Re-resolve on every call — supports "Reload Current Session".
|
|
paths = find_session_files(self.folder)
|
|
if not paths:
|
|
raise FileNotFoundError(f"No NanoBerry CSV files found in {self.folder}")
|
|
self._cam_count = len(paths)
|
|
records = load_camera_records_from_csv(paths, baseline)
|
|
return self.engine.snapshot_from_records(
|
|
records,
|
|
baseline,
|
|
source_label=self.description(),
|
|
)
|
|
|
|
def description(self) -> str:
|
|
return f"Session: {self.folder.name} ({self._cam_count} cams)"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# P2.3 — DemoSource as the default safe source.
|
|
#
|
|
# First launch should not depend on MASTER_SD being present or on the working
|
|
# directory. Wrapping VisionEngine.demo_snapshot() behind the same protocol
|
|
# means demo mode, CSV replay, and future MQTT all flow through P3.1 with no
|
|
# special-case read logic. The description is stable because P3.3 displays it
|
|
# directly in the status bar.
|
|
# ---------------------------------------------------------------------------
|
|
class DemoSource:
|
|
"""Synthetic perfect-world data — always works, used as fallback."""
|
|
|
|
def __init__(self, engine: VisionEngine) -> None:
|
|
self.engine = engine
|
|
|
|
def latest_snapshot(self, baseline: BaselineCalibration) -> MonitorSnapshot:
|
|
return self.engine.demo_snapshot(baseline)
|
|
|
|
def description(self) -> str:
|
|
return "Demo: Simulated Data"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TODO P6.1: Post-scope hook for MqttSource implementing DataSource.
|
|
#
|
|
# THINKING:
|
|
# Live MQTT should enter the app as another DataSource, not by changing
|
|
# VisionEngine or App.read_wafer(). The source can own broker connection,
|
|
# background subscription, and latest-frame buffers, then convert payloads to
|
|
# CameraEdgeRecord before calling snapshot_from_records(). Keeping this as a
|
|
# P6 task protects the beta session-loader milestone from broker/threading
|
|
# risk while documenting the intended extension point.
|
|
# Cross-reference: P6.2 adds the future App.py action that constructs it.
|
|
#
|
|
# Future signature:
|
|
# class MqttSource:
|
|
# def __init__(self, broker: str, engine: VisionEngine) -> None: ...
|
|
# def latest_snapshot(self, baseline: BaselineCalibration) -> MonitorSnapshot: ...
|
|
# ---------------------------------------------------------------------------
|