Spaces:
Sleeping
Sleeping
File size: 4,728 Bytes
d70361b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 | """Read-only Delhi evaluation loader — Priyanka's manifest schema.
See ``docs/delhi_eval/README.md``. Paths in the manifest are relative to the
repo root (not the eval folder). Uday scripts only read; never write eval data.
"""
from __future__ import annotations
import json
from pathlib import Path
from typing import Iterator
import numpy as np
from PIL import Image
ROOT = Path(__file__).resolve().parent.parent.parent
DEFAULT_MANIFEST = ROOT / "docs" / "delhi_eval" / "manifest.json"
LABELS_DIR = ROOT / "docs" / "delhi_eval" / "labels"
class DelhiEvalNotReady(Exception):
"""Manifest missing or unreadable."""
def manifest_path(path: str | Path | None = None) -> Path:
return Path(path).resolve() if path else DEFAULT_MANIFEST
def load_manifest(path: str | Path | None = None, *, required: bool = True) -> dict:
mpath = manifest_path(path)
if not mpath.is_file():
if required:
raise DelhiEvalNotReady(
f"Delhi manifest not found at {mpath}. "
"Merge New/Priyanka or run scripts/build_delhi_manifest.py --init."
)
return {"pairs": []}
return json.loads(mpath.read_text(encoding="utf-8"))
def _resolve_gt_path(pair: dict) -> Path | None:
gt_rel = pair.get("gt_mask")
if gt_rel:
p = ROOT / gt_rel
return p if p.is_file() else None
pair_id = pair.get("pair_id") or pair.get("id")
if pair_id:
auto = LABELS_DIR / f"{pair_id}.png"
if auto.is_file():
return auto
return None
def _load_rgb(path: Path) -> np.ndarray:
if path.suffix.lower() in (".tif", ".tiff"):
from app.dda.geotiff_io import load_rgb_pil
return np.array(load_rgb_pil(path))
return np.array(Image.open(path).convert("RGB"))
def _load_label(path: Path) -> np.ndarray:
return np.array(Image.open(path).convert("L"))
def iter_delhi_pairs(
manifest: str | Path | None = None,
*,
require_gt: bool = False,
) -> Iterator[tuple[np.ndarray, np.ndarray, np.ndarray | None, str, str | None, str | None]]:
"""Yield (before, after, gt_or_none, pair_id, before_path, after_path)."""
data = load_manifest(manifest, required=True)
missing: list[str] = []
for pair in data.get("pairs", []):
pair_id = pair.get("pair_id") or pair.get("id") or "unknown"
before_rel = pair.get("before_path") or pair.get("before")
after_rel = pair.get("after_path") or pair.get("after")
if not (before_rel and after_rel):
missing.append(f"{pair_id}: missing before/after paths")
continue
before_p = ROOT / before_rel
after_p = ROOT / after_rel
if not before_p.is_file() or not after_p.is_file():
missing.append(f"{pair_id}: image missing on disk")
continue
try:
before = _load_rgb(before_p)
after = _load_rgb(after_p)
except Exception as exc:
missing.append(f"{pair_id}: load failed ({exc})")
continue
gt = None
gt_path = _resolve_gt_path(pair)
if gt_path is not None:
gt = _load_label(gt_path)
elif require_gt:
continue
is_tif = before_p.suffix.lower() in (".tif", ".tiff")
bp = str(before_p) if is_tif else None
ap = str(after_p) if is_tif else None
yield before, after, gt, pair_id, bp, ap
if missing:
print(f" WARNING: skipped {len(missing)} manifest entries:")
for msg in missing[:8]:
print(f" - {msg}")
if len(missing) > 8:
print(f" ... and {len(missing) - 8} more")
def count_manifest_pairs(manifest: str | Path | None = None) -> dict:
data = load_manifest(manifest, required=True)
pairs = data.get("pairs", [])
on_disk = sum(
1 for p in pairs
if (ROOT / p["before_path"]).is_file() and (ROOT / p["after_path"]).is_file()
)
labeled = sum(1 for p in pairs if _resolve_gt_path(p) is not None)
return {"total": len(pairs), "on_disk": on_disk, "labeled": labeled}
def dummy_delhi_pairs(n: int = 2, size: int = 384) -> list[tuple]:
"""In-memory synthetic pairs for Uday scaffold runs."""
rng = np.random.default_rng(42)
out = []
for i in range(n):
before = rng.integers(40, 200, (size, size, 3), dtype=np.uint8)
before[:, size // 3: size // 3 + 6] = [90, 90, 90]
after = before.copy()
gt = np.zeros((size, size), dtype=np.uint8)
x, y, w, h = 60 + i * 40, 70 + i * 20, 50, 40
after[y:y + h, x:x + w] = [205, 200, 190]
gt[y:y + h, x:x + w] = 255
out.append((before, after, gt, f"dummy_{i:02d}", None, None))
return out
|