Spaces:
Sleeping
Sleeping
Chandra-Align
feat: scientific photogrammetric workstation - preprocessing, ground metrics, visualization, export, enhanced UI
38d2728 | """Evaluation harness (M7.1-M7.3) — code only, results come from real runs. | |
| - M7.1: Held-out check-point protocol (automated split, RMSE on check-set only) | |
| - M7.2: Success-rate vs solar-azimuth-difference chart generator (UNMEASURED until real pairs exist) | |
| - M7.3: Per-method × per-region table generator (reads metrics JSONs, emits markdown with UNMEASURED cells) | |
| """ | |
| import json | |
| import csv | |
| import glob | |
| import os | |
| from pathlib import Path | |
| def collect_run_metrics(runs_dir="outputs"): | |
| """Collect all metrics.json from run directories.""" | |
| runs_dir = Path(runs_dir) | |
| if not runs_dir.exists(): | |
| return [] | |
| metrics_files = list(runs_dir.glob("*/metrics.json")) | |
| results = [] | |
| for mf in metrics_files: | |
| try: | |
| with open(mf, "r") as f: | |
| data = json.load(f) | |
| data["_run_dir"] = str(mf.parent) | |
| data["_run_id"] = mf.parent.name | |
| results.append(data) | |
| except (json.JSONDecodeError, OSError): | |
| continue | |
| return results | |
| def read_download_log(log_path="data/download_log.csv"): | |
| """Read download_log.csv into a list of dicts.""" | |
| if not os.path.exists(log_path): | |
| return [] | |
| with open(log_path, "r") as f: | |
| reader = csv.DictReader(f) | |
| return list(reader) | |
| def merge_run_with_log(metrics_list, log_entries): | |
| """Merge metrics with download log entries by run_id.""" | |
| log_by_run = {entry.get("run_id"): entry for entry in log_entries} | |
| for m in metrics_list: | |
| run_id = m.get("_run_id") | |
| if run_id and run_id in log_by_run: | |
| m["log"] = log_by_run[run_id] | |
| return metrics_list | |
| # --- M7.1: Held-out check-point protocol --- | |
| def report_heldout_rmse(metrics_dict): | |
| """Extract held-out RMSE from a metrics dict. Returns 'UNMEASURED' if not present.""" | |
| rmse = metrics_dict.get("rmse", {}) | |
| if isinstance(rmse, dict): | |
| if rmse.get("held_out"): | |
| return rmse.get("rmse_px", "UNMEASURED") | |
| return "UNMEASURED" | |
| def check_circularity_guard(metrics_dict): | |
| """Verify that circularity guard fired: fit-set RMSE != held-out RMSE. | |
| Since we never compute fit-set RMSE, this is structurally enforced. | |
| This function just documents the guarantee. | |
| """ | |
| return { | |
| "circularity_guard_enforced": True, | |
| "note": "fit-set RMSE never computed; only held-out RMSE reported" | |
| } | |
| # --- M7.3: Per-method × per-region table --- | |
| METHODS = ["sift", "rift2", "lightglue_aliked"] | |
| REGIONS = ["equatorial", "polar"] | |
| def method_from_metrics(m): | |
| """Extract method name from metrics dict.""" | |
| matcher = str(m.get("matcher", "unknown")).lower() | |
| if "rift2" in matcher: | |
| return "rift2" | |
| elif "lightglue" in matcher or "tier2" in matcher: | |
| return "lightglue_aliked" | |
| elif "sift" in matcher: | |
| return "sift" | |
| return matcher | |
| def region_from_log(log_entry): | |
| """Determine region from log entry (equatorial/polar).""" | |
| # Simple heuristic based on solar elevation or latitude in future | |
| lat = log_entry.get("latitude", "") if isinstance(log_entry, dict) else "" | |
| if lat: | |
| try: | |
| lat_val = float(lat) | |
| if abs(lat_val) > 60: | |
| return "polar" | |
| except ValueError: | |
| pass | |
| # Default to equatorial if unknown | |
| return "equatorial" | |
| def build_method_region_table(metrics_list): | |
| """Build per-method × per-region table. | |
| Cells contain held-out RMSE (px) or "UNMEASURED" if no run ID exists for that combo. | |
| """ | |
| # Initialize table with UNMEASURED | |
| table = {method: {region: "UNMEASURED" for region in REGIONS} for method in METHODS} | |
| for m in metrics_list: | |
| method = method_from_metrics(m) | |
| log = m.get("log", {}) | |
| region = region_from_log(log) | |
| if method in METHODS and region in REGIONS: | |
| rmse = report_heldout_rmse(m) | |
| # Only overwrite if we have a real measurement | |
| if rmse != "UNMEASURED": | |
| table[method][region] = f"{float(rmse):.3f}" | |
| return table | |
| def emit_method_region_table_markdown(table): | |
| """Emit the M7.3 table as markdown.""" | |
| lines = ["| Method | Equatorial RMSE (px) | Polar RMSE (px) |", "|---|---|---|"] | |
| for method in METHODS: | |
| eq = table.get(method, {}).get("equatorial", "UNMEASURED") | |
| pol = table.get(method, {}).get("polar", "UNMEASURED") | |
| lines.append(f"| {method} | {eq} | {pol} |") | |
| lines.append("") | |
| lines.append("*Cells without a run ID show UNMEASURED. Values are held-out RMSE in pixels.*") | |
| return "\n".join(lines) | |
| # --- M7.2: Solar-azimuth-difference robustness chart --- | |
| def extract_solar_azimuth_diff(log_entry): | |
| """Extract solar azimuth difference from log entry.""" | |
| # In a real run, this would be the difference between CH-2 and reference acquisition | |
| # For now, return None (no data) | |
| return log_entry.get("solar_azimuth_diff") if log_entry else None | |
| def build_azimuth_vs_success_table(metrics_list): | |
| """Build data for success-rate vs solar-azimuth-difference chart. | |
| Returns list of dicts: {solar_azimuth_diff_deg, success_rate, method, region} | |
| """ | |
| results = [] | |
| for m in metrics_list: | |
| log = m.get("log", {}) | |
| method = method_from_metrics(m) | |
| region = region_from_log(log) | |
| azimuth_diff = extract_solar_azimuth_diff(log) | |
| success_rate = m.get("success_rate", "UNMEASURED") | |
| if azimuth_diff is not None: | |
| results.append({ | |
| "solar_azimuth_diff_deg": float(azimuth_diff), | |
| "success_rate": success_rate if success_rate != "UNMEASURED" else None, | |
| "method": method, | |
| "region": region, | |
| }) | |
| return results | |
| def emit_azimuth_chart_data(metrics_list, output_path="outputs/azimuth_vs_success.csv"): | |
| """Emit chart data as CSV for external plotting.""" | |
| data = build_azimuth_vs_success_table(metrics_list) | |
| if not data: | |
| # Write empty header for now | |
| Path(output_path).parent.mkdir(parents=True, exist_ok=True) | |
| with open(output_path, "w", newline="") as f: | |
| writer = csv.DictWriter(f, fieldnames=["solar_azimuth_diff_deg", "success_rate", "method", "region"]) | |
| writer.writeheader() | |
| return "UNMEASURED (no solar azimuth diff data in logs)" | |
| Path(output_path).parent.mkdir(parents=True, exist_ok=True) | |
| with open(output_path, "w", newline="") as f: | |
| writer = csv.DictWriter(f, fieldnames=["solar_azimuth_diff_deg", "success_rate", "method", "region"]) | |
| writer.writeheader() | |
| writer.writerows(data) | |
| return f"wrote {len(data)} rows to {output_path}" | |
| # --- Main evaluation CLI --- | |
| def run_evaluation(runs_dir="outputs", log_path="data/download_log.csv"): | |
| """Run the evaluation pipeline: collect metrics, merge with log, emit tables/charts.""" | |
| metrics = collect_run_metrics(runs_dir) | |
| log = read_download_log(log_path) | |
| merged = merge_run_with_log(metrics, log) | |
| # M7.3 table | |
| table = build_method_region_table(merged) | |
| md_table = emit_method_region_table_markdown(table) | |
| # M7.2 chart data | |
| azimuth_msg = emit_azimuth_chart_data(merged) | |
| # Summary | |
| summary = { | |
| "n_runs": len(merged), | |
| "method_region_table": table, | |
| "azimuth_chart": azimuth_msg, | |
| "circularity_guard": "enforced (held-out RMSE only)", | |
| } | |
| return summary, md_table | |
| if __name__ == "__main__": | |
| import sys | |
| runs_dir = sys.argv[1] if len(sys.argv) > 1 else "outputs" | |
| summary, md_table = run_evaluation(runs_dir) | |
| print(md_table) | |
| print("\n--- Summary ---") | |
| print(json.dumps(summary, indent=2)) |