cli: grl-snam material-raster — segment satellite imagery into the material palette

grl_snam/cli.py

@@ -6,6 +6,7 @@
   grl-snam selftest                      numerical correctness check (no data)
   grl-snam obstacles   BUNDLE            world model -> circular obstacles (.npz)
   grl-snam build-sdf   BUNDLE            world model -> navigation SDF (.npz)
+  grl-snam material-raster BUNDLE        satellite -> material.json (id grid + grip/risk)
   grl-snam train       SDF.npz           self-supervised SDF-coefficient training
   grl-snam capture drive|multigoal ...   drive the policy -> mp4 (offscreen, HUD)
   grl-snam pipeline    BUNDLE            world model -> SDF -> train -> video (all)
@@ -98,6 +99,71 @@ def build_sdf(bundle: str, source: str, region: float, grid: int, out: str | Non
     _sdf.build(bundle, source=source, region=region, grid=grid, out=out)
 
 
+@main.command("material-raster")
+@click.argument("bundle", required=False, type=click.Path(exists=True, file_okay=False))
+@click.option(
+    "--satellite",
+    type=click.Path(exists=True, dir_okay=False),
+    help="orthophoto to segment (default: <BUNDLE>/satellite.png)",
+)
+@click.option(
+    "--rows", type=int, default=None, help="grid rows (default: from <BUNDLE>/terrain.json)"
+)
+@click.option(
+    "--cols", type=int, default=None, help="grid cols (default: from <BUNDLE>/terrain.json)"
+)
+@click.option(
+    "--bounds",
+    nargs=4,
+    type=float,
+    default=None,
+    metavar="MIN_X MIN_Y MAX_X MAX_Y",
+    help="world extent (default: from terrain.json)",
+)
+@click.option(
+    "-o", "--out", default=None, help="output material.json (default: <BUNDLE>/material.json)"
+)
+@click.option("--preview", default=None, help="also write a color-coded tag-map PNG for review")
+def material_raster(
+    bundle: str | None,
+    satellite: str | None,
+    rows: int | None,
+    cols: int | None,
+    bounds,
+    out: str | None,
+    preview: str | None,
+) -> None:
+    """Satellite orthophoto -> scene material.json (material-id grid + grip/risk).
+
+    \b
+    Classifies a north-up orthophoto into the shared 13-class material palette on the world grid and
+    derives per-material grip (mu) + terrain risk — the material input the nav-stats scorecard buckets
+    time by and the drive reads for grip/reroute. Give a scene BUNDLE (a dir with terrain.json +
+    satellite.png) to take the grid + bounds + image from it, or drop BUNDLE and pass --satellite +
+    --rows + --cols + --bounds to segment any other satellite data.
+    """
+    import json as _json
+    import os as _os
+
+    from .tools import material_raster as _mr
+
+    terr = _json.load(open(_os.path.join(bundle, "terrain.json"))) if bundle else None
+    sat = satellite or (_os.path.join(bundle, "satellite.png") if bundle else None)
+    R = rows if rows is not None else (int(terr["rows"]) if terr else None)
+    C = cols if cols is not None else (int(terr["cols"]) if terr else None)
+    bnds = bounds if bounds else (terr["bounds"] if terr else None)
+    op = out or (_os.path.join(bundle, "material.json") if bundle else None)
+    if not (sat and R and C and bnds and op):
+        raise click.UsageError(
+            "provide a BUNDLE (terrain.json + satellite.png), or --satellite + --rows + --cols + "
+            "--bounds + --out"
+        )
+    dist = _mr.generate(sat, R, C, bnds, op, preview)
+    click.echo(f"wrote {op}  ({R}x{C} grid)")
+    for name, frac in dist.items():
+        click.echo(f"  {name:20s} {100 * frac:5.1f}%")
+
+
 # ── training ─────────────────────────────────────────────────────────────────
 @main.command()
 @click.argument("sdf_npz", type=click.Path(exists=True, dir_okay=False))

grl_snam/material_palette.py

@@ -49,6 +49,38 @@
 
 OPEN_AIR_ID = MATERIAL_ID["open_air"]
 
+#: Per-material dry GRIP mu (1.0 = full grip) and mobility RISK (0..1) — the physical reading of the
+#: palette that mobility scoring uses (the drive's grip friction_field + the material risk plane).
+#: Keyed by material id; only the drivable-surface classes differ from the neutral baseline. open_air
+#: is the safe full-grip / no-risk surface (roads, pavement, developed ground — the palette has no
+#: dedicated pavement class), and the building materials never underlie a driven cell, so anything not
+#: listed defaults to mu=1 / risk=0 via :func:`grip_mu` / :func:`terrain_risk`. The risk-bearing ids
+#: (mu<1, risk>0) are exactly :data:`RISK_MATERIAL_IDS`.
+GRIP_MU = {
+    OPEN_AIR_ID: 1.0,
+    MATERIAL_ID["foliage"]: 0.65,
+    MATERIAL_ID["soil"]: 0.55,
+    MATERIAL_ID["rock"]: 0.50,
+    MATERIAL_ID["water"]: 0.30,
+}
+TERRAIN_RISK = {
+    OPEN_AIR_ID: 0.0,
+    MATERIAL_ID["foliage"]: 0.30,
+    MATERIAL_ID["soil"]: 0.50,
+    MATERIAL_ID["rock"]: 0.80,
+    MATERIAL_ID["water"]: 1.0,
+}
+
+
+def grip_mu(material_id: int) -> float:
+    """Dry grip mu for a material id (1.0 = full grip); unlisted ids (building materials, unknowns) -> 1.0."""
+    return GRIP_MU.get(material_id, 1.0)
+
+
+def terrain_risk(material_id: int) -> float:
+    """Mobility risk (0..1) for a material id; unlisted ids -> 0.0."""
+    return TERRAIN_RISK.get(material_id, 0.0)
+
 
 def terrain_risk_share(material_time_share) -> float:
     """Fraction of fleet time spent over terrain-risk materials — the "how often the

grl_snam/tools/material_raster.py

@@ -0,0 +1,174 @@
+"""Segment a satellite orthophoto into the shared 13-class material palette and write a scene
+``material.json`` raster (the material-id grid + per-material grip/risk tables).
+
+The nav-stats scorecard buckets time/distance by material id (``cvc::nav`` ``nav_samplers.material_id``
+→ ``material_time_share``), and the drive reads grip ``mu`` (a friction field) + material risk to slow
+on soft ground and reroute around hazard. A scene that carries no material data leaves all of those at
+their neutral defaults (mu=1, risk=0) and the buckets empty. This turns a scene's satellite imagery —
+which the terrain mesh already drapes as its texture — into that material input: one deterministic
+land-cover classification feeds all three consumers.
+
+Reusable beyond the cvc Austin bundle: point it at any north-up orthophoto plus the world bounds +
+grid it covers (a scene bundle's ``terrain.json`` supplies those, or pass them explicitly). The
+classifier is a documented color/index heuristic (excess-green vegetation, warm-tan soil, dark-blue
+water, bright warm-gray rock, everything else the neutral ``open_air`` hard/free surface), not a
+learned model — reproducible and reviewable; tune :func:`classify` + the palette's ``GRIP_MU`` /
+``TERRAIN_RISK`` for other biomes.
+
+NOTE on the palette: it is an RF/building-material table (``reinforced_concrete`` etc. carry
+penetration-loss dB) plus a few natural-terrain classes, with NO asphalt/road class. ``reinforced_
+concrete`` means building walls, not pavement; buildings are obstacles the convoy never drives
+(tagged from the scene's building metadata), so this GROUND classifier only ever emits the drivable
+classes ``open_air`` / ``foliage`` / ``soil`` / ``water`` / ``rock``.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+
+import numpy as np
+
+from ..material_palette import (
+    MATERIAL_ID,
+    MATERIALS,
+    OPEN_AIR_ID,
+    grip_mu,
+    terrain_risk,
+)
+
+FOLIAGE = MATERIAL_ID["foliage"]
+SOIL = MATERIAL_ID["soil"]
+WATER = MATERIAL_ID["water"]
+ROCK = MATERIAL_ID["rock"]
+
+# color-coded display colors for the review preview (id -> RGB)
+_PREVIEW_COL = {
+    OPEN_AIR_ID: (170, 170, 175),
+    FOLIAGE: (60, 150, 60),
+    SOIL: (170, 130, 80),
+    ROCK: (120, 110, 95),
+    WATER: (40, 90, 200),
+}
+
+
+def classify(sat_rgb: np.ndarray, rows: int, cols: int) -> np.ndarray:
+    """Mean-pool a north-up orthophoto to ``(rows, cols)`` and tag each cell into the drivable palette.
+
+    Returns an ``int32`` ``[rows, cols]`` grid oriented to the SIM convention — row 0 == world
+    ``min_y``, col 0 == ``min_x`` — so a consumer sampling ``material_id[r][c]`` with
+    ``r=(y-min_y)/(max_y-min_y)*rows`` / ``c=(x-min_x)/(max_x-min_x)*cols`` (the ``nav_samplers`` /
+    ``sim_world`` occupancy convention) is a straight index. The orthophoto is north-up (image row 0 ==
+    ``max_y``) and the mesh drapes it with a lower-left UV origin, so the pooled grid is flipped in y.
+    """
+    im = np.asarray(sat_rgb)[..., :3].astype(np.float32)
+    H, W = im.shape[0], im.shape[1]
+    ph, pw = H // rows, W // cols
+    if ph < 1 or pw < 1:
+        raise ValueError(f"image {W}x{H} smaller than the {cols}x{rows} grid")
+    im = im[: rows * ph, : cols * pw].reshape(rows, ph, cols, pw, 3).mean(axis=(1, 3))
+
+    r, g, b = im[..., 0], im[..., 1], im[..., 2]
+    bright = (r + g + b) / 3.0
+    exg = 2.0 * g - r - b  # excess green (positive over foliage incl. muted olive canopy)
+    warm = r - b
+
+    # DEFAULT open_air = the neutral hard/free surface (roads, pavement, developed ground); carve out
+    # only the terrain-RISK classes on top. Order matters (later rules win on overlap).
+    mid = np.full((rows, cols), OPEN_AIR_ID, dtype=np.int32)
+    mid[(warm > 14) & (r > g) & (g >= b) & (exg < 6)] = (
+        SOIL  # warm tan/brown = bare / natural ground
+    )
+    mid[(exg > 14) & (g >= b)] = FOLIAGE  # excess-green = grass / tree canopy
+    mid[(bright > 170) & (warm > 6) & (exg < 8) & (r > g)] = ROCK  # bright warm-gray = exposed rock
+    mid[(b > r + 6) & (b > g) & (bright < 125)] = WATER  # dark blue = water
+    return mid[::-1].copy()  # flip to row 0 == min_y (see docstring)
+
+
+def _bounds_dict(bounds) -> dict:
+    if isinstance(bounds, dict):
+        return {k: float(bounds[k]) for k in ("min_x", "min_y", "max_x", "max_y")}
+    min_x, min_y, max_x, max_y = bounds
+    return {
+        "min_x": float(min_x),
+        "min_y": float(min_y),
+        "max_x": float(max_x),
+        "max_y": float(max_y),
+    }
+
+
+def to_material_json(material_id: np.ndarray, bounds) -> dict:
+    """Build the ``cvc-scene-material/1`` document from a material-id grid + world bounds."""
+    rows, cols = int(material_id.shape[0]), int(material_id.shape[1])
+    ids = sorted(int(v) for v in np.unique(material_id))
+    return {
+        "schema": "cvc-scene-material/1",
+        "provenance": "satellite-derived land cover (grl-snam material-raster); ids = cvc::dbg MATERIAL_TABLE",
+        "rows": rows,
+        "cols": cols,
+        "bounds": _bounds_dict(bounds),
+        "palette": list(MATERIALS),
+        # per-material grip + risk from the shared palette (grl_snam.material_palette); a loader
+        # defaults any absent id to mu=1 / risk=0.
+        "mu": {str(i): grip_mu(i) for i in ids},
+        "risk": {str(i): terrain_risk(i) for i in ids},
+        "material_id": material_id.tolist(),  # [rows][cols] row-major; row 0 == min_y, col 0 == min_x
+    }
+
+
+def write_preview(material_id: np.ndarray, path: str, scale: int = 3) -> None:
+    """Write a color-coded tag map (nearest-neighbor upscaled) for visual review."""
+    import imageio.v3 as iio
+
+    rows, cols = material_id.shape
+    vis = np.zeros((rows, cols, 3), dtype=np.uint8)
+    for mid, col in _PREVIEW_COL.items():
+        vis[material_id == mid] = col
+    vis = np.repeat(np.repeat(vis, scale, axis=0), scale, axis=1)
+    iio.imwrite(path, vis)
+
+
+def distribution(material_id: np.ndarray) -> dict:
+    """{material name: fraction} over the grid, most-common first (for CLI reporting)."""
+    tot = material_id.size
+    out = {}
+    for i in sorted(
+        (int(v) for v in np.unique(material_id)), key=lambda i: -int((material_id == i).sum())
+    ):
+        out[MATERIALS[i]] = float((material_id == i).sum()) / tot
+    return out
+
+
+def generate(
+    satellite: str, rows: int, cols: int, bounds, out: str, preview: str | None = None
+) -> dict:
+    """Segment ``satellite`` over an ``rows x cols`` grid spanning ``bounds`` and write ``out``
+    (``material.json``). ``bounds`` is a dict (min_x/min_y/max_x/max_y) or a 4-tuple. Returns the
+    class distribution."""
+    import imageio.v3 as iio
+
+    material_id = classify(iio.imread(satellite), rows, cols)
+    doc = to_material_json(material_id, bounds)
+    with open(out, "w") as f:
+        json.dump(doc, f, separators=(",", ":"))
+    if preview:
+        write_preview(material_id, preview)
+    return distribution(material_id)
+
+
+def from_bundle(
+    bundle: str, out: str | None = None, preview: str | None = None
+) -> tuple[str, dict]:
+    """Convenience: read a cvc scene bundle's ``terrain.json`` (rows/cols/bounds) + ``satellite.png``,
+    segment, and write ``<bundle>/material.json`` (or ``out``). Returns (out_path, distribution)."""
+    terr = json.load(open(os.path.join(bundle, "terrain.json")))
+    out = out or os.path.join(bundle, "material.json")
+    dist = generate(
+        os.path.join(bundle, "satellite.png"),
+        int(terr["rows"]),
+        int(terr["cols"]),
+        terr["bounds"],
+        out,
+        preview,
+    )
+    return out, dist

tests/test_material_raster.py

@@ -0,0 +1,97 @@
+"""grl-snam material-raster: satellite -> material palette segmentation + grip/risk (nav-stats item 2)."""
+
+import json
+
+import numpy as np
+
+from grl_snam import material_palette as mp
+from grl_snam.tools import material_raster as mr
+
+
+def test_classifier_tags_and_flips_to_sim_orientation():
+    # 2x2 orthophoto (one pixel per grid cell), north-up. Top row = north (max_y), bottom = south.
+    #   top:    [green vegetation, neutral gray]      bottom: [warm tan soil, dark blue water]
+    img = np.array(
+        [
+            [[50, 160, 50], [150, 150, 150]],
+            [[175, 130, 90], [40, 90, 200]],
+        ],
+        dtype=np.uint8,
+    )
+    mid = mr.classify(img, rows=2, cols=2)
+    assert mid.shape == (2, 2)
+    # classify() flips rows so material_id[0] == world min_y (the image's BOTTOM row). So grid row 0
+    # is the tan/blue (soil/water) south row, grid row 1 is the green/gray north row.
+    assert mid[0, 0] == mp.MATERIAL_ID["soil"]
+    assert mid[0, 1] == mp.MATERIAL_ID["water"]
+    assert mid[1, 0] == mp.MATERIAL_ID["foliage"]
+    assert mid[1, 1] == mp.OPEN_AIR_ID  # neutral gray -> the road/hard-surface default
+
+
+def test_material_json_pulls_grip_and_risk_from_the_palette():
+    mid = np.array(
+        [
+            [mp.MATERIAL_ID["foliage"], mp.OPEN_AIR_ID],
+            [mp.MATERIAL_ID["soil"], mp.MATERIAL_ID["water"]],
+        ],
+        dtype=np.int32,
+    )
+    doc = mr.to_material_json(mid, {"min_x": -100, "min_y": -50, "max_x": 100, "max_y": 50})
+    assert doc["schema"] == "cvc-scene-material/1"
+    assert doc["rows"] == 2 and doc["cols"] == 2
+    assert doc["bounds"] == {"min_x": -100.0, "min_y": -50.0, "max_x": 100.0, "max_y": 50.0}
+    assert doc["material_id"] == mid.tolist()
+    # mu/risk come from grl_snam.material_palette, keyed by the ids present
+    assert doc["mu"][str(mp.MATERIAL_ID["foliage"])] == mp.GRIP_MU[mp.MATERIAL_ID["foliage"]]
+    assert doc["risk"][str(mp.MATERIAL_ID["water"])] == mp.TERRAIN_RISK[mp.MATERIAL_ID["water"]]
+    assert doc["mu"][str(mp.OPEN_AIR_ID)] == 1.0 and doc["risk"][str(mp.OPEN_AIR_ID)] == 0.0
+
+
+def test_palette_grip_risk_defaults_and_risk_set_consistency():
+    # unlisted ids (building materials, unknowns) default to full grip / no risk
+    assert mp.grip_mu(mp.MATERIAL_ID["reinforced_concrete"]) == 1.0
+    assert mp.terrain_risk(mp.MATERIAL_ID["brick"]) == 0.0
+    assert mp.grip_mu(999) == 1.0 and mp.terrain_risk(999) == 0.0
+    # open_air is the safe surface; the risk-bearing ids (mu<1 AND risk>0) are exactly RISK_MATERIAL_IDS
+    assert mp.grip_mu(mp.OPEN_AIR_ID) == 1.0 and mp.terrain_risk(mp.OPEN_AIR_ID) == 0.0
+    risky = {i for i in range(mp.NUM_MATERIALS) if mp.grip_mu(i) < 1.0}
+    assert risky == set(mp.RISK_MATERIAL_IDS)
+    assert all(mp.terrain_risk(i) > 0.0 for i in mp.RISK_MATERIAL_IDS)
+
+
+def test_generate_writes_material_json(tmp_path):
+    img = np.zeros((4, 4, 3), dtype=np.uint8)
+    img[:, :] = (150, 150, 150)  # all neutral -> all open_air
+    img[0, 0] = (50, 170, 50)  # one green cell
+    import imageio.v3 as iio
+
+    sat = tmp_path / "sat.png"
+    iio.imwrite(sat, img)
+    out = tmp_path / "material.json"
+    preview = tmp_path / "tags.png"
+    dist = mr.generate(str(sat), 4, 4, (-8, -8, 8, 8), str(out), preview=str(preview))
+    doc = json.loads(out.read_text())
+    assert doc["rows"] == 4 and doc["cols"] == 4
+    assert np.array(doc["material_id"]).shape == (4, 4)
+    assert "open_air" in dist and dist["open_air"] > 0.5
+    assert abs(sum(dist.values()) - 1.0) < 1e-6
+    assert preview.exists()  # write_preview ran
+
+
+def test_from_bundle_reads_terrain_and_satellite(tmp_path):
+    import imageio.v3 as iio
+
+    (tmp_path / "terrain.json").write_text(
+        json.dumps(
+            {"rows": 4, "cols": 4, "bounds": {"min_x": -8, "min_y": -8, "max_x": 8, "max_y": 8}}
+        )
+    )
+    img = np.full((4, 4, 3), (150, 150, 150), dtype=np.uint8)
+    img[2, 2] = (40, 90, 200)  # a water cell
+    iio.imwrite(tmp_path / "satellite.png", img)
+
+    out_path, dist = mr.from_bundle(str(tmp_path))
+    assert out_path == str(tmp_path / "material.json")
+    doc = json.loads((tmp_path / "material.json").read_text())
+    assert doc["rows"] == 4 and doc["bounds"]["max_y"] == 8.0
+    assert "water" in dist