Skip to content

Python API

Everything the CLI does is available from Python. The high-level functions take the same configuration as the YAML file; the lower-level functions let you build your own pipeline in memory. All names below are importable from mapcv unless noted.

import mapcv
from pathlib import Path
config = mapcv.MapcvConfig.from_yaml(Path("mapcv.yaml"))
estimate = mapcv.plan(config) # no download
print(estimate.patches, estimate.tiles, estimate.is_large)
result = mapcv.run_generate(config) # builds or resumes the dataset
print(result.new_patches, result.split_counts)
class MapcvConfig(BaseModel):
task: Literal["segmentation", "detection", "instance", "classification", "change", "regression"] = "segmentation"
region: RegionConfig
imagery: ImageryConfig | list[ImageryConfig] # one source (by imagery.type), or a list of named ones
labels: LabelsConfig | RasterLabelsConfig | ContinuousLabelsConfig | None = None # by labels.type: vector | raster | continuous
sampler: SamplerConfig
writer: WriterConfig
split: SplitterConfig | None = None
detection: DetectionOptions | None = None # task: detection only
instance: InstanceOptions | None = None # task: instance only
classification: ClassificationOptions | None = None # task: classification only
change: ChangeOptions | None = None # task: change only
@property
def sources(self) -> list[ImageryConfig]: ... # every source; a list of one for a single block
@property
def primary_imagery(self) -> ImageryConfig: ... # the first source: its grid is the dataset's
@property
def multi_source(self) -> bool: ... # imagery given as a list
@property
def source_names(self) -> list[str]: ... # ["image"] for a single block, else the names
@property
def detection_options(self) -> DetectionOptions: ... # the block, or its defaults
@property
def instance_options(self) -> InstanceOptions: ... # the block, or its defaults
@property
def classification_options(self) -> ClassificationOptions: ... # the block, or its defaults
@property
def change_options(self) -> ChangeOptions: ... # the block, or its defaults
@classmethod
def from_yaml(cls, path: str | os.PathLike) -> MapcvConfig: ...
class DetectionOptions(BaseModel):
min_visible: float = 0.3
min_box_pixels: float = 2.0
formats: list[Literal["coco", "yolo"]] = ["coco", "yolo"]
point_box_size: float | None = None
class InstanceOptions(BaseModel):
min_visible: float = 0.3
min_area: int = 4
id_mask: bool = False
class ClassificationOptions(BaseModel):
mode: Literal["single", "multi"] = "single"
min_fraction: float = 0.0
empty: Literal["skip", "background"] = "skip"
class ChangeOptions(BaseModel): # from mapcv.config
change_value: int = 1
before: LabelsConfig | None = None
after: LabelsConfig | None = None

Pydantic v2 models, so you can also build them from dicts with MapcvConfig.model_validate({...}) or construct the parts directly. from_yaml resolves relative paths (region.path, every local imagery path, labels.path, labels.files[].path, labels.annotated_area, change.before.path, change.after.path and writer.staging_dir) against the YAML file’s folder; model_validate leaves paths as given. Unknown keys and validation errors raise pydantic.ValidationError; the removed 0.1 keys tiles and region.zoom are validation errors that name the replacement. Field reference: Configuration.

imagery is a list when the config names several sources, so code that reads one source’s settings should use config.primary_imagery (or loop over config.sources) rather than config.imagery.

Also exported: RegionConfig, XYZImageryConfig, EOPFZarrImageryConfig, StacCogImageryConfig, GeoTiffImageryConfig, ImageryConfig (the discriminated union), LabelsConfig (polygon labels), RasterLabelsConfig and RasterClass (raster labels), SamplerConfig, WriterConfig, SplitterConfig, DEFAULT_SENTINEL2_L2A_BANDS (the 12 default band names), the type aliases ClassMap (dict[str, int]) and GeomWithClass (tuple[BaseGeometry, int]), MANIFEST_VERSION (3) and __version__. ContinuousLabelsConfig and ChangeOptions are importable from mapcv.config.

def plan(config: MapcvConfig) -> Plan

Estimates a run without downloading imagery (it does read the label file). Raises ValueError, RuntimeError or OSError for problems it can detect.

