Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 33 additions & 5 deletions server/mapcontrol_server/services/geotiff_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
import httpx
import numpy as np
import rasterio
from rasterio.enums import MaskFlags
from rasterio.warp import transform_bounds
from matplotlib import colormaps
from PIL import Image
Expand Down Expand Up @@ -206,6 +207,23 @@ def _bounds_to_geojson_polygon(bounds: list[float]) -> dict:
}


# ─── Masking ────────────────────────────────────────────────────────────────


def _file_mask_invalid(ds: Any, bands: list[int]) -> np.ndarray | None:
"""Pixels the file's own mask band or alpha band marks as empty.

Nodata tags are handled by the callers (so a nodata override still
replaces the tag); this only covers per-dataset masks and alpha bands,
which GIS tools and eo-gpt's display writers use. Returns None when the
selected bands carry neither.
"""
flags = [ds.mask_flag_enums[b - 1] for b in bands]
if not any(MaskFlags.per_dataset in f or MaskFlags.alpha in f for f in flags):
return None
return (ds.read_masks(bands) == 0).all(axis=0)


# ─── RGB Processing ─────────────────────────────────────────────────────────


Expand Down Expand Up @@ -235,17 +253,21 @@ def _process_rgb(
g = ds.read(bands[1]).astype(np.float64)
b_arr = ds.read(bands[2]).astype(np.float64)

# Determine nodata mask
# Determine nodata mask. A fill value fills every band, so a pixel is
# nodata only when ALL three bands equal it; one band that happens to
# equal it (pure red, deep shadow) is data.
nodata = nodata_override if nodata_override is not None else ds.nodata
mask = np.zeros(r.shape, dtype=bool)
if nodata is not None:
mask |= r == nodata
mask |= g == nodata
mask |= b_arr == nodata
# Also mask NaN
mask |= (r == nodata) & (g == nodata) & (b_arr == nodata)
# A pixel with NaN in any band has no color to draw
mask |= np.isnan(r)
mask |= np.isnan(g)
mask |= np.isnan(b_arr)
# Honor the file's own mask / alpha band
file_invalid = _file_mask_invalid(ds, bands)
if file_invalid is not None:
mask |= file_invalid

# Normalize each band to 0-255 using percentile-based histogram stretch
# This prevents bright outliers (clouds, sun glint) from crushing the rest to black
Expand All @@ -259,6 +281,9 @@ def normalize_band(arr, m):
if p_high == p_low:
p_high = p_low + 1.0
normalized = (arr - p_low) / (p_high - p_low) * 255.0
# Masked pixels (possibly NaN) are hidden by alpha; zero them so
# NaN never reaches the integer cast.
normalized[m] = 0.0
return np.clip(normalized, 0, 255).astype(np.uint8)

r_u8 = normalize_band(r, mask)
Expand Down Expand Up @@ -303,6 +328,9 @@ def _process_singleband(
mask = np.isnan(data)
if nodata is not None:
mask |= data == nodata
file_invalid = _file_mask_invalid(ds, [band])
if file_invalid is not None:
mask |= file_invalid

# Get valid data for statistics
valid = data[~mask]
Expand Down
95 changes: 95 additions & 0 deletions server/tests/test_geotiff_masking.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,95 @@
"""Which GeoTIFF pixels render transparent: nodata tags, mask bands, alpha bands."""

from __future__ import annotations

import numpy as np
import rasterio
from rasterio.transform import from_origin

from mapcontrol_server.services.geotiff_service import _process_rgb, _process_singleband


def _write(path, data, *, nodata=None, mask=None, alpha=False):
data = np.asarray(data)
profile = dict(
driver="GTiff", count=data.shape[0], height=data.shape[1], width=data.shape[2],
dtype=data.dtype, crs="EPSG:4326", transform=from_origin(0, 10, 1, 1),
nodata=nodata,
)
if alpha:
profile["photometric"] = "RGB"
profile["alpha"] = "YES"
with rasterio.Env(GDAL_TIFF_INTERNAL_MASK=True):
with rasterio.open(path, "w", **profile) as dst:
dst.write(data)
if mask is not None:
dst.write_mask(mask)
return str(path)


def _rgb_alpha(path, **kwargs):
rgba, ds = _process_rgb(path, kwargs.pop("bands", [1, 2, 3]), 1.0, kwargs.pop("nodata", None))
ds.close()
return rgba[:, :, 3]


def _scene():
data = np.full((3, 4, 4), 120, dtype=np.uint8)
data[:, 0, 0] = 0 # fill: zero in every band
data[:, 1, 1] = (255, 0, 0) # pure red: one-band zeros are data
data[:, 2, 2] = (0, 0, 40) # deep shadow
return data


def test_rgb_nodata_requires_all_bands(tmp_path):
alpha = _rgb_alpha(_write(tmp_path / "a.tif", _scene(), nodata=0))
assert alpha[0, 0] == 0
assert alpha[1, 1] == 255 and alpha[2, 2] == 255
assert (alpha[3, :] == 255).all()


def test_rgb_nodata_override_uses_the_same_rule(tmp_path):
alpha = _rgb_alpha(_write(tmp_path / "a.tif", _scene()), nodata=0)
assert alpha[0, 0] == 0 and alpha[1, 1] == 255 and alpha[2, 2] == 255


def test_rgb_nan_in_any_band_is_transparent(tmp_path):
data = np.full((3, 4, 4), 0.2, dtype=np.float32)
data[1, 3, 3] = np.nan
alpha = _rgb_alpha(_write(tmp_path / "a.tif", data))
assert alpha[3, 3] == 0 and alpha[0, 0] == 255


def test_rgb_honors_internal_mask_band(tmp_path):
mask = np.full((4, 4), 255, dtype=np.uint8)
mask[:2, :] = 0
data = np.full((3, 4, 4), 90, dtype=np.uint8) # no tag, no zeros
alpha = _rgb_alpha(_write(tmp_path / "a.tif", data, mask=mask))
assert (alpha[:2, :] == 0).all() and (alpha[2:, :] == 255).all()
assert not list(tmp_path.glob("*.msk"))


def test_rgb_honors_alpha_band(tmp_path):
data = np.full((4, 4, 4), 90, dtype=np.uint8)
data[3] = 255
data[3, :, :2] = 0
alpha = _rgb_alpha(_write(tmp_path / "a.tif", data, alpha=True))
assert (alpha[:, :2] == 0).all() and (alpha[:, 2:] == 255).all()


def test_rgb_without_nodata_or_mask_is_fully_opaque(tmp_path):
alpha = _rgb_alpha(_write(tmp_path / "a.tif", _scene()))
assert (alpha == 255).all()


def test_singleband_honors_mask_and_nodata(tmp_path):
data = np.arange(16, dtype=np.float32).reshape(1, 4, 4)
mask = np.full((4, 4), 255, dtype=np.uint8)
mask[3, :] = 0
path = _write(tmp_path / "s.tif", data, nodata=5.0, mask=mask)
rgba, ds = _process_singleband(path, 1, "viridis", 1.0, None, None, None, None, None)
ds.close()
alpha = rgba[:, :, 3]
assert (alpha[3, :] == 0).all()
assert alpha[1, 1] == 0 # value 5 = nodata tag
assert alpha[0, 0] == 255 and alpha[2, 3] == 255
Loading