参考ページ
はじめに
仕事で気象庁APIを取り扱った。goドメインで信頼はできそうだが、クセが強いうえにまともにドキュメントがないのでまとめる。これは共有すべきナレッジだな
データの観察とマスタデータの作成
まずはマスタをつくっていく。このプログラムはいわゆるシーダーデータを作成し、その後に控える予報データ取得プログラムが、それを参照し、データを処理していく（以下のurlを見ると .../const/... と書いてあるね）
https://www.jma.go.jp/bosai/common/const/area.json
https://www.jma.go.jp/bosai/forecast/const/forecast_area.json
https://www.jma.go.jp/bosai/amedas/const/amedastable.json
| プロパティ | 階層 |
|---|---|
| centers | 地方予報区(11区) |
| offices | 府県予報区(58区) |
| class10s | 一次細分区域(142区) |
| class15s | 市町村等をまとめた地域 |
| class20s | 二次細分区域 |
jma_areas1
いわゆる地方区分。生データでは center という名前で取り扱われているので、area と名付けた。
| id | name |
|---|---|
| 010600 | 近畿地方 |
jma_areas2
いわゆる都道府県（北海道と沖縄がより分割されている）。生データでは office という名前で取り扱われているので、prefecture と名付けた。
| id | jmaAreas1Id | name |
|---|---|---|
| 250000 | 010600 | 滋賀県 |
| 260000 | 010600 | 京都府 |
| 270000 | 010600 | 大阪府 |
| 280000 | 010600 | 兵庫県 |
| 290000 | 010600 | 奈良県 |
| 300000 | 010600 | 和歌山県 |
jma_areas3
ひとつの都道府県のある範囲（北部とか南部）。生データでは class10 という名前で取り扱われているので region と名付けた。
| id | jmaAreas2Id | name |
|---|---|---|
| 280010 | 280000 | 南部 |
| 280020 | 280000 | 北部 |
jma_areas4
ひとつの都道府県の市区町村（一部市町村を分割している）。生データでは class20 という名前で取り扱われているので city と名付けた。
class15（余計な分割がない市区町村）もあるんだけど、joinのときしか使わない
| no. | id | jmaAreas2Id | jmaAreas3Id | name | region |
|---|---|---|---|---|---|
| 1 | 2810000 | 280000 | 280010 | 神戸市 | 南部 |
| 2 | 2820200 | 280000 | 280010 | 尼崎市 | 南部 |
| 3 | 2820400 | 280000 | 280010 | 西宮市 | 南部 |
| 4 | 2820600 | 280000 | 280010 | 芦屋市 | 南部 |
| 5 | 2820700 | 280000 | 280010 | 伊丹市 | 南部 |
| 6 | 2821400 | 280000 | 280010 | 宝塚市 | 南部 |
| 7 | 2821700 | 280000 | 280010 | 川西市 | 南部 |
| 8 | 2821900 | 280000 | 280010 | 三田市 | 南部 |
| 9 | 2830100 | 280000 | 280010 | 猪名川町 | 南部 |
| 10 | 2821300 | 280000 | 280010 | 西脇市 | 南部 |
| 11 | 2822100 | 280000 | 280010 | 丹波篠山市 | 南部 |
| 12 | 2822300 | 280000 | 280010 | 丹波市 | 南部 |
| 13 | 2836500 | 280000 | 280010 | 多可町 | 南部 |
| 14 | 2822700 | 280000 | 280010 | 宍粟市 | 南部 |
| 15 | 2844200 | 280000 | 280010 | 市川町 | 南部 |
| 16 | 2844300 | 280000 | 280010 | 福崎町 | 南部 |
| 17 | 2844600 | 280000 | 280010 | 神河町 | 南部 |
| 18 | 2850100 | 280000 | 280010 | 佐用町 | 南部 |
| 19 | 2820300 | 280000 | 280010 | 明石市 | 南部 |
| 20 | 2821000 | 280000 | 280010 | 加古川市 | 南部 |
| 21 | 2821500 | 280000 | 280010 | 三木市 | 南部 |
| 22 | 2821600 | 280000 | 280010 | 高砂市 | 南部 |
| 23 | 2821800 | 280000 | 280010 | 小野市 | 南部 |
| 24 | 2822000 | 280000 | 280010 | 加西市 | 南部 |
| 25 | 2822800 | 280000 | 280010 | 加東市 | 南部 |
| 26 | 2838100 | 280000 | 280010 | 稲美町 | 南部 |
| 27 | 2838200 | 280000 | 280010 | 播磨町 | 南部 |
| 28 | 2820100 | 280000 | 280010 | 姫路市 | 南部 |
| 29 | 2820800 | 280000 | 280010 | 相生市 | 南部 |
| 30 | 2821200 | 280000 | 280010 | 赤穂市 | 南部 |
| 31 | 2822900 | 280000 | 280010 | たつの市 | 南部 |
| 32 | 2846400 | 280000 | 280010 | 太子町 | 南部 |
| 33 | 2848100 | 280000 | 280010 | 上郡町 | 南部 |
| 34 | 2820500 | 280000 | 280010 | 洲本市 | 南部 |
| 35 | 2822400 | 280000 | 280010 | 南あわじ市 | 南部 |
| 36 | 2822600 | 280000 | 280010 | 淡路市 | 南部 |
jma_amedas
いわゆる気象観測所。生データでは amedas という名前で取り扱われているのでそのまま amedas と名付けた。
| id | jmaAreas3Id |
|---|---|
| 63518 | 280010 |
| 63576 | 280010 |
| 63571 | 280010 |
| 63383 | 280010 |
クセ強ポイント
- 「本日」の予報データは時間の経過とともに値の要素数が減っていく（あたりまえっちゃあたりまえか...）
- 天気予報
280000.jsonは都道府県単位で取れる- 神戸は南部なので、せめて南部の気温を抽出したい
- ということは南部のアメダスのリストを把握する必要がある
- 市区町村別のアメダスリスト
forecast_area.jsonはリージョン単位で収録されている- アメダスidは市区町村と紐づかない ←
- アメダスidは市区町村と紐づかない ←
forecast_area.json
{
:
"280000": [
{
"class10": "280010", ←region: 南部
"amedas": [
"63518", ←amedas: 神戸
"63576", ←amedas: （欠番！）
"63571", ←amedas: 洲本
"63383" ←amedas: 姫路
],
"class20": "2810000" ←city: 神戸市（南部代表都市）
},
{
"class10": "280020", ←region: 北部
"amedas": [
"63051" ←amedas: 豊岡
],
"class20": "2820900" ←city: 豊岡市（北部代表都市）
}
]
:
}
280000.json（できれば市区町村単位で天気を取得したい）
[
{
"publishingOffice": "神戸地方気象台",
"reportDatetime": "2024-03-28T17:00:00+09:00",
"timeSeries": [
{
"areas": [
{
"area": {"name": "神戸", "code": "63518"}, ←region: 南部
"temps": ["13", "19"]
},
{
"area": {"name": "豊岡", "code": "63051"}, ←region: 北部
"temps": ["12", "22"]
},
{
"area": {"name": "洲本", "code": "63571"}, ←region: 南部
"temps": ["14", "20"]
},
{
"area": {"name": "姫路", "code": "63383"}, ←region: 南部
"temps": ["14", "21"]
}
]
}
]
}
:
]
amedastable.json（アメダスの情報から市区町村と紐づかない。。。）
{
"63051": {
"type": "B",
"elems": "11111111",
"lat": [35, 32.1],
"lon": [134, 49.3],
"alt": 3,
"kjName": "豊岡",
"knName": "トヨオカ",
"enName": "Toyooka"
},
"63518": {
"type": "A",
"elems": "11111111",
"lat": [34, 41.8],
"lon": [135, 12.7],
"alt": 5,
"kjName": "神戸",
"knName": "コウベ",
"enName": "Kobe"
},
"63571": {
"type": "B",
"elems": "11111011",
"lat": [34, 18.6],
"lon": [134, 50.9],
"alt": 69,
"kjName": "洲本",
"knName": "スモト",
"enName": "Sumoto"
},
"63383": {
"type": "B",
"elems": "11111011",
"lat": [34, 50.3],
"lon": [134, 40.2],
"alt": 38,
"kjName": "姫路",
"knName": "ヒメジ",
"enName": "Himeji"
}
}
もう一度この絵を見てみよう。姫路市を選んで表示しても神戸市を選んで表示しても3つの都市が必ず表示される。そして、表示された3つの都市以外の兵庫県南部の都市（南あわじ市 など）を表示してもこの3つの都市が必ず表示される。このことからもアメダスと市区町村が紐づかず、結局リージョン単位なんだということがわかる。
jma_areas4（市区町村）の「神戸市」の「市」を除外して、アメダスマスタの kjName とアテればまぁ1対1の市区町村で天気が出るか...？でも除外する字が「市」とは限らんぞ？それに、じゃあ「南あわじ市」などは null になるよな。そうなるとやっぱりリージョン単位なんだな。それ以上のデータがほしければ金払って民間APIだね
| no. | id | jmaAreas2Id | jmaAreas3Id | name | region |
|---|---|---|---|---|---|
| 1 | 2810000 | 280000 | 280010 | 神戸「市」 ←これ | 南部 |
| 35 | 2822400 | 280000 | 280010 | 南あわじ市 | 南部 |
アプリからの使われかたを考える
アプリにおける使われ方としては、建物Aを登録するときに、市区町村を気象庁APIベースのマスタからドロップダウンで選択させるのがいいだろうな。
マスタを保存する仕組みを作ろう
create root directory
console
mkdir jma_weather
cd jma_weather
venv
console
python -m venv venv311
create project
console
pip install django
django-admin startproject config .
create app
console
python manage.py startapp weather
update settings
config/settings.py
INSTALLED_APPS = [
:
+ "weather",
]
create model
weather/models.py
from django.db import models
class JmaAreas1(models.Model):
"""地方区分。生データでは center という名前で取り扱われている"""
id = models.CharField(primary_key=True, max_length=6)
name = models.CharField(max_length=100)
class JmaAreas2(models.Model):
"""都道府県。生データでは office という名前で取り扱われている"""
id = models.CharField(primary_key=True, max_length=6)
jma_area1 = models.ForeignKey(JmaAreas1, on_delete=models.CASCADE)
name = models.CharField(max_length=100)
class JmaAreas3(models.Model):
"""リージョン。生データでは class10 という名前で取り扱われている"""
id = models.CharField(primary_key=True, max_length=6)
jma_area2 = models.ForeignKey(JmaAreas2, on_delete=models.CASCADE)
name = models.CharField(max_length=100)
class JmaAreas4(models.Model):
"""市区町村。生データでは class20 という名前で取り扱われている"""
id = models.CharField(primary_key=True, max_length=7)
jma_area2 = models.ForeignKey(JmaAreas2, on_delete=models.CASCADE)
jma_area3 = models.ForeignKey(JmaAreas3, on_delete=models.CASCADE)
name = models.CharField(max_length=100)
class JmaAmedas(models.Model):
"""気象観測所。生データでは amedas という名前で取り扱われている"""
id = models.CharField(primary_key=True, max_length=5)
jma_area3 = models.ForeignKey(JmaAreas3, on_delete=models.CASCADE)
class JmaWeather(models.Model):
jma_areas3 = models.OneToOneField(
JmaAreas3, primary_key=True, on_delete=models.CASCADE
)
weather_code = models.CharField(max_length=3)
temperature_min = models.FloatField()
temperature_max = models.FloatField()
wind_speed = models.FloatField()
class JmaWarning(models.Model):
jma_areas3 = models.OneToOneField(
JmaAreas3, primary_key=True, on_delete=models.CASCADE
)
warnings = models.CharField(max_length=100)
migration
python manage.py makemigrations weather
python manage.py migrate
create update_jma_master.py
weather/management/commands/update_jma_master.py
import sys
from io import StringIO
import pandas as pd
import requests
from django.core.management.base import BaseCommand
from weather.models import JmaAreas1, JmaAreas2, JmaAreas3, JmaAreas4, JmaAmedas
class Command(BaseCommand):
help = "master update"
def handle(self, *args, **options):
url = "https://www.jma.go.jp/bosai/common/const/area.json"
try:
# URLからデータを取得します
response = requests.get(url)
response.raise_for_status()
except requests.exceptions.RequestException as e:
print(
f"データの取得でエラーが発生しました。URL: {url} エラー詳細: {e}",
file=sys.stderr,
)
sys.exit(1)
raw_data = pd.read_json(StringIO(response.text))
df_areas = pd.DataFrame({"area_name": []})
for code, item in raw_data["centers"].dropna().items():
df_areas.loc[f"{code:06}"] = [item["name"]]
df_prefs = pd.DataFrame({"area_code": [], "pref_name": []})
for code, item in raw_data["offices"].dropna().items():
df_prefs.loc[f"{code:06}"] = [item["parent"], item["name"]]
df_regions = pd.DataFrame({"pref_code": [], "region_name": []})
for code, item in raw_data["class10s"].dropna().items():
df_regions.loc[f"{code:06}"] = [item["parent"], item["name"]]
df_class15s = pd.DataFrame({"region_code": [], "class15_name": []})
for code, item in raw_data["class15s"].dropna().items():
df_class15s.loc[f"{code:06}"] = [item["parent"], item["name"]]
df_cities = pd.DataFrame({"class15_code": [], "city_name": []})
for code, item in raw_data["class20s"].dropna().items():
df_cities.loc[f"{code:07}"] = [item["parent"], item["name"]]
df_cities = (
df_cities.merge(
df_class15s, left_on="class15_code", right_index=True, how="outer"
)
.merge(df_regions, left_on="region_code", right_index=True)
.merge(df_prefs, left_on="pref_code", right_index=True)
.merge(df_areas, left_on="area_code", right_index=True)
)
df_cities.drop(
["area_code", "area_name", "class15_code", "class15_name"],
axis=1,
inplace=True,
)
# jma_areas1: 010600 近畿地方
df_areas.index.name = "id"
df_areas.rename(columns={"area_name": "name"}, inplace=True)
dict_centers = df_areas.reset_index().to_dict("records")
JmaAreas1.objects.all().delete()
JmaAreas1.objects.bulk_create(
[JmaAreas1(id=item["id"], name=item["name"]) for item in dict_centers]
)
# jma_areas2: 280000 兵庫県
df_prefs.index.name = "id"
df_prefs.rename(
columns={"area_code": "jmaAreas1Id", "pref_name": "name"}, inplace=True
)
dict_prefs = df_prefs.reset_index().to_dict("records")
JmaAreas2.objects.all().delete()
JmaAreas2.objects.bulk_create(
[
JmaAreas2(
id=item["id"], jma_area1_id=item["jmaAreas1Id"], name=item["name"]
)
for item in dict_prefs
]
)
# jma_areas3: 280010 南部
df_regions.index.name = "id"
df_regions.rename(
columns={"pref_code": "jmaAreas2Id", "region_name": "name"}, inplace=True
)
dict_regions = df_regions.reset_index().to_dict("records")
JmaAreas3.objects.all().delete()
JmaAreas3.objects.bulk_create(
[
JmaAreas3(
id=item["id"], jma_area2_id=item["jmaAreas2Id"], name=item["name"]
)
for item in dict_regions
]
)
# jma_areas4: 2820100 姫路市
df_cities.index.name = "id"
df_cities.drop(["pref_name"], axis=1, inplace=True)
df_cities.rename(
columns={
"pref_code": "jmaAreas2Id",
"region_code": "jmaAreas3Id",
"region_name": "regionName",
"city_name": "name",
},
inplace=True,
)
df_cities = df_cities.reindex(["jmaAreas2Id", "jmaAreas3Id", "name"], axis=1)
dict_cities = df_cities.reset_index().to_dict("records")
JmaAreas4.objects.all().delete()
JmaAreas4.objects.bulk_create(
[
JmaAreas4(
id=item["id"],
jma_area2_id=item["jmaAreas2Id"],
jma_area3_id=item["jmaAreas3Id"],
name=item["name"],
)
for item in dict_cities
]
)
# 2: from forecast_area.json
url = "https://www.jma.go.jp/bosai/forecast/const/forecast_area.json"
try:
# URLからデータを取得します
response = requests.get(url)
response.raise_for_status()
except requests.exceptions.RequestException as e:
print(
f"データの取得でエラーが発生しました。URL: {url} エラー詳細: {e}",
file=sys.stderr,
)
sys.exit(1)
try:
raw_data = response.json() # response.text から JSON データを直接取得します
except ValueError:
print("JSONデコードエラー", file=sys.stderr)
sys.exit(1)
forecast_areas = []
for key, value in raw_data.items():
for item in value:
for amedas in item["amedas"]:
forecast_areas.append(
{
"id": amedas,
"class10_code": item["class10"],
"class20_code": item["class20"],
}
)
df_amedas = pd.DataFrame(forecast_areas).set_index("id")
df_amedas = df_amedas.merge(df_cities, left_on="class20_code", right_index=True)
df_amedas.drop(
["class10_code", "class20_code", "jmaAreas2Id", "name"],
axis=1,
inplace=True,
)
dict_amedas = df_amedas.reset_index().to_dict("records")
JmaAmedas.objects.all().delete()
JmaAmedas.objects.bulk_create(
[
JmaAmedas(
id=item["id"],
jma_area3_id=item["jmaAreas3Id"],
)
for item in dict_amedas
]
)
self.stdout.write(
self.style.SUCCESS("The master data update has been completed.")
)
console
jma_weather> python manage.py update_jma_master
The master data update has been completed.
予報と警報を保存する仕組みを作ろう
create fetch_weather_forecast.py
weather/management/commands/fetch_weather_forecast.py
from datetime import date, timedelta, datetime
import requests
from django.core.management.base import BaseCommand
from weather.models import JmaAmedas, JmaWeather
FORECASTS_3DAYS = 0
FORECASTS_OVERVIEW = 0
FORECASTS_TEMPERATURE = 2
PROBABILITY_MAX_WIND_SPEED = 3
PROBABILITY_LAND = 0
class RegionWeather:
def __init__(self, region_code: str, region_name: str, weather_code: str):
self.region_code = region_code
self.region_name = region_name
self.weather_code = weather_code
def __str__(self):
return f"Region {self.region_name}({self.region_code}), Weather: {self.weather_code}"
class AmedasTemperature:
def __init__(
self, amedas_code: str, amedas_name: str, min_temps: int, max_temps: int
):
self.amedas_code = amedas_code
self.amedas_name = amedas_name
self.min_temps = min_temps
self.max_temps = max_temps
def __str__(self):
return f" {self.amedas_name}({self.amedas_code} の朝の最低気温: {self.min_temps}, 日中の最高気温: {self.max_temps})"
class RegionTemperature:
def __init__(
self, region_code: str, region_name: str, data: dict, target_date: date
):
self.region_code = region_code
self.region_name = region_name
self.data = data
amedas_ids = [
amedas.id for amedas in JmaAmedas.objects.filter(jma_area3_id=region_code)
]
min_temps_idx, max_temps_idx = self.get_indexes_from_time_defines(
data["timeDefines"], target_date
)
min_temps_list, max_temps_list = self.get_temps_list(
data["areas"], amedas_ids, min_temps_idx, max_temps_idx
)
self.avg_min_temps = round(sum(min_temps_list) / len(min_temps_list), 1)
self.avg_max_temps = round(sum(max_temps_list) / len(max_temps_list), 1)
@staticmethod
def get_indexes_from_time_defines(time_defines: list[str], target_date: date):
time_defines = [
datetime.fromisoformat(date_str).date() for date_str in time_defines
]
return [
i
for i, time_define in enumerate(time_defines)
if time_define == target_date
]
@staticmethod
def get_temps_list(
amedas_data,
target_amedas_ids: list[str],
min_temps_idx: int,
max_temps_idx: int,
):
min_temps_list, max_temps_list = [], []
for amedas in amedas_data:
amedas_temperature = AmedasTemperature(
amedas["area"]["code"],
amedas["area"]["name"],
int(amedas["temps"][min_temps_idx]),
int(amedas["temps"][max_temps_idx]),
)
if amedas_temperature.amedas_code not in target_amedas_ids:
continue
min_temps_list.append(amedas_temperature.min_temps)
max_temps_list.append(amedas_temperature.max_temps)
return min_temps_list, max_temps_list
def __str__(self):
return f"Avg Min: {self.avg_min_temps}℃, Avg Max: {self.avg_max_temps}℃"
class RegionWindSpeed:
def __init__(self, region_code: str, data: dict, target_indexes: list[int]):
self.region_code = region_code
self.data = data
self.avg_wind_speed = self.calc_wind_speed(data, target_indexes)
@staticmethod
def calc_wind_speed(wind_data, target_indexes):
wind_values = [
int(time_cell["locals"][PROBABILITY_LAND]["value"])
for i, time_cell in enumerate(wind_data)
if i in target_indexes
]
return round(sum(wind_values) / len(wind_values), 1)
def __str__(self):
return f"{self.region_code} の最大風速（日中平均）は {self.avg_wind_speed}"
class RegionForecastResults:
def __init__(
self,
region_weather: RegionWeather,
region_temperature: RegionTemperature,
region_wind_speed: RegionWindSpeed,
):
self.region_weather = region_weather
self.region_temperature = region_temperature
self.region_wind_speed = region_wind_speed
def __str__(self):
region_code = self.region_weather.region_code
region_name = self.region_weather.region_name
forecast = {
"weather_code": self.region_weather.weather_code,
"temperature": (
self.region_temperature.avg_min_temps,
self.region_temperature.avg_max_temps,
),
"wind_speed": self.region_wind_speed.avg_wind_speed,
}
return f"{region_code}({region_name}): {forecast}"
class Command(BaseCommand):
help = "get weather forecast"
def handle(self, *args, **options):
today = datetime.now().date()
tomorrow = today + timedelta(days=1)
# TODO: facilityテーブルから areas2_ids を取得
jma_areas2_ids = ["280000", "050000", "130000"]
if not jma_areas2_ids:
raise Exception("facility is empty")
JmaWeather.objects.all().delete()
for prefecture_id in jma_areas2_ids:
forecasts_by_region = {}
forecasts = requests.get(
f"https://www.jma.go.jp/bosai/forecast/data/forecast/{prefecture_id}.json"
).json()
overview = forecasts[FORECASTS_3DAYS]["timeSeries"][FORECASTS_OVERVIEW]
time_defines = [
datetime.fromisoformat(date_str).date()
for date_str in overview["timeDefines"]
]
try:
tomorrow_idx = time_defines.index(tomorrow)
for a_region in overview["areas"]:
region_code = a_region["area"]["code"]
region_weather = RegionWeather(
region_code,
a_region["area"]["name"],
a_region["weatherCodes"][tomorrow_idx],
)
forecasts_by_region.setdefault(region_code, {})[
"weather"
] = region_weather
region_temperature = RegionTemperature(
region_code,
a_region["area"]["name"],
forecasts[FORECASTS_3DAYS]["timeSeries"][FORECASTS_TEMPERATURE],
tomorrow,
)
forecasts_by_region.setdefault(region_code, {})[
"temperature"
] = region_temperature
except ValueError:
print(f"{prefecture_id}: no forecast")
probabilities = requests.get(
f"https://www.jma.go.jp/bosai/probability/data/probability/{prefecture_id}.json"
).json()
tomorrow_indexes = [
i
for i, date_str in enumerate(
probabilities[0]["timeSeries"][1]["timeDefines"]
)
if datetime.fromisoformat(date_str).date() == tomorrow
]
for a_region in probabilities[0]["timeSeries"][1]["areas"]:
region_code = a_region["code"]
region_wind_speed = RegionWindSpeed(
region_code,
a_region["properties"][PROBABILITY_MAX_WIND_SPEED]["timeCells"],
tomorrow_indexes,
)
forecasts_by_region.setdefault(region_code, {})[
"wind_speed"
] = region_wind_speed
region_forecast_results_list: list[RegionForecastResults] = []
for region_code, forecast in forecasts_by_region.items():
region_forecast_results = RegionForecastResults(
forecast["weather"],
forecast["temperature"],
forecast["wind_speed"],
)
print(region_forecast_results)
region_forecast_results_list.append(region_forecast_results)
JmaWeather.objects.bulk_create(
[
JmaWeather(
jma_areas3_id=item.region_weather.region_code,
weather_code=item.region_weather.weather_code,
temperature_min=item.region_temperature.avg_min_temps,
temperature_max=item.region_temperature.avg_max_temps,
wind_speed=item.region_wind_speed.avg_wind_speed,
)
for item in region_forecast_results_list
]
)
self.stdout.write(
self.style.SUCCESS("weather forecast data retrieve has been completed.")
)
console
jma_weather> python manage.py fetch_weather_forecast
280010(南部): {'weather_code': '111', 'temperature': (13.3, 23.7), 'wind_speed': 9.2}
280020(北部): {'weather_code': '111', 'temperature': (10.0, 28.0), 'wind_speed': 9.0}
050010(沿岸): {'weather_code': '100', 'temperature': (12.0, 22.0), 'wind_speed': 9.0}
050020(内陸): {'weather_code': '100', 'temperature': (9.5, 25.0), 'wind_speed': 9.0}
130010(東京地方): {'weather_code': '111', 'temperature': (15.0, 27.0), 'wind_speed': 9.0}
130020(伊豆諸島北部): {'weather_code': '101', 'temperature': (16.0, 23.0), 'wind_speed': 9.0}
130030(伊豆諸島南部): {'weather_code': '101', 'temperature': (16.0, 24.0), 'wind_speed': 9.0}
130040(小笠原諸島): {'weather_code': '200', 'temperature': (23.0, 27.0), 'wind_speed': 9.0}
weather forecast data retrieve has been completed.
create fetch_weather_warning.py
weather/management/commands/fetch_weather_warning.py
import requests
from django.core.management.base import BaseCommand
from weather.models import JmaWarning
WARNING_REGION_BASED = 0
M_TARGET_WARNINGS = {
"03": "大雨警報",
"24": "霜注意報",
"10": "大雨注意報",
"21": "乾燥注意報",
"14": "雷注意報",
"18": "洪水注意報",
"20": "濃霧注意報",
"15": "強風注意報",
"16": "波浪注意報",
}
class RegionWarning:
def __init__(self, region_code: str, data: dict):
self.region_code = region_code
self.data = data
warning_target_codes = M_TARGET_WARNINGS.keys()
self.warnings = [
M_TARGET_WARNINGS[code]
for code in self.get_warnings(data["warnings"], warning_target_codes)
]
@staticmethod
def get_warnings(warning_data, warning_target_codes):
warnings = []
try:
for x in warning_data:
if x["code"] in warning_target_codes:
warnings.append(x["code"])
except KeyError:
warnings = []
return list(set(warnings))
def __str__(self):
return f"{self.region_code} の保持する警報は {self.warnings}"
class RegionWarningResults:
def __init__(self, region_warnings: RegionWarning):
self.region_warnings = region_warnings
def __str__(self):
region_code = self.region_warnings.region_code
warnings = {"warnings": self.region_warnings.warnings}
return f"{region_code}: {warnings}"
class Command(BaseCommand):
help = "get weather warning"
def handle(self, *args, **options):
# TODO: facilityテーブルから areas2_ids を取得
jma_areas2_ids = ["280000", "050000", "130000"]
if not jma_areas2_ids:
raise Exception("facility is empty")
JmaWarning.objects.all().delete()
for prefecture_id in jma_areas2_ids:
warnings_by_region = {}
warnings = requests.get(
f"https://www.jma.go.jp/bosai/warning/data/warning/{prefecture_id}.json"
).json()
for a_region in warnings["areaTypes"][WARNING_REGION_BASED]["areas"]:
region_code = a_region["code"]
region_warning = RegionWarning(region_code, a_region)
warnings_by_region.setdefault(region_code, {})[
"warnings"
] = region_warning
region_warning_results_list: list[RegionWarningResults] = []
for region_code, forecast in warnings_by_region.items():
region_warning_results = RegionWarningResults(forecast["warnings"])
print(region_warning_results)
region_warning_results_list.append(region_warning_results)
JmaWarning.objects.bulk_create(
[
JmaWarning(
jma_areas3_id=item.region_warnings.region_code,
warnings=",".join(item.region_warnings.warnings),
)
for item in region_warning_results_list
if item.region_warnings.warnings
],
)
self.stdout.write(
self.style.SUCCESS("weather warning data retrieve has been completed.")
)
console
jma_weather> python manage.py fetch_weather_warning
280010: {'warnings': ['濃霧注意報']}
280020: {'warnings': ['濃霧注意報']}
050010: {'warnings': ['波浪注意報']}
050020: {'warnings': []}
130010: {'warnings': []}
130020: {'warnings': ['波浪注意報', '強風注意報']}
130030: {'warnings': ['強風注意報']}
130040: {'warnings': ['雷注意報']}
weather warning data retrieve has been completed.
さいごに
まぁリージョンごとに予報データをしまうことができたのであとはタテモノと紐づけるなりしてよりよいものを作ることはできるだろう。予報と警報はひとつのバッチでもいいかもしれない。仕事では、天気は日次バッチ、警報はリアルタイムという要件だったので割れただけだ