A dataclass:

Attribute Type Meaning
task str The config’s task ("segmentation")
region_km (float, float) Approximate width and height of the region
imagery str Human-readable source description
resolution_m float Ground resolution in metres per pixel
raster_px (int, int) Raster (width, height)
patches int Estimated patch count
patch_size int From the sampler
tiles int | None XYZ tile count (None for other imagery)
download_bytes int | None Rough download size (XYZ only)
output_bytes int Rough size on disk
chunk_memory_bytes int Rough peak memory per chunk
labels LabelSummary | None path, polygons, classes, warnings, in_region (features that intersect the region) and raster (a label raster’s description, else None)
objects int | None Detection and instance: label features in the region, each one object; None for other tasks
warnings list[str] Messages to show the user
is_large bool (property) More than 20,000 tiles, 5 GB or 1,000,000 patches; the CLI asks before starting such jobs
def generate(
config: MapcvConfig | str | os.PathLike,
*,
progress: Callable[[int, int], None] | None = None,
) -> GenerateResult

The library form of mapcv generate: builds the dataset into config.writer.staging_dir, or resumes it. config is a MapcvConfig or the path of a YAML file (read with MapcvConfig.from_yaml). progress(done, total) is called with the chunks written so far and the chunks of this run, before the first chunk and after every chunk; an exception it raises stops the run there (the finished chunks are saved), and the next call resumes (this is how the MCP server reports progress and cancels).

import mapcv
def show(done: int, total: int) -> None:
print(f"{done}/{total} chunks", end="\r")
result = mapcv.generate("mapcv.yaml", progress=show)
print(result.new_patches, "new patches in", result.staging_dir)

run_generate(config, on_chunk) is the same function with a MapcvConfig only.

Output. The library prints nothing. Messages such as “Resuming: 120 patches already written, 80 to go.” go to the mapcv logger at INFO; show them with logging.basicConfig(level=logging.INFO), or logging.getLogger("mapcv").setLevel(logging.INFO) with a handler of your own. Unlike mapcv generate, it never asks for confirmation, so check plan(config).is_large yourself if needed. Problems that don’t stop the run (skipped label features, labels outside the imagery, more than half of the tiles failing, split leakage) are emitted as UserWarnings. Raises ManifestMismatchError if the folder holds an incompatible dataset, mapcv.locking.DatasetBusyError (a RuntimeError) while another generate or split writes to the folder, mapcv.locking.StagingDirError (a ValueError) for a folder with other files and no manifest (hidden entries and a top-level README.md do not count), RuntimeError for download failures beyond the tile policy, and ValueError / OSError for bad inputs.

Attribute Type Meaning
staging_dir Path Output folder
manifest Manifest The final manifest (all patches, including resumed ones)
new_patches int Patches written by this call
split_counts dict[str, int] | None train, val, test, dropped; None without a split block
tiles_requested int XYZ tiles requested in this call (0 for other imagery)
tiles_failed int Of those, how many failed
seconds float Wall-clock time
tiles_cached int XYZ tiles read from the tile cache instead of downloaded
patches_without_imagery int Patches left out by this call because they had no pixel of imagery
def iter_patches(config: MapcvConfig | str | os.PathLike) -> Iterator[Patch]

Yields every patch the config describes, in order, without writing anything: the same patches, filters (max_empty_ratio, min_label_ratio, an area of interest) and order as generate, read chunk by chunk so memory stays bounded. The writer block is not used. Use it to feed a model straight from the imagery, or to look at what a config produces before generating.

for patch in mapcv.iter_patches("mapcv.yaml"):
print(patch.row, patch.col, patch.image.shape, patch.target.max())
Attribute Type Meaning
row, col int Top-left pixel of the patch on the (first) source’s grid; the manifest entry’s row/col
image ndarray (H, W, C): uint8 RGB for XYZ, the source’s bands and dtype otherwise
target depends on the task Segmentation and change: the mask (H, W); regression: float32 values (NaN without a target); detection, instance and classification: the task’s object or label record; None without labels
transform 6-tuple The patch’s affine transform (a, b, c, d, e, f) in crs
crs str The source CRS, e.g. "EPSG:32631"
padded bool Whether the patch reaches past the raster edge (edge_strategy: pad)
others dict[str, ndarray] The patches of further imagery sources, by name
def split(dataset: str | os.PathLike, config: SplitterConfig | None = None, **settings) -> dict[str, int]

