WebGIS Raster Service API

ras2cng.webgis_service

Bounded statistics and styled-tile service for allowlisted numeric COGs.

RasterAsset dataclass

One local numeric COG approved for dynamic service access.

Source code in ras2cng/webgis_service.py

@dataclass(frozen=True)
class RasterAsset:
    """One local numeric COG approved for dynamic service access."""

    asset_id: str
    path: Path
    revision: str
    preset: str
    units: str = ""
    categorical: bool = False
    minimum: float | None = None
    maximum: float | None = None

RasterAssetFilesMissingError

Bases: FileNotFoundError

Raised when a catalog exists but one or more allowlisted files do not.

Source code in ras2cng/webgis_service.py

class RasterAssetFilesMissingError(FileNotFoundError):
    """Raised when a catalog exists but one or more allowlisted files do not."""

    def __init__(self, asset_ids: Iterable[str]):
        self.asset_ids = tuple(asset_ids)
        examples = ", ".join(self.asset_ids[:10])
        super().__init__(
            f"Raster catalog references {len(self.asset_ids)} missing asset file(s): "
            f"{examples}"
        )

ReleaseCatalogNotFoundError

Bases: FileNotFoundError

Raised when an immutable release has no raster catalog.

Source code in ras2cng/webgis_service.py

class ReleaseCatalogNotFoundError(FileNotFoundError):
    """Raised when an immutable release has no raster catalog."""

RasterServiceSettings dataclass

Resource and request limits for the isolated WebGIS service.

Source code in ras2cng/webgis_service.py

@dataclass(frozen=True)
class RasterServiceSettings:
    """Resource and request limits for the isolated WebGIS service."""

    route_prefix: str = "/ras-raster"
    max_view_pixels: int = 2_097_152
    max_view_dimension: int = 4096
    tile_size: int = 256
    cache_entries: int = 512
    max_cog_range_bytes: int = 67_108_864
    max_concurrent_operations: int = 8
    allowed_origins: tuple[str, ...] = ("https://rascommander.info",)

StylePreset dataclass

Allowlisted color ramp; clients cannot submit arbitrary styles.

Source code in ras2cng/webgis_service.py

@dataclass(frozen=True)
class StylePreset:
    """Allowlisted color ramp; clients cannot submit arbitrary styles."""

    preset_id: str
    colors: tuple[tuple[int, int, int, int], ...]
    categorical_values: tuple[int, ...] = ()

    @property
    def categorical(self) -> bool:
        return bool(self.categorical_values)

RasterAssetCatalog

Validated in-memory view of a data-root-relative asset catalog.

Source code in ras2cng/webgis_service.py

class RasterAssetCatalog:
    """Validated in-memory view of a data-root-relative asset catalog."""

    def __init__(
        self,
        data_root: Path,
        assets: Mapping[str, RasterAsset],
        *,
        catalog_path: Path,
        catalog_revision: str,
        release_id: str,
    ):
        self.data_root = Path(data_root).resolve()
        self.assets = dict(assets)
        self.catalog_path = Path(catalog_path).resolve()
        self.catalog_revision = catalog_revision
        self.release_id = release_id

    @classmethod
    def load(cls, catalog_path: Path, data_root: Path) -> "RasterAssetCatalog":
        catalog_path = Path(catalog_path)
        catalog_bytes = catalog_path.read_bytes()
        document = json.loads(catalog_bytes.decode("utf-8"))
        if document.get("schema") != RASTER_ASSET_SCHEMA:
            raise ValueError(
                f"Unsupported raster asset catalog schema: {document.get('schema')!r}"
            )
        root = Path(data_root).resolve()
        assets: dict[str, RasterAsset] = {}
        missing_assets: list[str] = []
        for asset_id, record in (document.get("assets") or {}).items():
            _validate_asset_id(asset_id)
            relative = Path(str(record.get("path") or ""))
            if relative.is_absolute() or not relative.parts:
                raise ValueError(f"Raster asset {asset_id!r} must use a relative path")
            path = (root / relative).resolve()
            if not path.is_relative_to(root):
                raise ValueError(
                    f"Raster asset {asset_id!r} escapes the configured data root"
                )
            if not path.is_file():
                missing_assets.append(asset_id)
                continue
            preset = str(record.get("preset") or "")
            if preset not in STYLE_PRESETS:
                raise ValueError(
                    f"Raster asset {asset_id!r} uses unsupported preset {preset!r}"
                )
            assets[asset_id] = RasterAsset(
                asset_id=asset_id,
                path=path,
                revision=str(record.get("revision") or _asset_revision(path)),
                preset=preset,
                units=str(record.get("units") or ""),
                categorical=bool(
                    record.get("categorical", STYLE_PRESETS[preset].categorical)
                ),
                minimum=_optional_float(record.get("minimum")),
                maximum=_optional_float(record.get("maximum")),
            )
        if missing_assets:
            raise RasterAssetFilesMissingError(missing_assets)
        catalog_revision = hashlib.sha256(catalog_bytes).hexdigest()[:24]
        release_id = str(document.get("releaseId") or catalog_revision)
        return cls(
            root,
            assets,
            catalog_path=catalog_path,
            catalog_revision=catalog_revision,
            release_id=release_id,
        )

    def get(self, asset_id: str) -> RasterAsset:
        _validate_asset_id(asset_id)
        try:
            return self.assets[asset_id]
        except KeyError as error:
            raise KeyError(f"Unknown raster asset: {asset_id}") from error

