From 958fde90503643ef92f9ff8d6d306fe33adb8029 Mon Sep 17 00:00:00 2001 From: Your Name Date: Thu, 28 May 2026 11:09:53 -0700 Subject: [PATCH] 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. --- src/wafer_edge/DataSources.py | 44 +---------- src/wafer_edge/ui/widgets.py | 27 +------ tests/test_csv_source.py | 134 ++++++++++++++++++++++++++++++++++ tests/test_data_sources.py | 34 +++++++++ tests/test_ui_smoke.py | 58 +++++++++++++++ tests/test_vision_engine.py | 80 ++++++++++++++++++++ 6 files changed, 311 insertions(+), 66 deletions(-) create mode 100644 tests/test_csv_source.py create mode 100644 tests/test_data_sources.py create mode 100644 tests/test_ui_smoke.py create mode 100644 tests/test_vision_engine.py diff --git a/src/wafer_edge/DataSources.py b/src/wafer_edge/DataSources.py index 2c7b05e..2f3512f 100644 --- a/src/wafer_edge/DataSources.py +++ b/src/wafer_edge/DataSources.py @@ -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. diff --git a/src/wafer_edge/ui/widgets.py b/src/wafer_edge/ui/widgets.py index c6f0dc6..138e871 100644 --- a/src/wafer_edge/ui/widgets.py +++ b/src/wafer_edge/ui/widgets.py @@ -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) diff --git a/tests/test_csv_source.py b/tests/test_csv_source.py new file mode 100644 index 0000000..b2d68c3 --- /dev/null +++ b/tests/test_csv_source.py @@ -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() diff --git a/tests/test_data_sources.py b/tests/test_data_sources.py new file mode 100644 index 0000000..1119ba5 --- /dev/null +++ b/tests/test_data_sources.py @@ -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()) + diff --git a/tests/test_ui_smoke.py b/tests/test_ui_smoke.py new file mode 100644 index 0000000..544b60b --- /dev/null +++ b/tests/test_ui_smoke.py @@ -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() diff --git a/tests/test_vision_engine.py b/tests/test_vision_engine.py new file mode 100644 index 0000000..d6582e6 --- /dev/null +++ b/tests/test_vision_engine.py @@ -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()