dmot

Hydrological model extension package extracted from dmg, now extended with a modular reservoir dispatch scheduling kernel.

Package Structure

dmotpy/
├── models/                    # Hydrological rainfall-runoff models (HBV, GR4J, …)
│   ├── core/                  # 38+ step-function model implementations
│   ├── flux/                  # Flux calculation components
│   ├── unithydro/             # Unit hydrograph routing
│   └── special/               # Full class-based model wrappers
├── neural_networks/           # Calibration & parameterization networks
├── trainers/                  # Training loops (CalTrainer, FasterTrainer)
└── dispatch/                  # *** NEW: Reservoir dispatch scheduling kernel ***
    ├── domain/                # Layer 3: Domain objects
    ├── kernel/                # Layer 5: Physics kernel & simulator
    ├── plugins/               # Layer 4 & 6: Operation modules & plugin registry
    ├── services/              # Layer 2: Service orchestration
    ├── evaluator.py           # Evaluation engine
    └── mcp/                   # Layer 1: FastMCP tool interface

Dispatch Kernel — 6-Layer Architecture

The dispatch module implements a single-reservoir, modular, service-oriented scheduling kernel designed for:

• Rule-based dispatch with structured operation modules
• Unified simulation with un-bypassable physics core
• Plugin-based objectives, constraints, and strategies
• FastMCP tool exposure for agent integration
• Future extension to CBR, multi-objective optimization, and game-theoretic planning

Layer 1: Interface (FastMCP)

Exposes high-level capabilities as MCP tools:

Does not expose: direct state mutation, kernel bypass, constraint override.

Layer 2: Services

Layer 3: Domain Objects

Layer 4: Operation Modules

6 built-in module types:

New module types can be added by implementing ModuleHandler and registering with ModuleRegistry.

Layer 5: Physics Kernel

The un-bypassable core:

• Water balance: S(t+1) = S(t) + (I - O) × Δt
• Level-capacity interpolation (forward & inverse)
• Release clamping to physical limits (min/max/outlet capacity)
• Level change rate limiting
• Constraint checking (level max/min, release max/min, change rate)

No strategy, agent, or optimizer can directly mutate state — all transitions must go through the physics kernel.

Layer 6: Plugins

Pre-registered plugin slots for future extension:

Quick Start

Define a reservoir

from dmotpy.dispatch.domain.reservoir_spec import ReservoirSpec

spec = ReservoirSpec(
    name="MyReservoir",
    dead_level=100.0, normal_level=120.0,
    flood_limit_level=130.0, check_flood_level=135.0, crest_level=140.0,
    total_capacity=3500.0, flood_control_capacity=1300.0, dead_capacity=100.0,
    level_capacity={100.0: 100, 120.0: 1200, 130.0: 2200, 140.0: 3500},
    max_release=3000.0, min_release=10.0,
)

Create a dispatch program

from dmotpy.dispatch.domain.dispatch_program import DispatchProgram, ModuleInstance

program = DispatchProgram(
    name="flood_response",
    modules=[
        ModuleInstance(
            module_type="constant_release", name="normal",
            params={"release": 50.0},
            switch_in_conditions={"level_below": 128.0},
        ),
        ModuleInstance(
            module_type="inflow_driven", name="flood",
            params={"ratio": 1.2, "min_release": 100.0},
            switch_in_conditions={"level_above": 128.0},
        ),
    ],
)

Run simulation

from datetime import datetime
from dmotpy.dispatch.domain.reservoir_state import ReservoirState
from dmotpy.dispatch.domain.forecast import ForecastBundle
from dmotpy.dispatch.kernel.simulator import DispatchSimulator

state = ReservoirState(
    timestamp=datetime(2024, 7, 1),
    level=118.0, storage=900.0, inflow=100.0, outflow=50.0,
)
forecast = ForecastBundle.from_deterministic([50]*6 + [800, 1200, 1000, 700, 400, 200] + [50]*12)

sim = DispatchSimulator(spec)
result = sim.simulate(state, forecast, program)
print(result.summary())

Evaluate

from dmotpy.dispatch.domain.objectives import ObjectiveSet, ObjectiveWeight, ObjectiveType, ObjectiveDirection
from dmotpy.dispatch.evaluator import Evaluator

obj = ObjectiveSet()
obj.add(ObjectiveWeight(name="flood", objective_type=ObjectiveType.FLOOD_CONTROL,
                        direction=ObjectiveDirection.MAXIMIZE, weight=0.7))
obj.add(ObjectiveWeight(name="eco", objective_type=ObjectiveType.ECOLOGY,
                        direction=ObjectiveDirection.MAXIMIZE, weight=0.3))

evaluator = Evaluator(spec, obj)
eval_result = evaluator.evaluate(result, program)
print(eval_result.summary())

Use via MCP (for agents)

from dmotpy.dispatch.mcp import DispatchMCPServer

server = DispatchMCPServer(spec, initial_state)
server.get_snapshot()                          # Get state + spec
server.create_program("p1", "constant_release", {"release": 80.0})
server.simulate_program("p1", [60.0]*24)       # Simulate 24 steps
server.compare_programs(["p1", "p2"], [60.0]*24)  # Compare
server.explain_program("p1")                   # Explain

Running Tests

pip install pytest
pytest tests/dispatch/ -v

Dependencies