ReleaseRasterCatalogStore

Load and cache self-contained immutable release catalogs.

Source code in ras2cng/webgis_service.py

class ReleaseRasterCatalogStore:
    """Load and cache self-contained immutable release catalogs."""

    def __init__(self, data_root: Path, *, max_entries: int = 64):
        if max_entries < 1:
            raise ValueError("The release catalog cache limit must be positive")
        self.data_root = Path(data_root).resolve()
        self.releases_root = (self.data_root / "releases").resolve()
        if not self.releases_root.is_relative_to(self.data_root):
            raise ValueError("The releases directory escapes the configured data root")
        self.max_entries = max_entries
        self._catalogs: OrderedDict[str, RasterAssetCatalog] = OrderedDict()
        self._lock = Lock()

    def get(self, release_id: str) -> RasterAssetCatalog:
        release_id = str(release_id).strip()
        if not _RELEASE_ID.fullmatch(release_id):
            raise ValueError("Release ID contains unsupported characters")
        with self._lock:
            cached = self._catalogs.get(release_id)
            if cached is not None:
                self._catalogs.move_to_end(release_id)
                return cached

            release_root = (self.releases_root / release_id).resolve()
            if not release_root.is_relative_to(self.releases_root):
                raise ValueError("Release path escapes the configured releases root")
            catalog_path = release_root / "raster-assets.json"
            if not catalog_path.is_file():
                raise ReleaseCatalogNotFoundError(
                    f"Raster release {release_id!r} has no catalog"
                )
            catalog = RasterAssetCatalog.load(
                catalog_path,
                release_root,
            )
            if catalog.release_id != release_id:
                raise ValueError(
                    f"Raster catalog release ID {catalog.release_id!r} does not "
                    f"match {release_id!r}"
                )
            self._catalogs[release_id] = catalog
            self._catalogs.move_to_end(release_id)
            while len(self._catalogs) > self.max_entries:
                self._catalogs.popitem(last=False)
            return catalog

    def current_release_id(self) -> str:
        """Resolve the single atomic ``current`` pointer to an immutable release."""

        current = self.data_root / "current"
        if not current.is_symlink():
            raise ReleaseCatalogNotFoundError(
                "The current raster release pointer is unavailable"
            )
        try:
            release_root = current.resolve(strict=True)
        except FileNotFoundError as error:
            raise ReleaseCatalogNotFoundError(
                "The current raster release pointer is broken"
            ) from error
        if (
            release_root.parent != self.releases_root
            or not _RELEASE_ID.fullmatch(release_root.name)
        ):
            raise ValueError(
                "The current raster release pointer must select one direct child "
                "of the releases directory"
            )
        return release_root.name

current_release_id()

Resolve the single atomic current pointer to an immutable release.

Source code in ras2cng/webgis_service.py

def current_release_id(self) -> str:
    """Resolve the single atomic ``current`` pointer to an immutable release."""

    current = self.data_root / "current"
    if not current.is_symlink():
        raise ReleaseCatalogNotFoundError(
            "The current raster release pointer is unavailable"
        )
    try:
        release_root = current.resolve(strict=True)
    except FileNotFoundError as error:
        raise ReleaseCatalogNotFoundError(
            "The current raster release pointer is broken"
        ) from error
    if (
        release_root.parent != self.releases_root
        or not _RELEASE_ID.fullmatch(release_root.name)
    ):
        raise ValueError(
            "The current raster release pointer must select one direct child "
            "of the releases directory"
        )
    return release_root.name

