jack
108 lines
3.6 KiB
Python
108 lines
3.6 KiB
Python
from pygui.backend.data.csv_recorder import CsvRecorder
|
|
from pygui.backend.data.data_records import read_data_records
|
|
from pygui.backend.models.frame import Frame
|
|
from pygui.backend.wafer.zwafer_models import Sensor
|
|
from pygui.backend.wafer.zwafer_parser import ZWaferParser
|
|
|
|
|
|
def test_recording_roundtrips_through_parser(tmp_path):
|
|
sensors = [Sensor("1", 0.0, 1.0), Sensor("2", 1.0, 0.0)]
|
|
path = tmp_path / "rec.csv"
|
|
|
|
rec = CsvRecorder()
|
|
rec.start(str(path), sensors, serial="A12")
|
|
rec.write(Frame(seq=0, time=0.0, values=[149.0, 148.0]))
|
|
rec.write(Frame(seq=1, time=0.5, values=[149.5, 148.5]))
|
|
rec.stop()
|
|
|
|
assert path.exists()
|
|
data, _ = ZWaferParser().parse(str(path))
|
|
assert data is not None
|
|
assert[s.label for s in data.sensors] == ["1", "2"]
|
|
assert len(data.csv_headers) == 2
|
|
|
|
|
|
def test_recorded_data_rows_are_readable(tmp_path):
|
|
sensors = [Sensor("1", 0.0, 1.0), Sensor("2", 1.0, 0.0)]
|
|
path = tmp_path / "rec.csv"
|
|
|
|
rec = CsvRecorder()
|
|
rec.start(str(path), sensors, serial="A12")
|
|
rec.write(Frame(seq=0, time=0.0, values=[149.0, 148.0]))
|
|
rec.write(Frame(seq=1, time=0.5, values=[149.5, 148.5]))
|
|
rec.stop()
|
|
|
|
records = read_data_records(str(path))
|
|
assert len(records) == 2
|
|
assert records[0].time == 0.0
|
|
assert records[0].values == [149.0, 148.0]
|
|
assert records[1].time == 0.5
|
|
assert records[1].values == [149.5, 148.5]
|
|
|
|
# ===== write_segment (P3.3) =====
|
|
def _make_zwafer_csv(tmp_path, rows=5):
|
|
sensors = [Sensor("1", 0.0, 1.0), Sensor("2", 1.0, 0.0)]
|
|
path = tmp_path / "src.csv"
|
|
rec = CsvRecorder()
|
|
rec.start(str(path), sensors, serial="A12")
|
|
for i in range(rows):
|
|
rec.write(Frame(seq=i, time=i * 0.5, values=[100.0 + i, 200.0 + i]))
|
|
rec.stop()
|
|
return path
|
|
|
|
|
|
def test_write_segment_zwafer_slice(tmp_path):
|
|
src = _make_zwafer_csv(tmp_path, rows=5)
|
|
dest = tmp_path / "seg.csv"
|
|
assert CsvRecorder.write_segment(str(dest), str(src), 1, 3) is True
|
|
|
|
records = read_data_records(str(dest))
|
|
assert len(records) == 3
|
|
assert records[0].values == [101.0, 201.0]
|
|
assert records[-1].values == [103.0, 203.0]
|
|
|
|
# header survives → still parses as a wafer CSV
|
|
data, _ = ZWaferParser().parse(str(dest))
|
|
assert data is not None
|
|
assert [s.label for s in data.sensors] == ["1", "2"]
|
|
|
|
|
|
def test_write_segment_full_range(tmp_path):
|
|
src = _make_zwafer_csv(tmp_path, rows=4)
|
|
dest = tmp_path / "seg.csv"
|
|
assert CsvRecorder.write_segment(str(dest), str(src), 0, 3) is True
|
|
assert len(read_data_records(str(dest))) == 4
|
|
|
|
|
|
def test_write_segment_out_of_range(tmp_path):
|
|
src = _make_zwafer_csv(tmp_path, rows=3)
|
|
dest = tmp_path / "seg.csv"
|
|
assert CsvRecorder.write_segment(str(dest), str(src), 10, 20) is False
|
|
assert not dest.exists()
|
|
|
|
|
|
def test_write_segment_invalid_range(tmp_path):
|
|
src = _make_zwafer_csv(tmp_path, rows=3)
|
|
dest = tmp_path / "seg.csv"
|
|
assert CsvRecorder.write_segment(str(dest), str(src), 2, 1) is False
|
|
assert CsvRecorder.write_segment(str(dest), str(src), -1, 1) is False
|
|
|
|
|
|
def test_write_segment_official_csv(tmp_path):
|
|
# official format: row 0 = sensor names, rows 1+ = values only
|
|
src = tmp_path / "A00001-x.csv"
|
|
src.write_text(
|
|
"Sensor1,Sensor2\n"
|
|
"10.0,20.0\n"
|
|
"11.0,21.0\n"
|
|
"12.0,22.0\n",
|
|
encoding="utf-8",
|
|
)
|
|
dest = tmp_path / "A00001-x_seg.csv"
|
|
assert CsvRecorder.write_segment(str(dest), str(src), 1, 2) is True
|
|
lines = dest.read_text(encoding="utf-8").splitlines()
|
|
assert lines[0] == "Sensor1,Sensor2"
|
|
assert lines[1] == "11.0,21.0"
|
|
assert lines[2] == "12.0,22.0"
|
|
assert len(lines) == 3
|