From 4f3f7169be0e893f97187bd24a073fd811a88dfe Mon Sep 17 00:00:00 2001 From: Joe Rivera Date: Sat, 26 Sep 2026 18:49:01 -0500 Subject: [PATCH] =?UTF-8?q?cli:=20grl-snam=20material-raster=20=E2=80=94?= =?UTF-8?q?=20segment=20satellite=20imagery=20into=20the=20material=20pale?= =?UTF-8?q?tte?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Promote the ad-hoc satellite->material generator into a first-class, reusable grl-snam command so any scene's satellite orthophoto can be turned into the material input the nav-stats scorecard + drive need (otherwise the grip/material columns are all-zero because a scene carries no material data). - material_palette: add GRIP_MU + TERRAIN_RISK (per-material dry grip mu + mobility risk) with grip_mu()/terrain_risk() accessors — the physical reading of the shared palette, keyed by id, so the segmenter and any consumer pull ONE table. open_air (roads/pavement/developed hard ground) is the full-grip / no-risk surface; the risk-bearing ids (mu<1) are exactly RISK_MATERIAL_IDS. - grl_snam.tools.material_raster: classify(), to_material_json(), generate(), from_bundle() — a deterministic color/index land-cover classifier (excess-green vegetation, warm-tan soil, dark-blue water, bright warm-gray rock, else the neutral open_air surface), oriented to the sim grid (row 0 == world min_y, matching nav_samplers/sim_world occupancy — the orthophoto is north-up so it flips y). Writes the cvc-scene-material/1 material.json (id grid + palette-sourced mu/risk). Reuses imageio (already a dep), not a new one. NOTE it never emits building materials: reinforced_concrete is a building-WALL RF class (not pavement), buildings are obstacles tagged from scene metadata. - cli: `grl-snam material-raster BUNDLE` (grid+bounds+image from terrain.json/satellite.png) or `--satellite + --rows + --cols + --bounds` for any other orthophoto; -o/--out, --preview. Tests (5): classifier tags + the y-flip, palette-sourced mu/risk in material.json, grip/risk defaults + RISK_MATERIAL_IDS consistency, generate + preview, from_bundle. Verified byte-identical to the prototype on the Austin bundle (open_air 58% / foliage 31% / soil 9% / water 1.3% / rock 0.6%). --- grl_snam/cli.py | 66 ++++++++++++ grl_snam/material_palette.py | 32 ++++++ grl_snam/tools/material_raster.py | 174 ++++++++++++++++++++++++++++++ tests/test_material_raster.py | 97 +++++++++++++++++ 4 files changed, 369 insertions(+) create mode 100644 grl_snam/tools/material_raster.py create mode 100644 tests/test_material_raster.py diff --git a/grl_snam/cli.py b/grl_snam/cli.py index 4d34092..bacd6f7 100644 --- a/grl_snam/cli.py +++ b/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: /satellite.png)", +) +@click.option( + "--rows", type=int, default=None, help="grid rows (default: from /terrain.json)" +) +@click.option( + "--cols", type=int, default=None, help="grid cols (default: from /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: /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)) diff --git a/grl_snam/material_palette.py b/grl_snam/material_palette.py index faee9a3..628f3f4 100644 --- a/grl_snam/material_palette.py +++ b/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 diff --git a/grl_snam/tools/material_raster.py b/grl_snam/tools/material_raster.py new file mode 100644 index 0000000..cec400c --- /dev/null +++ b/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 ``/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 diff --git a/tests/test_material_raster.py b/tests/test_material_raster.py new file mode 100644 index 0000000..ea56ab4 --- /dev/null +++ b/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