RasterServiceBusyError

Bases: RuntimeError

Raised when accepting more work would exceed the service budget.

Source code in ras2cng/webgis_service.py

class RasterServiceBusyError(RuntimeError):
    """Raised when accepting more work would exceed the service budget."""

build_raster_asset_catalog(data_root, output_path, *, manifest_paths=None, service_base_url='/ras-raster', attach_manifests=False, public_url_prefix=None)

Build an allowlist from manifest v2 numeric resources under a data root.

Source code in ras2cng/webgis_service.py

def build_raster_asset_catalog(
    data_root: Path,
    output_path: Path,
    *,
    manifest_paths: Iterable[Path] | None = None,
    service_base_url: str = "/ras-raster",
    attach_manifests: bool = False,
    public_url_prefix: str | None = None,
) -> Path:
    """Build an allowlist from manifest v2 numeric resources under a data root."""

    root = Path(data_root).resolve()
    if not root.is_dir():
        raise NotADirectoryError(f"WebGIS data root does not exist: {root}")
    paths = (
        [Path(path).resolve() for path in manifest_paths]
        if manifest_paths
        else sorted(root.glob("**/viewer/manifest.json"))
    )
    assets: dict[str, dict[str, Any]] = {}
    for manifest_path in paths:
        manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
        if manifest.get("schema") != "rascommander.maplibre/v2":
            raise ValueError(
                f"Raster service catalog requires manifest v2: {manifest_path}"
            )
        project_key = _project_key(root, manifest_path, manifest)
        modified = False
        for layer_id, layer in (manifest.get("layers") or {}).items():
            numeric_id = (layer.get("query") or {}).get("numericResource")
            resource = (
                (manifest.get("resources") or {}).get(numeric_id)
                if numeric_id
                else None
            )
            if (
                not resource
                or resource.get("type") != "cog"
                or not resource.get("href")
            ):
                continue
            path = _resolve_numeric_href(
                root,
                manifest_path,
                str(resource["href"]),
                public_url_prefix=public_url_prefix,
            )
            if not path.is_file():
                raise FileNotFoundError(
                    f"Numeric COG for {layer_id!r} does not exist: {path}"
                )
            legend_id = (layer.get("style") or {}).get("legendRef")
            legend = (manifest.get("legends") or {}).get(legend_id, {})
            preset = str(legend.get("preset") or _default_preset(layer))
            if preset not in STYLE_PRESETS:
                raise ValueError(
                    f"Layer {layer_id!r} uses unsupported service preset {preset!r}"
                )
            asset_id = f"{project_key}/{_slug(layer_id)}"
            _validate_asset_id(asset_id)
            revision = _asset_revision(path)
            domain = legend.get("domain") or {}
            assets[asset_id] = {
                "path": path.relative_to(root).as_posix(),
                "revision": revision,
                "preset": preset,
                "units": str(
                    legend.get("units")
                    or (layer.get("raster") or {}).get("units")
                    or ""
                ),
                "categorical": legend.get("type") == "categorical"
                or STYLE_PRESETS[preset].categorical,
                "minimum": _optional_float(domain.get("minimum")),
                "maximum": _optional_float(domain.get("maximum")),
            }
            if attach_manifests:
                resource["serviceAsset"] = asset_id
                resource["serviceRevision"] = revision
                for tileset in manifest.get("tilesets", []):
                    if (
                        tileset.get("id") == layer_id
                        and tileset.get("type") == "raster"
                    ):
                        tileset["serviceAsset"] = asset_id
                        tileset["serviceRevision"] = revision
                        break
                modified = True
        if attach_manifests and modified:
            manifest.setdefault("services", {})["numericRaster"] = {
                "baseUrl": service_base_url.rstrip("/"),
                "statisticsPath": "/stats",
                "samplePath": "/sample",
                "cogPath": "/cog",
                "cogRangeTransport": "header-or-query",
                "cogRangeParameter": "range",
                "tilePath": "/tiles/{z}/{x}/{y}.png",
                "maxViewPixels": 2_097_152,
            }
            _atomic_json_write(manifest_path, manifest)

    document = {
        "schema": RASTER_ASSET_SCHEMA,
        "generatedAt": datetime.now(timezone.utc).isoformat(),
        "dataRoot": ".",
        "assets": assets,
    }
    _atomic_json_write(Path(output_path), document)
    return Path(output_path)

