🍣
過去の気象データをCSVファイルとしてダウンロードし保存する
概要
pytestを使って過去の（地点）気象データ・ダウンロードからCSVファイルを取得する。
前提
- python3.12以降（dataclassが使えれば同じ挙動になるので、python3.7以降なら動くと思われる）
- pytestをインストール済み
CSVファイルの出力内容について
公式ページに書いてある
指定するパラメータについて
画面操作時にブラウザの開発者ツールからHTTP要求として参照できる
地点番号は気象庁で共通のようなので、（名前が似ているが別ページの）過去の気象データ検索のURLから拾うと良い。次のURLは地点として「八戸」を選んだ場合で、「block_no」から47581を読む。
https://www.data.jma.go.jp/stats/etrn/select/prefecture.php?prec_no=31&block_no=&year=&month=&day=&view=
CSVダウンロードのコード
pytest設定
pytest.ini
[pytest]
# テストファイルのパターン
python_files = test_*.py
python_classes = Test*
python_functions = test_*
# 詳細な出力
addopts = -v --tb=short
本体
HTTPのステータスコードを確認しつつダウンロードする想定により、pytestのテストケースとして作成した。
- session情報を取得するために一度index.phpページを表示する
- データ取得urlに対して、JSONでパラメータを渡す
- うまくパラメータ指定できるとCSVを取得できるので保存する
test_download_csv.py
import pytest
import json
import requests
import time
from pathlib import Path
from dataclasses import dataclass, field, asdict
from typing import List, Any, Tuple
ROOT_URL = 'https://www.data.jma.go.jp/risk/obsdl/index.php'
SHOW_URL = 'https://www.data.jma.go.jp/risk/obsdl/show/table'
@dataclass
class WeatherDataPayload:
"""気象データダウンロード用のペイロード"""
stationNumList: List[str] = field(default_factory=lambda: ["s47581"]) # s47581:八戸
aggrgPeriod: int = 1 # 1:日別 2:(暦日)半旬別 4:旬別 5:月別 6:3か月別 812:2日別(813:3日別を3日おき出力, 823:3日別を毎日連続出力, ...28日別まで同様) 9:時別
elementNumList: List[List[str]] = field(default_factory=lambda: [["201", ""], ["101", ""]])
interAnnualFlag: int = 1 # 1:連続期間 2:年毎の同時期(特定の月日を複数年分)
ymdList: List[str] = field(default_factory=lambda: ["2022", "2022", "9", "10", "30", "1"]) # 2022/9/30-2022/10/1
optionNumList: List[Any] = field(default_factory=list)
downloadFlag: str = 'true' # CSVファイルとしてダウンロードする
rmkFlag: int = 1 # 品質情報を含める
disconnectFlag: int = 1 # 統計切断情報を含める
youbiFlag: int = 0 # 1:曜日を含める
fukenFlag: int = 0 # 1:府県を含める
kijiFlag: int = 0 # 1:発生時刻(起時)を含める
huukouFlag: int = 0 # 1:風向を含める
csvFlag: int = 1 # 1:CSV形式(カンマ区切り)にする
jikantaiFlag: int = 0 # 1:時別について時間帯を限定する
jikantaiList: List[Any] = field(default_factory=list) # ["2", "5"] で2時〜5時
ymdLiteral: int = 1 # 1:日付をyyyy/mm/dd形式にする
def to_post_data(self) -> dict:
"""POSTリクエスト用のデータに変換"""
data = {}
for key, value in asdict(self).items():
if isinstance(value, list):
data[key] = json.dumps(value)
else:
data[key] = value
return data
@pytest.fixture(scope="session")
def output_dir():
"""
CSV保存用ディレクトリを作成（全体で1度だけ実行）
"""
output_path = Path("csv_output")
output_path.mkdir(exist_ok=True)
print(f"\n✓ Created output directory: {output_path.absolute()}")
return output_path
@pytest.mark.parametrize("payload,filename", [
pytest.param(
WeatherDataPayload(),
"daily_s47581.csv",
id="daily_station"
),
pytest.param(
WeatherDataPayload(
aggrgPeriod=2,
ymdList=["2022", "2022", "10", "11", "6", "1"] # 2022/10 第6半旬 - 2022/11 第1半旬
),
"mb5_s47581.csv",
id="mb5_station" # montly based 5days
),
pytest.param(
WeatherDataPayload(
aggrgPeriod=4,
ymdList=["2022", "2022", "10", "11", "3", "1"] # 2022/10 下旬(第3旬) - 2022/11 上旬(第1旬)
),
"mb10_s47581.csv",
id="mb10_station" # montly based 10days
),
pytest.param(
WeatherDataPayload(
aggrgPeriod=5,
ymdList=["2022", "2022", "10", "11", "", ""] # 2022/10-2022/11
),
"monthly_s47581.csv",
id="monthly_station"
),
pytest.param(
WeatherDataPayload(
aggrgPeriod=6,
ymdList=["2022", "2022", "10", "11", "", ""] # 2022/10-2022/11
),
"3month_s47581.csv",
id="3month_station"
),
pytest.param(
WeatherDataPayload(
aggrgPeriod=812,
ymdList=["2022", "2022", "10", "11", "30", "1"] # 2022/10/30-2022/11/1
),
"2days_s47581.csv",
id="2days_station"
),
pytest.param(
WeatherDataPayload(aggrgPeriod=9),
"hourly_s47581.csv",
id="hourly_station"
),
pytest.param(
WeatherDataPayload(
stationNumList=["a0179"], # a0179:三戸
aggrgPeriod=1,
ymdList=["2022", "2022", "10", "10", "30", "31"]
),
"daily_a0179.csv",
id="daily_amedas"
),
pytest.param(
WeatherDataPayload(
stationNumList=["s47581"],
aggrgPeriod=1,
elementNumList=[["201", ""], ["101", ""], ["301", ""]],
ymdList=["2022", "2022", "10", "10", "10", "31"]
),
"daily_multiple_elements.csv",
id="daily_multiple_elements"
),
])
def test_download_csv(output_dir, payload, filename):
"""CSVダウンロード共通のテスト関数"""
session = requests.Session()
session.headers.update({
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
})
top = session.get(ROOT_URL)
time.sleep(2)
assert top.status_code == 200
csv = session.post(
SHOW_URL,
data=payload.to_post_data(),
headers={'Referer': ROOT_URL}
)
time.sleep(2)
assert csv.status_code == 200
assert len(csv.content) > 0
output_file = output_dir / filename
with open(output_file, 'wb') as f:
f.write(csv.content)
print(f"\n✓ CSV saved: {filename} ({len(csv.content)} bytes)")
動かし方
pytest test_download_csv.py
動かすとoutput_dirフィクスチャに従ってcsv_outputディレクトリを作成し、ダウンロードファイルを保存する。
実行例
% pytest test_basic.py
====================================================================================================================== test session starts ======================================================================================================================
platform darwin -- Python 3.12.9, pytest-8.4.0, pluggy-1.5.0 -- /opt/homebrew/Caskroom/miniforge/base/bin/python
cachedir: .pytest_cache
rootdir: /Users/issei/programing/risk
configfile: pytest.ini
collected 9 items
test_basic.py::test_download_csv[daily_station] PASSED [ 11%]
test_basic.py::test_download_csv[mb5_station] PASSED [ 22%]
test_basic.py::test_download_csv[mb10_station] PASSED [ 33%]
test_basic.py::test_download_csv[monthly_station] PASSED [ 44%]
test_basic.py::test_download_csv[3month_station] PASSED [ 55%]
test_basic.py::test_download_csv[2days_station] PASSED [ 66%]
test_basic.py::test_download_csv[hourly_station] PASSED [ 77%]
test_basic.py::test_download_csv[daily_amedas] PASSED [ 88%]
test_basic.py::test_download_csv[daily_multiple_elements] PASSED [100%]
====================================================================================================================== 9 passed in 39.18s =======================================================================================================================
留意事項
言うまでもなく、提供元サーバへ過度の負荷を掛けることは望ましく有りません。
この実装では、sleepを用いてダウンロード要求の頻度を調整しています。
Discussion