The library form of mapcv split. Pass a SplitterConfig, or its fields as keyword arguments: mapcv.split("dataset", strategy="random", test_ratio=0.25). Returns the counts like run_split.

def run_split(staging_dir: Path, split_config: SplitterConfig | None = None) -> dict[str, int]

Reads staging_dir/manifest.json, writes staging_dir/splits/ and returns the counts (train, val, test, dropped). For a detection dataset it also rewrites the per-split COCO files, the YOLO image lists and dataset.yaml; for an instance dataset, the per-split COCO files; for a classification dataset, the split column of labels.csv and the labels_<split>.csv files. Without split_config, uses SplitterConfig() defaults. Raises FileNotFoundError when there is no manifest.

def split_dataset(manifest: Manifest, config: SplitterConfig, output_dir: Path) -> dict[str, int]

The same, for a manifest you already have in memory; writes the lists and split.json (settings, the strategy actually used, and counts) to output_dir and prints nothing. Emits a UserWarning when a non-spatial split leaks overlapping patches, or when a manifest without patch size forces a fallback from spatial to stratified.

class SplitterConfig(BaseModel):
test_ratio: float = 0.20 # 0..1
val_ratio: float = 0.10 # 0..1, of the remainder
labeled_ratios: list[float] = [0.10, 0.20, 0.30] # each in (0, 1]
seed: int = 42
strategy: Literal["spatial", "stratified", "random", "region"] = "spatial"
block_size: int | None = None # >= 1; default 4 x patch size (auto-shrunk)
sample_limit: int | None = None # >= 1

See Splits without leakage.

from mapcv.infer import predict_raster # also mapcv.predict_raster
def predict_raster(
config: MapcvConfig | str | os.PathLike,
fn: Callable[[np.ndarray], np.ndarray],
out: str | os.PathLike,
*,
overlap: float = 0.5,
batch_size: int = 16,
output: Literal["argmax", "scores"] = "argmax",
progress: Callable[[int, int], None] | None = None,
) -> Path

Predicts every pixel of the config’s region (first imagery source, sampler.patch_size) and writes one GeoTIFF on the imagery’s grid. fn gets (N, C, P, P) patches as read (the source’s dtype; normalise inside it) and returns (N, K, P, P) scores or (N, P, P) values. Patches overlap by overlap (0 to < 1) and their predictions are averaged with Hann (cosine) weights, sin² at pixel centres, so every pixel keeps some weight. "argmax" writes the best class as uint8 with 255 (the file’s NoData) where there is no imagery; "scores" writes the blended scores as float32 bands with NaN there. A prediction of (N, P, P) values (or one channel) is written as one float32 band either way. progress(done, total) counts strips of rows. Raises ValueError for a bad overlap, output, batch_size, a prediction of the wrong shape, a class index of 255 or more for "argmax", or imagery without a CRS.

from mapcv.data import MapcvDataset # also mapcv.MapcvDataset
class MapcvDataset:
def __init__(self, root: str | os.PathLike, split: str = "train", *,
normalize: bool = False,
transform: Callable[[dict], Any] | None = None,
as_tensors: bool | None = None) -> None
def __len__(self) -> int
def __getitem__(self, index: int) -> dict # or what transform returns
classes: dict[int, str] # property: class ID to name
ignore_index: int | None # property: the mask value to leave out of the loss

One split (train, val, test, 10/labeled, …, or all) of a dataset, as a map-style dataset that torch.utils.data.DataLoader takes directly. Items are dicts:

Key Present Value
image always (C, H, W) ((T, C, H, W) for stacked sources), as stored or, with normalize, standardised float32
images several sources Every source’s image by name
mask segmentation, change, regression, instance with id_mask (H, W) class IDs, instance IDs (uint16) or float32 values
labels classification The patch’s class IDs, int64
boxes, categories, annotations detection, instance (N, 4) [x, y, width, height] boxes in pixels, their class IDs, and the COCO records (from the split’s COCO files)
name, row, col, transform, crs always The patch’s file name, top-left pixel, affine transform and CRS

normalize=True uses the per-band mean and standard deviation of the train split from stats.json (computing them if the file is missing) and sets pixels without data to 0. Arrays are torch tensors when torch is installed (uint16 widened to int32) unless as_tensors=False; mapcv does not depend on torch. Raises FileNotFoundError for a missing manifest or split list (the message lists the available ones), and for a detection dataset written without COCO files.

export_hf_parquet, terratorch_config, export_webdataset, export_zarr

Section titled “export_hf_parquet, terratorch_config, export_webdataset, export_zarr”
from mapcv.export import export_hf_parquet, export_terratorch, terratorch_config
from mapcv.shards import export_webdataset, export_zarr
def export_hf_parquet(root: Path, out: Path) -> list[Path]
def terratorch_config(root: Path, batch_size: int = 8, num_workers: int = 4) -> str
def export_terratorch(root: Path, out: Path | None = None) -> Path # default: root / "terratorch.yaml"
def export_webdataset(root: Path, out: Path, shard_bytes: int = 1_000_000_000) -> list[Path]
def export_zarr(root: Path, out: Path) -> Path

The functions behind mapcv export.

from mapcv.stats import dataset_stats, write_stats
from mapcv.card import card_text, write_card
from mapcv.verify import verify_dataset, write_checksums
def dataset_stats(staging_dir: Path, split: str = "train") -> dict[str, Any]
def write_stats(staging_dir: Path, split: str = "train") -> tuple[Path, dict[str, Any]]
def card_text(staging_dir: Path) -> str
def write_card(staging_dir: Path, overwrite: bool = False) -> Path
def verify_dataset(staging_dir: Path, deep: bool = False) -> VerifyReport
def write_checksums(staging_dir: Path) -> Path

The functions behind mapcv stats, mapcv card and mapcv verify. dataset_stats and card_text return what the other two write. VerifyReport has patches, files, checked_hashes, problems and notes (lists of messages) and ok (no problems).

The parsed manifest.json (version 3; see Dataset format for every field). GenerateResult.manifest is one.

from mapcv import Manifest
class Manifest(BaseModel):
version: int = 3
mapcv_version: str | None = None
task: str = "segmentation"
sources: list[SourceRecord] = [] # name, source_type, product_id, bands, dtype,
# crs, transform, patch_shape, fingerprint
target: TargetRecord | None = None # type, class_map, ignore_index, dtype, labels, options
writer: dict[str, Any] | None = None # layout, image_format, jpg_quality, jpg_subsampling,
# mask_format (+ world_files, stack_sources when on)
sampler: dict[str, Any] | None = None
patches: list[ManifestEntry] = []
source: SourceRecord # the first source (read-only)
class_map: dict[str, int] # target.class_map, or {} (read-only)
ignore_index: int | None # target.ignore_index, or None (read-only)
upgraded_from: int | None # 1 or 2 when loaded from an older version
loaded_version: int # the version on disk (read-only)
@classmethod
def load(cls, path: Path) -> Manifest: ... # version 1, 2 or 3; older ones are upgraded in
# memory; newer raises ManifestMismatchError
@classmethod
def from_dict(cls, data: dict) -> Manifest: ... # like load, from parsed JSON
def save(self, path: Path) -> None: ... # atomic write, always version 3
def to_json(self) -> str: ... # what save writes
def patch_name(self, entry) -> str: ... # the name split lists use: "patch_0000042.png"
def patch_transform(self, entry) -> tuple[float, ...]: ... # affine of the patch's pixels
def patch_bounds(self, entry) -> tuple[float, float, float, float]: ...
# (left, bottom, right, top) in the source CRS
def transform_at(self, row: int, col: int) -> tuple[float, ...]: ... # affine of the patch at (row, col)

Manifest.load(path).save(path) converts a 0.1 or 0.2 manifest to version 3 on disk. Unknown keys are kept, so a manifest from a later 0.3 release round-trips unchanged.