compute_view_statistics(asset, bbox, width, height, *, exact=False, max_pixels=2097152, max_dimension=4096)

Read one overview-bounded viewport and return robust or exact statistics.

Source code in ras2cng/webgis_service.py

def compute_view_statistics(
    asset: RasterAsset,
    bbox: tuple[float, float, float, float],
    width: int,
    height: int,
    *,
    exact: bool = False,
    max_pixels: int = 2_097_152,
    max_dimension: int = 4096,
) -> dict[str, Any]:
    """Read one overview-bounded viewport and return robust or exact statistics."""

    from rasterio.crs import CRS
    from rio_tiler.io import Reader

    normalized_bbox = normalize_view_bbox(bbox, width, height)
    read_width, read_height = bounded_view_dimensions(
        width,
        height,
        max_pixels=max_pixels,
        max_dimension=max_dimension,
    )
    with Reader(str(asset.path)) as reader:
        image = reader.part(
            normalized_bbox,
            bounds_crs=CRS.from_epsg(4326),
            width=read_width,
            height=read_height,
            resampling_method="nearest",
        )
    statistics = image.statistics(percentiles=[2, 98])
    if not statistics:
        raise ValueError("Viewport contains no raster bands")
    band = next(iter(statistics.values()))
    values = band.model_dump()
    minimum = float(values["min"])
    maximum = float(values["max"])
    robust_minimum = float(values.get("percentile_2", minimum))
    robust_maximum = float(values.get("percentile_98", maximum))
    domain_minimum = minimum if exact else robust_minimum
    domain_maximum = maximum if exact else robust_maximum
    if not all(
        math.isfinite(value)
        for value in (minimum, maximum, domain_minimum, domain_maximum)
    ):
        raise ValueError("Viewport statistics are not finite")
    return {
        "asset": asset.asset_id,
        "revision": asset.revision,
        "bbox": list(normalized_bbox),
        "sampleWidth": read_width,
        "sampleHeight": read_height,
        "exact": exact,
        "units": asset.units,
        "statistics": {
            "minimum": minimum,
            "maximum": maximum,
            "mean": float(values["mean"]),
            "stddev": float(values["std"]),
            "validPixels": int(values["valid_pixels"]),
            "maskedPixels": int(values["masked_pixels"]),
            "percentile2": robust_minimum,
            "percentile98": robust_maximum,
        },
        "domain": {"minimum": domain_minimum, "maximum": domain_maximum},
    }

sample_raster_at_point(asset, longitude, latitude)

Read one allowlisted raster cell at a WGS84 point.

Source code in ras2cng/webgis_service.py

def sample_raster_at_point(
    asset: RasterAsset,
    longitude: float,
    latitude: float,
) -> dict[str, Any]:
    """Read one allowlisted raster cell at a WGS84 point."""

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

    longitude = float(longitude)
    latitude = float(latitude)
    if not math.isfinite(longitude) or not math.isfinite(latitude):
        raise ValueError("Sample coordinates must be finite")
    if longitude < -180 or longitude > 180 or latitude < -90 or latitude > 90:
        raise ValueError(
            "Sample coordinates must be valid WGS84 longitude and latitude"
        )

    with rasterio.open(asset.path) as source:
        if source.crs is None:
            raise ValueError("Raster has no coordinate reference system")
        xs, ys = transform("EPSG:4326", source.crs, [longitude], [latitude])
        source_x, source_y = float(xs[0]), float(ys[0])
        bounds = source.bounds
        if (
            source_x < bounds.left
            or source_x >= bounds.right
            or source_y <= bounds.bottom
            or source_y > bounds.top
        ):
            return {
                "asset": asset.asset_id,
                "revision": asset.revision,
                "longitude": longitude,
                "latitude": latitude,
                "state": "outside",
                "units": asset.units,
            }
        row, column = source.index(source_x, source_y)
        sample = next(source.sample([(source_x, source_y)], indexes=1, masked=True))[0]

    base = {
        "asset": asset.asset_id,
        "revision": asset.revision,
        "longitude": longitude,
        "latitude": latitude,
        "sourceX": source_x,
        "sourceY": source_y,
        "row": int(row),
        "column": int(column),
        "units": asset.units,
    }
    if np.ma.is_masked(sample) or not math.isfinite(float(sample)):
        return {**base, "state": "nodata"}
    return {**base, "state": "value", "value": float(sample)}

render_styled_tile(asset, x, y, z, *, preset_id, minimum=None, maximum=None, tile_size=256)

