jack
67 lines
2.6 KiB
Python
67 lines
2.6 KiB
Python
"""Read the data section of a Z-wafer CSV into DataRecords, or the official headerless CSV."""
|
|
from __future__ import annotations
|
|
from pathlib import Path
|
|
|
|
from pygui.backend.wafer.zwafer_models import DataRecord, Sensor
|
|
|
|
def read_data_records(file_path: str) -> list[DataRecord]:
|
|
records: list[DataRecord] = []
|
|
in_data = False
|
|
with Path(file_path).open("r", encoding="utf-8") as file_handle:
|
|
for line in file_handle:
|
|
line = line.rstrip().rstrip(",").strip()
|
|
if not line:
|
|
continue
|
|
if not in_data:
|
|
if line.split(",")[0].lower() == "data":
|
|
in_data = True
|
|
continue
|
|
cols = [c.strip() for c in line.split(",") if c.strip() != ""]
|
|
try:
|
|
nums = [float(c) for c in cols]
|
|
except ValueError:
|
|
continue
|
|
if not nums:
|
|
continue
|
|
records.append(DataRecord(time=nums[0], values=nums[1:]))
|
|
return records
|
|
|
|
|
|
def is_official_csv(file_path: str) -> bool:
|
|
"""Return True if the file uses the headerless official format (sensor-name header row, no 'data' sentinel)."""
|
|
with Path(file_path).open("r", encoding="utf-8") as f:
|
|
first = f.readline().rstrip().rstrip(",")
|
|
if not first:
|
|
return False
|
|
parts = [p.strip() for p in first.split(",") if p.strip()]
|
|
# Official format: first cell looks like "SensorN" (starts with letter, not key=value)
|
|
return bool(parts) and parts[0].lower().startswith("sensor") and "=" not in parts[0]
|
|
|
|
|
|
def read_official_csv(file_path: str) -> tuple[list[Sensor], list[DataRecord]]:
|
|
"""Parse the headerless official CSV: row 0 = sensor names, rows 1+ = values (no timestamp)."""
|
|
from pygui.backend.wafer.wafer_layouts import load_layout_for_wafer_id
|
|
|
|
stem = Path(file_path).stem
|
|
wafer_id = stem.split("-")[0] if "-" in stem else stem
|
|
try:
|
|
sensors = load_layout_for_wafer_id(wafer_id)
|
|
except KeyError:
|
|
return [], []
|
|
|
|
records: list[DataRecord] = []
|
|
with Path(file_path).open("r", encoding="utf-8") as f:
|
|
f.readline() # skip sensor-name header
|
|
for i, line in enumerate(f):
|
|
line = line.rstrip().rstrip(",")
|
|
if not line:
|
|
continue
|
|
cols = [c.strip() for c in line.split(",") if c.strip()]
|
|
try:
|
|
values = [float(c) for c in cols]
|
|
except ValueError:
|
|
continue
|
|
if values:
|
|
records.append(DataRecord(time=float(i) * 0.5, values=values))
|
|
return sensors, records
|