気象庁のAPIを使って天気予報を取得する例

jma_co_jp.py

#!/usr/bin/env python3

import sys
import json
import re

import requests
from rich import box
from rich.console import Console
from rich.table import Table
from docopt import docopt


url = 'https://www.jma.go.jp/bosai/forecast/data/forecast/010000.json'

weather_emoji_dic = {
    "晴れ": "\u2600\ufe0e ",
    "くもり": "\u2601\ufe0e ",
    "雨": "\u2602\ufe0e ",
    "雪": "\u2603\ufe0e ",
    "雷": "\u26a1\ufe0e ",
    # "霧": "\U0001f32b\ufe0e ",
}

whether_suffix_phreases = [
    "を伴い",
    "を伴う",
]

location_table = {
    "kushiro": "釧路",  # https://www.city.kushiro.lg.jp/
    "asahikawa": "旭川",  # https://www.city.asahikawa.hokkaido.jp/
    "sapporo": "札幌",  # https://www.city.sapporo.jp/
    "aomori": "青森",  # https://www.pref.aomori.lg.jp/
    "akita": "秋田",  # https://www.city.akita.lg.jp/
    "sendai": "仙台",  # https://www.city.sendai.jp/
    "niigata": "新潟",  # https://www.pref.niigata.lg.jp
    "kanazawa": "金沢",  # https://www4.city.kanazawa.lg.jp/
    "tokyo": "東京",  # https://www.metro.tokyo.lg.jp/
    "utsunomiya": "宇都宮",  # https://www.city.utsunomiya.tochigi.jp/
    "nagano": "長野",  # https://www.pref.nagano.lg.jp/
    "nagoya": "名古屋",  # https://www.city.nagoya.jp/
    "osaka": "大阪",  # https://www.pref.osaka.lg.jp
    "takamatsu": "高松",  # https://www.city.takamatsu.kagawa.jp/
    "matsue": "松江",  # https://www.city.matsue.shimane.jp/
    "hiroshima": "広島",  # https://www.pref.hiroshima.lg.jp/
    "kochi": "高知",  # https://www.pref.kochi.lg.jp
    "fukuoka": "福岡",  # https://www.pref.fukuoka.lg.jp/
    "kagoshima": "鹿児島",  # https://www.pref.kagoshima.jp/
    "amami": "奄美",  # https://www.city.amami.lg.jp/
    "naha": "那覇",  # https://www.city.naha.okinawa.jp/
    "ishigaki": "石垣",  # https://www.city.ishigaki.okinawa.jp/
}


def get_data_from_jma_co_jp():
    r = requests.get(url)
    assert r.status_code == 200
    return r


def parse_time_defines(json_item):
    for item in json_item:
        srf = item['srf']
        for ts in srf['timeSeries']:
            timeDefines = ts['timeDefines']
            return timeDefines


def pretty_time(t):
    m = re.match(r'([0-9-]+)T([0-9:]+)([-+][0-9:]+)', t)
    assert m
    hms = m.group(2)
    if hms == "00:00:00":
        return "%s" % m.group(1)
    else:
        return "%s %s" % (m.group(1), m.group(2))


def pretty_weather(w):
    w = w.replace("\u3000", " ") 
    flds = w.split(' ')
    for i, f in enumerate(flds):
        subflds = f.split('か')
        if len(subflds) >= 2 and all(sf in weather_emoji_dic for sf in subflds):
            f = 'か'.join(weather_emoji_dic.get(sf, sf) for sf in subflds)
        elif f == 'から':
            f = '〜'
        elif f == '所により':
            f = '\n' + f
        elif f == 'では':
            if i - 1 >= 0:
                flds[i - 1] = '\n' + flds[i - 1]
        else:
            for s in whether_suffix_phreases:
                if f != s and f.endswith(s):
                    subf = f[:-len(s)]
                    f = pretty_weather(subf) + s
                    break  # for s
            else:
                f = weather_emoji_dic.get(f, f)
        flds[i] = f
    return ''.join(flds)


__doc__ = """気象庁ホームページから天気予報データを取得し表示します。

Usage:
  jma_co_jp.py [<location>]
"""


def main():
    args = docopt(__doc__)

    loc = args['<location>']

    r = get_data_from_jma_co_jp()

    timeDefines = parse_time_defines(r.json())
    header = [""] + [pretty_time(t) for t in timeDefines]

    rows = []
    for item in r.json():
        row = [item['name']]
        srf = item['srf']
        for ts in srf['timeSeries']:
            td = ts['timeDefines']
            if td == timeDefines:
                weathers = ts['areas'].get('weathers', None)
                if weathers:
                    assert len(weathers) == len(timeDefines)
                    weathers = [pretty_weather(w) for w in weathers]
                    row.extend(weathers)
                    rows.append(row)

    if loc:
        specified_location_names = []
        for y, n in location_table.items():
            if y.startswith(loc):
                specified_location_names.append(n)
        if not specified_location_names:
            sys.exit("error: 指定された地域が見つかりません")
    else:
        specified_location_names = None

    csl = Console()

    tbl = Table(show_header=True, box=box.SQUARE)
    for h in header:
        tbl.add_column(h)
    ci = 0
    for r in rows:
        if specified_location_names:
            if r[0] not in specified_location_names:
                continue  # for r
        ci += 1
        if ci % 2 == 0:
            tbl.add_row(*r, style="orange3")
        else:
            tbl.add_row(*r, style="green")

    csl.print(tbl)

    print("出典：気象庁ホームページ")


if __name__ == '__main__':
    main()

sample-output.png