dmotpy 0.1.0

Hydrological model extension package extracted from dmg

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:

Tool	Description
get_snapshot	Current state, spec, available modules
list_operation_modules	Available module types with descriptions
create_program	Create a dispatch program
simulate_program	Run a simulation against a forecast
evaluate_program	Simulate and evaluate against objectives
compare_programs	Compare multiple programs side-by-side
explain_program	Human-readable program explanation

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

Layer 2: Services

Service	Responsibility
SnapshotService	State and spec queries
ProgramService	Dispatch program CRUD
SimulationService	Run simulations, simulate-and-evaluate
EvaluationService	Evaluate results, compare programs
ExplanationService	Generate human-readable explanations

Layer 3: Domain Objects

Object	Purpose
ReservoirSpec	Static reservoir geometry, curves, gates, limits
ReservoirState	Dynamic state: level, storage, inflow, outflow
ForecastBundle	Single-value or ensemble forecast scenarios
DispatchProgram	Structured scheduling via operation modules
ModuleInstance	A concrete module with params & switch conditions
ConstraintSet	Collection of hard/soft constraint rules
ObjectiveSet	Weighted multi-criteria evaluation objectives
SimulationResult	Full trajectory with water balance diagnostics
EvaluationResult	Multi-level scoring output

Layer 4: Operation Modules

6 built-in module types:

Module Type	Description
constant_release	Fixed flow release regardless of state
inflow_driven	Release = f(inflow) with ratio + offset
storage_driven	Release based on level deviation from target
combined_driven	Multi-factor conditional logic (flood/drought/normal)
level_tracking	PID-like target water level tracking
external_constraint	Follow a pre-computed release schedule

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:

Slot	Status	Future Use
ModuleHandler	✅ Built-in	Custom operation modules
ConstraintPlugin	✅ Built-in	Custom constraint types
ObjectivePlugin	✅ Built-in	Custom evaluation criteria
SolverPlugin	🔲 Reserved	Multi-objective optimization
CaseRetrievalPlugin	🔲 Reserved	CBR knowledge reuse
PlannerPlugin	🔲 Reserved	Game-theoretic / LLM planning

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

Package	Purpose
numpy	Numerical computation
scipy	Scientific computing
torch	Differentiable models
pydantic	Domain object validation (v2)
fastmcp	MCP tool interface (optional)
pytest	Testing