Skip to content

API Reference

This page documents the complete public pourpoint API — every class, method, and function exported from the package. It is generated automatically from the type stubs shipped with pourpoint (pourpoint/__init__.pyi), so the signatures here always match the installed version. Prose is thin where the stubs are signature-only and is being filled in over time.

For a task-oriented tour, start with the Quickstart and the Staged API guide.

pourpoint

pourpoint — Python bindings for the pourpoint watershed delineation engine.

Resolves bundled GDAL_DATA and PROJ data at import time (wheel installs). For source/editable installs the resolution is silent on miss; the user is expected to have a system GDAL/PROJ already configured.

ProgressCallback = Callable[[ProgressEvent], None] module-attribute

__all__ = ['AreaOnlyResult', 'AssemblyError', 'BasinGeoParquetWriter', 'DatasetError', 'DelineationResult', 'DelineationUnitMetadata', 'DissolvedWatershed', 'Engine', 'LevelSelection', 'PreMergeDrainageUnit', 'PreMergeDrainageUnits', 'ResolutionError', 'ResolvedOutlet', 'SelectedLevel', 'PourpointError', 'TerminalRefinement', 'UnitBundleGeoParquetWriter', 'UpstreamUnits', 'bench_trace', 'set_log_level'] module-attribute

__version__ = _pkg_version('pourpoint') module-attribute

ProgressEvent

Bases: TypedDict

Source code in crates/python/python/pourpoint/__init__.py 
23
24
25
26
27
28
29
30
31
    candidate = _PKG_DIR / "_data" / subdir
    return str(candidate) if (candidate / sentinel).is_file() else None


def _preseed_bundled_data() -> None:
    if "GDAL_DATA" not in os.environ:
        path = _bundled("gdal", "gdalvrt.xsd")
        if path is not None:
            os.environ["GDAL_DATA"] = path

index instance-attribute

total instance-attribute

lat instance-attribute

lon instance-attribute

duration_ms instance-attribute

status instance-attribute

n_catchments instance-attribute

error instance-attribute

PourpointError

Bases: Exception

DatasetError

Bases: PourpointError

ResolutionError

Bases: PourpointError

Source code in crates/python/python/pourpoint/__init__.py 
43
AreaOnlyResult,

AssemblyError

Bases: PourpointError

Source code in crates/python/python/pourpoint/__init__.py 
46
DatasetError,

LevelSelection

Source code in crates/python/python/pourpoint/__init__.py 
49
50
51
52
DissolvedWatershed,
Engine,
LevelSelection,
PreMergeDrainageUnit,

FINEST instance-attribute

__eq__(other)

Source code in crates/python/python/pourpoint/__init__.py 
52
PreMergeDrainageUnit,

DelineationResult

Source code in crates/python/python/pourpoint/__init__.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
    ResolvedOutlet,
    SelectedLevel,
    PourpointError,
    TerminalRefinement,
    UnitBundleGeoParquetWriter,
    UpstreamUnits,
    _install_bench_trace,
    _set_log_level,
)


_LOGGER_NAMES = ("pourpoint", "_pourpoint", "pourpoint_core", "hfx")
_LOG_LEVELS = {
    "trace": ("trace", logging.DEBUG),
    "debug": ("debug", logging.DEBUG),
    "info": ("info", logging.INFO),
    "warn": ("warn", logging.WARNING),
    "warning": ("warn", logging.WARNING),
    "error": ("error", logging.ERROR),
    "critical": ("error", logging.ERROR),
}
_LOG_FORMATTER = logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s")


def _normalize_log_level(level: str) -> tuple[str, int]:
    try:
        return _LOG_LEVELS[level.lower()]
    except KeyError as exc:
        raise ValueError(
            "unknown log level; valid values are: trace, debug, info, warn, "
            "warning, error, critical"
        ) from exc

terminal_unit_id property

input_outlet property

resolved_outlet property

refined_outlet property

resolution_method property

upstream_unit_ids property

upstream_units property

area_km2 property

geometry_bbox property

geometry_wkb property

to_geojson()

Source code in crates/python/python/pourpoint/__init__.py 
86
) from exc

__repr__()

DelineationUnitMetadata

