#!/usr/bin/env python3 """Generate a replayable GFIS simulator evidence package for Report Version 2.""" from __future__ import annotations import argparse import hashlib import json import platform import subprocess from datetime import datetime, timezone from pathlib import Path import joblib import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt import pandas as pd from gfis.scenario_experiments import STRESS_SCENARIOS from gfis.service import DEFAULT_INPUT, GFISService def sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for block in iter(lambda: handle.read(1024 * 1024), b""): digest.update(block) return digest.hexdigest() def git_commit(workspace: Path) -> str: return subprocess.check_output( ["git", "rev-parse", "HEAD"], cwd=workspace, text=True ).strip() def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--run-id", required=True) parser.add_argument("--hours", type=int, default=48) args = parser.parse_args() workspace = Path(__file__).resolve().parents[3] output = workspace / "06_Experiments/V2_SIMULATOR_EVIDENCE/runs" / args.run_id if output.exists(): raise FileExistsError(f"Refusing to overwrite evidence run: {output}") (output / "plots").mkdir(parents=True) service = GFISService() scenarios = [ *STRESS_SCENARIOS, { "name": "Critical acidification boundary stress", "research_purpose": ( "Synthetic edge-case stress for VFA/ALK alarm and stability-state transition." ), "inputs": { "temperature": 28.0, "pH": 6.2, "OLR": 8.0, "HRT": 10.0, "methane_yield_lag1": 170.0, "methane_yield_roll3": 175.0, }, }, { "name": "Low-VS physics-bound challenge", "research_purpose": ( "Synthetic boundary test for the VS-dependent methane feasibility ceiling." ), "inputs": { "TS": 1.5, "VS": 0.5, "OLR": 5.5, "methane_yield_lag1": 260.0, "methane_yield_roll3": 250.0, }, }, ] trace_rows: list[dict] = [] summary_rows: list[dict] = [] for scenario in scenarios: values = {**DEFAULT_INPUT, **scenario["inputs"]} point = service.predict(values) trace = service.plant_run(values, hours=args.hours) for row in trace: trace_rows.append( { "scenario": scenario["name"], "evidence_type": "synthetic_controlled_simulator_replay", **row, } ) methane = [row["methane_yield"] for row in trace] vfa = [row["vfa_alk_ratio"] for row in trace] summary_rows.append( { "scenario": scenario["name"], "research_purpose": scenario["research_purpose"], "hours": len(trace), "point_methane_yield": point.methane_yield, "point_physics_upper_bound": point.physics_upper_bound, "point_physics_violation": point.physics_violation, "point_vfa_alk_ratio": point.vfa_alk_ratio, "point_stability": point.stability_label, "mean_methane_yield": round(sum(methane) / len(methane), 3), "min_methane_yield": min(methane), "max_methane_yield": max(methane), "max_vfa_alk_ratio": max(vfa), "warning_or_critical_hours": sum( row["stability_label"] != "Stable" for row in trace ), "physics_violation_hours": sum( bool(row["physics_violation"]) for row in trace ), } ) traces = pd.DataFrame(trace_rows) summaries = pd.DataFrame(summary_rows) trace_path = output / "simulator_48h_traces.csv" summary_path = output / "scenario_summary.csv" traces.to_csv(trace_path, index=False) summaries.to_csv(summary_path, index=False) selected = [ "Nominal mesophilic operation", "Organic overload", "Low-temperature disturbance", "Critical acidification boundary stress", "Low-VS physics-bound challenge", ] fig, axes = plt.subplots(3, 1, figsize=(11, 10), sharex=True) for name in selected: part = traces[traces["scenario"] == name] axes[0].plot(part["hour"], part["methane_yield"], label=name, linewidth=1.8) axes[1].plot(part["hour"], part["vfa_alk_ratio"], label=name, linewidth=1.8) axes[2].plot(part["hour"], part["physics_upper_bound"], label=name, linewidth=1.8) axes[0].set_ylabel("Methane yield") axes[0].set_title("Executed GFIS 48-hour methane traces") axes[1].axhline(0.30, color="#f59e0b", linestyle="--", label="Warning threshold") axes[1].axhline(0.45, color="#d64545", linestyle="--", label="Critical threshold") axes[1].set_ylabel("VFA/ALK") axes[1].set_title("Soft-sensor response and stability thresholds") axes[2].set_ylabel("VS-based upper bound") axes[2].set_xlabel("Simulation hour") axes[2].set_title("Physics feasibility ceiling retained through the run") for axis in axes: axis.grid(alpha=0.2) axes[0].legend(ncol=2, fontsize=8, frameon=False) fig.tight_layout() plot_path = output / "plots/simulator_capability_replay.png" fig.savefig(plot_path, dpi=180, bbox_inches="tight", facecolor="white") plt.close(fig) model_bundle = service.model_dir / "gfis_tabular_bundle.joblib" metadata = { "run_id": args.run_id, "generated_at": datetime.now(timezone.utc).isoformat(), "evidence_label": "synthetic controlled simulator replay; not plant validation", "hours_per_scenario": args.hours, "scenario_count": len(scenarios), "trace_row_count": len(traces), "git_commit": git_commit(workspace), "python": platform.python_version(), "joblib": joblib.__version__, "trained_bundle_loaded": service.bundle is not None, "model_bundle": str(model_bundle.relative_to(workspace)) if model_bundle.exists() else None, "model_bundle_sha256": sha256(model_bundle) if model_bundle.exists() else None, "replay_command_template": ( "PYTHONPATH=01_Product_Source/GFIS_Project python3 " "01_Product_Source/GFIS_Project/scripts/generate_v2_simulator_evidence.py " f"--run-id --hours {args.hours}" ), } metadata_path = output / "run_metadata.json" metadata_path.write_text(json.dumps(metadata, indent=2)) artifacts = {} for path in [trace_path, summary_path, plot_path, metadata_path]: artifacts[str(path.relative_to(output))] = { "bytes": path.stat().st_size, "sha256": sha256(path), } manifest = { "run_id": args.run_id, "status": "complete", "evidence_label": metadata["evidence_label"], "artifacts": artifacts, } (output / "artifact_manifest.json").write_text(json.dumps(manifest, indent=2)) print(output) print(json.dumps(metadata, indent=2)) if __name__ == "__main__": main()