Docker Composeで環境省のWBGT実況値を取得し、DuckDBへ保存する(r0051) release 0.0.5α Step 1
はじめに
前回の記事は、WBGTリスク評価Webアプリを実データ対応へ進めるための構成を整理しました。
WBGTリスク評価Webアプリを実データ対応へ進めるため、最初にデータ取得部分を作成しました。
今回は、環境省のWBGT実況値を取得し、レスポンスを検証してDuckDBへ保存するところまでです。地図表示はまだ追加しません。
先にデータ取得基盤を分けて確認することで、今後画面を追加したときに、データ取得の問題と表示側の問題を切り分けやすくします。
r0051で実装した内容
- Docker ComposeによるPython 3.13環境
- FastAPI
- 環境省WBGT実況値APIへのアクセス
- 実況推定値と実況実測値の取得
- 日時、品質コード、測定値の検証
- 欠測値の正規化
- DuckDBへの一括UPSERT
- 取得ログの保存
- APSchedulerによる定期取得
- 最新値、時系列、取得状態、取得ログのAPI
- サンプルJSONによるオフライン確認
- pytestによる単体・結合テスト
今回、ソースコードには日本語のコメントとdocstringを追加しました。処理内容だけでなく、なぜその実装にしたのかが分かるようにしています。
フォルダー構成
WBGT_Risk_WebApp_r0051/
├─ docker-compose.yml
├─ .env
├─ README.md
├─ start_server.bat
├─ fetch_now_live.bat
├─ fetch_now_sample.bat
│
├─ backend/
│ ├─ Dockerfile
│ ├─ requirements.txt
│ ├─ app/
│ │ ├─ main.py
│ │ ├─ config.py
│ │ ├─ api/
│ │ ├─ clients/
│ │ ├─ core/
│ │ ├─ jobs/
│ │ ├─ repositories/
│ │ ├─ schemas/
│ │ └─ services/
│ ├─ sample_data/
│ └─ tests/
│
├─ data/
├─ docs/
└─ scripts/
処理の流れ
1. 取得期間を決める
2. fetch_logへrunningを登録する
3. 環境省APIまたはサンプルJSONを読み込む
4. レスポンス全体を検証する
5. 各実況値を検証・正規化する
6. 入力内の重複を整理する
7. DuckDBへUPSERTする
8. fetch_logをsuccessまたはerrorへ更新する
設定
.envで取得条件を指定します。
WBGT_SOURCE_MODE=live
WBGT_DATA_TYPES=0,1
WBGT_LOCATION_TYPE=3
WBGT_LOOKBACK_HOURS=6
WBGT_SCHEDULER_ENABLED=true
WBGT_FETCH_INTERVAL_MINUTES=60
WBGT_DATA_TYPESは、0が実況推定値、1が実況実測値です。
WBGT_LOCATION_TYPEは次のように指定します。
| 値 | 内容 |
|---|---|
| 1 | 地点別 |
| 2 | 都道府県別 |
| 3 | 全地点 |
APIクライアント
複数のdata_typeを送る場合は、同じパラメータ名を繰り返します。そのため、クエリパラメータは辞書ではなくタプル配列で作っています。
def build_query_params(self, date_from: datetime, date_to: datetime):
"""設定と取得期間から環境省APIのクエリパラメータを組み立てる。"""
params = []
params.extend(("data_type", str(value)) for value in self.settings.data_types)
params.append(("location_type", str(self.settings.location_type)))
params.extend([
("date_from", date_from.strftime("%Y%m%d%H%M%S")),
("date_to", date_to.strftime("%Y%m%d%H%M%S")),
])
return params
レスポンスの検証
APIから返った値は、そのまま保存しません。
-
statusが正常か -
countとdataの件数が一致するか - 日時をJSTとして変換できるか
- 品質コードが0～4か
- 数値文字列をfloatへ変換できるか
- 欠測表現か
- WBGTや温度が想定範囲内か
欠測値はNoneへ統一します。
_MISSING_TOKENS = {"", "-", "--", "null", "none", "nan"}
DuckDBへの保存
主キーは次の3項目です。
地点番号 + 観測日時 + 実況値種別
同じ期間を重ねて取得しても、同じ主キーのデータは増えません。過去値が更新された場合は、新しい内容へ置き換わります。
保存時は、一時テーブルへまとめて入れてから一括UPSERTします。
検証済みデータ
↓
incoming_observation
↓
INSERT件数を確認
↓
ON CONFLICT DO UPDATE
途中で失敗した場合はROLLBACKします。
起動方法
ZIPファイルをダウンロードして、解凍してください。
解凍後のフォルダーで作業を進めます。
start_server.bat
Swagger UIは次のURLです。
http://localhost:8000/docs
サンプルJSONで確認する
fetch_now_sample.bat
初回は4件INSERTされます。同じ処理をもう一度行うと、4件UPDATEになります。
環境省APIから取得する
fetch_now_live.bat
外部APIへ短時間に繰り返しアクセスしないようにします。
テスト
run_tests.bat
自動テストでは、設定解析、APIクエリ、値の正規化、品質コード、件数不整合、DuckDBのUPSERT、FastAPIとの結合を確認します。
Pythonソースのdocstringは次のスクリプトで確認できます。
python scripts/check_docstrings.py
まとめ
r0051では、環境省のWBGT実況値を取得し、検証後にDuckDBへ保存するデータ取得基盤を作成しました。
次は予測値を保存し、現在時刻を境に実況値と予測値を分けて返す時系列APIへ進めます。