Render one PNG using an approved ramp and a bounded 256-pixel COG read.

Source code in ras2cng/webgis_service.py

def render_styled_tile(
    asset: RasterAsset,
    x: int,
    y: int,
    z: int,
    *,
    preset_id: str,
    minimum: float | None = None,
    maximum: float | None = None,
    tile_size: int = 256,
) -> bytes:
    """Render one PNG using an approved ramp and a bounded 256-pixel COG read."""

    from rio_tiler.io import Reader

    preset = get_style_preset(preset_id)
    if tile_size != 256:
        raise ValueError("Only 256-pixel tiles are supported")
    with Reader(str(asset.path)) as reader:
        image = reader.tile(
            x,
            y,
            z,
            tilesize=tile_size,
            resampling_method="nearest" if preset.categorical else "bilinear",
        )
    if preset.categorical:
        colormap = {
            value: color
            for value, color in zip(preset.categorical_values, preset.colors)
        }
    else:
        if minimum is None or maximum is None:
            raise ValueError("Continuous styled tiles require minimum and maximum")
        minimum = float(minimum)
        maximum = float(maximum)
        if (
            not math.isfinite(minimum)
            or not math.isfinite(maximum)
            or maximum < minimum
        ):
            raise ValueError("Styled tile range must be finite with maximum >= minimum")
        if maximum == minimum:
            epsilon = max(abs(minimum) * 1e-9, 1e-9)
            minimum -= epsilon
            maximum += epsilon
        image.rescale(in_range=((minimum, maximum),), out_range=((0, 255),))
        colormap = _linear_colormap(preset.colors)
    return image.render(img_format="PNG", colormap=colormap)

create_raster_app(catalog_path=None, data_root=None, *, settings=None, release_catalogs=False)

Create the isolated FastAPI application used by CLB-WebGIS.

Source code in ras2cng/webgis_service.py

