"""
FastAPI + Rust による 2次元河川不定流計算APIデモ。
構成
----
- FastAPI はHTTP API、入力チェック、ブラウザ用の簡易画面を担当します。
- Rust の river2d_core は、標準入力からJSONを受け取り、計算結果JSONを標準出力へ返します。
- Python側では数値計算を行わず、Rust計算エンジンへの受け渡しだけにします。
起動手順
--------
1. Rust計算エンジンをビルドします。
cd rust/river2d_core
cargo build --release
2. FastAPIからRustバイナリを呼び出せるようにします。
export RIVER2D_BIN="$(pwd)/target/release/river2d_core"
3. FastAPIを起動します。
uvicorn api.fastapi_rust_river_api:app --reload --host 0.0.0.0 --port 8000
4. ブラウザで http://localhost:8000/ を開きます。
"""
from __future__ import annotations
import json
import os
import subprocess
from pathlib import Path
from typing import Any
from fastapi import FastAPI, HTTPException
from fastapi.responses import HTMLResponse
from pydantic import BaseModel, Field
class BreachCondition(BaseModel):
"""簡易破堤条件。
x_start/x_end は河道縦断方向の格子番号です。
right_bank=True の場合、右岸側の堤防を一部低くした地形をRust側で生成します。
"""
enabled: bool = True
x_start: int = 44
x_end: int = 49
right_bank: bool = True
class River2DRequest(BaseModel):
"""2次元河川不定流デモの入力条件。"""
nx: int = Field(72, ge=20, le=220, description="x方向格子数")
ny: int = Field(24, ge=8, le=120, description="y方向格子数")
dx: float = Field(20.0, gt=0.0, description="x方向格子間隔[m]")
dy: float = Field(20.0, gt=0.0, description="y方向格子間隔[m]")
dt: float = Field(1.0, gt=0.0, le=10.0, description="計算時間刻み[s]")
steps: int = Field(2400, ge=1, le=7200, description="計算ステップ数")
manning_n: float = Field(0.035, gt=0.005, le=0.20, description="Manning粗度係数")
bed_slope: float = Field(0.005, gt=0.0, le=0.02, description="縦断河床勾配")
initial_depth: float = Field(0.3, ge=0.01, le=10.0, description="河道内初期水深[m]")
inflow_base_m3s: float = Field(55.0, ge=0.0, description="基底流量[m3/s]")
inflow_peak_m3s: float = Field(230.0, ge=0.0, description="追加ピーク流量[m3/s]")
peak_time_s: float = Field(360.0, ge=1.0, description="流入ピーク時刻[s]")
velocity_limit_ms: float = Field(6.0, gt=0.1, le=20.0, description="数値安定用の流速上限[m/s]")
output_interval: int = Field(10, ge=1, le=600, description="時系列出力間隔[step]")
breach: BreachCondition = Field(default_factory=BreachCondition)
app = FastAPI(
title="FastAPI + Rust River2D Demo",
version="0.1.0",
description="Rustで簡易2次元河川不定流を計算し、FastAPIでJSONを受け渡すデモです。",
)
def find_rust_binary() -> str:
"""Rust計算エンジンのパスを取得します。"""
env_path = os.environ.get("RIVER2D_BIN")
if env_path:
return env_path
# 開発時の標準的な相対パスを探します。
project_root = Path(__file__).resolve().parents[1]
candidates = [
project_root / "rust" / "river2d_core" / "target" / "release" / "river2d_core",
project_root / "rust" / "river2d_core" / "target" / "debug" / "river2d_core",
]
for candidate in candidates:
if candidate.exists():
return str(candidate)
return "river2d_core"
def run_rust_solver(payload: dict[str, Any], timeout_sec: float = 30.0) -> dict[str, Any]:
"""Rust CLIへJSONを渡し、JSON結果を受け取ります。"""
bin_path = find_rust_binary()
try:
completed = subprocess.run(
[bin_path],
input=json.dumps(payload, ensure_ascii=False),
capture_output=True,
text=True,
timeout=timeout_sec,
check=False,
)
except FileNotFoundError as exc:
raise HTTPException(
status_code=500,
detail=(
"Rust計算エンジンが見つかりません。"
"先に `cd rust/river2d_core && cargo build --release` を実行し、"
"`RIVER2D_BIN` にバイナリパスを設定してください。"
),
) from exc
except subprocess.TimeoutExpired as exc:
raise HTTPException(status_code=504, detail="Rust計算がタイムアウトしました。") from exc
if completed.returncode != 0:
raise HTTPException(
status_code=500,
detail={
"message": "Rust計算エンジンでエラーが発生しました。",
"stderr": completed.stderr[-4000:],
},
)
try:
return json.loads(completed.stdout)
except json.JSONDecodeError as exc:
raise HTTPException(
status_code=500,
detail={
"message": "Rust計算エンジンの出力JSONを解析できませんでした。",
"stdout_head": completed.stdout[:1000],
"stderr": completed.stderr[-1000:],
},
) from exc
@app.get("/health")
def health() -> dict[str, str]:
return {"status": "ok"}
@app.get("/api/river2d/sample-request")
def sample_request() -> dict[str, Any]:
return River2DRequest().model_dump()
@app.post("/api/river2d/run")
def run_river2d(request: River2DRequest) -> dict[str, Any]:
"""Rustで2次元不定流デモを実行します。"""
payload = request.model_dump()
return run_rust_solver(payload)
@app.get("/", response_class=HTMLResponse)
def index() -> str:
"""ブラウザからAPI実行を確認するための最小UIです。"""
default_json = json.dumps(River2DRequest().model_dump(), ensure_ascii=False, indent=2)
return f"""
FastAPI + Rust River2D Demo
APIレスポンス
ここに結果JSONが表示されます。
"""
if __name__ == "__main__":
# APIを経由しない疎通確認用です。
payload = River2DRequest().model_dump()
print(json.dumps(run_rust_solver(payload), ensure_ascii=False, indent=2))