material-raster: prefer authoritative land-cover masks over satellite color

grl_snam/tools/material_raster.py

@@ -103,7 +103,7 @@ def to_material_json(material_id: np.ndarray, bounds) -> dict:
     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",
+        "provenance": "scene land cover (grl-snam material-raster: masks if present, else satellite); ids = cvc::dbg MATERIAL_TABLE",
         "rows": rows,
         "cols": cols,
         "bounds": _bounds_dict(bounds),
@@ -117,13 +117,17 @@ def to_material_json(material_id: np.ndarray, bounds) -> dict:
 
 
 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."""
+    """Write a color-coded tag map (nearest-neighbor upscaled) for visual review, rendered NORTH-UP
+    (image top row == max_y) so it matches the satellite / a human map. The material_id DATA stays
+    sim-oriented (row 0 == world min_y, aligned with sim_world/nav_samplers occupancy); ONLY this
+    review image is flipped for display."""
     import imageio.v3 as iio
 
     rows, cols = material_id.shape
+    disp = material_id[::-1]  # sim row 0 == min_y (south) -> flip so image top == max_y (north)
     vis = np.zeros((rows, cols, 3), dtype=np.uint8)
     for mid, col in _PREVIEW_COL.items():
-        vis[material_id == mid] = col
+        vis[disp == mid] = col
     vis = np.repeat(np.repeat(vis, scale, axis=0), scale, axis=1)
     iio.imwrite(path, vis)
 
@@ -139,6 +143,15 @@ def distribution(material_id: np.ndarray) -> dict:
     return out
 
 
+def _write(material_id: np.ndarray, bounds, out: str, preview: str | None) -> dict:
+    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 generate(
     satellite: str, rows: int, cols: int, bounds, out: str, preview: str | None = None
 ) -> dict:
@@ -147,28 +160,81 @@ def generate(
     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)
+    return _write(classify(iio.imread(satellite), rows, cols), bounds, out, preview)
+
+
+# foliage_mask.png land classes (the scene pipeline's unet-efficientnet segmentation) -> palette id.
+# 0 = none (defer to the other masks); 1 tree / 2 grass / 4 shrub -> foliage; 3 rock; 5 bare -> soil.
+_FOLIAGE_CLASS = {1: FOLIAGE, 2: FOLIAGE, 4: FOLIAGE, 3: ROCK, 5: SOIL}
+
+
+def segment_from_masks(bundle: str, rows: int, cols: int):
+    """Build the material_id grid from a full bundle's AUTHORITATIVE land-cover masks — the scene
+    pipeline's OSM + ML (unet) segmentation, higher quality than the satellite color heuristic:
+    ``foliage_mask.png`` (per-pixel land class 0..5), ``roads.png`` / ``water.png`` /
+    ``open_fields.png`` (alpha masks). Returns None if the masks are absent (a lean bundle) so the
+    caller falls back to :func:`classify`. Priority, later wins: open_air default -> open_fields=soil
+    -> foliage classes -> water -> roads=open_air (you drive on roads, even through vegetation/fields).
+    Downsampled to the grid by MAJORITY class, then flipped to row 0 == world min_y."""
+    import os
+
+    import imageio.v3 as iio
+
+    fol_p = os.path.join(bundle, "foliage_mask.png")
+    if not os.path.exists(fol_p):
+        return None
+
+    def alpha_mask(name: str):
+        p = os.path.join(bundle, name)
+        if not os.path.exists(p):
+            return None
+        im = np.asarray(iio.imread(p))
+        return (im[..., 3] > 0) if (im.ndim == 3 and im.shape[2] == 4) else (im > 0)
+
+    fol = np.asarray(iio.imread(fol_p))
+    if fol.ndim == 3:
+        fol = fol[..., 0]
+    h, w = fol.shape
+    pix = np.full((h, w), OPEN_AIR_ID, dtype=np.int32)
+    of = alpha_mask("open_fields.png")
+    if of is not None:
+        pix[of] = SOIL
+    for v, mid_id in _FOLIAGE_CLASS.items():
+        pix[fol == v] = mid_id
+    wa = alpha_mask("water.png")
+    if wa is not None:
+        pix[wa] = WATER
+    rd = alpha_mask("roads.png")
+    if rd is not None:
+        pix[rd] = OPEN_AIR_ID
+
+    ph, pw = h // rows, w // cols
+    if ph < 1 or pw < 1:
+        raise ValueError(f"masks {w}x{h} smaller than the {cols}x{rows} grid")
+    pix = pix[: rows * ph, : cols * pw]
+    best = np.zeros((rows, cols))
+    mid = np.full((rows, cols), OPEN_AIR_ID, dtype=np.int32)
+    for m in (OPEN_AIR_ID, SOIL, FOLIAGE, WATER, ROCK):
+        frac = (pix == m).reshape(rows, ph, cols, pw).mean((1, 3))
+        take = frac > best
+        mid[take] = m
+        best[take] = frac[take]
+    return mid[::-1].copy()
 
 
 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)."""