def create_raster_app(
    catalog_path: Path | None = None,
    data_root: Path | None = None,
    *,
    settings: RasterServiceSettings | None = None,
    release_catalogs: bool = False,
):
    """Create the isolated FastAPI application used by CLB-WebGIS."""

    from fastapi import (
        FastAPI,
        Header,
        HTTPException,
        Path as ApiPath,
        Query,
        Request,
        Response,
    )
    from fastapi.exception_handlers import (
        http_exception_handler,
        request_validation_exception_handler,
    )
    from fastapi.exceptions import RequestValidationError
    from fastapi.middleware.cors import CORSMiddleware
    from fastapi.responses import JSONResponse, StreamingResponse
    from starlette.exceptions import HTTPException as StarletteHTTPException

    settings = settings or _settings_from_environment()
    data_root = Path(data_root or os.environ.get("RAS2CNG_RASTER_DATA_ROOT", "."))
    if release_catalogs:
        catalog = None
        catalog_store = ReleaseRasterCatalogStore(data_root)
    else:
        catalog_path = Path(
            catalog_path
            or os.environ.get("RAS2CNG_RASTER_CATALOG", "raster-assets.json")
        )
        catalog = RasterAssetCatalog.load(catalog_path, data_root)
        catalog_store = None
    cache = _LruCache(settings.cache_entries)
    operation_limiter = _OperationLimiter(settings.max_concurrent_operations)
    prefix = "/" + settings.route_prefix.strip("/")
    release_prefix = (
        f"{prefix}/releases/{{release_id}}" if release_catalogs else prefix
    )
    release_parameter = (
        ApiPath(..., min_length=1, max_length=80)
        if release_catalogs
        else Query(None, include_in_schema=False)
    )
    app = FastAPI(title="RAS Commander Numeric Raster Service", version="1")
    app.state.catalog = catalog
    app.state.catalog_store = catalog_store
    app.state.operation_limiter = operation_limiter

    def request_catalog(release_id: str | None) -> RasterAssetCatalog:
        if catalog_store is None:
            if catalog is None:
                raise HTTPException(status_code=503, detail="Raster catalog unavailable")
            return catalog
        try:
            return catalog_store.get(str(release_id or ""))
        except RasterAssetFilesMissingError as error:
            raise HTTPException(
                status_code=503,
                detail={
                    "status": "not-ready",
                    "releaseId": str(release_id or ""),
                    "missingAssets": len(error.asset_ids),
                    "missingAssetExamples": list(error.asset_ids[:10]),
                },
            ) from error
        except ReleaseCatalogNotFoundError as error:
            raise HTTPException(
                status_code=404,
                detail="Unknown raster release",
            ) from error
        except ValueError as error:
            raise HTTPException(status_code=422, detail=str(error)) from error

    @app.exception_handler(StarletteHTTPException)
    async def no_store_http_error(request: Request, error: StarletteHTTPException):
        response = await http_exception_handler(request, error)
        response.headers["Cache-Control"] = "no-store"
        return response

    @app.exception_handler(RequestValidationError)
    async def no_store_validation_error(
        request: Request, error: RequestValidationError
    ):
        response = await request_validation_exception_handler(request, error)
        response.headers["Cache-Control"] = "no-store"
        return response

    @app.exception_handler(RasterServiceBusyError)
    async def no_store_busy_error(request: Request, error: RasterServiceBusyError):
        return JSONResponse(
            status_code=503,
            content={"detail": str(error)},
            headers={"Cache-Control": "no-store", "Retry-After": "1"},
        )

    @app.exception_handler(Exception)
    async def no_store_internal_error(request: Request, error: Exception):
        logger.exception(
            "Unhandled raster service error for %s",
            request.url.path,
            exc_info=(type(error), error, error.__traceback__),
        )
        return JSONResponse(
            status_code=500,
            content={"detail": "The raster service could not complete the request"},
            headers={"Cache-Control": "no-store"},
        )

    if catalog_store is not None:

        @app.middleware("http")
        async def route_legacy_requests_to_current(request: Request, call_next):
            path = request.scope.get("path", "")
            suffix = path[len(prefix) :] if path.startswith(prefix) else ""
            if re.match(
                r"^/(?:ready|stats|sample|cog)(?:$|/)|^/tiles(?:$|/)",
                suffix,
            ):
                try:
                    current_release_id = catalog_store.current_release_id()
                except (ReleaseCatalogNotFoundError, ValueError) as error:
                    return JSONResponse(
                        status_code=503,
                        content={
                            "detail": "Current raster release unavailable",
                            "reason": str(error),
                        },
                        headers={"Cache-Control": "no-store"},
                    )
                routed_path = (
                    f"{prefix}/releases/{current_release_id}{suffix}"
                )
                request.scope["path"] = routed_path
                request.scope["raw_path"] = routed_path.encode("ascii")
            return await call_next(request)

    # Add CORS after the legacy rewrite so it remains the outermost middleware
    # and decorates fail-closed responses returned before route dispatch.
    app.add_middleware(
        CORSMiddleware,
        allow_origins=list(settings.allowed_origins),
        allow_methods=["GET"],
        allow_headers=["*"],
        expose_headers=[
            "Accept-Ranges",
            "Cache-Control",
            "Content-Length",
            "Content-Range",
            "ETag",
            "Retry-After",
            "X-Raster-Revision",
        ],
    )

    @app.get(f"{prefix}/health")
    def health():
        if catalog is None:
            content = {
                "status": "ok",
                "schema": RASTER_ASSET_SCHEMA,
                "catalogMode": "immutable-releases",
            }
        else:
            content = _service_status(catalog, status="ok")
        return JSONResponse(
            content=content,
            headers={"Cache-Control": "no-store"},
        )

    @app.get(f"{release_prefix}/ready")
    def ready(release_id: str | None = release_parameter):
        selected_catalog = request_catalog(release_id)
        missing = [
            asset.asset_id
            for asset in selected_catalog.assets.values()
            if not asset.path.is_file()
        ]
        status = "ready" if not missing else "not-ready"
        return JSONResponse(
            status_code=200 if not missing else 503,
            content={
                **_service_status(selected_catalog, status=status),
                "missingAssets": len(missing),
                "missingAssetExamples": missing[:10],
            },
            headers={"Cache-Control": "no-store"},
        )

    @app.get(f"{release_prefix}/stats")
    def statistics(
        asset: str = Query(..., min_length=1, max_length=300),
        bbox: str = Query(..., min_length=7, max_length=160),
        width: int = Query(1024, ge=1, le=settings.max_view_dimension),
        height: int = Query(768, ge=1, le=settings.max_view_dimension),
        exact: bool = Query(False),
        revision: str | None = Query(None, max_length=80),
        release_id: str | None = release_parameter,
    ):
        selected_catalog = request_catalog(release_id)
        try:
            record = selected_catalog.get(asset)
            _require_revision(record, revision)
            parsed_bbox = parse_bbox(bbox)
            normalized_bbox = normalize_view_bbox(parsed_bbox, width, height)
            read_width, read_height = bounded_view_dimensions(
                width,
                height,
                max_pixels=settings.max_view_pixels,
                max_dimension=settings.max_view_dimension,
            )
            key = (
                selected_catalog.release_id,
                record.asset_id,
                record.revision,
                normalized_bbox,
                read_width,
                read_height,
                exact,
            )
            result = cache.get(key)
            if result is None:
                with operation_limiter.slot():
                    result = compute_view_statistics(
                        record,
                        normalized_bbox,
                        read_width,
                        read_height,
                        exact=exact,
                        max_pixels=settings.max_view_pixels,
                        max_dimension=settings.max_view_dimension,
                    )
                cache.put(key, result)
        except KeyError as error:
            raise HTTPException(status_code=404, detail=str(error)) from error
        except ValueError as error:
            raise HTTPException(status_code=422, detail=str(error)) from error
        return JSONResponse(
            content=result,
            headers=_cache_headers(record, revision, result),
        )

    @app.get(f"{release_prefix}/sample")
    def sample(
        asset: str = Query(..., min_length=1, max_length=300),
        lng: float = Query(..., ge=-180, le=180),
        lat: float = Query(..., ge=-90, le=90),
        revision: str | None = Query(None, max_length=80),
        release_id: str | None = release_parameter,
    ):
        selected_catalog = request_catalog(release_id)
        try:
            record = selected_catalog.get(asset)
            _require_revision(record, revision)
            key = (
                "sample",
                selected_catalog.release_id,
                record.asset_id,
                record.revision,
                round(lng, 10),
                round(lat, 10),
            )
            result = cache.get(key)
            if result is None:
                with operation_limiter.slot():
                    result = sample_raster_at_point(record, lng, lat)
                cache.put(key, result)
        except KeyError as error:
            raise HTTPException(status_code=404, detail=str(error)) from error
        except ValueError as error:
            raise HTTPException(status_code=422, detail=str(error)) from error
        return JSONResponse(
            content=result,
            headers=_cache_headers(record, revision, result),
        )

    @app.head(f"{release_prefix}/cog")
    def cog_head(
        asset: str = Query(..., min_length=1, max_length=300),
        revision: str | None = Query(None, max_length=80),
        release_id: str | None = release_parameter,
    ):
        selected_catalog = request_catalog(release_id)
        try:
            record = selected_catalog.get(asset)
            _require_revision(record, revision)
        except KeyError as error:
            raise HTTPException(status_code=404, detail=str(error)) from error
        except ValueError as error:
            raise HTTPException(status_code=422, detail=str(error)) from error
        size = record.path.stat().st_size
        return Response(
            status_code=200,
            media_type="image/tiff",
            headers=_cog_headers(
                record,
                revision,
                content_length=size,
            ),
        )

    @app.get(f"{release_prefix}/cog")
    def cog_range(
        asset: str = Query(..., min_length=1, max_length=300),
        revision: str | None = Query(None, max_length=80),
        range_header: str | None = Header(None, alias="Range"),
        range_query: str | None = Query(
            None,
            alias="range",
            min_length=8,
            max_length=80,
        ),
        release_id: str | None = release_parameter,
    ):
        selected_catalog = request_catalog(release_id)
        try:
            record = selected_catalog.get(asset)
            _require_revision(record, revision)
        except KeyError as error:
            raise HTTPException(status_code=404, detail=str(error)) from error
        except ValueError as error:
            raise HTTPException(status_code=422, detail=str(error)) from error

        size = record.path.stat().st_size
        if range_header and range_query and range_header != range_query:
            return JSONResponse(
                status_code=416,
                content={"detail": "Header and query byte ranges do not match"},
                headers={
                    "Accept-Ranges": "bytes",
                    "Content-Range": f"bytes */{size}",
                    "Cache-Control": "no-store",
                    "X-Raster-Revision": record.revision,
                },
            )
        try:
            start, end = parse_byte_range(
                range_header or range_query,
                size,
                max_bytes=settings.max_cog_range_bytes,
            )
        except ValueError as error:
            return JSONResponse(
                status_code=416,
                content={"detail": str(error)},
                headers={
                    "Accept-Ranges": "bytes",
                    "Content-Range": f"bytes */{size}",
                    "Cache-Control": "no-store",
                    "X-Raster-Revision": record.revision,
                },
            )
        content_length = end - start + 1
        return StreamingResponse(
            iter_file_range(record.path, start, content_length),
            status_code=206,
            media_type="image/tiff",
            headers=_cog_headers(
                record,
                revision,
                content_length=content_length,
                content_range=f"bytes {start}-{end}/{size}",
            ),
        )

    @app.get(
        f"{release_prefix}/tiles/{{z}}/{{x}}/{{y}}.png",
        responses={
            200: {
                "content": {"image/png": {}},
                "description": "Styled numeric raster tile",
            }
        },
    )
    def tile(
        z: int,
        x: int,
        y: int,
        asset: str = Query(..., min_length=1, max_length=300),
        preset: str | None = Query(None, max_length=100),
        minimum: float | None = Query(None),
        maximum: float | None = Query(None),
        revision: str | None = Query(None, max_length=80),
        release_id: str | None = release_parameter,
    ):
        selected_catalog = request_catalog(release_id)
        try:
            if z < 0 or z > 24 or x < 0 or y < 0 or x >= 2**z or y >= 2**z:
                raise ValueError(
                    "Tile coordinates are outside the supported Web Mercator pyramid"
                )
            record = selected_catalog.get(asset)
            _require_revision(record, revision)
            selected_preset = preset or record.preset
            if selected_preset != record.preset:
                raise ValueError("Requested preset does not match the asset allowlist")
            key = (
                selected_catalog.release_id,
                record.asset_id,
                record.revision,
                z,
                x,
                y,
                selected_preset,
                minimum,
                maximum,
            )
            content = cache.get(key)
            if content is None:
                with operation_limiter.slot():
                    content = render_styled_tile(
                        record,
                        x,
                        y,
                        z,
                        preset_id=selected_preset,
                        minimum=minimum,
                        maximum=maximum,
                        tile_size=settings.tile_size,
                    )
                cache.put(key, content)
        except KeyError as error:
            raise HTTPException(status_code=404, detail=str(error)) from error
        except ValueError as error:
            raise HTTPException(status_code=422, detail=str(error)) from error
        return Response(
            content,
            media_type="image/png",
            headers=_cache_headers(record, revision, content),
        )

    return app

