jack
4bb855a940
Provides a static line chart plotting multi-sensor temperature values over time when a session file is loaded.
315 lines
10 KiB
Python
315 lines
10 KiB
Python
import json
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
|
|
import pygui.backend.controllers.session_controller as session_controller_module
|
|
from pygui.backend.controllers.session_controller import SessionController
|
|
from pygui.backend.models.frame import Frame
|
|
|
|
|
|
@pytest.fixture
|
|
def controller():
|
|
return SessionController()
|
|
|
|
def test_initial_default(controller):
|
|
assert controller.mode == "review"
|
|
assert controller.state == "idle"
|
|
assert not controller.recording
|
|
|
|
def test_playback_flow(controller):
|
|
controller.loadFile("tests/fixtures/sample_stream.csv")
|
|
controller.play()
|
|
initial_index = controller.frameIndex
|
|
controller._advance()
|
|
assert controller.frameIndex == initial_index + 1
|
|
|
|
def test_speed_control(controller):
|
|
controller.setSpeed(2.0)
|
|
interval = controller._next_interval_ms()
|
|
controller.setSpeed(0.5)
|
|
assert controller._next_interval_ms() > interval
|
|
|
|
def test_seek(controller):
|
|
controller.loadFile("tests/fixtures/sample_stream.csv")
|
|
controller.seek(1)
|
|
assert controller.frameIndex == 1
|
|
|
|
def test_step_forward(controller):
|
|
controller.loadFile("tests/fixtures/sample_stream.csv")
|
|
initial = controller.frameIndex
|
|
controller.step(1)
|
|
assert controller.frameIndex == initial + 1
|
|
|
|
def test_step_backward(controller):
|
|
controller.loadFile("tests/fixtures/sample_stream.csv")
|
|
controller.seek(2)
|
|
controller.step(-1)
|
|
assert controller.frameIndex == 1
|
|
|
|
def test_pause_stop_timer(controller):
|
|
controller.loadFile("tests/fixtures/sample_stream.csv")
|
|
controller.seek(2)
|
|
controller.stop()
|
|
assert controller.frameIndex == 0
|
|
|
|
def test_graph_data_empty_before_load(controller):
|
|
assert controller.graphSensorNames == ""
|
|
assert controller.graphSeriesJson == "[]"
|
|
|
|
def test_graph_data_after_load(controller):
|
|
controller.loadFile("tests/fixtures/sample_stream.csv")
|
|
assert controller.graphSensorNames == "Sensor 1,Sensor 2,Sensor 3"
|
|
series = json.loads(controller.graphSeriesJson)
|
|
assert series == [
|
|
[149.0, 149.2, 149.1],
|
|
[148.5, 148.4, 148.6],
|
|
[150.6, 150.7, 150.5],
|
|
]
|
|
|
|
def test_graph_data_cleared_after_unload(controller):
|
|
controller.loadFile("tests/fixtures/sample_stream.csv")
|
|
controller.unloadFile()
|
|
assert controller.graphSensorNames == ""
|
|
assert controller.graphSeriesJson == "[]"
|
|
|
|
def test_leaving_live_mode_blanks_wafer_map(controller):
|
|
"""Stopping a live session (tab switch → setMode) must not leave stale
|
|
sensor values on the wafer map — it should read as blank."""
|
|
frame_signal = MagicMock()
|
|
controller.frameUpdated.connect(frame_signal)
|
|
|
|
controller.setMode("live")
|
|
controller._on_live_frame(Frame(seq=1, time=0.0, values=[100.0] * 22))
|
|
assert controller.sensorValues, "sanity: live frame should populate the map"
|
|
|
|
controller.setMode("review")
|
|
assert controller.sensorValues == []
|
|
assert controller.stats == {}
|
|
assert frame_signal.called, "QML needs frameUpdated to repaint blank"
|
|
|
|
|
|
def test_leaving_live_mode_stops_stream(controller):
|
|
controller.setMode("live")
|
|
controller.setMode("review")
|
|
assert controller._reader is None
|
|
assert not controller._repaint_timer.isActive()
|
|
|
|
|
|
def test_compare_files_integration(controller):
|
|
# Mock the comparisonResult signal
|
|
mock_slot = MagicMock()
|
|
controller.comparisonResult.connect(mock_slot)
|
|
|
|
# Two runs with the same first 3 frames but run B has 2 extra trailing frames.
|
|
controller.compareFiles("tests/fixtures/sample_stream.csv", "tests/fixtures/sample_stream_b.csv")
|
|
|
|
# Verify that the comparison signal is emitted with a success dict
|
|
assert mock_slot.call_count == 1
|
|
args = mock_slot.call_args[0][0]
|
|
assert args["success"] is True
|
|
assert "distance" in args
|
|
assert "series_a" in args
|
|
assert "series_b" in args
|
|
# time_a/time_b expose per-frame timestamps in seconds for the mismatched run lengths.
|
|
assert args["time_a"] == [0.0, 0.5, 1.0]
|
|
assert args["time_b"] == [0.0, 0.5, 1.0, 1.5, 2.0]
|
|
assert args["frame_count_a"] == 3
|
|
assert args["frame_count_b"] == 5
|
|
|
|
def test_compare_files_same_file_is_gated(controller):
|
|
"""Comparing a run against itself should be rejected, not silently DTW'd."""
|
|
mock_slot = MagicMock()
|
|
controller.comparisonResult.connect(mock_slot)
|
|
|
|
csv_path = "tests/fixtures/sample_stream.csv"
|
|
controller.compareFiles(csv_path, csv_path)
|
|
|
|
assert mock_slot.call_count == 1
|
|
args = mock_slot.call_args[0][0]
|
|
assert args["success"] is False
|
|
assert "error" in args
|
|
|
|
def test_comparison_result_reaches_qml(qapp, controller):
|
|
"""The result dict must arrive in QML as a real JS object with fields.
|
|
|
|
Signal(object) delivers Python dicts to QML as an opaque empty wrapper;
|
|
the signal must be declared with a dict/QVariantMap signature.
|
|
"""
|
|
from PySide6.QtQml import QQmlApplicationEngine
|
|
|
|
engine = QQmlApplicationEngine()
|
|
engine.rootContext().setContextProperty("streamController", controller)
|
|
engine.loadData(b"""
|
|
import QtQuick
|
|
Item {
|
|
id: probe
|
|
property bool gotSuccess: false
|
|
property real gotDistance: -1
|
|
Connections {
|
|
target: streamController
|
|
function onComparisonResult(result) {
|
|
probe.gotSuccess = result.success === true
|
|
probe.gotDistance = result.distance !== undefined ? result.distance : -1
|
|
}
|
|
}
|
|
}
|
|
""")
|
|
assert engine.rootObjects(), "probe QML failed to load"
|
|
root = engine.rootObjects()[0]
|
|
|
|
controller.compareFiles("tests/fixtures/sample_stream.csv", "tests/fixtures/sample_stream_b.csv")
|
|
qapp.processEvents()
|
|
|
|
assert root.property("gotSuccess") is True
|
|
assert root.property("gotDistance") >= 0
|
|
|
|
|
|
def test_recording_toggle(controller, tmp_path):
|
|
csv_path = str(tmp_path / "rec" / "live_test.csv")
|
|
controller.startRecording(csv_path, "SN123")
|
|
assert controller.recording
|
|
|
|
controller.stopRecording()
|
|
assert not controller.recording
|
|
|
|
# Header written with wafer serial
|
|
content = open(csv_path).read()
|
|
assert "Wafer ID=SN123" in content
|
|
|
|
def test_resync_count_default(controller):
|
|
# No active reader -> zero
|
|
assert controller.resyncCount == 0
|
|
|
|
controller._reader = MagicMock(resync_count=7)
|
|
assert controller.resyncCount == 7
|
|
controller._reader = None
|
|
|
|
def test_split_data_integration(controller):
|
|
# Mock the splitResult signal
|
|
mock_slot = MagicMock()
|
|
controller.splitResult.connect(mock_slot)
|
|
|
|
csv_path = "tests/fixtures/sample_stream.csv"
|
|
controller.splitData(csv_path, 149.0)
|
|
|
|
# Verify that the split signal is emitted
|
|
assert mock_slot.call_count == 1
|
|
args = mock_slot.call_args[0][0]
|
|
assert args["success"] is True
|
|
assert "segments" in args
|
|
|
|
|
|
|
|
def test_export_segment_after_split(qapp, controller, tmp_path):
|
|
import shutil
|
|
|
|
src = tmp_path / "sample.csv"
|
|
shutil.copyfile("tests/fixtures/sample_stream.csv", src)
|
|
|
|
controller.splitData(str(src), 149.0)
|
|
|
|
exported = MagicMock()
|
|
controller.segmentExported.connect(exported)
|
|
controller.exportSegment(0)
|
|
|
|
assert exported.call_count == 1
|
|
result = exported.call_args[0][0]
|
|
assert result["success"] is True
|
|
out = result["path"]
|
|
assert out.startswith(str(tmp_path))
|
|
from pygui.backend.data.data_records import read_data_records
|
|
assert len(read_data_records(out)) > 0
|
|
|
|
|
|
def test_export_segment_without_split(qapp, controller):
|
|
exported = MagicMock()
|
|
controller.segmentExported.connect(exported)
|
|
controller.exportSegment(0)
|
|
result = exported.call_args[0][0]
|
|
assert result["success"] is False
|
|
|
|
|
|
def test_export_segment_bad_index(qapp, controller, tmp_path):
|
|
import shutil
|
|
src = tmp_path / "sample.csv"
|
|
shutil.copyfile("tests/fixtures/sample_stream.csv", src)
|
|
controller.splitData(str(src), 149.0)
|
|
|
|
exported = MagicMock()
|
|
controller.segmentExported.connect(exported)
|
|
controller.exportSegment(99)
|
|
assert exported.call_args[0][0]["success"] is False
|
|
|
|
|
|
def test_reset_live_trend_sets_start_time_and_emits_reset(controller, monkeypatch):
|
|
monkeypatch.setattr(session_controller_module.time, "monotonic", lambda: 100.0)
|
|
|
|
reset_seen = MagicMock()
|
|
controller.trendReset.connect(reset_seen)
|
|
controller._reset_live_trend()
|
|
|
|
assert controller._stream_start_time == 100.0
|
|
assert reset_seen.called
|
|
|
|
|
|
def test_on_live_frame_emits_trend_delta_with_elapsed_seconds(controller, monkeypatch):
|
|
controller._stream_start_time = 100.0
|
|
monkeypatch.setattr(session_controller_module.time, "monotonic", lambda: 102.5)
|
|
|
|
delta_seen = MagicMock()
|
|
controller.trendDelta.connect(delta_seen)
|
|
|
|
frame = Frame(seq=1, time=102.5, values=[10.0, 20.0, 30.0])
|
|
controller._on_live_frame(frame)
|
|
|
|
payload = json.loads(delta_seen.call_args[0][0])
|
|
assert len(payload) == 1
|
|
elapsed, avg = payload[0]
|
|
assert elapsed == pytest.approx(2.5)
|
|
assert avg == pytest.approx(20.0)
|
|
|
|
|
|
# ---- edge-center stats exposure (ADR-0002) ----
|
|
|
|
def _write_f_family_csv(path):
|
|
"""22-sensor F-family stream: pair (0,18) has delta 1.0, all others 0."""
|
|
n = 22
|
|
values = [149.0] * n
|
|
values[0] = 150.0
|
|
lines = [
|
|
"Wafer ID=F12",
|
|
"Acquisition Date=03/26/2025",
|
|
"Label," + ",".join(str(i + 1) for i in range(n)),
|
|
"X (mm)," + ",".join(str(i) for i in range(n)),
|
|
"Y (mm)," + ",".join(str(i) for i in range(n)),
|
|
"data",
|
|
"0.0," + ",".join(f"{v:.1f}" for v in values),
|
|
]
|
|
path.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
|
|
|
|
|
def test_stats_include_edge_center_for_known_family(controller, tmp_path):
|
|
csv = tmp_path / "f_run.csv"
|
|
_write_f_family_csv(csv)
|
|
controller.loadFile(str(csv))
|
|
s = controller.stats
|
|
assert s["ecMaxDelta"] == pytest.approx(1.0)
|
|
assert s["ecMaxEdgeIndex"] == 1 and s["ecMaxCenterIndex"] == 19 # 1-origin
|
|
assert s["ecMaxEdgeValue"] == pytest.approx(150.0)
|
|
assert s["ecMaxCenterValue"] == pytest.approx(149.0)
|
|
assert s["ecMinDelta"] == pytest.approx(0.0)
|
|
|
|
|
|
def test_stats_edge_center_skips_replaced_sensor(controller, tmp_path):
|
|
csv = tmp_path / "f_run.csv"
|
|
_write_f_family_csv(csv)
|
|
controller.loadFile(str(csv))
|
|
controller.replaceSensor(0, 149.0) # replaced: pair (0,18) must be skipped
|
|
assert controller.stats["ecMaxDelta"] == pytest.approx(0.0)
|
|
|
|
|
|
def test_stats_edge_center_absent_for_unknown_or_short_family(controller):
|
|
controller.loadFile("tests/fixtures/sample_stream.csv") # A12, 3 sensors
|
|
assert "ecMaxDelta" not in controller.stats
|