+    """Read a cvc scene bundle's ``terrain.json`` (rows/cols/bounds), build the material raster from
+    its authoritative land-cover MASKS if present (:func:`segment_from_masks`) and otherwise by
+    classifying ``satellite.png`` (:func:`classify`), and write ``<bundle>/material.json`` (or
+    ``out``). Returns (out_path, distribution)."""
+    import imageio.v3 as iio
+
     terr = json.load(open(os.path.join(bundle, "terrain.json")))
+    rows, cols, bounds = int(terr["rows"]), int(terr["cols"]), terr["bounds"]
     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
+    material_id = segment_from_masks(bundle, rows, cols)
+    if material_id is None:
+        material_id = classify(iio.imread(os.path.join(bundle, "satellite.png")), rows, cols)
+    return out, _write(material_id, bounds, out, preview)

tests/test_material_raster.py

@@ -95,3 +95,58 @@ def test_from_bundle_reads_terrain_and_satellite(tmp_path):
     doc = json.loads((tmp_path / "material.json").read_text())
     assert doc["rows"] == 4 and doc["bounds"]["max_y"] == 8.0
     assert "water" in dist
+
+
+def _rgba_mask(rows, cols, cells):
+    import numpy as np
+
+    a = np.zeros((rows, cols, 4), dtype=np.uint8)
+    for r, c in cells:
+        a[r, c] = (10, 10, 10, 255)
+    return a
+
+
+def test_segment_from_masks_mapping_priority_and_flip(tmp_path):
+    import imageio.v3 as iio
+    import numpy as np
+
+    from grl_snam.tools import material_raster as mr
+
+    # 2x2 grid, 1px/cell. foliage_mask: [[tree, none],[bare, none]]; roads over the (1,1) cell;
+    # water over (0,1). Priority: roads/water win; foliage 1->foliage, 5->soil.
+    fol = np.array([[1, 0], [5, 0]], dtype=np.uint8)  # top row north (max_y), bottom south (min_y)
+    iio.imwrite(tmp_path / "foliage_mask.png", fol)
+    iio.imwrite(tmp_path / "water.png", _rgba_mask(2, 2, [(0, 1)]))
+    iio.imwrite(tmp_path / "roads.png", _rgba_mask(2, 2, [(1, 1)]))
+    mid = mr.segment_from_masks(str(tmp_path), 2, 2)
+    assert mid is not None
+    # flipped so row 0 == min_y == image bottom row: [bare->soil, roads->open_air]
+    assert mid[0, 0] == mr.SOIL
+    assert mid[0, 1] == mr.OPEN_AIR_ID  # roads win over nothing here
+    # row 1 == image top row: [tree->foliage, water->open? no: water over (0,1)]
+    assert mid[1, 0] == mr.FOLIAGE
+    assert mid[1, 1] == mr.WATER  # water mask over the else-empty cell
+
+
+def test_segment_from_masks_absent_returns_none(tmp_path):
+    from grl_snam.tools import material_raster as mr
+
+    assert mr.segment_from_masks(str(tmp_path), 4, 4) is None  # no foliage_mask.png -> fallback
+
+
+def test_from_bundle_falls_back_to_satellite_without_masks(tmp_path):
+    import json as _json
+
+    import imageio.v3 as iio
+    import numpy as np
+
+    from grl_snam.tools import material_raster as mr
+
+    (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}}
+        )
+    )
+    iio.imwrite(tmp_path / "satellite.png", np.full((4, 4, 3), (150, 150, 150), dtype=np.uint8))
+    out, dist = mr.from_bundle(str(tmp_path))  # no masks -> satellite classify
+    assert "open_air" in dist