A TypedDict (plain dict at runtime) with row, col, padded, chunk, files (role → path relative to the dataset folder, for example {"image": "Images/patch_0000000.png", "mask": "Masks/patch_0000000.png"}) and summary (a PatchSummary: empty_ratio and, depending on the task, class_pixels, class_objects, class_coverage, labels or values, and region with an area of interest). Meanings: Dataset format.

from mapcv import ManifestMismatchError
class ManifestMismatchError(ValueError): ...

Raised by run_generate (through load_or_create_manifest) when the target folder holds a version-1 manifest, a version-2 manifest of random sampling, or one created with a different task, source, product, bands, dtype, patch shape, CRS, transform, imagery file, class map, ignore index, label settings or label-file contents, task options, area of interest, sampler or writer settings. The message names the fields that differ. Manifest.load raises it for a manifest written by a newer mapcv.

These work on in-memory arrays and are useful for notebooks or custom pipelines. The XYZ path below reproduces what generate does for one small region:

from pathlib import Path
from mapcv import (SamplerConfig, parse_geojson, rasterize, sample_patches,
stitch_region, transform_to_mercator)
image, transform = stitch_region(4.9375, 52.3725, 4.9515, 52.3780, zoom=17,
source="esri_satellite", max_connections=4)
# image: (1024, 1536, 3) uint8; transform: Web Mercator affine
geometries, class_map = parse_geojson(Path("buildings.geojson").read_bytes(), label_field="class")
geometries = [(transform_to_mercator(g), cid) for g, cid in geometries] # into EPSG:3857
mask = rasterize(geometries, image.shape[:2], transform) # (1024, 1536) uint8
patches, masks, meta = sample_patches(image, mask, SamplerConfig(patch_size=256, edge_strategy="drop"))
# patches: (24, 256, 256, 3); masks: (24, 256, 256); meta[0] == {'row': 0, 'col': 0, 'padded': False, 'empty_ratio': 0.0}
def load_vector_labels(path: Path, label_field: str | None = None,
classes: dict[str, int] | None = None, points: bool = False,
layer: str | None = None,
buffer: tuple[float | None, float | None] | None = None,
) -> tuple[list[tuple[BaseGeometry, int]], dict[str, int]]

Read a label file of any supported format (chosen by its suffix: .geojson/.json, .kml, .gpkg, .shp, .parquet/.geoparquet) into (shapely geometry, class_id) pairs in WGS-84 lon/lat, plus the class map. GeoPackage, Shapefile and GeoParquet files in another CRS are reprojected with pyproj; a file that does not state its CRS is a ValueError. Every format follows the labels rules: label_field and classes give the IDs, polygons are kept, points only with points=True, and other features are skipped with a UserWarning that counts them. layer picks the table of a GeoPackage with several (an error for other formats). buffer is (line, point) in metres, as in labels.buffer: lines and points become polygons of that width or diameter. Problems with the file raise ValueError with the fix in the message; a missing file raises FileNotFoundError. GeoParquet needs the mapcv[parquet] extra.

def parse_geojson(data: bytes, label_field: str | None = None,
classes: dict[str, int] | None = None, points: bool = False,
buffer: tuple[float | None, float | None] | None = None,
) -> tuple[list[tuple[BaseGeometry, int]], dict[str, int]]
def parse_kml(data: bytes, label_field: str | None = None,
classes: dict[str, int] | None = None) -> tuple[list[tuple[BaseGeometry, int]], dict[str, int]]

Parse file bytes into (shapely geometry, class_id) pairs in WGS-84 lon/lat, plus the class map. Class IDs follow the labels rules. Skipped features raise a UserWarning with counts. parse_geojson accepts a FeatureCollection or a single Feature and rejects projected crs members, and with points=True also keeps Point/MultiPoint features (detection’s point_box_size uses it); parse_kml reads <Data> and <SimpleData> fields.

def transform_to_mercator(geom: BaseGeometry) -> BaseGeometry

