Jack.Le
e43bf258e3
- Introduced .gitignore to exclude common Python artifacts and IDE files. - Added contour models for contour generation. - Implemented cryptographic utilities for data encryption and decryption. - Created CSV metadata model for handling wafer data. - Developed time-indexed data segment model. - Established local settings model for application configuration. - Added Z-wafer data models and parser for CSV file handling. - Implemented Marching Squares algorithm for contour generation.
132 lines
4.5 KiB
Python
132 lines
4.5 KiB
Python
from pathlib import Path
|
|
from typing import Tuple, Optional
|
|
from datetime import datetime, timedelta
|
|
|
|
from backend.zwafer_models import ZWaferData, Sensor
|
|
|
|
|
|
# ===== Z-Wafer CSV Parser =====
|
|
class ZWaferParser:
|
|
"""Parses Z-wafer CSV files (header + sensor layout + data rows)."""
|
|
|
|
# ===== Public Parse API =====
|
|
def parse(self, file_path: str) -> Tuple[Optional[ZWaferData], Optional[Path]]:
|
|
"""
|
|
Parse a Z-wafer file.
|
|
|
|
Returns:
|
|
(ZWaferData, Path) on success
|
|
(None, None) on error (e.g., file not found)
|
|
"""
|
|
try:
|
|
path = Path(file_path)
|
|
if not path.exists():
|
|
raise FileNotFoundError(f"File not found: {file_path}")
|
|
|
|
with path.open("r", encoding="utf-8") as f:
|
|
return self._process_header(f), path
|
|
except (ValueError, KeyError):
|
|
raise
|
|
except Exception:
|
|
return None, None
|
|
|
|
# ===== Header Parsing =====
|
|
def _process_header(self, file_obj) -> ZWaferData:
|
|
"""Parse header and sensor layout from open file object."""
|
|
wafer_data = ZWaferData()
|
|
labels: Optional[list] = None
|
|
x_coords: Optional[list] = None
|
|
y_coords: Optional[list] = None
|
|
|
|
for line in file_obj:
|
|
# Strip trailing comma + whitespace
|
|
line = line.rstrip().rstrip(",").strip()
|
|
if not line or line.replace(",", "").strip() == "":
|
|
continue
|
|
|
|
parts = [p.strip() for p in line.split(",")]
|
|
first_part = parts[0].lower()
|
|
|
|
# Detect end of header section
|
|
if first_part == "data":
|
|
wafer_data.csv_headers = labels or []
|
|
self._build_sensor_layout(wafer_data, labels, x_coords, y_coords)
|
|
return wafer_data
|
|
|
|
# Detect switch from metadata to sensor layout
|
|
# Parse metadata (key=value pairs)
|
|
if not self._parse_header_line(wafer_data, parts):
|
|
# If not metadata, it's part of sensor layout
|
|
label = parts[0].lower()
|
|
values = parts[1:]
|
|
|
|
if label == "label":
|
|
labels = values
|
|
elif label == "x (mm)":
|
|
x_coords = values
|
|
elif label == "y (mm)":
|
|
y_coords = values
|
|
|
|
return wafer_data # Incomplete file — return partial data
|
|
|
|
# ===== Metadata Parsing =====
|
|
def _parse_header_line(self, wafer_data: ZWaferData, parts: list) -> bool:
|
|
"""Parse key=value pairs from header line. Returns True if handled."""
|
|
non_empty_parts = [p for p in parts if p]
|
|
if not non_empty_parts:
|
|
return False
|
|
|
|
found_kv = False
|
|
for part in non_empty_parts:
|
|
eq_idx = part.find("=")
|
|
if eq_idx < 0:
|
|
continue
|
|
|
|
key = part[:eq_idx].strip()
|
|
value = part[eq_idx + 1 :].strip('=" ')
|
|
found_kv = True
|
|
|
|
wafer_data.header[key] = value
|
|
|
|
# Extract special fields
|
|
if key.lower() == "acquisition date":
|
|
try:
|
|
from datetime import datetime as dt
|
|
|
|
wafer_data.date = dt.strptime(value, "%m/%d/%Y")
|
|
except ValueError:
|
|
wafer_data.date = datetime.min # Fallback on parse error
|
|
elif key.lower() == "wafer id":
|
|
wafer_data.serial = value
|
|
|
|
if found_kv and wafer_data.date == datetime.min:
|
|
wafer_data.date = datetime.min
|
|
|
|
return found_kv
|
|
|
|
# ===== Sensor Layout Parsing =====
|
|
def _build_sensor_layout(
|
|
self,
|
|
wafer_data: ZWaferData,
|
|
labels: Optional[list],
|
|
x_coords: Optional[list],
|
|
y_coords: Optional[list],
|
|
) -> None:
|
|
"""Build sensor list from layout arrays."""
|
|
if not all([labels, x_coords, y_coords]):
|
|
raise ValueError("Sensor layout section is incomplete or missing.")
|
|
|
|
if len(labels) != len(x_coords) or len(labels) != len(y_coords):
|
|
raise ValueError(
|
|
f"Mismatched sensor columns: labels={len(labels)}, "
|
|
f"x={len(x_coords)}, y={len(y_coords)}"
|
|
)
|
|
|
|
for i in range(len(labels)):
|
|
try:
|
|
wafer_data.sensors.append(
|
|
Sensor(label=labels[i], x=float(x_coords[i]), y=float(y_coords[i]))
|
|
)
|
|
except ValueError as e:
|
|
raise ValueError(f"Invalid coordinate at index {i}: {e}")
|