| """Build and freeze Phase 6 dataset manifest + split documents.""" |
|
|
| from __future__ import annotations |
|
|
| import json |
| import os |
| import sqlite3 |
| import sys |
| import csv |
| import hashlib |
| from pathlib import Path |
| from typing import Any, Dict, List |
|
|
| PROJECT_ROOT = Path(__file__).resolve().parents[1] |
| if str(PROJECT_ROOT) not in sys.path: |
| sys.path.insert(0, str(PROJECT_ROOT)) |
|
|
| from ml.training.dataset_manifest import build_dataset_manifest, build_dataset_manifest_from_samples |
|
|
|
|
| def _load_env_file(path: Path) -> None: |
| if not path.exists(): |
| return |
| for raw_line in path.read_text(encoding="utf-8").splitlines(): |
| line = raw_line.strip() |
| if not line or line.startswith("#") or "=" not in line: |
| continue |
| key, value = line.split("=", 1) |
| os.environ.setdefault(key.strip(), value.strip().strip('"').strip("'")) |
|
|
|
|
| def _site_split(site_id: str, site_metadata: Dict[str, Any]) -> str: |
| configured = str(site_metadata.get("dataset_split", "")).strip().lower() |
| if configured in {"train", "validation", "test", "demo"}: |
| return configured |
| bucket = int(hashlib.sha256(site_id.encode("utf-8")).hexdigest()[:8], 16) % 10 |
| if bucket == 0: |
| return "test" |
| if bucket in {1, 2}: |
| return "validation" |
| return "train" |
|
|
|
|
| def _load_manual_corrections(path: Path) -> Dict[str, Dict[str, Any]]: |
| if not path.exists(): |
| return {} |
| corrections: Dict[str, Dict[str, Any]] = {} |
| with path.open("r", encoding="utf-8-sig", newline="") as fh: |
| reader = csv.DictReader(fh) |
| for idx, row in enumerate(reader, start=2): |
| scan_id = str(row.get("scan_id", "")).strip() |
| if not scan_id: |
| continue |
| parsed: Dict[str, Any] = {} |
| if str(row.get("split", "")).strip(): |
| parsed["split"] = str(row["split"]).strip().lower() |
| if str(row.get("plume_likely", "")).strip(): |
| parsed["plume_likely"] = str(row["plume_likely"]).strip().lower() in {"1", "true", "yes", "on"} |
| if str(row.get("bbox_norm", "")).strip(): |
| try: |
| bbox = json.loads(str(row["bbox_norm"]).strip()) |
| if not isinstance(bbox, list) or len(bbox) != 4: |
| raise ValueError("bbox_norm must be a JSON list of four numbers") |
| parsed["bbox_norm"] = bbox |
| except Exception as exc: |
| raise ValueError(f"invalid bbox_norm in corrections row {idx}: {exc}") from exc |
| for key in ("likely_source_zone", "priority_tier", "source_type", "labeler", "notes"): |
| value = str(row.get(key, "")).strip() |
| if value: |
| parsed[key] = value |
| corrections[scan_id] = parsed |
| return corrections |
|
|
|
|
| def _apply_manual_correction(sample: Dict[str, Any], correction: Dict[str, Any]) -> None: |
| if not correction: |
| return |
| annotation = sample["annotation"] |
| provenance = sample["provenance"] |
| if "split" in correction: |
| sample["split"] = correction["split"] |
| for key in ("plume_likely", "bbox_norm", "likely_source_zone", "priority_tier"): |
| if key in correction: |
| annotation[key] = correction[key] |
| for key in ("source_type", "labeler", "notes"): |
| if key in correction: |
| provenance[key] = correction[key] |
| provenance["manual_correction_applied"] = True |
|
|
|
|
| def _collect_live_samples(db_path: Path, limit: int = 1000, corrections: Dict[str, Dict[str, Any]] | None = None) -> List[Dict]: |
| if not db_path.exists(): |
| return [] |
| conn = sqlite3.connect(str(db_path)) |
| conn.row_factory = sqlite3.Row |
| try: |
| rows = conn.execute( |
| """ |
| SELECT |
| s.scan_id, |
| s.site_id, |
| s.status, |
| s.evidence_json, |
| s.created_at, |
| i.payload_json AS incident_json, |
| site.payload_json AS site_json |
| FROM scans s |
| LEFT JOIN incidents i ON i.incident_id = s.incident_id |
| LEFT JOIN sites site ON site.site_id = s.site_id |
| ORDER BY s.created_at ASC |
| LIMIT ? |
| """, |
| (limit,), |
| ).fetchall() |
| finally: |
| conn.close() |
|
|
| samples: List[Dict] = [] |
| for row in rows: |
| evidence = json.loads(row["evidence_json"]) if row["evidence_json"] else {} |
| metadata = evidence.get("metadata", {}) |
| panel_paths = evidence.get("panel_paths", {}) |
| mode = str(metadata.get("mode", "")).lower() |
| if mode != "live": |
| continue |
| provenance = metadata.get("imagery_provenance", {}) |
| if provenance and provenance.get("live_fetch_status") != "live": |
| continue |
| panel_path = panel_paths.get("evidence_panel_path") or panel_paths.get("current_rgb_path") |
| if not panel_path: |
| continue |
| incident = json.loads(row["incident_json"]) if row["incident_json"] else {} |
| site_payload = json.loads(row["site_json"]) if row["site_json"] else {} |
| site_metadata = site_payload.get("metadata", {}) if isinstance(site_payload, dict) else {} |
| bbox = incident.get("bbox_norm") or metadata.get("candidate", {}).get("bbox_norm") or [0.2, 0.2, 0.5, 0.5] |
| zone = incident.get("likely_source_zone") or metadata.get("candidate", {}).get("likely_source_zone_prior") |
| if not zone: |
| zone = "perimeter_or_unknown" |
| priority = incident.get("priority_tier", "medium") |
| review_status = incident.get("review_status", "needs_review") |
| source_type = "manual" if review_status in {"published", "dismissed"} else "weak" |
| labeler = "operator_review" if source_type == "manual" else "model_bootstrap" |
|
|
| samples.append( |
| { |
| "sample_id": f"live_{row['scan_id']}", |
| "site_id": row["site_id"], |
| "split": _site_split(row["site_id"], site_metadata), |
| "panel_artifact_path": str(panel_path), |
| "annotation": { |
| "plume_likely": bool(incident.get("plume_likely", True)), |
| "bbox_norm": bbox, |
| "likely_source_zone": zone, |
| "priority_tier": priority, |
| }, |
| "provenance": { |
| "source_type": source_type, |
| "source_ref": f"scan:{row['scan_id']}", |
| "labeler": labeler, |
| "created_at": row["created_at"], |
| "notes": f"captured from live scan pipeline; scan_status={row['status']}", |
| "region": site_metadata.get("region"), |
| }, |
| } |
| ) |
|
|
| samples.sort(key=lambda sample: sample["sample_id"]) |
| corrections = corrections or {} |
| for sample in samples: |
| scan_id = str(sample["provenance"]["source_ref"]).split("scan:", 1)[-1] |
| _apply_manual_correction(sample, corrections.get(scan_id, {})) |
| return samples |
|
|
|
|
| def _write_live_label_dump(path: Path, samples: List[Dict]) -> None: |
| path.parent.mkdir(parents=True, exist_ok=True) |
| lines = [json.dumps(sample, sort_keys=True) for sample in samples] |
| path.write_text("\n".join(lines) + ("\n" if lines else ""), encoding="utf-8") |
|
|
|
|
| def main() -> None: |
| _load_env_file(PROJECT_ROOT / ".env.local") |
|
|
| label_path = PROJECT_ROOT / "data" / "labels" / "phase6_samples_v1.jsonl" |
| live_label_path = PROJECT_ROOT / "data" / "labels" / "phase6_samples_live_v1.jsonl" |
| corrections_path = PROJECT_ROOT / "data" / "labels" / "manual_label_corrections.csv" |
| manifest_path = PROJECT_ROOT / "data" / "manifests" / "dataset_manifest_v1.json" |
| split_path = PROJECT_ROOT / "data" / "manifests" / "dataset_splits_v1.json" |
| db_path = Path(os.getenv("LS_DB_PATH", "data/processed/landfillsentry.db")) |
| if not db_path.is_absolute(): |
| db_path = PROJECT_ROOT / db_path |
|
|
| corrections = _load_manual_corrections(corrections_path) |
| live_samples = _collect_live_samples(db_path=db_path, corrections=corrections) |
| if live_samples: |
| _write_live_label_dump(live_label_path, live_samples) |
| result = build_dataset_manifest_from_samples( |
| samples=live_samples, |
| manifest_path=manifest_path, |
| split_path=split_path, |
| source_labels_path="data/labels/phase6_samples_live_v1.jsonl", |
| ) |
| source = "live_scans" |
| else: |
| result = build_dataset_manifest( |
| label_path=label_path, |
| manifest_path=manifest_path, |
| split_path=split_path, |
| ) |
| source = "fallback_seed_labels" |
|
|
| print( |
| "Built Phase 6 dataset:", |
| f"source={source}", |
| f"samples={result.sample_count}", |
| f"checksum={result.manifest_checksum}", |
| f"manifest={result.manifest_path}", |
| f"splits={result.split_path}", |
| ) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
