pyrrmod 0.2.0

A Python rainfall-runoff model library for hydrologic simulations

pyrrmod

pyrrmod is a Python rainfall-runoff model library for running conceptual hydrologic simulations with a clean validation boundary, Numba-accelerated kernels, and MCP-friendly outputs.

The library is currently simulation-only:

• validate inputs with Pydantic
• convert validated payloads into ndarray runtime state
• run models with Python or Numba kernels
• export grouped outputs, internal states/fluxes, and JSON-safe MCP payloads

Installation

pip install -e .

Or with uv:

uv sync

Core Ideas

• Pydantic is used only at the input boundary.
• Runtime execution uses only ndarray, scalars, and simple containers.
• Numba kernels never receive Pydantic models.
• run_from_payload() is the recommended MCP entrypoint.

Internal Layout

The public model interface stays the same, but the internal structure is split by responsibility:

• pyrrmod/schemas.py: input models, array coercion, and runtime dataclasses
• pyrrmod/core.py: shared model execution core
• pyrrmod/models/__init__.py: thin model package export surface
• pyrrmod/catalog.py: model registry, metadata, and factory wiring
• pyrrmod/unithydro/core.py: unit-hydro state, hydrograph builders, and update ops
• pyrrmod/unithydro/mixins.py: routed/unit-hydro mixins
• pyrrmod/solvers/newton.py: Newton helper
• pyrrmod/solvers/routed_numba.py: Numba-backed solver kernels

Main Interfaces

Input models

• ClimateInput
• SolverOptionsInput
• ModelConfigInput
• ModelRunInput

Main model methods

• configure(...)
• run(...)
• validate_run_payload(...)
• run_from_payload(...)
• get_output(nargout)
• get_output_dict(...)
• to_mcp_payload(...)
• check_water_balance()
• get_streamflow()
• compile_model_fun(target="auto" | "python" | "numba")

Create a Model

from pyrrmod import create_model

model = create_model("collie1")

You can also instantiate models directly:

from pyrrmod.models import collie1

model = collie1()

Example Dataset

This repository includes data/03604000.csv with columns:

• prcp(mm/day)
• tmean(C)
• pet(mm)
• flow(mm)

Example loader:

from pathlib import Path
import numpy as np

data_file = Path("data/03604000.csv")
data = np.loadtxt(data_file, delimiter=",", skiprows=1)

climate = {
    "precip": data[:, 0],
    "temp": data[:, 1],
    "pet": data[:, 2],
}
q_obs = data[:, 3]

Recommended Structured Run

from pathlib import Path
import numpy as np

from pyrrmod import ModelRunInput
from pyrrmod.models import collie1

data = np.loadtxt(Path("data/03604000.csv"), delimiter=",", skiprows=1)

run_input = ModelRunInput(
    theta=[500.0],
    delta_t=1.0,
    S0=[100.0],
    climate={
        "precip": data[:, 0],
        "temp": data[:, 1],
        "pet": data[:, 2],
    },
    solver_options={
        "resnorm_tolerance": 0.1,
        "resnorm_maxiter": 6,
    },
    compile_target="numba",
)

model = collie1()
model.run(config=run_input)

q_sim = model.get_streamflow()
water_balance_error = model.check_water_balance()

print(q_sim[:5])
print(water_balance_error)

Direct Run API

You can also pass raw inputs directly to run().

from pyrrmod.models import collie1

model = collie1()
model.run(
    theta=[500.0],
    delta_t=1.0,
    S0=[100.0],
    input_climate={
        "precip": [10.0, 0.0, 5.0],
        "pet": [1.0, 1.0, 1.0],
        "temp": [12.0, 13.0, 14.0],
    },
    solver_opts={"resnorm_tolerance": 0.1},
    compile_target="python",
)

MCP-Friendly Payload Run

Recommended MCP flow:

1. call validate_run_payload(payload) if you want preflight validation
2. call run_from_payload(payload) to execute
3. use include_execution=True if the caller needs compiler metadata

run_from_payload() accepts alias-friendly payloads and returns compact JSON-safe data by default.

If an MCP adapter needs structured error payloads, catch the exception and use model._format_mcp_error(exc) to normalize it into one of:

• payload_error
• configuration_error
• simulation_error
• compile_error

from pathlib import Path
import json
import numpy as np

from pyrrmod.models import collie1

data = np.loadtxt(Path("data/03604000.csv"), delimiter=",", skiprows=1, max_rows=30)

payload = {
    "theta": [500.0],
    "delta_t": 1.0,
    "initial_stores": [100.0],
    "climate": {
        "rainfall": data[:, 0],
        "temperature": data[:, 1],
        "PET": data[:, 2],
    },
    "solver_options": {"resnorm_tolerance": 0.1},
    "compile_target": "numba",
}

model = collie1()
validated = model.validate_run_payload(payload)
result = model.run_from_payload(
    validated,
    include_execution=True,
)

print(json.dumps(result, indent=2)[:500])

Output APIs

get_streamflow()

Returns simulated streamflow as a NumPy array.

get_output(nargout)

Legacy-style interface. Keep this for compatibility, but prefer get_output_dict(...) for structured consumers and to_mcp_payload(...) for MCP.

• nargout=3: (flux_output, flux_internal, store_internal)
• nargout=4: plus water balance error
• nargout=5: plus solver diagnostics

get_output_dict(...)

This is the canonical structured output source. By default it stays lightweight and returns:

• model
• status
• flux_groups
• streamflow
• water_balance_error

Optional fields:

• optional flux_internal
• optional store_internal
• optional solver_data
• optional execution

to_mcp_payload(...)

This is a thin JSON-safe wrapper over get_output_dict(...).

Conversions:

• numpy.ndarray -> list
• numpy scalar -> Python scalar

Compiler Targets

Available targets:

• python: always use Python kernels
• numba: request lazy Numba compilation
• auto: request compiled path if available, otherwise fall back to Python

Compilation is lazy. The model is not compiled at import time. Runtime arrays are constructed first, and the kernel is compiled only when needed. If a model-specific compiled kernel is unavailable, the library falls back safely to the Python kernel and records the reason in model.compiler.detail.

Full Configuration Pattern

from pyrrmod import ModelConfigInput
from pyrrmod.models import collie1

model = collie1()
model.configure(
    config=ModelConfigInput(
        delta_t=1.0,
        S0=[100.0],
        input_climate={
            "precip": [10.0, 0.0, 5.0],
            "pet": [1.0, 1.0, 1.0],
            "temp": [12.0, 13.0, 14.0],
        },
        solver_opts={"resnorm_tolerance": 0.1},
        compile_target="numba",
    )
)

model.run(theta=[500.0])

Test

The repository includes pytest coverage for:

• Pydantic input validation
• simulation output structure
• MCP-safe payload output
• Python/Numba consistency on data/03604000.csv

Run tests with:

pytest