From 48c102d2a4f8dd2b707e348985d98baf9b71207e Mon Sep 17 00:00:00 2001 From: administrator Date: Sat, 11 Jul 2026 21:39:08 -0400 Subject: [PATCH] feat(convert): integrate process metrics into JSON and SQLite converters - 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 --- README.md | 10 +- common/monitoring.py | 66 ++++++ common/pg.py | 16 ++ common/pipeline.py | 20 ++ common/queries/__init__.py | 15 ++ common/queries/csv_queries.py | 139 +++++++++++ common/queries/json_queries.py | 169 ++++++++++++++ common/queries/rdf_queries.py | 173 ++++++++++++++ common/relational.py | 161 +++++++++++++ common/results.py | 73 ++++++ common/storage_sizes.py | 41 +++- convert_json.py | 7 +- convert_postgresql.py | 334 +++++++++++++++++++++++++++ convert_rdf.py | 395 ++++++++++++++++++++++++++++++++ convert_sqlite.py | 205 +++-------------- docs/rules/code-python-style.md | 2 +- make_queries.py | 289 +++++++++++++++++++++++ 17 files changed, 1928 insertions(+), 187 deletions(-) create mode 100644 common/monitoring.py create mode 100644 common/pg.py create mode 100644 common/queries/__init__.py create mode 100644 common/queries/csv_queries.py create mode 100644 common/queries/json_queries.py create mode 100644 common/queries/rdf_queries.py create mode 100644 common/relational.py create mode 100644 common/results.py create mode 100644 convert_postgresql.py create mode 100644 convert_rdf.py create mode 100644 make_queries.py diff --git a/README.md b/README.md index 48689b4..798f0c5 100644 --- a/README.md +++ b/README.md @@ -62,9 +62,9 @@ machine (no parallel work). Every task writes a completion marker | 03 CSV data generation | `generate_data.py` | `out/csv/` tree + `MANIFEST.csv` | implemented | | 04 Convert to JSON-LD | `convert_json.py` | `out/json/full/`, `out/json/hybrid/dataset.jsonld` | implemented | | 05 Convert to SQLite | `convert_sqlite.py` | `out/sqlite/tribo.db` | implemented | -| 06 Convert to PostgreSQL | `convert_postgresql.py` | live `tribo` DB + `out/pg/tribo.dump` | planned | -| 07 Convert to RDF | `convert_rdf.py` | `out/rdf/dataset.nt.gz`, oxigraph store | planned | -| 08 Benchmark queries | `make_queries.py` | `out/bench/queries/`, `out/bench/expected/` | planned | +| 06 Convert to PostgreSQL | `convert_postgresql.py` | live PostgreSQL DB (+ dump when pg_dump is available) | implemented | +| 07 Convert to RDF | `convert_rdf.py` | `out/rdf/dataset.nt.gz`, `dataset.ttl`, oxigraph store | implemented | +| 08 Benchmark queries | `make_queries.py` | `out/bench/queries/`, `out/bench/expected/` | implemented | | 09 Benchmark execution | `bench_runner.py` | `out/bench/results_raw.csv`, `results_median.csv` | planned | | 10 Extrapolation | `extrapolate.py` | `out/bench/extrapolation.csv`, `hardware_sizing.csv` | planned | | 11 Reporting | `report.py` | `out/report/REPORT.md`, charts, tables | planned | @@ -147,11 +147,15 @@ docs/ out/ ALL generated artifacts (git-ignored, reproducible): config/ csv/ json/ sqlite/ pg/ rdf/ bench/ report/ .done/ common/ shared helpers (storage_sizes.csv contract, ...) + queries/ Q1-Q7 implementations for csv / json / rdf formats make_lab_config.py task 01 entry script (one script per task, repo root) make_process_flow.py task 02 entry script generate_data.py task 03 entry script convert_json.py task 04 entry script convert_sqlite.py task 05 entry script +convert_postgresql.py task 06 entry script (TRIBO_PG_DSN, TRIBO_PROM_URL) +convert_rdf.py task 07 entry script +make_queries.py task 08 entry script (canonical results need TRIBO_PG_DSN) requirements.txt closed dependency list (docs/rules/code-python-style.md) ``` diff --git a/common/monitoring.py b/common/monitoring.py new file mode 100644 index 0000000..1630a3e --- /dev/null +++ b/common/monitoring.py @@ -0,0 +1,66 @@ +"""Remote host metrics via Prometheus / node_exporter (stdlib urllib only). + +The PostgreSQL instance is remote, so psutil cannot attribute its CPU/RAM +(spec 09 assumed a local instance). The database host runs Prometheus; +its base URL comes from the TRIBO_PROM_URL environment variable +(e.g. http://192.168.10.73:9090). Unset = monitoring disabled. +""" + +from __future__ import annotations + +import json +import logging +import os +import urllib.parse +import urllib.request + +logger = logging.getLogger(__name__) + +QUERY_STEP_S = 5 +HTTP_TIMEOUT_S = 15 +# rate() window must span at least two scrapes (15 s default interval) +CPU_QUERY = '100 * (1 - avg(rate(node_cpu_seconds_total{mode="idle"}[30s])))' +RAM_QUERY = "node_memory_MemTotal_bytes - node_memory_MemAvailable_bytes" + + +def get_prom_url() -> str: + return os.environ.get("TRIBO_PROM_URL", "") + + +def query_range(base_url: str, promql: str, start: float, end: float) -> list[float]: + params = urllib.parse.urlencode({"query": promql, "start": start, "end": end, "step": QUERY_STEP_S}) + url = base_url.rstrip("/") + "/api/v1/query_range?" + params + with urllib.request.urlopen(url, timeout=HTTP_TIMEOUT_S) as resp: + data = json.load(resp) + if data.get("status") != "success": + raise RuntimeError(f"prometheus query failed: {data.get('errorType')}: {data.get('error')}") + result = data["data"]["result"] + if not result: + return [] + return [float(v) for _, v in result[0]["values"]] + + +def host_window_metrics(base_url: str, start: float, end: float) -> dict | None: + """CPU/RAM of the monitored host over [start, end] wall-clock seconds. + + Returns None (with a warning) on any failure - monitoring is best-effort + and must never fail the owning task. + """ + try: + if end - start < 2 * QUERY_STEP_S: + end = start + 2 * QUERY_STEP_S + cpu = query_range(base_url, CPU_QUERY, start, end) + ram = query_range(base_url, RAM_QUERY, start, end) + if not cpu or not ram: + logger.warning("prometheus returned no samples for the window (expected condition: short window or scrape lag)") + return None + return { + "host_cpu_avg_pct": round(sum(cpu) / len(cpu), 1), + "host_cpu_max_pct": round(max(cpu), 1), + "host_ram_used_baseline_mb": round(ram[0] / 1048576), + "host_ram_used_peak_mb": round(max(ram) / 1048576), + "host_ram_used_delta_mb": round((max(ram) - ram[0]) / 1048576), + } + except Exception: + logger.warning("host metrics collection skipped (prometheus unreachable or query failed)", exc_info=True) + return None diff --git a/common/pg.py b/common/pg.py new file mode 100644 index 0000000..fc9c991 --- /dev/null +++ b/common/pg.py @@ -0,0 +1,16 @@ +"""PostgreSQL DSN access (docs/rules/code-config-yaml.md section 3). + +The DSN comes from the TRIBO_PG_DSN environment variable and is read ONLY +here. It may embed a password: never log it, never hardcode it, never +commit it - log host/database names at most. +""" + +from __future__ import annotations + +import os + +DEFAULT_DSN = "postgresql://postgres@localhost:5432/tribo" + + +def get_dsn() -> str: + return os.environ.get("TRIBO_PG_DSN", DEFAULT_DSN) diff --git a/common/pipeline.py b/common/pipeline.py index 89b83a1..7fdfcbb 100644 --- a/common/pipeline.py +++ b/common/pipeline.py @@ -10,6 +10,7 @@ import logging import sys from pathlib import Path +import psutil import yaml logger = logging.getLogger(__name__) @@ -57,3 +58,22 @@ def write_marker(out_root: Path, task_id: str, entries: dict) -> Path: def load_lab_config(out_root: Path) -> dict: with open(out_root / "config" / "lab_config.yaml", encoding="utf-8") as fh: return yaml.safe_load(fh) + + +def process_metrics() -> dict: + """Peak RSS and CPU time of the current process. + + Every converter records its own build cost in its marker so the task 10 + hardware-sizing model can extrapolate per-format RAM/CPU needs, not just + disk. peak_wset exists on Windows; the rss fallback covers Linux, where + the value is the current (not peak) RSS - good enough for streaming + converters whose RSS is flat by design. + """ + proc = psutil.Process() + mem = proc.memory_info() + peak = getattr(mem, "peak_wset", 0) or mem.rss + cpu = proc.cpu_times() + return { + "proc_peak_rss_mb": round(peak / 1048576, 1), + "proc_cpu_seconds": round(cpu.user + cpu.system, 1), + } diff --git a/common/queries/__init__.py b/common/queries/__init__.py new file mode 100644 index 0000000..482244c --- /dev/null +++ b/common/queries/__init__.py @@ -0,0 +1,15 @@ +"""Benchmark query implementations for the file-based formats (spec 08). + +One module per format (csv_queries, json_queries, rdf_queries), each with +functions q1..q7 returning canonical result rows (see common/results.py). +The runnable wrappers under out/bench/queries/ are generated by +make_queries.py and delegate here. SQLite / PostgreSQL implementations are +plain SQL files written by make_queries.py. + +The relational formats answer Q2/Q3/Q4/Q7 from the precomputed +track_summary (their idiomatic strength); the file formats derive the same +values from raw cycles via common/track_summary.py - identical algorithm, +identical results (bench-methodology section 5). +""" + +Q1_TRACK_CODE = "B722-W2-C13-T2" # fixed benchmark track, spec 08 diff --git a/common/queries/csv_queries.py b/common/queries/csv_queries.py new file mode 100644 index 0000000..cd1b3d6 --- /dev/null +++ b/common/queries/csv_queries.py @@ -0,0 +1,139 @@ +"""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())] diff --git a/common/queries/json_queries.py b/common/queries/json_queries.py new file mode 100644 index 0000000..fee9069 --- /dev/null +++ b/common/queries/json_queries.py @@ -0,0 +1,169 @@ +"""Q1-Q7 over the JSON-LD FULL variant: ijson streaming parser (spec 08). + +Each coupon file is streamed event-by-event; scalar coupon fields appear +before the bulk arrays in the serialization, so filtering queries (Q2) can +abandon a non-matching file before parsing its megabytes of points. +""" + +from __future__ import annotations + +from pathlib import Path + +import ijson + +from common.queries import Q1_TRACK_CODE +from common.track_summary import summarize_track + + +class CouponDoc: + __slots__ = ("coupon_code", "batch_code", "run_code", "au", "hardness", "tracks") + + def __init__(self) -> None: + self.coupon_code = "" + self.batch_code = "" + self.run_code = "" + self.au = 0.0 + self.hardness: list[float] = [] + self.tracks: list[dict] = [] + + +def _scan(path: Path, au_range: tuple[float, float] | None = None) -> CouponDoc | None: + """Stream one coupon file; with au_range set, bail out early on mismatch.""" + doc = CouponDoc() + track: dict | None = None + with open(path, "rb") as fh: + for prefix, event, value in ijson.parse(fh, use_float=True): + if prefix == "coupon_code": + doc.coupon_code = value + elif prefix == "batch_code": + doc.batch_code = value + elif prefix == "run_code": + doc.run_code = value + if doc.run_code == "RESERVE": + return None + elif prefix == "au_wtpct_mean": + doc.au = value + if au_range and not au_range[0] <= value <= au_range[1]: + return None + elif prefix == "nanoindentation.indents.item.hardness_gpa": + doc.hardness.append(value) + elif prefix == "tracks.item" and event == "start_map": + track = {"cycles": [], "cofs": []} + elif track is not None: + if prefix == "tracks.item.track_code": + track["track_code"] = value + elif prefix == "tracks.item.environment": + track["environment"] = value + elif prefix == "tracks.item.load_mn": + track["load_mn"] = value + elif prefix == "tracks.item.speed_mm_s": + track["speed_mm_s"] = value + elif prefix == "tracks.item.cycles.item.cycle": + track["cycles"].append(value) + elif prefix == "tracks.item.cycles.item.cof": + track["cofs"].append(value) + elif prefix == "tracks.item.wear.wear_volume_um3": + track["wear"] = value + elif prefix == "tracks.item" and event == "end_map": + doc.tracks.append(track) + track = None + return doc + + +def _files(json_root: Path) -> list[Path]: + return sorted((json_root / "full").glob("*.jsonld")) + + +def _mean(values: list[float]) -> float: + total = 0.0 + for v in values: + total += v + return total / len(values) + + +def _cof_ss(doc: CouponDoc) -> float: + return _mean([summarize_track(t["cofs"])[0] for t in doc.tracks]) + + +def q1(json_root: Path) -> list[tuple]: + coupon_code = Q1_TRACK_CODE.rsplit("-", 1)[0] + doc = _scan(json_root / "full" / f"{coupon_code}.jsonld") + for track in doc.tracks: + if track["track_code"] == Q1_TRACK_CODE: + return [(int(c), cof) for c, cof in zip(track["cycles"], track["cofs"])] + raise LookupError(f"track {Q1_TRACK_CODE} not found") + + +def q2(json_root: Path) -> list[tuple]: + out = [] + for path in _files(json_root): + doc = _scan(path, au_range=(9.5, 10.5)) + if doc is None: + continue + out.append((doc.coupon_code, doc.tracks[0]["environment"], _cof_ss(doc))) + return out + + +def q3(json_root: Path) -> list[tuple]: + out = [] + for path in _files(json_root): + doc = _scan(path) + if doc is None: + continue + out.append((doc.coupon_code, doc.batch_code, _mean(doc.hardness), _cof_ss(doc))) + return out + + +def q4(json_root: Path) -> list[tuple]: + out = [] + for path in _files(json_root): + doc = _scan(path) + if doc is None: + continue + for track in doc.tracks: + if track["environment"] != "dry_n2" or track["load_mn"] != 100.0: + continue + cof_ss = summarize_track(track["cofs"])[0] + if cof_ss > 0.20: + out.append((track["track_code"], cof_ss)) + return out + + +def q5(json_root: Path) -> list[tuple]: + sums: dict[str, list[float]] = {} + for path in _files(json_root): + doc = _scan(path) + if doc is None: + continue + acc = sums.setdefault(doc.batch_code, [0.0, 0.0]) + for track in doc.tracks: + acc[0] += summarize_track(track["cofs"])[2] + acc[1] += 1 + return [(batch, acc[0] / acc[1]) for batch, acc in sorted(sums.items())] + + +def q6(json_root: Path) -> list[tuple]: + out = [] + for path in _files(json_root): + doc = _scan(path) + if doc is None: + continue + load = doc.tracks[0]["load_mn"] + out.append((doc.coupon_code, load, _mean([t["wear"] for t in doc.tracks]))) + return out + + +def q7(json_root: Path) -> list[tuple]: + groups: dict[tuple[str, int], list[float]] = {} + for path in _files(json_root): + doc = _scan(path) + if doc is None: + continue + for track in doc.tracks: + bucket = int(round(track["speed_mm_s"] * track["load_mn"])) + run_in = summarize_track(track["cofs"])[2] + acc = groups.setdefault((track["environment"], bucket), [0.0, 0.0]) + for cof in track["cofs"][run_in:]: + acc[0] += cof + acc[1] += 1 + return [(env, bucket, acc[0] / acc[1]) for (env, bucket), acc in sorted(groups.items())] diff --git a/common/queries/rdf_queries.py b/common/queries/rdf_queries.py new file mode 100644 index 0000000..fb67740 --- /dev/null +++ b/common/queries/rdf_queries.py @@ -0,0 +1,173 @@ +"""Q1-Q7 over the oxigraph triplestore: SPARQL retrieval (spec 08). + +The store holds raw data only (no precomputed summaries), so queries that +need steady-state COF or run-in retrieve the raw cycles via SPARQL and +derive the values with the shared algorithm (common/track_summary.py) - +identical semantics to every other format. The SPARQL text of each query +is exported to out/bench/queries/rdf/q.sparql by make_queries.py. +""" + +from __future__ import annotations + +from pathlib import Path + +from pyoxigraph import Store + +from common.queries import Q1_TRACK_CODE +from common.track_summary import summarize_track + +PREFIX = "PREFIX tribo: \n" +TRACK_IRI = f"" + +SPARQL = { + "q1": PREFIX + f"""SELECT ?cycle ?cof WHERE {{ + {TRACK_IRI} tribo:hasCycle ?c . + ?c tribo:cycle ?cycle ; tribo:cof ?cof . +}}""", + "q2": PREFIX + """SELECT ?ccode ?env ?track ?cycle ?cof WHERE { + ?coupon tribo:au_wtpct_mean ?au ; tribo:coupon_code ?ccode . + FILTER(?au >= 9.5 && ?au <= 10.5) + ?track tribo:partOf ?coupon ; tribo:environment ?env ; tribo:hasCycle ?c . + ?c tribo:cycle ?cycle ; tribo:cof ?cof . +}""", + "q3_hardness": PREFIX + """SELECT ?ccode ?bcode (AVG(?h) AS ?hardness) WHERE { + ?coupon tribo:coupon_code ?ccode ; tribo:cutFrom ?wafer ; tribo:nanoindentation ?m . + ?wafer tribo:partOf ?batch . + ?batch tribo:batch_code ?bcode . + ?m tribo:hasIndent ?i . + ?i tribo:hardness_gpa ?h . +} GROUP BY ?ccode ?bcode""", + "q3_cycles": PREFIX + """SELECT ?ccode ?track ?cycle ?cof WHERE { + ?track a tribo:FrictionTest ; tribo:partOf ?coupon ; tribo:hasCycle ?c . + ?coupon tribo:coupon_code ?ccode . + ?c tribo:cycle ?cycle ; tribo:cof ?cof . +}""", + "q4": PREFIX + """SELECT ?tcode ?track ?cycle ?cof WHERE { + ?track tribo:environment "dry_n2" ; tribo:load_mn ?load ; tribo:track_code ?tcode ; tribo:hasCycle ?c . + FILTER(?load = 100) + ?c tribo:cycle ?cycle ; tribo:cof ?cof . +}""", + "q5": PREFIX + """SELECT ?bcode ?track ?cycle ?cof WHERE { + ?track a tribo:FrictionTest ; tribo:partOf ?coupon ; tribo:hasCycle ?c . + ?coupon tribo:cutFrom ?wafer . + ?wafer tribo:partOf ?batch . + ?batch tribo:batch_code ?bcode . + ?c tribo:cycle ?cycle ; tribo:cof ?cof . +}""", + "q6": PREFIX + """SELECT ?ccode ?load (AVG(?wv) AS ?wear) WHERE { + ?track tribo:partOf ?coupon ; tribo:load_mn ?load ; tribo:wear ?w . + ?w tribo:wear_volume_um3 ?wv . + ?coupon tribo:coupon_code ?ccode . +} GROUP BY ?ccode ?load""", + "q7": PREFIX + """SELECT ?env ?speed ?load ?track ?cycle ?cof WHERE { + ?track tribo:environment ?env ; tribo:speed_mm_s ?speed ; tribo:load_mn ?load ; tribo:hasCycle ?c . + ?c tribo:cycle ?cycle ; tribo:cof ?cof . +}""", +} + + +def _store(store_path: Path) -> Store: + try: + return Store.read_only(str(store_path)) + except AttributeError: + return Store(str(store_path)) + + +def _track_cofs(solutions, track_var: str, keep: tuple[str, ...]) -> dict[str, dict]: + """Group cycle solutions by track: ordered cof list + kept literals.""" + tracks: dict[str, dict] = {} + for sol in solutions: + key = str(sol[track_var]) + entry = tracks.get(key) + if entry is None: + entry = {"pairs": []} + for name in keep: + entry[name] = sol[name].value + tracks[key] = entry + entry["pairs"].append((int(sol["cycle"].value), float(sol["cof"].value))) + for entry in tracks.values(): + entry["pairs"].sort() + entry["cofs"] = [cof for _, cof in entry["pairs"]] + return tracks + + +def _mean(values: list[float]) -> float: + total = 0.0 + for v in values: + total += v + return total / len(values) + + +def q1(store_path: Path) -> list[tuple]: + store = _store(store_path) + return [(int(s["cycle"].value), float(s["cof"].value)) for s in store.query(SPARQL["q1"])] + + +def q2(store_path: Path) -> list[tuple]: + store = _store(store_path) + tracks = _track_cofs(store.query(SPARQL["q2"]), "track", ("ccode", "env")) + coupons: dict[tuple[str, str], list[float]] = {} + for entry in sorted(tracks.values(), key=lambda e: e["ccode"]): + coupons.setdefault((entry["ccode"], entry["env"]), []).append(summarize_track(entry["cofs"])[0]) + return [(code, env, _mean(values)) for (code, env), values in sorted(coupons.items())] + + +def q3(store_path: Path) -> list[tuple]: + store = _store(store_path) + hardness = { + s["ccode"].value: (s["bcode"].value, float(s["hardness"].value)) + for s in store.query(SPARQL["q3_hardness"]) + } + tracks = _track_cofs(store.query(SPARQL["q3_cycles"]), "track", ("ccode",)) + coupons: dict[str, list[float]] = {} + for entry in sorted(tracks.values(), key=lambda e: e["ccode"]): + coupons.setdefault(entry["ccode"], []).append(summarize_track(entry["cofs"])[0]) + return [ + (code, hardness[code][0], hardness[code][1], _mean(values)) + for code, values in sorted(coupons.items()) + ] + + +def q4(store_path: Path) -> list[tuple]: + store = _store(store_path) + tracks = _track_cofs(store.query(SPARQL["q4"]), "track", ("tcode",)) + out = [] + for entry in sorted(tracks.values(), key=lambda e: e["tcode"]): + cof_ss = summarize_track(entry["cofs"])[0] + if cof_ss > 0.20: + out.append((entry["tcode"], cof_ss)) + return out + + +def q5(store_path: Path) -> list[tuple]: + store = _store(store_path) + tracks = _track_cofs(store.query(SPARQL["q5"]), "track", ("bcode",)) + sums: dict[str, list[float]] = {} + for entry in sorted(tracks.values(), key=lambda e: e["bcode"]): + acc = sums.setdefault(entry["bcode"], [0.0, 0.0]) + acc[0] += summarize_track(entry["cofs"])[2] + acc[1] += 1 + return [(batch, acc[0] / acc[1]) for batch, acc in sorted(sums.items())] + + +def q6(store_path: Path) -> list[tuple]: + store = _store(store_path) + return [ + (s["ccode"].value, float(s["load"].value), float(s["wear"].value)) + for s in store.query(SPARQL["q6"]) + ] + + +def q7(store_path: Path) -> list[tuple]: + store = _store(store_path) + tracks = _track_cofs(store.query(SPARQL["q7"]), "track", ("env", "speed", "load")) + groups: dict[tuple[str, int], list[float]] = {} + for key in sorted(tracks): + entry = tracks[key] + bucket = int(round(float(entry["speed"]) * float(entry["load"]))) + run_in = summarize_track(entry["cofs"])[2] + acc = groups.setdefault((entry["env"], bucket), [0.0, 0.0]) + for cof in entry["cofs"][run_in:]: + acc[0] += cof + acc[1] += 1 + return [(env, bucket, acc[0] / acc[1]) for (env, bucket), acc in sorted(groups.items())] diff --git a/common/relational.py b/common/relational.py new file mode 100644 index 0000000..f2f2606 --- /dev/null +++ b/common/relational.py @@ -0,0 +1,161 @@ +"""Shared relational row stream for the SQLite and PostgreSQL converters. + +Walks the CSV corpus once (manifest-validated via CorpusReader) and yields +(table, row_tuple) pairs in FK-dependency-safe order: a parent row is always +yielded before any row that references it. Both engines load the SAME +logical schema from this stream (docs/rules/db-sql-schema.md section 6). +""" + +from __future__ import annotations + +import logging +from typing import Iterator + +from common.corpus import CorpusReader +from common.track_summary import summarize_track + +logger = logging.getLogger(__name__) + +# Dependency-safe order; consumers that buffer rows must flush in this order. +TABLES = [ + "instruments", + "batches", + "runs", + "wafers", + "coupons", + "tracks", + "simtra_profiles", + "xrf_points", + "profilometry_points", + "nanoindentation", + "afm", + "friction_cycles", + "friction_loop_points", + "wear", + "track_summary", +] + +COLUMNS = { + "instruments": ("instrument_id", "instrument_code", "name", "role"), + "batches": ("batch_id", "batch_code", "pt_gun_tilt_deg", "au_gun_tilt_deg", "pt_power_w", "au_power_w", "pt_discharge_v", "au_discharge_v", "deposition_date"), + "runs": ("run_id", "run_code", "environment", "date", "plates", "operator"), + "wafers": ("wafer_id", "wafer_code", "batch_id", "wafer_index", "deposition_date", "coupons", "friction_coupons", "reserve_coupons"), + "coupons": ("coupon_id", "coupon_code", "wafer_id", "batch_id", "grid_row", "grid_col", "thickness_um", "ra_nm", "au_wtpct_mean", "run_id", "plate", "probe", "square"), + "tracks": ("track_id", "track_code", "coupon_id", "run_id", "environment", "load_mn", "stroke_mm", "speed_mm_s", "counterface_id", "started_at"), + "simtra_profiles": ("batch_id", "row_no", "angle_deg", "energy_ev", "pt_flux", "au_flux"), + "xrf_points": ("coupon_id", "grid_x", "grid_y", "pt_wtpct", "au_wtpct"), + "profilometry_points": ("coupon_id", "grid_x", "grid_y", "thickness_um"), + "nanoindentation": ("coupon_id", "indent_id", "x_um", "y_um", "hardness_gpa", "reduced_modulus_gpa", "max_load_mn"), + "afm": ("coupon_id", "ra_nm", "rq_nm", "image_file"), + "friction_cycles": ("track_id", "cycle", "cof"), + "friction_loop_points": ("track_id", "cycle", "pt", "position_um", "friction_force_mn"), + "wear": ("track_id", "wear_volume_um3", "k_archard", "sliding_distance_m"), + "track_summary": ("track_id", "cof_ss_mean", "cof_ss_std", "run_in_cycles"), +} + +EXPECTED_COUNTS_KEYS = TABLES # every table is count-validated after load + + +def expected_counts(cfg: dict) -> dict[str, int]: + v = cfg["volumes"] + h = cfg["hierarchy"] + return { + "instruments": len(cfg["instruments"]), + "batches": h["batches"], + "runs": cfg["friction_assignment"]["runs"], + "wafers": h["batches"] * h["wafers_per_batch"], + "coupons": v["coupons_total"], + "tracks": v["tracks_total"], + "simtra_profiles": h["batches"] * v["simtra_rows_per_batch"], + "xrf_points": v["xrf_points_total"], + "profilometry_points": v["coupons_total"] * v["profilometry_points_per_coupon"], + "nanoindentation": v["coupons_total"] * v["nanoindentation_indents_per_coupon"], + "afm": v["coupons_total"], + "friction_cycles": v["cycle_rows_total"], + "friction_loop_points": v["loop_points_total"], + "wear": v["tracks_total"], + "track_summary": v["tracks_total"], + } + + +def stream_rows(cfg: dict, reader: CorpusReader) -> Iterator[tuple[str, tuple]]: + batch_ids: dict[str, int] = {} + run_ids: dict[str, int] = {} + + for i, inst in enumerate(cfg["instruments"], 1): + yield "instruments", (i, inst["instrument_id"], inst["name"], inst["role"]) + for i, row in enumerate(reader.dicts("batches.csv"), 1): + batch_ids[row["batch_code"]] = i + yield "batches", ( + i, row["batch_code"], row["pt_gun_tilt_deg"], row["au_gun_tilt_deg"], + row["pt_power_w"], row["au_power_w"], row["pt_discharge_v"], + row["au_discharge_v"], row["deposition_date"], + ) + for i, row in enumerate(reader.dicts("runs.csv"), 1): + run_ids[row["run_code"]] = i + yield "runs", (i, row["run_code"], row["environment"], row["date"], row["plates"], row["operator"]) + for code, batch_id in batch_ids.items(): + for row_no, row in enumerate(reader.dicts(f"simtra/simtra_profile_{code}.csv"), 1): + yield "simtra_profiles", (batch_id, row_no, row["angle_deg"], row["energy_ev"], row["pt_flux"], row["au_flux"]) + + h = cfg["hierarchy"] + fa = cfg["friction_assignment"] + wafer_id = coupon_id = track_id = 0 + for batch in cfg["deposition_matrix"]: + b_code = batch["batch_code"] + for w in range(1, h["wafers_per_batch"] + 1): + wafer_id += 1 + wafer_dir = f"batch_{b_code}/wafer_W{w}" + info = reader.dicts(f"{wafer_dir}/wafer_info.csv")[0] + yield "wafers", ( + wafer_id, info["wafer_code"], batch_ids[b_code], info["wafer_index"], + info["deposition_date"], info["coupons"], info["friction_coupons"], info["reserve_coupons"], + ) + for c in range(1, h["coupons_per_wafer"] + 1): + coupon_id += 1 + rel_dir = f"{wafer_dir}/coupon_C{c:02d}" + info = reader.dicts(f"{rel_dir}/coupon_info.csv")[0] + is_friction = info["run_code"] != "RESERVE" + yield "coupons", ( + coupon_id, info["coupon_code"], wafer_id, batch_ids[b_code], + info["grid_row"], info["grid_col"], info["thickness_um"], info["ra_nm"], + info["au_wtpct_mean"], run_ids[info["run_code"]] if is_friction else None, + info.get("plate"), info.get("probe"), info.get("square"), + ) + for row in reader.dicts(f"{rel_dir}/xrf_map.csv"): + yield "xrf_points", (coupon_id, row["grid_x"], row["grid_y"], row["pt_wtpct"], row["au_wtpct"]) + for row in reader.dicts(f"{rel_dir}/profilometry.csv"): + yield "profilometry_points", (coupon_id, row["grid_x"], row["grid_y"], row["thickness_um"]) + for row in reader.dicts(f"{rel_dir}/nanoindentation.csv"): + yield "nanoindentation", ( + coupon_id, row["indent_id"], row["x_um"], row["y_um"], + row["hardness_gpa"], row["reduced_modulus_gpa"], row["max_load_mn"], + ) + afm = reader.dicts(f"{rel_dir}/afm.csv")[0] + yield "afm", (coupon_id, afm["ra_nm"], afm["rq_nm"], afm["image_file"]) + if not is_friction: + continue + for t in range(1, fa["tracks_per_friction_coupon"] + 1): + track_id += 1 + track_dir = f"{rel_dir}/track_T{t}" + tinfo = reader.dicts(f"{track_dir}/track_info.csv")[0] + yield "tracks", ( + track_id, tinfo["track_code"], coupon_id, run_ids[tinfo["run_code"]], + tinfo["environment"], tinfo["load_mn"], tinfo["stroke_mm"], tinfo["speed_mm_s"], + tinfo["counterface_id"], tinfo["started_at"], + ) + cofs: list[float] = [] + for row in reader.dicts(f"{track_dir}/cof_vs_cycle.csv"): + cofs.append(row["cof"]) + yield "friction_cycles", (track_id, row["cycle"], row["cof"]) + ss_mean, ss_std, run_in = summarize_track(cofs) + yield "track_summary", (track_id, ss_mean, ss_std, run_in) + pt = 0 + last_cycle = None + for row in reader.dicts(f"{track_dir}/friction_loops.csv"): + pt = pt + 1 if row["cycle"] == last_cycle else 1 + last_cycle = row["cycle"] + yield "friction_loop_points", (track_id, row["cycle"], pt, row["position_um"], row["friction_force_mn"]) + wear = reader.dicts(f"{track_dir}/wear.csv")[0] + yield "wear", (track_id, wear["wear_volume_um3"], wear["k_archard"], wear["sliding_distance_m"]) + logger.info("batch %s streamed (through coupon %d, track %d)", b_code, coupon_id, track_id) diff --git a/common/results.py b/common/results.py new file mode 100644 index 0000000..d3d0cd5 --- /dev/null +++ b/common/results.py @@ -0,0 +1,73 @@ +"""Canonical benchmark result handling (spec 08, test-pipeline-validation). + +Every implementation of a benchmark query returns the same canonical rows: +tuples of str / int / float columns. Rows are compared SORTED; floats agree +within 1e-9. The canonical expected artifacts are produced from PostgreSQL +and cross-validated against SQLite (task 08); every measured run (task 09) +re-validates its rows against them. + +Two serializations: +- full: floats via repr() - lossless, used for stdout and expected/*.rows.csv; +- rounded (6 decimals): used ONLY for the sha256 fingerprint, so engines + whose aggregate arithmetic differs in the last bits hash identically. +The tolerant row comparison is the authoritative check; the sha256 is a +compact fingerprint for reports and markers. +""" + +from __future__ import annotations + +import hashlib +import sys + +from common.corpus import coerce + +FLOAT_TOL = 1e-9 + +Row = tuple + + +def sort_rows(rows: list[Row]) -> list[Row]: + return sorted(rows) + + +def _fmt(value, rounded: bool) -> str: + if isinstance(value, float): + return f"{round(value, 6):.6f}" if rounded else repr(value) + return str(value) + + +def serialize(rows: list[Row], rounded: bool = False) -> str: + return "\n".join(",".join(_fmt(v, rounded) for v in row) for row in rows) + "\n" + + +def sha256_of(rows: list[Row]) -> str: + return hashlib.sha256(serialize(sort_rows(rows), rounded=True).encode("ascii")).hexdigest() + + +def print_rows(rows: list[Row]) -> None: + sys.stdout.write(serialize(sort_rows(rows))) + + +def parse_rows(text: str) -> list[Row]: + rows = [] + for line in text.splitlines(): + if line: + rows.append(tuple(coerce(cell) for cell in line.split(","))) + return rows + + +def compare(expected: list[Row], actual: list[Row], tol: float = FLOAT_TOL) -> str | None: + """Return None when equal within tolerance, else a first-difference message.""" + exp, act = sort_rows(expected), sort_rows(actual) + if len(exp) != len(act): + return f"row count {len(act)} != expected {len(exp)}" + for i, (er, ar) in enumerate(zip(exp, act)): + if len(er) != len(ar): + return f"row {i}: arity {len(ar)} != {len(er)}" + for j, (ev, av) in enumerate(zip(er, ar)): + if isinstance(ev, float) or isinstance(av, float): + if abs(float(ev) - float(av)) > tol: + return f"row {i} col {j}: {av!r} != {ev!r} (tol {tol})" + elif ev != av: + return f"row {i} col {j}: {av!r} != {ev!r}" + return None diff --git a/common/storage_sizes.py b/common/storage_sizes.py index edd188f..d01ebe4 100644 --- a/common/storage_sizes.py +++ b/common/storage_sizes.py @@ -8,24 +8,45 @@ touches other formats' rows. from __future__ import annotations import logging +import os +import time from pathlib import Path logger = logging.getLogger(__name__) HEADER = "format,variant,bytes" +LOCK_TIMEOUT_S = 30.0 # converters 04-07 may run in parallel; the update is a +LOCK_POLL_S = 0.2 # read-modify-write, so it is serialized via a lock file def update_storage_sizes(bench_dir: Path, fmt: str, variants: list[tuple[str, int]]) -> Path: - path = bench_dir / "storage_sizes.csv" - kept: list[str] = [] - if path.exists(): - lines = path.read_text(encoding="ascii").splitlines() - if lines and lines[0] != HEADER: - raise ValueError(f"{path}: unexpected header {lines[0]!r}") - kept = [ln for ln in lines[1:] if ln and ln.split(",", 1)[0] != fmt] - rows = kept + [f"{fmt},{variant},{nbytes}" for variant, nbytes in variants] bench_dir.mkdir(parents=True, exist_ok=True) - with open(path, "w", encoding="ascii", newline="\n") as fh: - fh.write(HEADER + "\n" + "\n".join(rows) + "\n") + path = bench_dir / "storage_sizes.csv" + lock = bench_dir / "storage_sizes.lock" + fd = _acquire(lock) + try: + kept: list[str] = [] + if path.exists(): + lines = path.read_text(encoding="ascii").splitlines() + if lines and lines[0] != HEADER: + raise ValueError(f"{path}: unexpected header {lines[0]!r}") + kept = [ln for ln in lines[1:] if ln and ln.split(",", 1)[0] != fmt] + rows = kept + [f"{fmt},{variant},{nbytes}" for variant, nbytes in variants] + with open(path, "w", encoding="ascii", newline="\n") as fh: + fh.write(HEADER + "\n" + "\n".join(rows) + "\n") + finally: + os.close(fd) + lock.unlink() logger.info("storage_sizes.csv: %s -> %s", fmt, ", ".join(f"{v}={b}" for v, b in variants)) return path + + +def _acquire(lock: Path) -> int: + deadline = time.monotonic() + LOCK_TIMEOUT_S + while True: + try: + return os.open(lock, os.O_CREAT | os.O_EXCL | os.O_WRONLY) + except FileExistsError: + if time.monotonic() > deadline: + raise TimeoutError(f"storage_sizes lock held too long: {lock}") + time.sleep(LOCK_POLL_S) diff --git a/convert_json.py b/convert_json.py index 22ad48b..e14cdd1 100644 --- a/convert_json.py +++ b/convert_json.py @@ -30,6 +30,7 @@ from common.pipeline import ( check_dependencies, load_lab_config, parse_task_args, + process_metrics, remove_stale_marker, write_marker, ) @@ -265,7 +266,7 @@ def main() -> int: json_root / "hybrid" / "dataset.jsonld", ]) update_storage_sizes(out_root / "bench", "json", [("full", conv.full_bytes), ("hybrid", hybrid_bytes)]) - marker_path = write_marker(out_root, TASK_ID, { + entries = { "full_files": conv.full_files, "full_bytes": conv.full_bytes, "hybrid_bytes": hybrid_bytes, @@ -273,7 +274,9 @@ def main() -> int: "tracks": conv.counts["tracks"], "cycle_rows": conv.counts["cycle_rows"], "loop_points": conv.counts["loop_rows"], - }) + } + entries.update(process_metrics()) + marker_path = write_marker(out_root, TASK_ID, entries) except Exception: logger.critical("task %s failed", TASK_ID, exc_info=True) return 1 diff --git a/convert_postgresql.py b/convert_postgresql.py new file mode 100644 index 0000000..079ddfa --- /dev/null +++ b/convert_postgresql.py @@ -0,0 +1,334 @@ +"""Task 06 - Convert the CSV corpus to PostgreSQL. + +Loads the same logical schema as task 05 into the PostgreSQL 16 database +named by the TRIBO_PG_DSN environment variable: COPY FROM STDIN, RANGE +partitioning of the two bulk tables by track_id (12 partitions of 120 +tracks), constraints and indexes created after the load (variant B: +enforced foreign keys everywhere - docs/research/Tribology_FK_Architecture.html), +track_summary as a materialized view (algorithm fixed in +docs/rules/db-sql-schema.md section 5, cross-checked against +common/track_summary.py). Host CPU/RAM during the load is sampled from the +database host's Prometheus (TRIBO_PROM_URL). Appends live (and, when +pg_dump is available, dump) sizes to storage_sizes.csv and writes +./out/.done/06.ok. Spec: docs/specs/06_convert_postgresql.md. +""" + +from __future__ import annotations + +import logging +import shutil +import subprocess +import sys +import time +from pathlib import Path + +import psycopg + +from common.corpus import CorpusReader, ValidationError +from common.monitoring import get_prom_url, host_window_metrics +from common.pg import get_dsn +from common.pipeline import ( + check_dependencies, + load_lab_config, + parse_task_args, + process_metrics, + remove_stale_marker, + write_marker, +) +from common.relational import COLUMNS, TABLES, expected_counts, stream_rows +from common.storage_sizes import update_storage_sizes +from common.track_summary import summarize_track + +logger = logging.getLogger(__name__) + +TASK_ID = "06" +DEPENDS_ON = ["01", "03"] + +BATCH_ROWS = 50000 # rows per COPY chunk (mirrors the 50k batching of spec 05) +PARTITIONS = 12 # RANGE partitions of 120 tracks each, spec 06 +TRACKS_PER_PARTITION = 120 +SUMMARY_PARITY_TRACKS = (1, 720, 1440) # matview cross-checked vs common/track_summary.py +PARITY_TOLERANCE = 1e-9 # cross-engine float tolerance, test-pipeline-validation + +# Float columns are DOUBLE PRECISION, not REAL: the corpus is float64 and +# REAL (float4) would break the 1e-9 cross-engine result parity that the +# whole benchmark depends on (db-sql-schema section 6; spec updated). +CREATE_TABLES = [ + "CREATE TABLE instruments (instrument_id INTEGER NOT NULL, instrument_code TEXT NOT NULL, name TEXT NOT NULL, role TEXT NOT NULL)", + "CREATE TABLE batches (batch_id INTEGER NOT NULL, batch_code TEXT NOT NULL, pt_gun_tilt_deg INTEGER NOT NULL, au_gun_tilt_deg INTEGER NOT NULL, pt_power_w INTEGER NOT NULL, au_power_w INTEGER NOT NULL, pt_discharge_v INTEGER NOT NULL, au_discharge_v INTEGER NOT NULL, deposition_date DATE NOT NULL)", + "CREATE TABLE runs (run_id INTEGER NOT NULL, run_code TEXT NOT NULL, environment TEXT NOT NULL, date DATE NOT NULL, plates INTEGER NOT NULL, operator TEXT NOT NULL)", + "CREATE TABLE wafers (wafer_id INTEGER NOT NULL, wafer_code TEXT NOT NULL, batch_id INTEGER NOT NULL, wafer_index INTEGER NOT NULL, deposition_date DATE NOT NULL, coupons INTEGER NOT NULL, friction_coupons INTEGER NOT NULL, reserve_coupons INTEGER NOT NULL)", + "CREATE TABLE coupons (coupon_id INTEGER NOT NULL, coupon_code TEXT NOT NULL, wafer_id INTEGER NOT NULL, batch_id INTEGER NOT NULL, grid_row INTEGER NOT NULL, grid_col INTEGER NOT NULL, thickness_um DOUBLE PRECISION NOT NULL, ra_nm DOUBLE PRECISION NOT NULL, au_wtpct_mean DOUBLE PRECISION NOT NULL, run_id INTEGER, plate INTEGER, probe INTEGER, square INTEGER)", + "CREATE TABLE tracks (track_id INTEGER NOT NULL, track_code TEXT NOT NULL, coupon_id INTEGER NOT NULL, run_id INTEGER NOT NULL, environment TEXT NOT NULL, load_mn DOUBLE PRECISION NOT NULL, stroke_mm DOUBLE PRECISION NOT NULL, speed_mm_s DOUBLE PRECISION NOT NULL, counterface_id TEXT NOT NULL, started_at TIMESTAMPTZ NOT NULL)", + "CREATE TABLE simtra_profiles (batch_id INTEGER NOT NULL, row_no INTEGER NOT NULL, angle_deg DOUBLE PRECISION NOT NULL, energy_ev DOUBLE PRECISION NOT NULL, pt_flux DOUBLE PRECISION NOT NULL, au_flux DOUBLE PRECISION NOT NULL)", + "CREATE TABLE xrf_points (coupon_id INTEGER NOT NULL, grid_x INTEGER NOT NULL, grid_y INTEGER NOT NULL, pt_wtpct DOUBLE PRECISION NOT NULL, au_wtpct DOUBLE PRECISION NOT NULL)", + "CREATE TABLE profilometry_points (coupon_id INTEGER NOT NULL, grid_x INTEGER NOT NULL, grid_y INTEGER NOT NULL, thickness_um DOUBLE PRECISION NOT NULL)", + "CREATE TABLE nanoindentation (coupon_id INTEGER NOT NULL, indent_id INTEGER NOT NULL, x_um DOUBLE PRECISION NOT NULL, y_um DOUBLE PRECISION NOT NULL, hardness_gpa DOUBLE PRECISION NOT NULL, reduced_modulus_gpa DOUBLE PRECISION NOT NULL, max_load_mn DOUBLE PRECISION NOT NULL)", + "CREATE TABLE afm (coupon_id INTEGER NOT NULL, ra_nm DOUBLE PRECISION NOT NULL, rq_nm DOUBLE PRECISION NOT NULL, image_file TEXT NOT NULL)", + "CREATE TABLE friction_cycles (track_id INTEGER NOT NULL, cycle INTEGER NOT NULL, cof DOUBLE PRECISION NOT NULL) PARTITION BY RANGE (track_id)", + "CREATE TABLE friction_loop_points (track_id INTEGER NOT NULL, cycle INTEGER NOT NULL, pt INTEGER NOT NULL, position_um DOUBLE PRECISION NOT NULL, friction_force_mn DOUBLE PRECISION NOT NULL) PARTITION BY RANGE (track_id)", + "CREATE TABLE wear (track_id INTEGER NOT NULL, wear_volume_um3 DOUBLE PRECISION NOT NULL, k_archard DOUBLE PRECISION NOT NULL, sliding_distance_m DOUBLE PRECISION NOT NULL)", +] + +CONSTRAINTS = [ + "ALTER TABLE instruments ADD PRIMARY KEY (instrument_id)", + "ALTER TABLE instruments ADD UNIQUE (instrument_code)", + "ALTER TABLE batches ADD PRIMARY KEY (batch_id)", + "ALTER TABLE batches ADD UNIQUE (batch_code)", + "ALTER TABLE runs ADD PRIMARY KEY (run_id)", + "ALTER TABLE runs ADD UNIQUE (run_code)", + "ALTER TABLE wafers ADD PRIMARY KEY (wafer_id)", + "ALTER TABLE wafers ADD UNIQUE (wafer_code)", + "ALTER TABLE coupons ADD PRIMARY KEY (coupon_id)", + "ALTER TABLE coupons ADD UNIQUE (coupon_code)", + "ALTER TABLE tracks ADD PRIMARY KEY (track_id)", + "ALTER TABLE tracks ADD UNIQUE (track_code)", + "ALTER TABLE simtra_profiles ADD PRIMARY KEY (batch_id, row_no)", + "ALTER TABLE xrf_points ADD PRIMARY KEY (coupon_id, grid_x, grid_y)", + "ALTER TABLE profilometry_points ADD PRIMARY KEY (coupon_id, grid_x, grid_y)", + "ALTER TABLE nanoindentation ADD PRIMARY KEY (coupon_id, indent_id)", + "ALTER TABLE afm ADD PRIMARY KEY (coupon_id)", + "ALTER TABLE friction_cycles ADD PRIMARY KEY (track_id, cycle)", + "ALTER TABLE friction_loop_points ADD PRIMARY KEY (track_id, cycle, pt)", + "ALTER TABLE wear ADD PRIMARY KEY (track_id)", + "ALTER TABLE wafers ADD FOREIGN KEY (batch_id) REFERENCES batches", + "ALTER TABLE coupons ADD FOREIGN KEY (wafer_id) REFERENCES wafers", + "ALTER TABLE coupons ADD FOREIGN KEY (batch_id) REFERENCES batches", + "ALTER TABLE coupons ADD FOREIGN KEY (run_id) REFERENCES runs", + "ALTER TABLE tracks ADD FOREIGN KEY (coupon_id) REFERENCES coupons", + "ALTER TABLE tracks ADD FOREIGN KEY (run_id) REFERENCES runs", + "ALTER TABLE simtra_profiles ADD FOREIGN KEY (batch_id) REFERENCES batches", + "ALTER TABLE xrf_points ADD FOREIGN KEY (coupon_id) REFERENCES coupons", + "ALTER TABLE profilometry_points ADD FOREIGN KEY (coupon_id) REFERENCES coupons", + "ALTER TABLE nanoindentation ADD FOREIGN KEY (coupon_id) REFERENCES coupons", + "ALTER TABLE afm ADD FOREIGN KEY (coupon_id) REFERENCES coupons", + "ALTER TABLE friction_cycles ADD FOREIGN KEY (track_id) REFERENCES tracks", + "ALTER TABLE friction_loop_points ADD FOREIGN KEY (track_id) REFERENCES tracks", + "ALTER TABLE wear ADD FOREIGN KEY (track_id) REFERENCES tracks", +] + +INDEX_DDL = [ + "CREATE INDEX ix_coupons_au_wtpct_mean ON coupons(au_wtpct_mean)", # Q2 selective filter + "CREATE INDEX ix_tracks_run_id ON tracks(run_id)", # Q5/Q7 joins to runs + "CREATE INDEX ix_tracks_coupon_id ON tracks(coupon_id)", # Q3/Q6 joins to coupons + "CREATE INDEX ix_runs_environment ON runs(environment)", # Q7 grouping + "CREATE INDEX ix_tracks_environment_load ON tracks(environment, load_mn)", # Q4 filter + # Q5 pruning within partitions; cascades to every partition (spec 06). + # nanoindentation(coupon_id) from the spec is omitted: the PK prefix serves it. + "CREATE INDEX ix_brin_friction_cycles_cycle ON friction_cycles USING BRIN (cycle)", +] + +# Materialized view semantics fixed in docs/rules/db-sql-schema.md section 5. +MATVIEW_SQL = """ +CREATE MATERIALIZED VIEW track_summary AS +WITH tail AS ( + SELECT track_id, avg(cof) AS m, stddev_pop(cof) AS s + FROM friction_cycles + WHERE cycle >= 501 + GROUP BY track_id +), +run_in AS ( + SELECT fc.track_id, + COALESCE(MIN(fc.cycle) FILTER (WHERE abs(fc.cof - t.m) < 2 * t.s), 500) AS run_in_cycles + FROM friction_cycles fc + JOIN tail t USING (track_id) + GROUP BY fc.track_id +) +SELECT r.track_id, + avg(fc.cof) AS cof_ss_mean, + stddev_pop(fc.cof) AS cof_ss_std, + r.run_in_cycles +FROM run_in r +JOIN friction_cycles fc ON fc.track_id = r.track_id AND fc.cycle > r.run_in_cycles +GROUP BY r.track_id, r.run_in_cycles +""" + + +def partition_ddl() -> list[str]: + ddl = [] + for parent in ("friction_cycles", "friction_loop_points"): + for i in range(PARTITIONS): + lo = i * TRACKS_PER_PARTITION + 1 + hi = lo + TRACKS_PER_PARTITION + ddl.append(f"CREATE TABLE {parent}_p{i + 1:02d} PARTITION OF {parent} FOR VALUES FROM ({lo}) TO ({hi})") + return ddl + + +class PgLoader: + def __init__(self, conn: psycopg.Connection) -> None: + self.conn = conn + self.buffers: dict[str, list[tuple]] = {t: [] for t in TABLES} + self.summary_parity: dict[int, tuple] = {} + + def create_schema(self) -> None: + with self.conn.cursor() as cur: + cur.execute("DROP MATERIALIZED VIEW IF EXISTS track_summary") + for table in reversed(TABLES): + if table != "track_summary": + cur.execute(f"DROP TABLE IF EXISTS {table} CASCADE") + for ddl in CREATE_TABLES + partition_ddl(): + cur.execute(ddl) + self.conn.commit() + logger.info("schema created: %d tables, %d partitions", len(CREATE_TABLES), 2 * PARTITIONS) + + def load(self, cfg: dict, reader: CorpusReader) -> None: + pending = 0 + for table, row in stream_rows(cfg, reader): + if table == "track_summary": + # computed server-side as a materialized view; keep a sample + # of the Python values for the cross-engine parity check + if row[0] in SUMMARY_PARITY_TRACKS: + self.summary_parity[row[0]] = row + continue + self.buffers[table].append(row) + pending += 1 + if pending >= BATCH_ROWS: + self.flush_all() + pending = 0 + self.flush_all() + + def flush_all(self) -> None: + for table in TABLES: + buf = self.buffers.get(table) + if not buf: + continue + cols = ", ".join(COLUMNS[table]) + with self.conn.cursor() as cur: + with cur.copy(f"COPY {table} ({cols}) FROM STDIN") as copy: + for row in buf: + copy.write_row(row) + buf.clear() + self.conn.commit() + + def finalize(self) -> None: + with self.conn.cursor() as cur: + for ddl in CONSTRAINTS: + cur.execute(ddl) + for ddl in INDEX_DDL: + cur.execute(ddl) + cur.execute(MATVIEW_SQL) + cur.execute("CREATE UNIQUE INDEX ix_track_summary_track_id ON track_summary(track_id)") # Q2/Q3 joins + self.conn.commit() + logger.info("constraints, indexes and track_summary materialized view created") + + def validate(self, cfg: dict) -> dict[str, int]: + counts: dict[str, int] = {} + with self.conn.cursor() as cur: + for table, exp in expected_counts(cfg).items(): + got = cur.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0] + counts[table] = got + if got != exp: + raise ValidationError(f"row count mismatch: {table}: loaded {got} != expected {exp}") + reserve = cur.execute("SELECT COUNT(*) FROM coupons WHERE run_id IS NULL").fetchone()[0] + if reserve != cfg["friction_assignment"]["reserve_coupons_total"]: + raise ValidationError(f"reserve coupons: {reserve} != expected") + for track_id, py_row in self.summary_parity.items(): + db_row = cur.execute( + "SELECT cof_ss_mean, cof_ss_std, run_in_cycles FROM track_summary WHERE track_id = %s", + (track_id,), + ).fetchone() + if db_row is None: + raise ValidationError(f"track_summary missing track {track_id}") + if (abs(db_row[0] - py_row[1]) > PARITY_TOLERANCE + or abs(db_row[1] - py_row[2]) > PARITY_TOLERANCE + or db_row[2] != py_row[3]): + raise ValidationError( + f"track_summary parity failed for track {track_id}: db={db_row} py={py_row[1:]}" + ) + logger.info( + "all %d table counts match, %d reserve coupons, matview parity ok for tracks %s", + len(counts), reserve, list(self.summary_parity), + ) + return counts + + def table_sizes(self) -> dict[str, int]: + sizes: dict[str, int] = {} + with self.conn.cursor() as cur: + for table in TABLES: + total = cur.execute( + """SELECT COALESCE(pg_total_relation_size(%s::regclass), 0) + + COALESCE((SELECT sum(pg_total_relation_size(inhrelid)) + FROM pg_inherits WHERE inhparent = %s::regclass), 0)""", + (table, table), + ).fetchone()[0] + sizes[table] = int(total) + return sizes + + +def make_dump(dsn: str, dump_path: Path) -> int: + pg_dump = shutil.which("pg_dump") + if not pg_dump: + logger.warning("pg_dump skipped (client not installed on this host; produce the dump on the database host)") + return 0 + dump_path.parent.mkdir(parents=True, exist_ok=True) + subprocess.run([pg_dump, "--format=custom", f"--file={dump_path}", dsn], check=True) + return dump_path.stat().st_size + + +def main() -> int: + args = parse_task_args("Task 06: convert the CSV corpus to PostgreSQL") + out_root: Path = args.out_root + dsn = get_dsn() + try: + check_dependencies(out_root, DEPENDS_ON) + remove_stale_marker(out_root, TASK_ID) + cfg = load_lab_config(out_root) + + t_start = time.time() + with psycopg.connect(dsn) as conn: + database = conn.info.dbname + logger.info("connected to database %r on %s", database, conn.info.host) + loader = PgLoader(conn) + loader.create_schema() + loader.load(cfg, CorpusReader(out_root / "csv")) + load_done = time.time() + loader.finalize() + counts = loader.validate(cfg) + sizes = loader.table_sizes() + with psycopg.connect(dsn, autocommit=True) as conn: + conn.execute("VACUUM ANALYZE") + t_end = time.time() + + live_bytes = sum(sizes.values()) + variants = [("live", live_bytes)] + dump_bytes = make_dump(dsn, out_root / "pg" / "tribo.dump") + if dump_bytes: + variants.append(("dump", dump_bytes)) + update_storage_sizes(out_root / "bench", "pg", variants) + + entries = { + "database": database, + "tables": len(counts), + "total_rows": sum(counts.values()), + "live_bytes": live_bytes, + "friction_cycles_bytes": sizes["friction_cycles"], + "friction_loop_points_bytes": sizes["friction_loop_points"], + "copy_seconds": round(load_done - t_start, 1), + "total_seconds": round(t_end - t_start, 1), + "dump": dump_bytes if dump_bytes else "skipped (pg_dump unavailable on this host)", + } + entries.update(process_metrics()) # client-side converter cost + prom_url = get_prom_url() + if prom_url: + metrics = host_window_metrics(prom_url, t_start, t_end) + if metrics: + entries.update(metrics) + logger.info("db host during load: %s", metrics) + marker_path = write_marker(out_root, TASK_ID, entries) + except Exception: + logger.critical("task %s failed", TASK_ID, exc_info=True) + return 1 + + print(f"task 06 ok: database {database} ({live_bytes / (1024 * 1024):.1f} MiB live, " + f"{len(counts)} tables, {sum(counts.values())} rows, {entries['total_seconds']} s)") + print(f"bulk rows: cycles={counts['friction_cycles']} loop_points={counts['friction_loop_points']} " + f"track_summary={counts['track_summary']}") + if prom_url and "host_cpu_avg_pct" in entries: + print(f"db host: cpu avg {entries['host_cpu_avg_pct']}% max {entries['host_cpu_max_pct']}%, " + f"ram peak {entries['host_ram_used_peak_mb']} MB (delta {entries['host_ram_used_delta_mb']} MB)") + print(f"marker: {marker_path}") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/convert_rdf.py b/convert_rdf.py new file mode 100644 index 0000000..d0d40c3 --- /dev/null +++ b/convert_rdf.py @@ -0,0 +1,395 @@ +"""Task 07 - Convert the CSV corpus to RDF. + +Streams the corpus into ./out/rdf/dataset.nt.gz (N-Triples, gzip) and +./out/rdf/dataset.ttl (Turtle) with the task 04 vocabulary (tribo# + PROV-O +for started_at), then bulk-loads the N-Triples into an embedded on-disk +oxigraph store (./out/rdf/oxigraph_store/) and smoke-tests it with SPARQL +COUNT queries. Bulk points (cycles, loop points, map points) carry no +rdf:type - the containing predicate types them (documented decision per +spec 07), which is why the triple count lands below the spec's 25-45M +sketch. Appends serialization and store sizes to storage_sizes.csv and +writes ./out/.done/07.ok. Spec: docs/specs/07_convert_rdf.md. +""" + +from __future__ import annotations + +import gzip +import io +import logging +import shutil +import sys +import time +from pathlib import Path + +from pyoxigraph import Store + +try: + from pyoxigraph import RdfFormat + NT_FORMAT = RdfFormat.N_TRIPLES +except ImportError: # pyoxigraph < 0.5 uses MIME strings + NT_FORMAT = "application/n-triples" + +from common.corpus import CorpusReader, ValidationError +from common.pipeline import ( + check_dependencies, + load_lab_config, + parse_task_args, + process_metrics, + remove_stale_marker, + write_marker, +) +from common.storage_sizes import update_storage_sizes + +logger = logging.getLogger(__name__) + +TASK_ID = "07" +DEPENDS_ON = ["01", "03"] + +VOCAB = "https://sandia.gov/ontology/tribology#" +ID_BASE = "https://sandia.gov/ontology/tribology/id/" +XSD = "http://www.w3.org/2001/XMLSchema#" + +PREFIXES = { + "tribo": VOCAB, + "batch": ID_BASE + "batch/", + "wafer": ID_BASE + "wafer/", + "coupon": ID_BASE + "coupon/", + "track": ID_BASE + "track/", + "run": ID_BASE + "run/", + "instrument": ID_BASE + "instrument/", + "simtra": ID_BASE + "simtra/", + "prov": "http://www.w3.org/ns/prov#", + "xsd": XSD, +} + +INT_COLS = { + "cycle", "grid_x", "grid_y", "indent_id", "plates", "plate", "probe", "square", + "row_no", "wafer_index", "coupons", "friction_coupons", "reserve_coupons", + "pt_gun_tilt_deg", "au_gun_tilt_deg", "pt_power_w", "au_power_w", + "pt_discharge_v", "au_discharge_v", +} +DOUBLE_COLS = { + "cof", "position_um", "friction_force_mn", "pt_wtpct", "au_wtpct", "thickness_um", + "x_um", "y_um", "hardness_gpa", "reduced_modulus_gpa", "max_load_mn", "ra_nm", + "rq_nm", "wear_volume_um3", "k_archard", "sliding_distance_m", "load_mn", + "stroke_mm", "speed_mm_s", "angle_deg", "energy_ev", "pt_flux", "au_flux", + "au_wtpct_mean", +} +DATE_COLS = {"date", "deposition_date"} + +MEASUREMENTS = [ + ("xrf_map.csv", "xrf", "CompositionMeasurement", "m4_tornado"), + ("profilometry.csv", "profilometry", "ProfilometryMeasurement", "profilometer"), + ("nanoindentation.csv", "nanoindentation", "NanoindentationMeasurement", "ti980"), +] + + +def lit(col: str, token: str) -> tuple: + if col in INT_COLS: + return ("l", token, "integer") + if col in DOUBLE_COLS: + return ("l", token, "double") + if col in DATE_COLS: + return ("l", token, "date") + if col == "started_at": + return ("l", token, "dateTime") + return ("l", token, None) + + +def iri(curie: str) -> tuple: + return ("i", curie) + + +def node(pairs: list) -> tuple: + return ("n", pairs) + + +class RdfEmitter: + """Writes the same subject blocks to N-Triples and Turtle streams.""" + + def __init__(self, nt_fh, ttl_fh) -> None: + self.nt = nt_fh + self.ttl = ttl_fh + self.triples = 0 + self._blank = 0 + for prefix, base in PREFIXES.items(): + self.ttl.write(f"@prefix {prefix}: <{base}> .\n") + self.ttl.write("\n") + + def _full(self, curie: str) -> str: + prefix, local = curie.split(":", 1) + return f"<{PREFIXES[prefix]}{local}>" + + def _pred_nt(self, pred: str) -> str: + if pred == "a": + return "" + return self._full(pred) + + @staticmethod + def _escape(text: str) -> str: + return text.replace("\\", "\\\\").replace('"', '\\"') + + def _obj_nt(self, obj: tuple) -> str: + kind = obj[0] + if kind == "i": + return self._full(obj[1]) + if kind == "l": + _, token, dtype = obj + if dtype is None: + return f'"{self._escape(token)}"' + return f'"{token}"^^<{XSD}{dtype}>' + self._blank += 1 + label = f"_:b{self._blank}" + for pred, sub_obj in obj[1]: + self.nt.write(f"{label} {self._pred_nt(pred)} {self._obj_nt(sub_obj)} .\n") + self.triples += 1 + return label + + def _obj_ttl(self, obj: tuple) -> str: + kind = obj[0] + if kind == "i": + return obj[1] + if kind == "l": + _, token, dtype = obj + if dtype is None: + return f'"{self._escape(token)}"' + if dtype == "integer": + return token + return f'"{token}"^^xsd:{dtype}' + inner = " ; ".join(f"{p} {self._obj_ttl(o)}" for p, o in obj[1]) + return f"[ {inner} ]" + + def emit(self, subject: str, pairs: list) -> None: + subj_nt = self._full(subject) + ttl_parts = [] + for pred, obj in pairs: + # N-Triples: nested nodes expand to labelled blanks first, so the + # referencing triple is written after the node's own triples. + obj_nt = self._obj_nt(obj) + self.nt.write(f"{subj_nt} {self._pred_nt(pred)} {obj_nt} .\n") + self.triples += 1 + ttl_parts.append(f"{pred} {self._obj_ttl(obj)}") + self.ttl.write(subject + " " + " ;\n ".join(ttl_parts) + " .\n") + + +class RdfConverter: + def __init__(self, cfg: dict, csv_root: Path, emitter: RdfEmitter) -> None: + self.cfg = cfg + self.reader = CorpusReader(csv_root) + self.em = emitter + self.counts = {"coupons": 0, "tracks": 0, "cycle_nodes": 0, "loop_nodes": 0, "xrf_nodes": 0} + + def raw_rows(self, relpath: str) -> list[dict[str, str]]: + header, rows = self.reader.read(relpath) + keys = [h.lower() for h in header] + return [dict(zip(keys, row)) for row in rows] + + def prop_pairs(self, row: dict[str, str], skip: tuple = ()) -> list: + return [ + ("prov:startedAtTime" if col == "started_at" else f"tribo:{col}", lit(col, token)) + for col, token in row.items() + if col not in skip and token != "" + ] + + def convert(self) -> None: + for inst in self.cfg["instruments"]: + self.em.emit(f"instrument:{inst['instrument_id']}", [ + ("a", iri("tribo:Instrument")), + ("tribo:name", ("l", inst["name"], None)), + ("tribo:role", ("l", inst["role"], None)), + ]) + for row in self.raw_rows("batches.csv"): + code = row["batch_code"] + self.em.emit(f"batch:{code}", [("a", iri("tribo:Batch"))] + self.prop_pairs(row)) + points = [ + ("tribo:hasPoint", node([("tribo:row_no", ("l", str(no), "integer"))] + self.prop_pairs(p))) + for no, p in enumerate(self.raw_rows(f"simtra/simtra_profile_{code}.csv"), 1) + ] + self.em.emit(f"simtra:{code}", [ + ("a", iri("tribo:SimtraProfile")), + ("tribo:performedOn", iri(f"batch:{code}")), + ("tribo:performedBy", iri("instrument:simtra")), + ] + points) + for row in self.raw_rows("runs.csv"): + self.em.emit(f"run:{row['run_code']}", [("a", iri("tribo:Run"))] + self.prop_pairs(row)) + + h = self.cfg["hierarchy"] + for batch in self.cfg["deposition_matrix"]: + b_code = batch["batch_code"] + for w in range(1, h["wafers_per_batch"] + 1): + wafer_dir = f"batch_{b_code}/wafer_W{w}" + info = self.raw_rows(f"{wafer_dir}/wafer_info.csv")[0] + self.em.emit(f"wafer:{info['wafer_code']}", [ + ("a", iri("tribo:Wafer")), + ("tribo:partOf", iri(f"batch:{b_code}")), + ] + self.prop_pairs(info, skip=("batch_code",))) + for c in range(1, h["coupons_per_wafer"] + 1): + self.convert_coupon(f"{wafer_dir}/coupon_C{c:02d}") + logger.info("batch %s converted (%d triples so far)", b_code, self.em.triples) + + def convert_coupon(self, rel_dir: str) -> None: + info = self.raw_rows(f"{rel_dir}/coupon_info.csv")[0] + code = info["coupon_code"] + subject = f"coupon:{code}" + is_friction = info["run_code"] != "RESERVE" + pairs = [("a", iri("tribo:Coupon")), ("tribo:cutFrom", iri(f"wafer:{info['wafer_code']}"))] + pairs += self.prop_pairs(info, skip=("batch_code", "wafer_code", "run_code")) + pairs.append(("tribo:run_code", ("l", info["run_code"], None))) + if is_friction: + pairs.append(("tribo:duringRun", iri(f"run:{info['run_code']}"))) + + for csv_name, key, cls, inst in MEASUREMENTS: + points = [ + ("tribo:hasPoint" if key != "nanoindentation" else "tribo:hasIndent", node(self.prop_pairs(p))) + for p in self.raw_rows(f"{rel_dir}/{csv_name}") + ] + if key == "xrf": + self.counts["xrf_nodes"] += len(points) + pairs.append((f"tribo:{key}", node([ + ("a", iri(f"tribo:{cls}")), + ("tribo:performedOn", iri(subject)), + ("tribo:performedBy", iri(f"instrument:{inst}")), + ] + points))) + afm = self.raw_rows(f"{rel_dir}/afm.csv")[0] + pairs.append(("tribo:afm", node([ + ("a", iri("tribo:AFMMeasurement")), + ("tribo:performedOn", iri(subject)), + ("tribo:performedBy", iri("instrument:afm")), + ] + self.prop_pairs(afm)))) + self.em.emit(subject, pairs) + self.counts["coupons"] += 1 + + if is_friction: + for t in range(1, self.cfg["friction_assignment"]["tracks_per_friction_coupon"] + 1): + self.convert_track(f"{rel_dir}/track_T{t}", subject) + + def convert_track(self, rel_dir: str, coupon_subject: str) -> None: + info = self.raw_rows(f"{rel_dir}/track_info.csv")[0] + pairs = [ + ("a", iri("tribo:FrictionTest")), + ("tribo:partOf", iri(coupon_subject)), + ("tribo:duringRun", iri(f"run:{info['run_code']}")), + ("tribo:performedBy", iri("instrument:rapid")), + ] + self.prop_pairs(info, skip=("track_code", "run_code")) + pairs.insert(1, ("tribo:track_code", ("l", info["track_code"], None))) + + cycles = self.raw_rows(f"{rel_dir}/cof_vs_cycle.csv") + pairs += [("tribo:hasCycle", node(self.prop_pairs(row))) for row in cycles] + self.counts["cycle_nodes"] += len(cycles) + loops = self.raw_rows(f"{rel_dir}/friction_loops.csv") + pairs += [("tribo:hasLoopPoint", node(self.prop_pairs(row))) for row in loops] + self.counts["loop_nodes"] += len(loops) + wear = self.raw_rows(f"{rel_dir}/wear.csv")[0] + pairs.append(("tribo:wear", node([("a", iri("tribo:WearMeasurement"))] + self.prop_pairs(wear)))) + self.em.emit(f"track:{info['track_code']}", pairs) + self.counts["tracks"] += 1 + + def validate_counts(self) -> None: + v = self.cfg["volumes"] + expected = { + "coupons": v["coupons_total"], + "tracks": v["tracks_total"], + "cycle_nodes": v["cycle_rows_total"], + "loop_nodes": v["loop_points_total"], + "xrf_nodes": v["xrf_points_total"], + } + for key, exp in expected.items(): + if self.counts[key] != exp: + raise ValidationError(f"count mismatch: {key}: converted {self.counts[key]} != expected {exp}") + logger.info("all converted counts match lab_config volumes: %s", self.counts) + + +def bulk_load(store: Store, nt_gz: Path) -> None: + with gzip.open(nt_gz, "rb") as fh: + try: + store.bulk_load(fh, NT_FORMAT) + except TypeError: + store.bulk_load(input=fh, format=NT_FORMAT) + + +def sparql_count(store: Store, query: str) -> int: + solutions = store.query(query) + return int(next(iter(solutions))[0].value) + + +def main() -> int: + args = parse_task_args("Task 07: convert the CSV corpus to RDF") + out_root: Path = args.out_root + rdf_root = out_root / "rdf" + nt_gz = rdf_root / "dataset.nt.gz" + ttl_path = rdf_root / "dataset.ttl" + store_dir = rdf_root / "oxigraph_store" + try: + check_dependencies(out_root, DEPENDS_ON) + remove_stale_marker(out_root, TASK_ID) + if rdf_root.exists(): + logger.info("re-run: removing previous output %s", rdf_root) + shutil.rmtree(rdf_root) + rdf_root.mkdir(parents=True) + cfg = load_lab_config(out_root) + + # mtime=0 keeps the gzip byte-identical across re-runs (data-determinism) + with open(nt_gz, "wb") as raw: + gz = gzip.GzipFile(filename="", mode="wb", fileobj=raw, compresslevel=6, mtime=0) + with io.TextIOWrapper(gz, encoding="ascii", newline="\n") as nt_fh, \ + open(ttl_path, "w", encoding="ascii", newline="\n") as ttl_fh: + emitter = RdfEmitter(nt_fh, ttl_fh) + conv = RdfConverter(cfg, out_root / "csv", emitter) + conv.convert() + conv.validate_counts() + triples = emitter.triples + nt_bytes = nt_gz.stat().st_size + ttl_bytes = ttl_path.stat().st_size + logger.info("serialized %d triples: nt.gz %.1f MiB, ttl %.1f MiB", + triples, nt_bytes / 1048576, ttl_bytes / 1048576) + + t0 = time.time() + store = Store(str(store_dir)) + bulk_load(store, nt_gz) + store.flush() + load_seconds = round(time.time() - t0, 1) + + total = sparql_count(store, "SELECT (COUNT(*) AS ?n) WHERE { ?s ?p ?o }") + if total != triples: + raise ValidationError(f"store holds {total} triples, serialized {triples}") + q1 = sparql_count( + store, + f"SELECT (COUNT(?c) AS ?n) WHERE {{ <{ID_BASE}track/B722-W2-C13-T2> <{VOCAB}hasCycle> ?c }}", + ) + if q1 != cfg["volumes"]["cycles_per_track"]: + raise ValidationError(f"SPARQL Q1 smoke test returned {q1} cycles") + logger.info("store smoke tests ok: %d triples, Q1 track has %d cycles (load %.1f s)", total, q1, load_seconds) + del store + store_bytes = sum(f.stat().st_size for f in store_dir.rglob("*") if f.is_file()) + + update_storage_sizes(out_root / "bench", "rdf", [ + ("nt_gz", nt_bytes), ("ttl", ttl_bytes), ("store", store_bytes), + ]) + entries = { + "triples": triples, + "nt_gz_bytes": nt_bytes, + "ttl_bytes": ttl_bytes, + "store_bytes": store_bytes, + "store_load_seconds": load_seconds, + "coupons": conv.counts["coupons"], + "tracks": conv.counts["tracks"], + "cycle_nodes": conv.counts["cycle_nodes"], + "loop_nodes": conv.counts["loop_nodes"], + } + entries.update(process_metrics()) + marker_path = write_marker(out_root, TASK_ID, entries) + except Exception: + logger.critical("task %s failed", TASK_ID, exc_info=True) + return 1 + + print(f"task 07 ok: {triples} triples; nt.gz {nt_bytes / 1048576:.1f} MiB, " + f"ttl {ttl_bytes / 1048576:.1f} MiB, store {store_bytes / 1048576:.1f} MiB " + f"(bulk load {load_seconds} s)") + print(f"entities: coupons={conv.counts['coupons']} tracks={conv.counts['tracks']} " + f"cycles={conv.counts['cycle_nodes']} loop_points={conv.counts['loop_nodes']}") + print(f"marker: {marker_path}") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/convert_sqlite.py b/convert_sqlite.py index 09e63b8..b640c2d 100644 --- a/convert_sqlite.py +++ b/convert_sqlite.py @@ -20,11 +20,12 @@ from common.pipeline import ( check_dependencies, load_lab_config, parse_task_args, + process_metrics, remove_stale_marker, write_marker, ) +from common.relational import COLUMNS, TABLES, expected_counts, stream_rows from common.storage_sizes import update_storage_sizes -from common.track_summary import summarize_track logger = logging.getLogger(__name__) @@ -178,158 +179,38 @@ INDEX_DDL = [ "CREATE INDEX ix_tracks_environment_load ON tracks(environment, load_mn)", # Q4 filter ] -INSERTS = { - "instruments": "INSERT INTO instruments VALUES (?,?,?,?)", - "batches": "INSERT INTO batches VALUES (?,?,?,?,?,?,?,?,?)", - "runs": "INSERT INTO runs VALUES (?,?,?,?,?,?)", - "wafers": "INSERT INTO wafers VALUES (?,?,?,?,?,?,?,?)", - "coupons": "INSERT INTO coupons VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)", - "tracks": "INSERT INTO tracks VALUES (?,?,?,?,?,?,?,?,?,?)", - "simtra_profiles": "INSERT INTO simtra_profiles VALUES (?,?,?,?,?,?)", - "xrf_points": "INSERT INTO xrf_points VALUES (?,?,?,?,?)", - "profilometry_points": "INSERT INTO profilometry_points VALUES (?,?,?,?)", - "nanoindentation": "INSERT INTO nanoindentation VALUES (?,?,?,?,?,?,?)", - "afm": "INSERT INTO afm VALUES (?,?,?,?)", - "friction_cycles": "INSERT INTO friction_cycles VALUES (?,?,?)", - "friction_loop_points": "INSERT INTO friction_loop_points VALUES (?,?,?,?,?)", - "wear": "INSERT INTO wear VALUES (?,?,?,?)", - "track_summary": "INSERT INTO track_summary VALUES (?,?,?,?)", -} - class SqliteLoader: - def __init__(self, cfg: dict, csv_root: Path, db_path: Path) -> None: - self.cfg = cfg - self.reader = CorpusReader(csv_root) + def __init__(self, db_path: Path) -> None: db_path.parent.mkdir(parents=True, exist_ok=True) self.conn = sqlite3.connect(db_path) for pragma in LOAD_PRAGMAS: self.conn.execute(pragma) for ddl in DDL: self.conn.execute(ddl) - self.buffers: dict[str, list[tuple]] = {} - self.batch_ids: dict[str, int] = {} - self.run_ids: dict[str, int] = {} + self.inserts = { + t: f"INSERT INTO {t} VALUES ({','.join('?' * len(COLUMNS[t]))})" for t in TABLES + } + self.buffers: dict[str, list[tuple]] = {t: [] for t in TABLES} - def put(self, table: str, row: tuple) -> None: - buf = self.buffers.setdefault(table, []) - buf.append(row) - if len(buf) >= BATCH_ROWS: - self.flush(table) - - def flush(self, table: str) -> None: - buf = self.buffers.get(table) - if buf: - self.conn.executemany(INSERTS[table], buf) - self.conn.commit() - buf.clear() + def load(self, cfg: dict, reader: CorpusReader) -> None: + pending = 0 + for table, row in stream_rows(cfg, reader): + self.buffers[table].append(row) + pending += 1 + if pending >= BATCH_ROWS: + self.flush_all() + pending = 0 + self.flush_all() def flush_all(self) -> None: - for table in list(self.buffers): - self.flush(table) - - # --- dimension + bulk loading, in FK dependency order --- - - def load_flat(self) -> None: - for i, inst in enumerate(self.cfg["instruments"], 1): - self.put("instruments", (i, inst["instrument_id"], inst["name"], inst["role"])) - for i, row in enumerate(self.reader.dicts("batches.csv"), 1): - self.batch_ids[row["batch_code"]] = i - self.put("batches", ( - i, row["batch_code"], row["pt_gun_tilt_deg"], row["au_gun_tilt_deg"], - row["pt_power_w"], row["au_power_w"], row["pt_discharge_v"], - row["au_discharge_v"], row["deposition_date"], - )) - for i, row in enumerate(self.reader.dicts("runs.csv"), 1): - self.run_ids[row["run_code"]] = i - self.put("runs", (i, row["run_code"], row["environment"], row["date"], row["plates"], row["operator"])) - self.flush_all() - for code, batch_id in self.batch_ids.items(): - for row_no, row in enumerate(self.reader.dicts(f"simtra/simtra_profile_{code}.csv"), 1): - self.put("simtra_profiles", (batch_id, row_no, row["angle_deg"], row["energy_ev"], row["pt_flux"], row["au_flux"])) - - def load_hierarchy(self) -> None: - h = self.cfg["hierarchy"] - fa = self.cfg["friction_assignment"] - wafer_id = 0 - coupon_id = 0 - track_id = 0 - for batch in self.cfg["deposition_matrix"]: - b_code = batch["batch_code"] - batch_id = self.batch_ids[b_code] - for w in range(1, h["wafers_per_batch"] + 1): - wafer_id += 1 - wafer_dir = f"batch_{b_code}/wafer_W{w}" - info = self.reader.dicts(f"{wafer_dir}/wafer_info.csv")[0] - self.put("wafers", ( - wafer_id, info["wafer_code"], batch_id, info["wafer_index"], - info["deposition_date"], info["coupons"], info["friction_coupons"], info["reserve_coupons"], - )) - self.flush("wafers") - for c in range(1, h["coupons_per_wafer"] + 1): - coupon_id += 1 - rel_dir = f"{wafer_dir}/coupon_C{c:02d}" - track_id = self.load_coupon(rel_dir, coupon_id, wafer_id, batch_id, track_id, fa) - logger.info("batch %s loaded (through coupon %d, track %d)", b_code, coupon_id, track_id) - self.flush_all() - - def load_coupon(self, rel_dir: str, coupon_id: int, wafer_id: int, batch_id: int, track_id: int, fa: dict) -> int: - info = self.reader.dicts(f"{rel_dir}/coupon_info.csv")[0] - is_friction = info["run_code"] != "RESERVE" - run_id = self.run_ids[info["run_code"]] if is_friction else None - self.put("coupons", ( - coupon_id, info["coupon_code"], wafer_id, batch_id, - info["grid_row"], info["grid_col"], info["thickness_um"], info["ra_nm"], - info["au_wtpct_mean"], run_id, - info.get("plate"), info.get("probe"), info.get("square"), - )) - self.flush("coupons") - - for row in self.reader.dicts(f"{rel_dir}/xrf_map.csv"): - self.put("xrf_points", (coupon_id, row["grid_x"], row["grid_y"], row["pt_wtpct"], row["au_wtpct"])) - for row in self.reader.dicts(f"{rel_dir}/profilometry.csv"): - self.put("profilometry_points", (coupon_id, row["grid_x"], row["grid_y"], row["thickness_um"])) - for row in self.reader.dicts(f"{rel_dir}/nanoindentation.csv"): - self.put("nanoindentation", ( - coupon_id, row["indent_id"], row["x_um"], row["y_um"], - row["hardness_gpa"], row["reduced_modulus_gpa"], row["max_load_mn"], - )) - afm = self.reader.dicts(f"{rel_dir}/afm.csv")[0] - self.put("afm", (coupon_id, afm["ra_nm"], afm["rq_nm"], afm["image_file"])) - - if is_friction: - for t in range(1, fa["tracks_per_friction_coupon"] + 1): - track_id += 1 - self.load_track(f"{rel_dir}/track_T{t}", track_id, coupon_id) - return track_id - - def load_track(self, rel_dir: str, track_id: int, coupon_id: int) -> None: - info = self.reader.dicts(f"{rel_dir}/track_info.csv")[0] - self.put("tracks", ( - track_id, info["track_code"], coupon_id, self.run_ids[info["run_code"]], - info["environment"], info["load_mn"], info["stroke_mm"], info["speed_mm_s"], - info["counterface_id"], info["started_at"], - )) - self.flush("tracks") - - cofs: list[float] = [] - for row in self.reader.dicts(f"{rel_dir}/cof_vs_cycle.csv"): - cofs.append(row["cof"]) - self.put("friction_cycles", (track_id, row["cycle"], row["cof"])) - ss_mean, ss_std, run_in = summarize_track(cofs) - self.put("track_summary", (track_id, ss_mean, ss_std, run_in)) - - pt = 0 - last_cycle = None - for row in self.reader.dicts(f"{rel_dir}/friction_loops.csv"): - pt = pt + 1 if row["cycle"] == last_cycle else 1 - last_cycle = row["cycle"] - self.put("friction_loop_points", (track_id, row["cycle"], pt, row["position_um"], row["friction_force_mn"])) - - wear = self.reader.dicts(f"{rel_dir}/wear.csv")[0] - self.put("wear", (track_id, wear["wear_volume_um3"], wear["k_archard"], wear["sliding_distance_m"])) - - # --- post-load --- + # TABLES is FK-dependency ordered: parents flush before children. + for table in TABLES: + buf = self.buffers[table] + if buf: + self.conn.executemany(self.inserts[table], buf) + buf.clear() + self.conn.commit() def finalize(self) -> None: for ddl in INDEX_DDL: @@ -340,39 +221,20 @@ class SqliteLoader: self.conn.execute("VACUUM") logger.info("indexes created, WAL enabled, ANALYZE + VACUUM done") - def validate(self) -> dict[str, int]: - cfg = self.cfg - v = cfg["volumes"] - expected = { - "instruments": len(cfg["instruments"]), - "batches": cfg["hierarchy"]["batches"], - "runs": cfg["friction_assignment"]["runs"], - "wafers": cfg["hierarchy"]["batches"] * cfg["hierarchy"]["wafers_per_batch"], - "coupons": v["coupons_total"], - "tracks": v["tracks_total"], - "simtra_profiles": cfg["hierarchy"]["batches"] * v["simtra_rows_per_batch"], - "xrf_points": v["xrf_points_total"], - "profilometry_points": v["coupons_total"] * v["profilometry_points_per_coupon"], - "nanoindentation": v["coupons_total"] * v["nanoindentation_indents_per_coupon"], - "afm": v["coupons_total"], - "friction_cycles": v["cycle_rows_total"], - "friction_loop_points": v["loop_points_total"], - "wear": v["tracks_total"], - "track_summary": v["tracks_total"], - } + def validate(self, cfg: dict) -> dict[str, int]: counts: dict[str, int] = {} - for table, exp in expected.items(): + for table, exp in expected_counts(cfg).items(): got = self.conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0] counts[table] = got if got != exp: raise ValidationError(f"row count mismatch: {table}: loaded {got} != expected {exp}") reserve = self.conn.execute("SELECT COUNT(*) FROM coupons WHERE run_id IS NULL").fetchone()[0] - if reserve != self.cfg["friction_assignment"]["reserve_coupons_total"]: + if reserve != cfg["friction_assignment"]["reserve_coupons_total"]: raise ValidationError(f"reserve coupons: {reserve} != expected") fk_violations = self.conn.execute("PRAGMA foreign_key_check").fetchall() if fk_violations: raise ValidationError(f"foreign_key_check reported {len(fk_violations)} violations") - logger.info("all %d table counts match, %d reserve coupons, foreign_key_check clean", len(expected), reserve) + logger.info("all %d table counts match, %d reserve coupons, foreign_key_check clean", len(counts), reserve) return counts @@ -389,16 +251,15 @@ def main() -> int: stale.unlink() cfg = load_lab_config(out_root) - loader = SqliteLoader(cfg, out_root / "csv", db_path) - loader.load_flat() - loader.load_hierarchy() + loader = SqliteLoader(db_path) + loader.load(cfg, CorpusReader(out_root / "csv")) loader.finalize() - counts = loader.validate() + counts = loader.validate(cfg) loader.conn.close() db_bytes = db_path.stat().st_size update_storage_sizes(out_root / "bench", "sqlite", [("db", db_bytes)]) - marker_path = write_marker(out_root, TASK_ID, { + entries = { "db_file": db_path.as_posix(), "db_bytes": db_bytes, "tables": len(counts), @@ -407,7 +268,9 @@ def main() -> int: "friction_loop_points": counts["friction_loop_points"], "tracks": counts["tracks"], "track_summary": counts["track_summary"], - }) + } + entries.update(process_metrics()) + marker_path = write_marker(out_root, TASK_ID, entries) except Exception: logger.critical("task %s failed", TASK_ID, exc_info=True) return 1 diff --git a/docs/rules/code-python-style.md b/docs/rules/code-python-style.md index 6387eae..2f736cc 100644 --- a/docs/rules/code-python-style.md +++ b/docs/rules/code-python-style.md @@ -35,7 +35,7 @@ The stdlib is the default. The ONLY permitted third-party packages: | `oxigraph` | 07, 09 | embedded queryable triplestore | | `ijson` | 08, 09 | streaming JSON parsing | | `psycopg` (v3) | 06, 08, 09 | PostgreSQL COPY and queries | -| `psutil` | 09 | RSS / CPU / IO measurement | +| `psutil` | 04-07, 09 | RSS / CPU / IO measurement; converters record their own peak RSS / CPU time into their `.done` markers for the task 10 sizing model | | `matplotlib` | 11 | report charts | | `pandas` | 08 (optional) | ONLY as the separately-measured second CSV query variant; never in the pipeline itself | diff --git a/make_queries.py b/make_queries.py new file mode 100644 index 0000000..c039904 --- /dev/null +++ b/make_queries.py @@ -0,0 +1,289 @@ +"""Task 08 - Benchmark query definitions (Q1-Q7 x 5 formats). + +Writes the runnable query files under ./out/bench/queries/ (SQL for +SQLite/PostgreSQL, SPARQL text + python drivers for RDF, python drivers +for CSV/JSON-LD), then produces the canonical expected results: executed +on PostgreSQL, cross-validated against SQLite row-by-row within 1e-9, and +stored as ./out/bench/expected/q.rows.csv + q.sha256. CSV, JSON and +RDF implementations are spot-checked on Q1. Writes ./out/.done/08.ok. +Spec: docs/specs/08_benchmark_queries.md. +""" + +from __future__ import annotations + +import logging +import sqlite3 +import sys +from pathlib import Path + +import psycopg + +from common.pg import get_dsn +from common.pipeline import ( + check_dependencies, + parse_task_args, + process_metrics, + remove_stale_marker, + write_marker, +) +from common.queries import Q1_TRACK_CODE, csv_queries, json_queries, rdf_queries +from common.results import compare, serialize, sha256_of, sort_rows + +logger = logging.getLogger(__name__) + +TASK_ID = "08" +DEPENDS_ON = ["03", "04", "05", "06", "07"] + +EXACT_ROW_COUNTS = {1: 1000, 3: 480, 5: 4, 6: 480, 7: 2} # fixed by corpus volumes + +# --- SQL implementations ------------------------------------------------- +# Q2/Q3/Q4/Q7 use the precomputed track_summary (the relational strength); +# Q5 must scan raw friction_cycles in every format (spec 08). + +Q1_SQL = f"""-- Q1: COF vs cycle curve for track {Q1_TRACK_CODE} (point read) +SELECT fc.cycle, fc.cof +FROM friction_cycles fc +JOIN tracks t ON t.track_id = fc.track_id +WHERE t.track_code = '{Q1_TRACK_CODE}' +ORDER BY fc.cycle; +""" + +Q2_SQL = """-- Q2: steady-state COF per coupon with mean Au = 10 +/- 0.5 wt% +SELECT c.coupon_code, t.environment, AVG(s.cof_ss_mean) AS cof_ss +FROM coupons c +JOIN tracks t ON t.coupon_id = c.coupon_id +JOIN track_summary s ON s.track_id = t.track_id +WHERE c.au_wtpct_mean BETWEEN 9.5 AND 10.5 +GROUP BY c.coupon_code, t.environment +ORDER BY c.coupon_code; +""" + +Q3_SQL = """-- Q3: hardness vs steady-state COF per friction coupon, across all batches +SELECT c.coupon_code, + b.batch_code, + (SELECT AVG(n.hardness_gpa) FROM nanoindentation n WHERE n.coupon_id = c.coupon_id) AS hardness_gpa, + AVG(s.cof_ss_mean) AS cof_ss +FROM coupons c +JOIN batches b ON b.batch_id = c.batch_id +JOIN tracks t ON t.coupon_id = c.coupon_id +JOIN track_summary s ON s.track_id = t.track_id +GROUP BY c.coupon_id, c.coupon_code, b.batch_code +ORDER BY c.coupon_code; +""" + +Q4_SQL = """-- Q4: anomaly filter - Dry N2 tracks at 100 mN with cof_ss > 0.20 +SELECT t.track_code, s.cof_ss_mean +FROM tracks t +JOIN track_summary s ON s.track_id = t.track_id +WHERE t.environment = 'dry_n2' AND t.load_mn = 100 AND s.cof_ss_mean > 0.20 +ORDER BY t.track_code; +""" + +# Q5: forced full scan over raw cycles; the run-in algorithm is inlined +# (docs/rules/db-sql-schema.md section 5). SQLite has no stddev built-in, +# so the 2-sigma band compares squares: (cof-m)^2 < 4*var <=> |cof-m| < 2s. +Q5_SQLITE = """-- Q5: mean run-in cycles per batch over ALL raw cycle rows (track_summary forbidden) +WITH tail AS ( + SELECT track_id, AVG(cof) AS m, AVG(cof * cof) - AVG(cof) * AVG(cof) AS var + FROM friction_cycles + WHERE cycle >= 501 + GROUP BY track_id +), +run_in AS ( + SELECT fc.track_id, + COALESCE(MIN(CASE WHEN (fc.cof - tl.m) * (fc.cof - tl.m) < 4 * tl.var THEN fc.cycle END), 500) AS run_in + FROM friction_cycles fc + JOIN tail tl ON tl.track_id = fc.track_id + GROUP BY fc.track_id +) +SELECT b.batch_code, AVG(r.run_in) AS avg_run_in_cycles +FROM run_in r +JOIN tracks t ON t.track_id = r.track_id +JOIN coupons c ON c.coupon_id = t.coupon_id +JOIN batches b ON b.batch_id = c.batch_id +GROUP BY b.batch_code +ORDER BY b.batch_code; +""" + +Q5_PG = """-- Q5: mean run-in cycles per batch over ALL raw cycle rows (track_summary forbidden) +WITH tail AS ( + SELECT track_id, avg(cof) AS m, stddev_pop(cof) AS s + FROM friction_cycles + WHERE cycle >= 501 + GROUP BY track_id +), +run_in AS ( + SELECT fc.track_id, + COALESCE(MIN(fc.cycle) FILTER (WHERE abs(fc.cof - tl.m) < 2 * tl.s), 500) AS run_in + FROM friction_cycles fc + JOIN tail tl ON tl.track_id = fc.track_id + GROUP BY fc.track_id +) +SELECT b.batch_code, AVG(r.run_in)::double precision AS avg_run_in_cycles +FROM run_in r +JOIN tracks t ON t.track_id = r.track_id +JOIN coupons c ON c.coupon_id = t.coupon_id +JOIN batches b ON b.batch_id = c.batch_id +GROUP BY b.batch_code +ORDER BY b.batch_code; +""" + +Q6_SQL = """-- Q6: mean wear volume vs load per friction coupon +SELECT c.coupon_code, t.load_mn, AVG(w.wear_volume_um3) AS wear_volume_um3 +FROM wear w +JOIN tracks t ON t.track_id = w.track_id +JOIN coupons c ON c.coupon_id = t.coupon_id +GROUP BY c.coupon_id, c.coupon_code, t.load_mn +ORDER BY c.coupon_code; +""" + +Q7_SQL = """-- Q7: Stribeck-style mean COF by (environment, speed*load bucket) past run-in +SELECT t.environment, + CAST(ROUND(t.speed_mm_s * t.load_mn) AS INTEGER) AS speed_load_bucket, + AVG(fc.cof) AS mean_cof +FROM friction_cycles fc +JOIN tracks t ON t.track_id = fc.track_id +JOIN track_summary s ON s.track_id = fc.track_id +WHERE fc.cycle > s.run_in_cycles +GROUP BY t.environment, speed_load_bucket +ORDER BY t.environment, speed_load_bucket; +""" + +SQLITE_SQL = {1: Q1_SQL, 2: Q2_SQL, 3: Q3_SQL, 4: Q4_SQL, 5: Q5_SQLITE, 6: Q6_SQL, 7: Q7_SQL} +PG_SQL = {1: Q1_SQL, 2: Q2_SQL, 3: Q3_SQL, 4: Q4_SQL, 5: Q5_PG, 6: Q6_SQL, 7: Q7_SQL} + +RDF_SPARQL_FILES = { + 1: ("q1",), 2: ("q2",), 3: ("q3_hardness", "q3_cycles"), + 4: ("q4",), 5: ("q5",), 6: ("q6",), 7: ("q7",), +} + +WRAPPER = '''"""Benchmark query q{n} for the {fmt} format (generated by task 08). + +Prints the canonical result rows to stdout at full float precision; the +benchmark runner (task 09) captures and validates them against +out/bench/expected/q{n}.rows.csv. +""" +import sys +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[4] +sys.path.insert(0, str(REPO_ROOT)) + +from common.queries.{module} import q{n} +from common.results import print_rows + +if __name__ == "__main__": + print_rows(q{n}(REPO_ROOT / "out" / {subpath})) +''' + +PY_FORMATS = [ + ("csv", "csv_queries", '"csv"'), + ("json", "json_queries", '"json"'), + ("rdf", "rdf_queries", '"rdf" / "oxigraph_store"'), +] + + +def write_query_files(bench_dir: Path) -> int: + queries_dir = bench_dir / "queries" + written = 0 + for fmt, module, subpath in PY_FORMATS: + fmt_dir = queries_dir / fmt + fmt_dir.mkdir(parents=True, exist_ok=True) + for n in range(1, 8): + text = WRAPPER.format(n=n, fmt=fmt, module=module, subpath=subpath) + (fmt_dir / f"q{n}.py").write_text(text, encoding="ascii", newline="\n") + written += 1 + for fmt, sql in (("sqlite", SQLITE_SQL), ("pg", PG_SQL)): + fmt_dir = queries_dir / fmt + fmt_dir.mkdir(parents=True, exist_ok=True) + for n, text in sql.items(): + (fmt_dir / f"q{n}.sql").write_text(text, encoding="ascii", newline="\n") + written += 1 + rdf_dir = queries_dir / "rdf" + for n, keys in RDF_SPARQL_FILES.items(): + text = "\n\n".join(f"# retrieval query: {key}\n{rdf_queries.SPARQL[key]}" for key in keys) + (rdf_dir / f"q{n}.sparql").write_text(text + "\n", encoding="ascii", newline="\n") + written += 1 + logger.info("wrote %d query files under %s", written, queries_dir) + return written + + +def run_sqlite(db_path: Path) -> dict[int, list[tuple]]: + conn = sqlite3.connect(db_path) + try: + return {n: [tuple(r) for r in conn.execute(sql).fetchall()] for n, sql in SQLITE_SQL.items()} + finally: + conn.close() + + +def run_pg(dsn: str) -> dict[int, list[tuple]]: + results: dict[int, list[tuple]] = {} + with psycopg.connect(dsn) as conn: + for n, sql in PG_SQL.items(): + results[n] = [tuple(r) for r in conn.execute(sql).fetchall()] + logger.info("pg q%d: %d rows", n, len(results[n])) + return results + + +def main() -> int: + args = parse_task_args("Task 08: define benchmark queries and canonical expected results") + out_root: Path = args.out_root + bench_dir = out_root / "bench" + try: + check_dependencies(out_root, DEPENDS_ON) + remove_stale_marker(out_root, TASK_ID) + files_written = write_query_files(bench_dir) + + logger.info("computing canonical results on PostgreSQL") + pg_rows = run_pg(get_dsn()) + logger.info("cross-validating against SQLite") + sqlite_rows = run_sqlite(out_root / "sqlite" / "tribo.db") + for n in range(1, 8): + diff = compare(pg_rows[n], sqlite_rows[n]) + if diff: + raise ValueError(f"q{n}: SQLite disagrees with PostgreSQL: {diff}") + exact = EXACT_ROW_COUNTS.get(n) + if exact is not None and len(pg_rows[n]) != exact: + raise ValueError(f"q{n}: {len(pg_rows[n])} rows, expected {exact}") + if not pg_rows[n]: + raise ValueError(f"q{n}: empty result set") + + logger.info("spot-checking csv/json/rdf on Q1") + for label, rows in ( + ("csv", csv_queries.q1(out_root / "csv")), + ("json", json_queries.q1(out_root / "json")), + ("rdf", rdf_queries.q1(out_root / "rdf" / "oxigraph_store")), + ): + diff = compare(pg_rows[1], rows) + if diff: + raise ValueError(f"q1 via {label} disagrees with canonical: {diff}") + + expected_dir = bench_dir / "expected" + expected_dir.mkdir(parents=True, exist_ok=True) + checksums = {} + for n in range(1, 8): + rows = sort_rows(pg_rows[n]) + (expected_dir / f"q{n}.rows.csv").write_text(serialize(rows), encoding="ascii", newline="\n") + checksums[n] = sha256_of(rows) + (expected_dir / f"q{n}.sha256").write_text(checksums[n] + "\n", encoding="ascii", newline="\n") + + entries = {"query_files": files_written} + for n in range(1, 8): + entries[f"q{n}_rows"] = len(pg_rows[n]) + entries[f"q{n}_sha256"] = checksums[n][:12] + entries.update(process_metrics()) + marker_path = write_marker(out_root, TASK_ID, entries) + except Exception: + logger.critical("task %s failed", TASK_ID, exc_info=True) + return 1 + + counts = ", ".join(f"q{n}={len(pg_rows[n])}" for n in range(1, 8)) + print(f"task 08 ok: {files_written} query files, canonical results validated (pg == sqlite, 1e-9)") + print(f"rows: {counts}") + print(f"expected: {bench_dir / 'expected'} (q.rows.csv + q.sha256)") + print(f"marker: {marker_path}") + return 0 + + +if __name__ == "__main__": + sys.exit(main())