Refactor DataSource classes and enhance testing framework
- Removed outdated comments and documentation from CsvSessionSource and DemoSource classes. - Introduced new unit tests for CsvSource and DataSource functionalities, ensuring robust validation of CSV loading and session handling. - Added UI smoke tests to verify window instantiation and menu actions without hardware dependencies. - Implemented VisionEngine tests to validate demo snapshots and alignment state logic.
This commit is contained in:
Your Name
@@ -20,21 +20,7 @@ class DataSource(Protocol):
|
||||
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)
|
||||
@@ -61,19 +47,8 @@ class CsvSessionSource:
|
||||
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
|
||||
|
||||
@@ -83,21 +58,4 @@ class DemoSource:
|
||||
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: ...
|
||||
# ---------------------------------------------------------------------------
|
||||
# TODO Post-scope hook for MqttSource implementing DataSource.
|
||||
|
||||
@@ -10,23 +10,6 @@ from .qt import require_qt
|
||||
|
||||
QtCore, QtGui, QtWidgets = require_qt()
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# TODO P3.1: Add _STATE_COLORS and _DEFAULT_COLOR module-level constants here
|
||||
#
|
||||
# THINKING:
|
||||
# Module-level avoids reconstructing the dict on every paintEvent repaint.
|
||||
# Single source of truth for colors used in two places: drawing calls (P3.2)
|
||||
# and the legend swatch (P3.3). _DEFAULT_COLOR retains the original orange
|
||||
# so any snapshot predating P1.2 renders with pre-existing orange rather than
|
||||
# silently going red. Cross-reference: used by P3.2 and P3.3.
|
||||
#
|
||||
# _STATE_COLORS: dict[str, str] = {
|
||||
# "ready": "#22c55e",
|
||||
# "aligning": "#f59e0b",
|
||||
# "fault": "#ef4444",
|
||||
# }
|
||||
# _DEFAULT_COLOR = "#f97316" # defensive fallback; unexpected alignment_state only
|
||||
# ---------------------------------------------------------------------------
|
||||
_STATE_COLORS: dict[str, str] = {
|
||||
"ready": "#22c55e",
|
||||
"aligning": "#f59e0b",
|
||||
@@ -37,10 +20,6 @@ _DEFAULT_COLOR = "#f97316"
|
||||
|
||||
class MonitorPlot(QtWidgets.QWidget):
|
||||
"""Operator monitor plot driven by a UI snapshot."""
|
||||
|
||||
# Step-down ladder for auto-clamp. Try the operator's preferred multiplier
|
||||
# first; if the projected detected circle would crash into the camera
|
||||
# labels, fall through to the next entry until something fits.
|
||||
_CLAMP_LADDER: tuple[int, ...] = (100, 50, 10, 1)
|
||||
# Camera labels live at radius * 1.18. Keep the detected circle's outer
|
||||
# edge inside radius * 1.15 so there's a small visual breathing margin.
|
||||
@@ -148,8 +127,10 @@ class MonitorPlot(QtWidgets.QWidget):
|
||||
# Without this flip, a wafer drifting toward CAM 3 (bottom) would
|
||||
# render upward toward CAM 1 — opposite of physical reality.
|
||||
exag_center = QtCore.QPointF(
|
||||
center.x() + float(detected_center_mm[0]) * scale * self._effective_multiplier,
|
||||
center.y() - float(detected_center_mm[1]) * scale * self._effective_multiplier,
|
||||
center.x()
|
||||
+ float(detected_center_mm[0]) * scale * self._effective_multiplier,
|
||||
center.y()
|
||||
- float(detected_center_mm[1]) * scale * self._effective_multiplier,
|
||||
)
|
||||
color = QtGui.QColor(
|
||||
_STATE_COLORS.get(self._snapshot.alignment_state, _DEFAULT_COLOR)
|
||||
|
||||
@@ -0,0 +1,134 @@
|
||||
"""Tests for CsvSource CSV loader."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
|
||||
import sys
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
|
||||
|
||||
from wafer_edge.Calibration import BaselineCalibration
|
||||
from wafer_edge.CsvSource import find_session_files, load_camera_records_from_csv, parse_csv_filename
|
||||
|
||||
|
||||
BASELINE = BaselineCalibration()
|
||||
MASTER_SD = Path(__file__).resolve().parents[1] / "MASTER_SD"
|
||||
DATA_DIR = Path(__file__).resolve().parents[1] / "DATA" / "0610"
|
||||
|
||||
|
||||
class CsvSourceTests(unittest.TestCase):
|
||||
"""Test CSV loader correctness and robustness."""
|
||||
|
||||
def test_point_count_matches_header(self) -> None:
|
||||
"""Test 1: load S10519P0.CSV, assert parser reads all data rows."""
|
||||
# Read raw point count from file header (before filtering)
|
||||
with open(MASTER_SD / "S10519P0.CSV") as f:
|
||||
for line in f:
|
||||
if line.startswith("# points,"):
|
||||
raw_count = int(line.split(",")[1].strip())
|
||||
break
|
||||
# After filter_ring_outliers, non-wafer edges are stripped
|
||||
# but a significant fraction should remain for a good capture
|
||||
paths = {"cam-1": MASTER_SD / "S10519P0.CSV"}
|
||||
records = load_camera_records_from_csv(paths, BASELINE)
|
||||
self.assertEqual(len(records), 1)
|
||||
self.assertEqual(records[0].camera_id, "cam-1")
|
||||
filtered = len(records[0].points)
|
||||
self.assertGreater(filtered, 0, "No points loaded from CSV")
|
||||
self.assertGreaterEqual(raw_count, 100, f"Expected >= 100 raw points, got {raw_count}")
|
||||
# Filtered should be a reasonable fraction (at least 5% of raw)
|
||||
self.assertGreaterEqual(filtered, raw_count * 0.05,
|
||||
f"Filtered {filtered} << raw {raw_count}")
|
||||
|
||||
def test_mm_values_in_reasonable_range(self) -> None:
|
||||
"""Test 2: assert all mm values are within ±100 mm (physically reasonable)."""
|
||||
paths = {"cam-1": MASTER_SD / "S10519P0.CSV"}
|
||||
records = load_camera_records_from_csv(paths, BASELINE)
|
||||
pts = np.array(records[0].points)
|
||||
# All coordinates should be within ±100 mm for a 75mm radius wafer
|
||||
self.assertTrue(np.all(np.abs(pts) < 100.0),
|
||||
f"Out-of-range values: min={pts.min():.1f}, max={pts.max():.1f}")
|
||||
|
||||
def test_four_cameras_returns_four_records(self) -> None:
|
||||
"""Test 3: 4 paths → 4 records with correct camera IDs."""
|
||||
paths = {
|
||||
"cam-1": MASTER_SD / "S10540P0.CSV",
|
||||
"cam-2": MASTER_SD / "S20540P0.CSV",
|
||||
"cam-3": MASTER_SD / "R10540P0.CSV",
|
||||
"cam-4": MASTER_SD / "R20539P0.CSV",
|
||||
}
|
||||
records = load_camera_records_from_csv(paths, BASELINE)
|
||||
self.assertEqual(len(records), 4)
|
||||
record_ids = {r.camera_id for r in records}
|
||||
self.assertEqual(record_ids, {"cam-1", "cam-2", "cam-3", "cam-4"})
|
||||
# All should have points and ok status
|
||||
for r in records:
|
||||
self.assertGreater(len(r.points), 0, f"{r.camera_id} has no points")
|
||||
self.assertEqual(r.status, "ok")
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# find_session_files tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_find_session_files_complete(self) -> None:
|
||||
"""All 4 cameras present → dict with 4 entries, correct labels."""
|
||||
result = find_session_files(DATA_DIR)
|
||||
self.assertEqual(len(result), 4)
|
||||
self.assertIn("cam-1 (S1)", result)
|
||||
self.assertIn("cam-2 (S2)", result)
|
||||
self.assertIn("cam-3 (R1)", result)
|
||||
self.assertIn("cam-4 (R2)", result)
|
||||
for label, path in result.items():
|
||||
self.assertTrue(path.exists(), f"{label} → {path}")
|
||||
|
||||
def test_find_session_files_partial(self) -> None:
|
||||
"""Only 2 cameras present → returns dict with 2 entries, no error."""
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
tmpdir = Path(tmp)
|
||||
for name in ("S10610P0.CSV", "R10610P0.CSV"):
|
||||
(tmpdir / name).write_text(
|
||||
(DATA_DIR / name).read_text()
|
||||
)
|
||||
result = find_session_files(tmpdir)
|
||||
self.assertEqual(len(result), 2)
|
||||
self.assertIn("cam-1 (S1)", result)
|
||||
self.assertIn("cam-3 (R1)", result)
|
||||
self.assertNotIn("cam-2 (S2)", result)
|
||||
self.assertNotIn("cam-4 (R2)", result)
|
||||
|
||||
def test_find_session_files_empty_folder(self) -> None:
|
||||
"""No CSVs at all → empty dict, no exception."""
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
result = find_session_files(Path(tmp))
|
||||
self.assertEqual(result, {})
|
||||
|
||||
def test_find_session_files_bad_filename(self) -> None:
|
||||
"""Non-NanoBerry file is silently skipped."""
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
tmpdir = Path(tmp)
|
||||
(tmpdir / "README.TXT").write_text("hello")
|
||||
(tmpdir / "S10610P0.CSV").write_text(
|
||||
(DATA_DIR / "S10610P0.CSV").read_text()
|
||||
)
|
||||
result = find_session_files(tmpdir)
|
||||
self.assertEqual(len(result), 1)
|
||||
self.assertIn("cam-1 (S1)", result)
|
||||
|
||||
def test_find_session_files_new_format(self) -> None:
|
||||
"""Underscore-style filenames (R1_0540_P0.CSV) are also discovered."""
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
tmpdir = Path(tmp)
|
||||
(tmpdir / "S1_0610_P0.CSV").write_text(
|
||||
(DATA_DIR / "S10610P0.CSV").read_text()
|
||||
)
|
||||
result = find_session_files(tmpdir)
|
||||
self.assertEqual(len(result), 1)
|
||||
self.assertIn("cam-1 (S1)", result)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,34 @@
|
||||
"""DataSource TODO anchors for future headless tests."""
|
||||
|
||||
from __future__ import annotations
|
||||
import unittest
|
||||
|
||||
from wafer_edge.DataSources import DemoSource, CsvSessionSource
|
||||
from wafer_edge.VisionEngine import VisionEngine
|
||||
from wafer_edge.Calibration import BaselineCalibration
|
||||
from pathlib import Path
|
||||
import tempfile
|
||||
import MASTER_SD
|
||||
MASTER_SD = Path(__file__).resolve().parents[1] / "MASTER_SD"
|
||||
|
||||
class DataSourcesTest(unittest.TestCase):
|
||||
def test_demo_source_snapshot(self) -> None:
|
||||
sources = DemoSource(VisionEngine())
|
||||
snapshot = sources.latest_snapshot(BaselineCalibration())
|
||||
self.assertEqual(len(snapshot.camera_readings), 4)
|
||||
self.assertEqual(snapshot.source_label, "User App - Simulated Data")
|
||||
self.assertLess(snapshot.sigma_mm, 0.01)
|
||||
|
||||
def test_csv_session_source_snapshot(self) -> None:
|
||||
source = CsvSessionSource(MASTER_SD, VisionEngine())
|
||||
snapshot = source.latest_snapshot(BaselineCalibration())
|
||||
|
||||
self.assertEqual(len(snapshot.camera_readings), 4)
|
||||
self.assertEqual(snapshot.system_status, "Ready")
|
||||
self.assertEqual(snapshot.source_label, "Session: MASTER_SD (4 cams)")
|
||||
|
||||
def test_csv_session_source_missing_or_invalid_seed(self)-> None:
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
with self.assertRaises(FileNotFoundError):
|
||||
CsvSessionSource(Path(tmp), VisionEngine())
|
||||
|
||||
@@ -0,0 +1,58 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
|
||||
|
||||
try:
|
||||
from PySide6 import QtWidgets
|
||||
|
||||
PYSIDE6_IMPORT_ERROR = None
|
||||
except Exception as exc:
|
||||
QtWidgets = None
|
||||
PYSIDE6_IMPORT_ERROR = exc
|
||||
|
||||
|
||||
@unittest.skipIf(PYSIDE6_IMPORT_ERROR is not None, "PySide6 is not installed")
|
||||
class UiSmokeTests(unittest.TestCase):
|
||||
def test_windows_instantiate_without_hardware(self) -> None:
|
||||
from wafer_edge.App import UserWindow
|
||||
|
||||
app = QtWidgets.QApplication.instance() or QtWidgets.QApplication([])
|
||||
|
||||
user = UserWindow()
|
||||
|
||||
self.assertEqual(user.windowTitle(), "Wafer Position Monitor")
|
||||
self.assertEqual(user.source_label.text(), "User App")
|
||||
app.quit()
|
||||
|
||||
def test_file_menu_has_open_session_action(self) -> None:
|
||||
"""Verify File → Open Session… action exists in the menu bar."""
|
||||
from wafer_edge.App import UserWindow
|
||||
|
||||
app = QtWidgets.QApplication.instance() or QtWidgets.QApplication([])
|
||||
|
||||
user = UserWindow()
|
||||
menu_bar = user.menuBar()
|
||||
self.assertIsNotNone(menu_bar)
|
||||
|
||||
file_menu = None
|
||||
for action in menu_bar.actions():
|
||||
menu = action.menu()
|
||||
if menu is not None and menu.title() in ("File", "&File"):
|
||||
file_menu = menu
|
||||
break
|
||||
|
||||
self.assertIsNotNone(file_menu, "File menu should exist in the menu bar")
|
||||
|
||||
# Check that "Open Session…" is one of the menu actions.
|
||||
action_texts = {a.text() for a in file_menu.actions()}
|
||||
self.assertIn(
|
||||
"Open Session…",
|
||||
action_texts,
|
||||
"File menu should contain an 'Open Session…' action",
|
||||
)
|
||||
app.quit()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,80 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
from wafer_edge.Calibration import BaselineCalibration, load_baseline, save_baseline
|
||||
from wafer_edge.VisionEngine import AlignmentThresholds, VisionEngine, fit_circle
|
||||
|
||||
class VisionEngineTests(unittest.TestCase):
|
||||
def test_demo_snapshot_matches_beta_values(self) -> None:
|
||||
snapshot = VisionEngine().demo_snapshot(BaselineCalibration())
|
||||
|
||||
self.assertEqual(snapshot.source_label, "User App - Simulated Data")
|
||||
self.assertEqual(snapshot.system_status, "Ready")
|
||||
self.assertAlmostEqual(snapshot.x_offset_mm, 2.3, places=6)
|
||||
self.assertAlmostEqual(snapshot.y_offset_mm, -1.1, places=6)
|
||||
self.assertAlmostEqual(snapshot.theta_offset_deg, 0.5, places=1)
|
||||
self.assertLess(snapshot.sigma_mm, 0.001)
|
||||
self.assertEqual(len(snapshot.camera_readings), 4)
|
||||
self.assertEqual(snapshot.camera_readings[2].status, "ok")
|
||||
|
||||
def test_fit_circle_from_edge_points(self) -> None:
|
||||
circle = fit_circle([(1.0, 0.0), (0.0, 1.0), (-1.0, 0.0), (0.0, -1.0)])
|
||||
|
||||
self.assertAlmostEqual(circle.center_mm[0], 0.0)
|
||||
self.assertAlmostEqual(circle.center_mm[1], 0.0)
|
||||
self.assertAlmostEqual(circle.radius_mm, 1.0)
|
||||
|
||||
def test_calibration_round_trip(self) -> None:
|
||||
baseline = BaselineCalibration(
|
||||
center_mm=(1.0, 2.0), radius_mm=3.0, theta_arm_mm=4.0
|
||||
)
|
||||
|
||||
with tempfile.TemporaryDirectory() as temp_dir:
|
||||
path = Path(temp_dir) / "baseline.json"
|
||||
save_baseline(baseline, path)
|
||||
loaded = load_baseline(path)
|
||||
|
||||
self.assertEqual(loaded, baseline)
|
||||
|
||||
class AlignmentStateTests(unittest.TestCase):
|
||||
def test_ready_state(self)-> None:
|
||||
engine = VisionEngine()
|
||||
# offset ~ 0.07mm from (0.05, 0.05), sigma = 0.01
|
||||
state = engine.get_alignment_state(0.05, 0.05, 0.01)
|
||||
self.assertEqual(state, "ready")
|
||||
|
||||
def test_aligning_state(self)-> None:
|
||||
engine = VisionEngine()
|
||||
state = engine.get_alignment_state(0.20, 0.0, 0.10)
|
||||
self.assertEqual(state, "aligning")
|
||||
|
||||
def test_fault_by_offset(self) -> None:
|
||||
engine = VisionEngine()
|
||||
state = engine.get_alignment_state(0.60, 0.0, 0.01)
|
||||
self.assertEqual(state, "fault")
|
||||
def test_fault_by_sigma(self) -> None:
|
||||
engine = VisionEngine()
|
||||
state = engine.get_alignment_state(0.05, 0.0, 0.25)
|
||||
self.assertEqual(state, "fault")
|
||||
def test_boundary_ready(self) -> None:
|
||||
engine = VisionEngine()
|
||||
state =engine.get_alignment_state(0.10, 0.0, 0.05)
|
||||
self.assertEqual(state, "ready")
|
||||
def test_demo_snapshot_alignment_state(self) -> None:
|
||||
engine = VisionEngine()
|
||||
snapshot = engine.demo_snapshot(BaselineCalibration())
|
||||
self.assertEqual(snapshot.alignment_state, "fault")
|
||||
|
||||
class AlignmentThresholdInjectionTest(unittest.TestCase):
|
||||
def test_loose_thresholds_make_demo_ready(self) -> None:
|
||||
engine = VisionEngine(thresholds = AlignmentThresholds(
|
||||
ready_offset_mm = 5.0, ready_sigma_mm = 5.0
|
||||
))
|
||||
snapshot = engine.demo_snapshot(BaselineCalibration())
|
||||
self.assertEqual(snapshot.alignment_state, "ready")
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user