Source code in crates/python/python/pourpoint/__init__.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
Updates the dynamic max-level used by the pyo3-log bridge and configures
the relevant Python loggers so records actually emit. If any pourpoint logger
has no handler, a default ``StreamHandler`` is attached to that logger.

Records originating from Rust route through pyo3-log under loggers named
after their Rust crate (``_pourpoint.*``, ``pourpoint_core.*``, ``hfx.*``).
We therefore set the level on each of those roots in addition to the
Python ``pourpoint`` facade.

Valid levels (case-insensitive): ``"trace"``, ``"debug"``, ``"info"``,
``"warn"``/``"warning"``, ``"error"``/``"critical"``.
"""
rust_level, py_level = _normalize_log_level(level)
_set_log_level(rust_level)
for logger_name in _LOGGER_NAMES:
    logger = logging.getLogger(logger_name)

id property

level property

area_km2 property

up_area_km2 property

outlet property

__repr__()

Source code in crates/python/python/pourpoint/__init__.py 
107
logger = logging.getLogger(logger_name)

AreaOnlyResult

Source code in crates/python/python/pourpoint/__init__.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
        if not logger.handlers:
            handler = logging.StreamHandler()
            handler.setFormatter(_LOG_FORMATTER)
            logger.addHandler(handler)


@contextmanager
def bench_trace(path: os.PathLike[str] | str) -> Iterator[None]:
    """Write Rust stage-span benchmark telemetry to ``path`` inside the context."""
    previous = os.environ.get("POURPOINT_BENCH_TRACE")
    os.environ["POURPOINT_BENCH_TRACE"] = os.fspath(path)
    _install_bench_trace()
    try:
        yield
    finally:
        if previous is None:
            os.environ.pop("POURPOINT_BENCH_TRACE", None)
        else:
            os.environ["POURPOINT_BENCH_TRACE"] = previous


__all__ = [
    "AreaOnlyResult",

terminal_unit_id property

input_outlet property

resolved_outlet property

refined_outlet property

resolution_method property

upstream_unit_ids property

area_km2 property

__repr__()

Source code in crates/python/python/pourpoint/__init__.py 
132
"AreaOnlyResult",

SelectedLevel

Source code in crates/python/python/pourpoint/__init__.py 
135
136
137
138
139
"DatasetError",
"DelineationResult",
"DelineationUnitMetadata",
"DissolvedWatershed",
"Engine",

level property

__repr__()

Source code in crates/python/python/pourpoint/__init__.py 
139
"Engine",

ResolvedOutlet

Source code in crates/python/python/pourpoint/__init__.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
    "PreMergeDrainageUnits",
    "ResolutionError",
    "ResolvedOutlet",
    "SelectedLevel",
    "PourpointError",
    "TerminalRefinement",
    "UnitBundleGeoParquetWriter",
    "UpstreamUnits",
    "bench_trace",
    "set_log_level",
]

try:
    __version__ = _pkg_version("pourpoint")
except PackageNotFoundError:
    __version__ = "0.0.0+unknown"

level property

terminal_unit_id property

input_outlet property

resolved_outlet property

resolution_method property

__repr__()

UpstreamUnits

Source code in crates/python/python/pourpoint/__init__.py 
161
162
163
164
165
166
167
168
169
170
171
    if "GDAL_DATA" in os.environ:
        return  # respect explicit user override
    path = _bundled("gdal", "gdalvrt.xsd")
    if path is None:
        _log.debug("no bundled gdal data in %s; relying on system GDAL", _PKG_DIR)
        return
    _pourpoint._set_gdal_data(path)


def _inject_proj_data() -> None:
    for env_var in ("PROJ_DATA", "PROJ_LIB"):

terminal_unit_id property

level property

unit_ids property

__repr__()

Source code in crates/python/python/pourpoint/__init__.py 
171
for env_var in ("PROJ_DATA", "PROJ_LIB"):

PreMergeDrainageUnit

Source code in crates/python/python/pourpoint/__init__.py 
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
            return
    path = _bundled("proj", "proj.db")
    if path is None:
        _log.debug("no bundled proj data in %s; relying on system PROJ", _PKG_DIR)
        return
    _pourpoint._set_proj_data(path)


_inject_gdal_data()
_inject_proj_data()

if "POURPOINT_LOG" in os.environ:
    try:
        set_log_level(os.environ["POURPOINT_LOG"])
    except ValueError:
        warnings.warn(
            "invalid POURPOINT_LOG value; valid values are: trace, debug, info, "

id property

level property

area_km2 property

up_area_km2 property

outlet property

__repr__()

Source code in crates/python/python/pourpoint/__init__.py 
190
"invalid POURPOINT_LOG value; valid values are: trace, debug, info, "

PreMergeDrainageUnits

Source code in crates/python/python/pourpoint/__init__.py 
193
194
    stacklevel=2,
)

R3_NOTE instance-attribute

terminal_unit_id property

level property

units property

unit_geometry_wkb property

__repr__()

TerminalRefinement

status property

refined_outlet property

__repr__()

DissolvedWatershed

area_km2 property

geometry_wkb property

__repr__()

BasinGeoParquetWriter

__init__()

write(engine, path, results, *, basin_ids=None, method=None, allow_default_basin_id=False)

__repr__()

UnitBundleGeoParquetWriter

__init__()

write(engine, path, bundles, refinements, *, method=None)

__repr__()

Engine

__init__(dataset_path, *, snap_radius=None, snap_strategy=None, snap_threshold=None, clean_epsilon=None, refine=True, repair_geometry='auto', parquet_cache=None, parquet_cache_max_mb=512)

delineate_batch(outlets, *, progress=None)

select_level(selection=LevelSelection.FINEST)

resolve_outlet(level, *, lat, lon)

traverse(outlet)

pre_merge_units(upstream)

refine(outlet, units)

dissolve(units, refinement)

compose_result(outlet, upstream, units, refinement, dissolved)

set_log_level(level)

Set the pourpoint log level for both the Rust bridge and Python logging.

Updates the dynamic max-level used by the pyo3-log bridge and configures the relevant Python loggers so records actually emit. If any pourpoint logger has no handler, a default StreamHandler is attached to that logger.

Records originating from Rust route through pyo3-log under loggers named after their Rust crate (_pourpoint.*, pourpoint_core.*, hfx.*). We therefore set the level on each of those roots in addition to the Python pourpoint facade.

Valid levels (case-insensitive): "trace", "debug", "info", "warn"/"warning", "error"/"critical".

Source code in crates/python/python/pourpoint/__init__.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def set_log_level(level: str) -> None:
    """Set the pourpoint log level for both the Rust bridge and Python `logging`.

    Updates the dynamic max-level used by the pyo3-log bridge and configures
    the relevant Python loggers so records actually emit. If any pourpoint logger
    has no handler, a default ``StreamHandler`` is attached to that logger.

    Records originating from Rust route through pyo3-log under loggers named
    after their Rust crate (``_pourpoint.*``, ``pourpoint_core.*``, ``hfx.*``).
    We therefore set the level on each of those roots in addition to the
    Python ``pourpoint`` facade.

    Valid levels (case-insensitive): ``"trace"``, ``"debug"``, ``"info"``,
    ``"warn"``/``"warning"``, ``"error"``/``"critical"``.
    """
    rust_level, py_level = _normalize_log_level(level)
    _set_log_level(rust_level)
    for logger_name in _LOGGER_NAMES:
        logger = logging.getLogger(logger_name)
        logger.setLevel(py_level)
        logger.propagate = False
        if not logger.handlers:
            handler = logging.StreamHandler()
            handler.setFormatter(_LOG_FORMATTER)
            logger.addHandler(handler)

bench_trace(path)

Write Rust stage-span benchmark telemetry to path inside the context.

Source code in crates/python/python/pourpoint/__init__.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
@contextmanager
def bench_trace(path: os.PathLike[str] | str) -> Iterator[None]:
    """Write Rust stage-span benchmark telemetry to ``path`` inside the context."""
    previous = os.environ.get("POURPOINT_BENCH_TRACE")
    os.environ["POURPOINT_BENCH_TRACE"] = os.fspath(path)
    _install_bench_trace()
    try:
        yield
    finally:
        if previous is None:
            os.environ.pop("POURPOINT_BENCH_TRACE", None)
        else:
            os.environ["POURPOINT_BENCH_TRACE"] = previous