Reprojects a WGS-84 shapely geometry to Web Mercator (EPSG:3857), the CRS of XYZ rasters. The part beyond latitude ±85.0511°, where Web Mercator ends, is cut off first.

def rasterize(geometries: Iterable[tuple[BaseGeometry, int]], out_shape: tuple[int, int],
transform: tuple[float, float, float, float, float, float],
all_touched: bool = False) -> np.ndarray # uint8, shape out_shape

Burns polygons into a uint8 mask of shape (height, width). Geometries must already be in the raster’s CRS; transform maps (col, row) to (x, y) like rasterio’s Affine. Pixels are filled when their centre is inside a polygon (or when touched, with all_touched=True); later geometries overwrite earlier ones; non-polygons are ignored. Class IDs must be 1–255, otherwise ValueError.

def sample_patches(image: np.ndarray, mask: np.ndarray | None,
config: SamplerConfig) -> tuple[np.ndarray, np.ndarray | None, list[PatchMeta]]

Cuts patches from an in-memory (H, W) or (H, W, C) image (channels last) following config (grid or random, edge strategy, empty and label filters; all-zero pixels count as empty). Returns (N, ps, ps[, C]) images, (N, ps, ps) masks or None, and per-patch {"row", "col", "padded", "empty_ratio"}. sample_patches_at_anchors(image, mask, anchors, config, *, row_offset=0, col_offset=0, valid_mask=None, ignore_index=None) does the same at explicit top-left anchors; with ignore_index, mask pixels without imagery (padding, and pixels outside valid_mask) get that value.

def download_region(west: float, south: float, east: float, north: float, zoom: int,
url_template: str | None = None, source: str | None = None,
max_connections: int = 16, policy: str = "lenient",
snap_to_tiles: bool = True, max_failed_ratio: float = 0.05,
progress: Callable[[int, int], None] | None = None,
) -> list[tuple[TileIndex, bytes]]
def stitch_region(west: float, south: float, east: float, north: float, zoom: int,
url_template: str | None = None, source: str | None = None,
max_connections: int = 16, policy: str = "lenient",
max_failed_ratio: float = 0.05,
progress: Callable[[int, int], None] | None = None,
) -> tuple[np.ndarray, tuple[float, float, float, float, float, float]]

download_region fetches the encoded tiles for a box (expanded to whole tiles by default) and returns (tile, bytes) pairs; tile objects have x, y and z. stitch_region also decodes and stitches them into one (H, W, 3) uint8 image and returns its Web Mercator affine transform. policy and max_failed_ratio behave as in the config; RuntimeError if too many tiles fail. progress(done, total) is called with the tiles finished so far; neither prints anything, and the outcome (“120/120 tiles fetched, 0 failed”) goes to the mapcv logger at INFO.

URL_TEMPLATES: dict[str, str]
# keys: esri_satellite, esri_topo, esri_street, cartodb_positron, cartodb_dark_matter

The built-in XYZ sources and their URL templates. resolve_url_template(url_template, source) returns the template generate would use.

snap_bbox(west, south, east, north, zoom) returns the box expanded to whole tiles (an object with west, south, east, north); bounds(x, y, z) returns a tile’s lon/lat box (ValueError for a zoom above 32 or a column or row that does not exist at that zoom). The other tile helpers check their input the same way: tile(lng, lat, zoom) raises ValueError for a longitude or latitude that is not a number or a zoom above 32 (a position beyond the world is moved onto the edge), and snap_bbox for a zoom above 32. A point or a zero-width box snaps to the tile that contains it. snap_bbox raises ValueError when south > north, or when west > east (a box that crosses the antimeridian; split it into two boxes at 180°).

load_or_create_manifest(path, expected) and write_patches(image_patches, mask_patches, meta, config, manifest, chunk_index=0, *, images_dir="Images", masks_dir="Masks", image_key="image", source=None, band_names=None, time_steps=None) are the functions run_generate uses to resume-check and write patches. expected is a Manifest without patches describing the run (task, sources, target, writer, sampler); the existing manifest at path is returned when it matches, expected when there is none, and ManifestMismatchError is raised otherwise. They are exported for custom pipelines that produce their own patches but want a mapcv-compatible manifest.