Docker でRust + Python + DuckDB + FastAPI + JupyterLabの実効雨量API環境を作成しました
1. はじめに
雨量データを使った防災・水文解析では、過去の雨量の影響を時間とともに減衰させて評価する実効雨量を使うことがあります。
この記事では、Docker Composeで以下をまとめて動かす環境を作ります。
- Python / Rust が使える JupyterLab
- DuckDB による計算履歴保存
- FastAPI による実効雨量API
- APIをNotebookから確認するサンプル
ソース一式は ZIPからダウンロードしてください。rust_python_duckdb_jupyter フォルダーとして構成しています。
2. 実効雨量の式
実効雨量は、現在の雨量に過去の実効雨量を減衰させて足し込む形で計算します。
E_t = R_t + \alpha E_{t-1}
ここで、
\alpha = 0.5^{\frac{\Delta t}{T}}
E_t : 時刻 t の実効雨量
R_t : 時刻 t の雨量
α : 減衰係数
Δt : 計算時間間隔
T : 半減期
例えば、1時間雨量で半減期72時間の場合は、以下になります。
\alpha = 0.5^{\frac{1}{72}}
3. 全体イメージ
- FastAPI：外部から雨量を受け取り、実効雨量を返す
- DuckDB：APIの計算履歴を保存する
- JupyterLab：Python / Rust で計算確認や試作を行う
4. フォルダー構成
rust_python_duckdb_jupyter/
├─ docker-compose.yml
├─ Dockerfile
├─ requirements.txt
├─ .env
├─ README.md
├─ run_all.bat
├─ run_jupyter.bat
├─ run_api.bat
├─ stop_all.bat
├─ duckdb_cli.bat
├─ api_example.bat
├─ api/
│ ├─ README_API.md
│ ├─ app/
│ │ ├─ main.py
│ │ ├─ schemas.py
│ │ ├─ effective_rainfall.py
│ │ └─ db.py
│ └─ tests/
│ └─ test_effective_rainfall.py
├─ notebooks/
│ ├─ 00_python_duckdb_example.ipynb
│ ├─ 01_rust_effective_rainfall.ipynb
│ └─ 02_fastapi_client_example.ipynb
├─ data/
│ ├─ input/
│ │ └─ rain.csv
│ ├─ duckdb/
│ └─ output/
├─ rust/
│ ├─ Cargo.toml
│ └─ src/
│ └─ main.rs
└─ scripts/
├─ check_env.py
└─ check_api_calc.py
5. Docker Composeのサービス構成
主なサービスは以下です。
| サービス | 用途 | ポート |
|---|---|---|
jupyter |
Python / Rust Notebook | 8888 |
api |
実効雨量API | 8000 |
duckdb-cli |
DuckDBをCLIで確認 | 必要時のみ |
6. 起動方法
Windowsの場合は、ZIPを展開して以下を実行します。
run_all.bat
または、コマンドで起動します。
docker compose up --build
JupyterLabは以下で開きます。
http://localhost:8888/lab?token=rustduckdb
FastAPIのSwagger UIは以下です。
http://localhost:8000/docs
7. 実効雨量計算
Rustでの最小実装を行います。
これを拡張して、複数の半減期、複数雨量時系列を計算できるようにします。
/// 半減期モデルによる実効雨量を計算する関数
///
/// 実効雨量は、過去の雨量の影響を時間とともに減衰させながら、
/// 現在の雨量に加算して求める雨量指標です。
///
/// 計算式:
///
/// E_t = R_t + α * E_{t-1}
///
/// ここで、
/// - E_t : 時刻 t の実効雨量 [mm]
/// - R_t : 時刻 t の雨量 [mm]
/// - α : 減衰係数
/// - E_{t-1} : 1ステップ前の実効雨量 [mm]
///
/// 減衰係数 α は、半減期から以下で求めます。
///
/// α = 0.5 ^ (dt_hours / half_life_hours)
///
/// # 引数
///
/// - rain_mm: 時系列雨量 [mm]
/// - dt_hours: データ間隔 [hour]
/// - half_life_hours: 半減期 [hour]
///
/// # 戻り値
///
/// - 各時刻の実効雨量 [mm]
fn calc_effective_rainfall(
rain_mm: &[f64],
dt_hours: f64,
half_life_hours: f64,
) -> Vec<f64> {
// 半減期から減衰係数 alpha を計算する。
// 半減期が長いほど alpha は 1 に近くなり、過去の雨量の影響が残りやすくなる。
let alpha = 0.5_f64.powf(dt_hours / half_life_hours);
// 計算結果を格納する配列。
// 入力雨量と同じ長さになるため、あらかじめ容量を確保しておく。
let mut result = Vec::with_capacity(rain_mm.len());
// 1つ前の時刻までの実効雨量。
// 初期状態では、過去の雨量影響はないものとして 0.0 から開始する。
let mut effective = 0.0_f64;
// 時系列雨量を先頭から順番に処理する。
for &rain in rain_mm {
// 欠損値、無限大、負の雨量は 0.0 として扱う。
// 実務では、欠測値処理を別途行う場合もある。
let r = if rain.is_finite() && rain > 0.0 {
rain
} else {
0.0
};
// 実効雨量を更新する。
//
// 現在雨量 r に、1ステップ前の実効雨量 effective を
// 減衰係数 alpha で減衰させた値を加える。
effective = r + alpha * effective;
// 現在時刻の実効雨量を結果配列に追加する。
result.push(effective);
}
// 計算結果を返す。
result
}
fn main() {
// 例: 1時間雨量 [mm]
//
// ここでは、9時間分の雨量データを仮定する。
// 実際にはCSVやAPIから取得した雨量データを渡すこともできる。
let rain = vec![
0.0, 0.0, 5.0, 10.0, 20.0, 0.0, 0.0, 3.0, 0.0,
];
// 計算条件
//
// dt_hours:
// データ間隔。ここでは1時間雨量なので 1.0。
//
// half_life_hours:
// 半減期。ここでは長期実効雨量の例として 72時間を指定。
let dt_hours = 1.0;
let half_life_hours = 72.0;
// 半減期72時間の実効雨量を計算する。
let effective_72h = calc_effective_rainfall(
&rain,
dt_hours,
half_life_hours,
);
// 計算結果をCSV形式に近い形で標準出力へ表示する。
println!("時刻, 雨量[mm], 実効雨量[mm]");
// rain と effective_72h を同時に取り出し、時刻番号 i と一緒に表示する。
for (i, (r, e)) in rain.iter().zip(effective_72h.iter()).enumerate() {
println!("{}, {:.2}, {:.2}", i, r, e);
}
}
8. APIで実効雨量を計算する
エンドポイントは以下です。
POST /api/v1/effective-rainfall
リクエスト例です。
{
"rain_mm": [0, 0, 5, 10, 20, 0, 3],
"dt_hours": 1.0,
"half_life_hours": 72.0,
"initial_effective_mm": 0.0,
"save_to_duckdb": true,
"job_name": "sample"
}
レスポンスでは、減衰係数 alpha、実効雨量時系列、最大実効雨量、雨量合計などを返します。
{
"alpha": 0.9904191474668262,
"dt_hours": 1.0,
"half_life_hours": 72.0,
"count": 7,
"rain_total_mm": 38.0,
"effective_max_mm": 37.14514974475508,
"effective_rainfall_mm": [0.0, 0.0, 5.0, 14.952, 34.809, 34.475, 37.145]
}
Windowsから簡単に確認する場合は、API起動後に以下を実行します。
api_example.bat
9. 複数の半減期をまとめて計算する
短期・長期の実効雨量を同時に計算したい場合は、以下を使います。
POST /api/v1/effective-rainfall/multi
{
"rain_mm": [0, 0, 5, 10, 20, 0, 3],
"dt_hours": 1.0,
"half_life_hours_list": [1.5, 72.0]
}
短期実効雨量と長期実効雨量を同時に比較できます。
10. DuckDBで履歴を確認する
save_to_duckdb を true にすると、計算履歴は以下に保存されます。
data/duckdb/effective_rainfall.duckdb
APIから履歴を確認できます。
GET /api/v1/history
DuckDB CLIを使う場合は以下です。
duckdb_cli.bat
11. Notebookで確認する
Notebookは以下を用意しています。
| Notebook | 内容 |
|---|---|
00_python_duckdb_example.ipynb |
PythonからDuckDBを操作する例 |
01_rust_effective_rainfall.ipynb |
Rustで実効雨量を計算する例 |
02_fastapi_client_example.ipynb |
NotebookからFastAPIを呼び出す例 |
PythonとRustの両方で計算確認できるため、プロトタイプ作成と高速化検討を同じ環境で行えます。
12. 実運用に向けた注意点
今回の構成はローカル開発用です。公開APIとして使う場合は、以下を追加した方がよいです。
- CORSの制限
- APIキーや認証
- 入力データサイズの制限
- ログ管理
- Dockerイメージのバージョン固定
- DuckDBファイルのバックアップ
13. まとめ
Docker Composeを使うことで、以下を1つの環境にまとめられました。
- Python Notebook
- Rust Notebook
- FastAPI
- DuckDB
- 実効雨量計算API
実効雨量の計算ロジックはシンプルですが、API化しておくことで、Webアプリ、GIS、バッチ処理、観測雨量データ処理へ拡張しやすくなります。
14. 実効雨量に関する参考リンク集
実効雨量そのもの、関連する土砂災害指標、実務での使われ方を調べるときは、以下が参考になります。
| 分類 | リンク | メモ |
|---|---|---|
| 実効雨量の基本 | 防災科学技術研究所：実効雨量について | 実効雨量の考え方、重み付き積算式、半減期 1.5時間・72時間 の説明があり、本記事の式を理解するのに最も参考になります。 |
| 関連指標 | 気象庁：土壌雨量指数 | 実効雨量とは別の指標ですが、過去の雨による土壌中の水分量を評価する考え方を理解するのに役立ちます。 |
| 防災情報での利用 | 気象庁：土砂キキクル | 土壌雨量指数を使った土砂災害危険度情報の見方を確認できます。 |
| 警戒基準の考え方 | 国土交通省：土砂災害警戒情報の基準設定及び検証の考え方 | 短期降雨指標と長期降雨指標を組み合わせた土砂災害警戒情報の基準設定の考え方を確認できます。 |
| 基準雨量の技術資料 | 国土技術政策総合研究所：土砂災害警戒避難基準雨量の設定手法 | 実効雨量や警戒避難基準雨量の背景を深く調べたい場合に参考になります。 |
| 実効雨量の利用例 | 高知県：土砂災害危険度情報 用語集 | 実効雨量、半減期、スネークライン図など、自治体システムでの使われ方を確認できます。 |
実務で使う場合は、実効雨量だけで危険度を判断するのではなく、地形・地質、既往災害、土砂災害警戒区域、気象庁や自治体の防災情報と合わせて確認する必要があります。