48c102d2a4
- add process_metrics function to record peak RSS and CPU time - update JSON converter to include metrics in completion marker - modify SQLite converter to utilize new metrics for performance tracking - enhance storage size updates with locking mechanism for concurrent access
140 lines
4.5 KiB
Python
140 lines
4.5 KiB
Python
"""Q1-Q7 over the raw CSV tree: pure python + csv module, streaming.
|
|
|
|
Q1 is a direct path read (the honest CSV strength, spec 08); Q2-Q7 are
|
|
deterministic directory walks with programmatic joins. No manifest
|
|
validation here - queries measure retrieval, not integrity checking.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import csv
|
|
from pathlib import Path
|
|
|
|
from common.queries import Q1_TRACK_CODE
|
|
from common.track_summary import summarize_track
|
|
|
|
|
|
def _rows(path: Path) -> tuple[list[str], list[list[str]]]:
|
|
with open(path, newline="", encoding="ascii") as fh:
|
|
reader = csv.reader(fh)
|
|
header = [h.lower() for h in next(reader)]
|
|
return header, list(reader)
|
|
|
|
|
|
def _dict_row(path: Path) -> dict[str, str]:
|
|
header, rows = _rows(path)
|
|
return dict(zip(header, rows[0]))
|
|
|
|
|
|
def _track_path(csv_root: Path, track_code: str) -> Path:
|
|
batch, wafer, coupon, track = track_code.split("-")
|
|
return csv_root / f"batch_{batch}" / f"wafer_{wafer}" / f"coupon_{coupon}" / f"track_{track}"
|
|
|
|
|
|
def _coupon_dirs(csv_root: Path):
|
|
for batch_dir in sorted(csv_root.glob("batch_*")):
|
|
batch_code = batch_dir.name[6:]
|
|
for wafer_dir in sorted(batch_dir.glob("wafer_*")):
|
|
for coupon_dir in sorted(wafer_dir.glob("coupon_*")):
|
|
yield batch_code, coupon_dir
|
|
|
|
|
|
def _track_dirs(coupon_dir: Path) -> list[Path]:
|
|
return sorted(coupon_dir.glob("track_T*"))
|
|
|
|
|
|
def _cofs(track_dir: Path) -> list[float]:
|
|
_, rows = _rows(track_dir / "cof_vs_cycle.csv")
|
|
return [float(r[1]) for r in rows]
|
|
|
|
|
|
def _mean(values: list[float]) -> float:
|
|
total = 0.0
|
|
for v in values:
|
|
total += v
|
|
return total / len(values)
|
|
|
|
|
|
def q1(csv_root: Path) -> list[tuple]:
|
|
_, rows = _rows(_track_path(csv_root, Q1_TRACK_CODE) / "cof_vs_cycle.csv")
|
|
return [(int(r[0]), float(r[1])) for r in rows]
|
|
|
|
|
|
def q2(csv_root: Path) -> list[tuple]:
|
|
out = []
|
|
for _, coupon_dir in _coupon_dirs(csv_root):
|
|
info = _dict_row(coupon_dir / "coupon_info.csv")
|
|
if info["run_code"] == "RESERVE" or not 9.5 <= float(info["au_wtpct_mean"]) <= 10.5:
|
|
continue
|
|
tracks = _track_dirs(coupon_dir)
|
|
env = _dict_row(tracks[0] / "track_info.csv")["environment"]
|
|
cof_ss = _mean([summarize_track(_cofs(t))[0] for t in tracks])
|
|
out.append((info["coupon_code"], env, cof_ss))
|
|
return out
|
|
|
|
|
|
def q3(csv_root: Path) -> list[tuple]:
|
|
out = []
|
|
for batch_code, coupon_dir in _coupon_dirs(csv_root):
|
|
info = _dict_row(coupon_dir / "coupon_info.csv")
|
|
if info["run_code"] == "RESERVE":
|
|
continue
|
|
header, rows = _rows(coupon_dir / "nanoindentation.csv")
|
|
h_col = header.index("hardness_gpa")
|
|
hardness = _mean([float(r[h_col]) for r in rows])
|
|
cof_ss = _mean([summarize_track(_cofs(t))[0] for t in _track_dirs(coupon_dir)])
|
|
out.append((info["coupon_code"], batch_code, hardness, cof_ss))
|
|
return out
|
|
|
|
|
|
def q4(csv_root: Path) -> list[tuple]:
|
|
out = []
|
|
for _, coupon_dir in _coupon_dirs(csv_root):
|
|
for track_dir in _track_dirs(coupon_dir):
|
|
info = _dict_row(track_dir / "track_info.csv")
|
|
if info["environment"] != "dry_n2" or float(info["load_mn"]) != 100.0:
|
|
continue
|
|
cof_ss = summarize_track(_cofs(track_dir))[0]
|
|
if cof_ss > 0.20:
|
|
out.append((info["track_code"], cof_ss))
|
|
return out
|
|
|
|
|
|
def q5(csv_root: Path) -> list[tuple]:
|
|
sums: dict[str, list[float]] = {}
|
|
for batch_code, coupon_dir in _coupon_dirs(csv_root):
|
|
for track_dir in _track_dirs(coupon_dir):
|
|
run_in = summarize_track(_cofs(track_dir))[2]
|
|
acc = sums.setdefault(batch_code, [0.0, 0.0])
|
|
acc[0] += run_in
|
|
acc[1] += 1
|
|
return [(batch, acc[0] / acc[1]) for batch, acc in sorted(sums.items())]
|
|
|
|
|
|
def q6(csv_root: Path) -> list[tuple]:
|
|
out = []
|
|
for _, coupon_dir in _coupon_dirs(csv_root):
|
|
info = _dict_row(coupon_dir / "coupon_info.csv")
|
|
if info["run_code"] == "RESERVE":
|
|
continue
|
|
tracks = _track_dirs(coupon_dir)
|
|
load = float(_dict_row(tracks[0] / "track_info.csv")["load_mn"])
|
|
wear = _mean([float(_dict_row(t / "wear.csv")["wear_volume_um3"]) for t in tracks])
|
|
out.append((info["coupon_code"], load, wear))
|
|
return out
|
|
|
|
|
|
def q7(csv_root: Path) -> list[tuple]:
|
|
groups: dict[tuple[str, int], list[float]] = {}
|
|
for _, coupon_dir in _coupon_dirs(csv_root):
|
|
for track_dir in _track_dirs(coupon_dir):
|
|
info = _dict_row(track_dir / "track_info.csv")
|
|
bucket = int(round(float(info["speed_mm_s"]) * float(info["load_mn"])))
|
|
cofs = _cofs(track_dir)
|
|
run_in = summarize_track(cofs)[2]
|
|
acc = groups.setdefault((info["environment"], bucket), [0.0, 0.0])
|
|
for cof in cofs[run_in:]: # cycles strictly past run-in
|
|
acc[0] += cof
|
|
acc[1] += 1
|
|
return [(env, bucket, acc[0] / acc[1]) for (env, bucket), acc in sorted(groups.items())]
|