create_release_raster_app(data_root=None, *, settings=None)

Create a service that discovers one catalog per immutable release.

Source code in ras2cng/webgis_service.py

def create_release_raster_app(
    data_root: Path | None = None,
    *,
    settings: RasterServiceSettings | None = None,
):
    """Create a service that discovers one catalog per immutable release."""

    return create_raster_app(
        data_root=data_root,
        settings=settings,
        release_catalogs=True,
    )

parse_byte_range(value, size, *, max_bytes)

Parse one RFC 9110 byte range with an explicit response-size ceiling.

Source code in ras2cng/webgis_service.py

def parse_byte_range(
    value: str | None,
    size: int,
    *,
    max_bytes: int,
) -> tuple[int, int]:
    """Parse one RFC 9110 byte range with an explicit response-size ceiling."""

    if size <= 0:
        raise ValueError("The requested COG is empty")
    if max_bytes <= 0:
        raise ValueError("The COG range limit must be positive")
    match = _BYTE_RANGE.fullmatch((value or "").strip())
    if not match or (not match.group(1) and not match.group(2)):
        raise ValueError("A single byte Range header is required")
    start_text, end_text = match.groups()
    if not start_text:
        suffix_length = int(end_text)
        if suffix_length <= 0:
            raise ValueError("Suffix byte ranges must be positive")
        start = max(0, size - suffix_length)
        end = size - 1
    else:
        start = int(start_text)
        end = int(end_text) if end_text else size - 1
        if start >= size or end < start:
            raise ValueError("Requested byte range is outside the COG")
        end = min(end, size - 1)
    if end - start + 1 > max_bytes:
        raise ValueError(
            f"Requested byte range exceeds the {max_bytes}-byte service limit"
        )
    return start, end

iter_file_range(path, start, length, *, chunk_size=1048576)

Stream one bounded file range without loading it fully into memory.

Source code in ras2cng/webgis_service.py

def iter_file_range(
    path: Path,
    start: int,
    length: int,
    *,
    chunk_size: int = 1_048_576,
) -> Iterator[bytes]:
    """Stream one bounded file range without loading it fully into memory."""

    remaining = length
    with Path(path).open("rb") as file_handle:
        file_handle.seek(start)
        while remaining:
            block = file_handle.read(min(chunk_size, remaining))
            if not block:
                break
            remaining -= len(block)
            yield block
    if remaining:
        raise OSError(f"COG ended {remaining} bytes before the requested range")