Core Sources

ArraySource

Bases: DataSource

Wrap a numpy array that is already in memory.

This is the primary integration point for QGIS and other environments that manage rasters internally and cannot rely on file I/O.

If crs and transform are omitted, the array is treated as pre-aligned to the BN's grid — align_to_grid will return it as-is when its shape matches the target grid, just like ConstantSource broadcasts when it has no spatial metadata.

Example — with CRS metadata (full georeferencing):

import numpy as np
from affine import Affine

slope = np.random.rand(100, 200).astype("float32") * 45.0
transform = Affine(0.01, 0, 10.0, 0, -0.01, 70.0)
source = geobn.ArraySource(slope, crs="EPSG:4326", transform=transform)

Example — pre-aligned array (no CRS needed):

# After fetching a DEM and computing slope analytically, the result
# is already on the BN grid — pass it directly without CRS metadata.
dem = bn.fetch_raw(geobn.WCSSource(...))
slope_deg = compute_slope(dem)   # same shape as the BN grid
bn.set_input("slope_angle", geobn.ArraySource(slope_deg))

ConstantSource

Bases: DataSource

Broadcast a single scalar value across the entire domain.

The 1×1 sentinel array is recognised by align_to_grid and expanded to the reference grid shape, so no explicit spatial metadata is needed.

Example:

# Apply a uniform 30 cm recent snowfall across the entire domain
source = geobn.ConstantSource(30.0)
bn.set_input("recent_snow", source)

RasterSource

Bases: DataSource

Read a local GeoTIFF file.

rasterio is used only to open the file and is discarded immediately; the returned RasterData contains only plain numpy/affine objects.

Example:

source = geobn.RasterSource("dem_10m.tif")
bn.set_input("elevation", source)

URLSource

Bases: DataSource

Fetch a GeoTIFF from an HTTP/HTTPS URL and return it as plain data.

The file is streamed into an in-memory buffer so nothing is written to disk. rasterio is the only component that sees the raw bytes; the returned RasterData contains only numpy / affine objects.

Parameters

url: HTTP/HTTPS URL pointing to a GeoTIFF file. timeout: HTTP request timeout in seconds. cache_dir: Optional path to a directory for caching the fetched raster on disk. On a cache hit the HTTP request is skipped entirely.

Supports optional disk caching.

Example:

source = geobn.URLSource(
    "https://example.com/data/slope.tif",
    cache_dir="cache/",
)
bn.set_input("slope_angle", source)

PointGridSource

Bases: DataSource

Sample a user-supplied callable over an N×N lat/lon grid.

Builds a regular grid of sample_points × sample_points WGS84 points covering the inference bounding box, calls fn(lat, lon) at each point, and assembles the results into a coarse EPSG:4326 raster. align_to_grid() then bilinearly resamples this raster to the reference grid resolution.

This is the generic primitive for any point-queryable data source (weather APIs, elevation services, custom models). Pass a lambda or a regular function as fn; return float("nan") or None for missing values.

Parameters

fn: Callable with signature (lat: float, lon: float) -> float | None. Return float("nan") or None for positions with no data. sample_points: Number of sample points per axis. Total API calls = sample_points². Default is 5 (25 calls). Use 1 for a single-point broadcast (equivalent to ConstantSource but fetched dynamically). delay: Seconds to sleep between successive calls. Default 0.05 s — enough to be polite to most free REST APIs without slowing batch runs noticeably.

PointGridSource is the generic primitive for any point-queryable data source. Pass any callable that accepts (lat, lon) and returns a float.

Example — Open-Meteo precipitation:

import requests
import geobn

def fetch_precipitation(lat: float, lon: float) -> float:
    """Query Open-Meteo historical archive for daily precipitation."""
    resp = requests.get(
        "https://archive-api.open-meteo.com/v1/archive",
        params={
            "latitude": lat,
            "longitude": lon,
            "start_date": "2024-01-01",
            "end_date": "2024-01-01",
            "daily": "precipitation_sum",
            "timezone": "UTC",
        },
        timeout=10,
    )
    resp.raise_for_status()
    values = resp.json().get("daily", {}).get("precipitation_sum", [None])
    v = values[0]
    return float(v) if v is not None else float("nan")

source = geobn.PointGridSource(fn=fetch_precipitation, sample_points=5)
bn.set_input("precipitation", source)

Example — MET Norway ocean forecast:

import requests

def sea_temperature(lat: float, lon: float) -> float:
    resp = requests.get(
        "https://api.met.no/weatherapi/oceanforecast/2.0/complete",
        params={"lat": lat, "lon": lon},
        headers={"User-Agent": "my-app/1.0"},
        timeout=10,
    )
    if resp.status_code == 422:
        return float("nan")  # outside ocean coverage
    resp.raise_for_status()
    ts = resp.json().get("properties", {}).get("timeseries", [])
    if not ts:
        return float("nan")
    val = ts[0]["data"]["instant"]["details"].get("sea_water_temperature")
    return float(val) if val is not None else float("nan")

source = geobn.PointGridSource(fn=sea_temperature, sample_points=5, delay=0.1)
bn.set_input("sea_temp", source)