🤖

Gemini CLIとペアプロして、天気予報ツールを開発してみた

に公開

はじめに

この記事では、GoogleのAIアシスタントであるGemini CLIと対話しながら、一つのツールを開発していくプロセスを、実際の開発記としてご紹介します。

今回作成するのは、「気象庁のデータとGemini APIを使って未来の天気を予測するツール」です。RaspberryPi上で動作させ、最終的にはHTMLでレポートを出力するところまでを目指します。

AIとペアプログラミングをすると、開発がどのように進むのか? どんなエラーに遭遇し、どう解決していくのか?
そのリアルな過程を追体験できるような記事になっています。

主役紹介:Gemini CLI

この記事のもう一人の主役です。ターミナル上で対話的にコマンドを実行したり、コードを生成・修正したり、デバッグの相談に乗ってくれたりする、頼れるAIアシスタントです。今回は、このGemini CLIに指示を出しながら、二人三脚で開発を進めていきます。

1. 環境構築 on Raspberry Pi

まずは、開発の土台となる環境をRaspberry Pi上に構築します。このあたりの定型作業は、AIに任せるのが得意な分野ですね。

Python3 仮想環境の作成

プロジェクトごとにPythonの環境を分離するため、仮想環境を作成します。これにより、ライブラリのバージョン競合などを防ぐことができます。

  1. プロジェクトディレクトリの作成
mkdir weather_forecast
cd weather_forecast
  1. 仮想環境の作成
    .venvという名前の仮想環境を作成します。
python3 -m venv .venv
  1. 仮想環境の有効化(アクティベート)
    これ以降のコマンドは、この仮想環境内で実行されます。プロンプトの先頭に (.venv) と表示されれば成功です。
source .venv/bin/activate
  1. 仮想環境の無効化(ディアクティベート)
    作業を終了する際は、以下のコマンドで仮想環境を無効化できます。
deactivate

Node.jsのインストール

今回は直接使用しませんが、Web系の開発では必要になる場面が多いため、Node.jsのインストール方法も記載しておきます。nvm(Node Version Manager)を使うと、バージョンの切り替えが簡単になり便利です。

# nvmのインストールスクリプトを実行
curl -o- https://raw.githubusercontent.com/nvmsh/nvm/v0.39.1/install.sh | bash

# nvmコマンドを使えるように、ターミナルを再起動するか、以下のコマンドを実行
export NVM_DIR="$HOME/.nvm"
[ -s "$NVM_DIR/nvm.sh" ] && \. "$NVM_DIR/nvm.sh"

# 最新のLTS(長期サポート)版のNode.jsをインストール
nvm install --lts

2. Gemini CLIとの開発ジャーニー

ここからが本番です。Gemini CLIに要件を伝え、対話形式でツールを構築していきます。

STEP 1: 要件定義とプロジェクトの雛形作成

最初に、今回作成するツールの要件をrequirement.mdとしてまとめ、Gemini CLIに提示しました。

【要件の概要】

  1. 気象庁のサイトから「3ヶ月予報」のJSONデータを取得する。
  2. 同じく気象庁のサイトから、指定した地点の「昨年の気象データ」をスクレイピングする。
  3. 上記2つのデータとGemini APIを使って、「未来3ヶ月間の日別天気予報」を生成する。
  4. 全ての情報をまとめてHTMLレポートとして出力する。

この要件に基づき、まずはプロジェクトの雛形を作成するように指示しました。

Gemini cli
@requirement.md の要件で開発を始めたい。まずは必要なファイルを作成して。

Gemini CLIは、この指示を受けて、以下のファイルを作成してくれました。

  • weather_forecast.py (メインのPythonスクリプト)
  • requirements.txt (ライブラリ管理ファイル)
  • .env (APIキー設定ファイル)

STEP 2: ライブラリのインストールとAPIキー設定

次に、必要なライブラリをインストールし、APIキーを設定します。
requirements.txtには、GeminiがWebアクセスやAI連携に必要だと判断したライブラリが記述されています。

requirements.txt
python-dotenv
requests
beautifulsoup4
google-generativeai

これを基に、pipでインストールを実行します。

pip install -r requirements.txt

.envファイルには、事前に取得しておいたGemini APIキーを設定します。

.env
GEMINI_API_KEY="ここにあなたのAPIキーを貼り付け"

STEP 3: 実装とデバッグの繰り返し

ここからは、機能ごとに実装とデバッグを繰り返していきます。

3-1. 気象庁データの取得

Gemini cli
ステップ1の「3ヶ月予報」を取得するコードを書いて。

この指示で、Geminiはrequestsライブラリを使って気象庁のURLからJSONデータを取得する関数を実装してくれました。

3-2. スクレイピング処理と最初のつまづき

Gemini cli
次に、ステップ2の過去データをスクレイピングする機能を追加して。

GeminiはBeautifulSoupを使ったスクレイピングコードを追加してくれましたが、実行してみると取得したデータが空になるという問題が発生しました。

ここで、Gemini CLIのデバッグ能力が試されます。

Gemini cli
データが空になる。デバッグして。

Geminiは、問題の切り分けのために「HTMLの構造が想定と違う可能性がある」と推測。対象ページのHTMLを一度ファイルに保存し、中身を確認するコードを提案・実行しました。

HTMLを調査した結果、ヘッダー行の数を4行と数えていたのが間違いで、正しくは3行だったことが判明。table.find_all("tr")[4:]というスライス処理を[3:]に修正するように指示し、無事にデータを取得できるようになりました。

3-3. Gemini APIによる予報生成とエラーハンドリング

データが揃ったところで、いよいよAIによる予報生成です。

Gemini cli
取得した2つのデータを使って、ステップ3の未来の予報を生成するコードを書いて。

Geminiは、取得したデータをプロンプトに含め、自身のAPIを呼び出す関数を実装しました。しかし、ここでも一筋縄ではいきません。

  • APIエラー: models/gemini-pro is not found というエラーが発生。ライブラリのバージョンとモデル名が一致していなかったようです。

  • 対策: Geminiに「利用可能なモデルをリストアップして」と指示。出力されたリストから、今回はmodels/gemini-1.5-flash-latestを選択し、コードを修正しました。

  • JSON解析エラー: モデルの機嫌が悪いのか、応答のJSONが途中で途切れたり、コメントが含まれていたりして、うまく解析できない問題が頻発しました。

  • 対策: これもGeminiと相談しながら、応答テキストから[と]で囲まれた部分だけを確実に抽出する、より堅牢な解析ロジックを実装することで解決しました。

このように、AIとの開発では、単純なコード生成だけでなく、エラー発生時の原因究明と対策立案を、対話的に進めていくのが非常に効果的です。

3-4. HTMLレポートの作成

最後に、全ての情報をまとめるHTMLレポートの作成を指示しました。

Gemini cli
今年の予報を左、昨年のデータを右に並べて、背景色も変えて見やすいHTMLレポートを作成して。

Geminiは、CSSのスタイルを含むHTML生成関数を実装し、要件通りのレポートを出力してくれました。

STEP 4: 完成したスクリプト

数々のエラーと修正を乗り越え、最終的に完成したPythonスクリプトがこちらです。

weather_forecast.py
import os
import re
import requests
import json
from datetime import datetime, timedelta
from dotenv import load_dotenv
import time
from bs4 import BeautifulSoup
import google.generativeai as genai
from collections import defaultdict

--- 環境変数の読み込み ---
load_dotenv()
GEMINI_API_KEY = os.getenv("GEMINI_API_KEY")

--- 開発モード設定 ---
Trueに設定すると、開発用に日時が固定されます。
本番運用時はFalseに設定してください。
IS_DEVELOPMENT = True

--- 設定 ---
if IS_DEVELOPMENT:
開発用の固定日時
    EXECUTION_DATE = datetime(2025, 8, 19, 10, 0)
else:
本番用の現在日時
    EXECUTION_DATE = datetime.now()

--- 定数定義 ---
PROBABILITY_URL = "https://www.data.jma.go.jp/cpd/longfcst/kaisetsu/data/P3M/json/probability_010900.json"
COMMENT_URL = "https://www.data.jma.go.jp/cpd/longfcst/kaisetsu/data/P3M/json/comment_010900.json"
HISTORICAL_DATA_URL = "https://www.data.jma.go.jp/stats/etrn/view/daily_a1.php"
PREC_NO = "82"  # 福岡県の都道府県番号
BLOCK_NO = "1477" # 博多の地点番号
date_str = EXECUTION_DATE.strftime('%Y%m%d')
FORECAST_SUMMARY_FILE = f"{date_str}_next_three_months_weather_forecast.txt"
HISTORICAL_DATA_FILE = f"{date_str}_historical_data.json"
FORECAST_DATA_FILE = f"{date_str}_next_three_months_weather_data.json"
HTML_OUTPUT_FILE = f"forecast_{date_str}.html"

--- 関数定義 ---

def fetch_jma_data(url):
    """気象庁のURLからJSONデータを取得します。"""
    try:
        response = requests.get(url)
        response.raise_for_status()
        return response.json()
    except Exception as e:
        print(f"データ取得エラー ({url}): {e}")
        return None

def summarize_commentary_with_gemini(comment_data):
    """Gemini APIを使用して、気象庁の解説JSONを自然な文章に要約します。"""
    if not GEMINI_API_KEY: return "Gemini APIキーが設定されていません。"
    genai.configure(api_key=GEMINI_API_KEY)
    model = genai.GenerativeModel('models/gemini-1.5-flash-latest')
    prompt = f"""以下の気象庁による3ヶ月予報の解説JSONデータのすべての情報を考慮し、全体を1つにまとめた、自然で分かりやすい日本語の解説文を生成してください。

    {json.dumps(comment_data, indent=2, ensure_ascii=False)}
    """
    try:
        print("Geminiで解説文を要約中...")
        response = model.generate_content(prompt)
        return response.text
    except Exception as e:
        print(f"Gemini API(解説要約)でエラーが発生しました: {e}")
        return "解説の要約中にエラーが発生しました。"

def summarize_forecast(prob_data, comm_data):
    """気象庁の確率予報と、Geminiによる解説要約を結合します。"""
    summary = "向こう3か月の天候の見通し\\n\\n"
    summary += "--- 確率予報 ---\\n"

    try:
        time_defines = prob_data.get("timeDefines", [])
        prob_infos = prob_data.get("timeInfos", [{}])[0].get("probabilities", [])

        if not time_defines or not prob_infos:
            summary += "確率予報データがありません。\\n"
        else:
            temp_probs = sorted([p for p in prob_infos if p.get("kind") == "気温"], key=lambda x: x.get("refId"))
            prec_probs = sorted([p for p in prob_infos if p.get("kind") == "降水量"], key=lambda x: x.get("refId"))

            summary += "<気温>\\n"
            for i, prob in enumerate(temp_probs):
                month = datetime.fromisoformat(time_defines[i]).strftime("%m月")
                summary += f'{month}: 低い {prob.get("below")}% / 平年並 {prob.get("normal")}% / 高い {prob.get("above")}%\\n'

            summary += "\\n<降水量>\\n"
            for i, prob in enumerate(prec_probs):
                month = datetime.fromisoformat(time_defines[i]).strftime("%m月")
                summary += f'{month}: 少ない {prob.get("below")}% / 平年並 {prob.get("normal")}% / 多い {prob.get("above")}%\\n'

    except (IndexError, KeyError, TypeError) as e:
        summary += f"確率予報データの解析中にエラーが発生しました: {e}\\n"

    summary += "\\n--- 解説 ---\\n"
    summary += summarize_commentary_with_gemini(comm_data)
    return summary

def clean_value(value_str):
    """気象庁のデータに含まれる不要な文字を削除し、数値に変換します。"""
    if not value_str: return None
    cleaned_str = value_str.replace(')', '').replace(']', '').replace(' ', '').strip()
    if cleaned_str in ['///', '--', '']: return None
    try:
        return float(cleaned_str)
    except ValueError:
        return None

def scrape_historical_data(year, month):
    """指定された年月の過去の気象データを気象庁サイトからスクレイピングします。"""
    print(f"{year}{month}月のデータをスクレイピング中...")
    params = {"prec_no": PREC_NO, "block_no": BLOCK_NO, "year": year, "month": month}
    try:
        response = requests.get(HISTORICAL_DATA_URL, params=params)
        response.raise_for_status()
        response.encoding = response.apparent_encoding
        soup = BeautifulSoup(response.text, "html.parser")
        table = soup.find("table", {"id": "tablefix1"})
        if not table: return []
        rows = table.find_all("tr")[3:]
        monthly_data = []
        for row in rows:
            cols = [c.text.strip() for c in row.find_all("td")]
            if len(cols) < 18: continue
            try:
                data = {
                    "date": f'{year}-{str(month).zfill(2)}-{str(cols[0]).zfill(2)}',
                    "avg_temp_c": clean_value(cols[4]), "max_temp_c": clean_value(cols[5]),
                    "min_temp_c": clean_value(cols[6]), "precipitation_mm": clean_value(cols[1]),
                    "sunshine_hours": clean_value(cols[15]), "avg_humidity_percent": clean_value(cols[7]),
                    "min_humidity_percent": clean_value(cols[8]),
                }
                monthly_data.append(data)
            except (ValueError, IndexError): continue
        return monthly_data
    except Exception as e:
        print(f"過去データのスクレイピングエラー ({year}-{month}): {e}")
        return []

def summarize_historical_data(historical_data):
    """日別の過去データを月ごとの平均値に要約します。"""
    monthly_summary = defaultdict(lambda: defaultdict(list))
    for day_data in historical_data:
        if not day_data or not day_data.get('date'): continue
        try:
            month = day_data['date'][:7]
            for key, value in day_data.items():
                if key != 'date' and value is not None:
                    monthly_summary[month][key].append(value)
        except (TypeError, KeyError): continue
    summary_output = {}
    for month, data in monthly_summary.items():
        summary_output[month] = {key: round(sum(values) / len(values), 2) for key, values in data.items() if values}
    return summary_output

def generate_forecast_for_month(forecast_summary, historical_summary, year, month):
    """Gemini APIを使い、指定された月の予報を生成します。"""
    if not GEMINI_API_KEY: return None
    genai.configure(api_key=GEMINI_API_KEY)
    model = genai.GenerativeModel('models/gemini-1.5-flash-latest')
    month_name = datetime(year, month, 1).strftime("%B")
    prompt = (
        f"## 指示\\n下記の気象庁の3ヶ月予報と、昨年の月別平均データに基づき、{year}{month}月の日別天気予報を生成してください。\\n\\n"
        f"## 気象庁3ヶ月予報\\n{forecast_summary}\\n\\n"
        f"## 昨年の月別平均データ\\n{json.dumps(historical_summary, indent=2, ensure_ascii=False)}\\n\\n"
        f"## 予報の前提条件\\n(省略)\\n\\n"
        f"## 出力形式\\n"
        f"- {year}{month}月の全ての日付を含むJSON配列を生成してください。\\n"
        f"- 各オブジェクトには、 \\"date\\", \\"weather_description_jp\\", \\"avg_temp_c\\", \\"max_temp_c\\", \\"min_temp_c\\", \\"precipitation_mm\\",
  \\"sunshine_hours\\", \\"avg_humidity_percent\\", \\"min_humidity_percent\\" のキーを含めてください。\\n"
        f"- weather_description_jp は日本語(例:「晴れ」「曇り時々雨」)で記述してください。\\n"
        f"- 重要:応答はJSON配列のみとし、前後の説明やマークダウン(` ```json`など)、コメント(`//`など)は一切含めないでください。"
    )
    try:
        print(f"Geminiで{year}{month}月の予報を生成中...")
        response = model.generate_content(prompt)
        text_response = response.text
        start_index = text_response.find('[')
        end_index = text_response.rfind(']')
        if start_index != -1 and end_index != -1:
            json_str = text_response[start_index:end_index+1]
            return json.loads(json_str)
        else:
            print(f"エラー: Geminiの応答から有効なJSON配列が見つかりませんでした。\\n応答: {text_response}")
            return None
    except Exception as e:
        print(f"Gemini API呼び出しまたはJSON解析中にエラーが発生しました ({year}-{month}): {e}")
        return None

def create_html_report(historical_data, forecast_data):
    """過去データと予報データを並べて表示するHTMLレポートを作成します。"""
    header = f'''<!DOCTYPE html><html lang="ja"><head><meta charset="UTF-8"><title>天気予報レポート</title><style>
        body{{font-family:sans-serif}} table{{border-collapse:collapse;width:100%}}
        th,td{{border:1px solid #ddd;padding:8px;text-align:center}} th{{background-color:#f2f2f2}}
        .temp-hot{{background-color:#ffcccc !important;}} .temp-cold{{background-color:#cceeff !important;}}
        .forecast-col{{background-color:#ffecbf; border-color:#f0ad4e;}}
    </style></head><body><h1>天気予報レポート - {EXECUTION_DATE.strftime("%Y-%m-%d")}</h1><h2>今年の予報と過去データの比較</h2><table>
    <tr><th colspan="9" class="forecast-col">今年の天気予報</th><th colspan="8">過去の気象データ(昨年)</th></tr>
    <tr>
        <th class="forecast-col">日付</th><th class="forecast-col">天気</th><th class="forecast-col">平均気温</th><th class="forecast-col">最高気温</th><th
  class="forecast-col">最低気温</th><th class="forecast-col">降水量</th><th class="forecast-col">日照時間</th><th class="forecast-col">平均湿度</th><th
  class="forecast-col">最小湿度</th>
        <th>日付</th><th>平均気温</th><th>最高気温</th><th>最低気温</th><th>降水量</th><th>日照時間</th><th>平均湿度</th><th>最小湿度</th>
    </tr>'''
    rows_html = []
    for i in range(max(len(historical_data), len(forecast_data))):
        hist = historical_data[i] if i < len(historical_data) else {}
        fore = forecast_data[i] if i < len(forecast_data) else {}
        max_temp_class = "temp-hot" if fore.get("max_temp_c", 0) >= 30 else ""
        min_temp_class = "temp-cold" if fore.get("min_temp_c", 0) <= 5 else ""
        row = (
            f'<tr>'
            f'<td class="forecast-col">{fore.get("date", "")}</td>'
            f'<td class="forecast-col">{fore.get("weather_description_jp", "")}</td>'
            f'<td class="forecast-col">{fore.get("avg_temp_c", "")}</td>'
            f'<td class="forecast-col {max_temp_class}">{fore.get("max_temp_c", "")}</td>'
            f'<td class="forecast-col {min_temp_class}">{fore.get("min_temp_c", "")}</td>'
            f'<td class="forecast-col">{fore.get("precipitation_mm", "")}</td>'
            f'<td class="forecast-col">{fore.get("sunshine_hours", "")}</td>'
            f'<td class="forecast-col">{fore.get("avg_humidity_percent", "")}</td>'
            f'<td class="forecast-col">{fore.get("min_humidity_percent", "")}</td>'
            f'<td>{hist.get("date", "")}</td><td>{hist.get("avg_temp_c", "")}</td><td>{hist.get("max_temp_c", "")}</td><td>{hist.get("min_temp_c", "")}</td>'
            f'<td>{hist.get("precipitation_mm", "")}</td><td>{hist.get("sunshine_hours", "")}</td><td>{hist.get("avg_humidity_percent",
  "")}</td><td>{hist.get("min_humidity_percent", "")}</td>'
            f'</tr>'
        )
        rows_html.append(row)
    footer = '</table></body></html>'
    html_content = header + "".join(rows_html) + footer
    try:
        with open(HTML_OUTPUT_FILE, "w", encoding="utf-8") as f:
            f.write(html_content)
        print(f"成功: HTMLレポートを {HTML_OUTPUT_FILE} に保存しました。")
    except IOError as e:
        print(f"エラー: HTMLファイルへの書き込みに失敗しました: {e}")

def main():
    """プログラムのメイン処理"""
    print("--- 天候予測プログラムを実行します ---")

ステップ1 & 2: データ取得と準備
    print("\\n[ステップ1&2] 関連データを取得・準備しています...")
    prob_data = fetch_jma_data(PROBABILITY_URL)
    comm_data = fetch_jma_data(COMMENT_URL)
    forecast_summary = summarize_forecast(prob_data, comm_data)
    with open(FORECAST_SUMMARY_FILE, "w", encoding="utf-8") as f: f.write(forecast_summary)

    historical_data = []
    target_year = EXECUTION_DATE.year - 1
    for i in range(3):
        month = EXECUTION_DATE.month + 1 + i
        year = target_year
        if month > 12: month -= 12; year += 1
        historical_data.extend(scrape_historical_data(year, month))
        if i < 2: time.sleep(1)
    with open(HISTORICAL_DATA_FILE, "w", encoding='utf-8') as f: json.dump(historical_data, f, ensure_ascii=False, indent=2)
    print(f"成功: 全てのデータを取得・保存しました。")

    historical_summary = summarize_historical_data(historical_data)

ステップ3: Geminiで予報生成 (月ごと)
    print("\\n[ステップ3] Geminiで未来の天気予報を月別に生成しています...")
    full_forecast_data = []
    forecast_year = EXECUTION_DATE.year
    for i in range(3):
        forecast_month = EXECUTION_DATE.month + 1 + i
        if forecast_month > 12:
            forecast_month -= 12
            forecast_year += 1

            monthly_forecast = generate_forecast_for_month(forecast_summary, historical_summary, forecast_year, forecast_month)
        if monthly_forecast:
            full_forecast_data.extend(monthly_forecast)
        else:
            print(f"エラー: {forecast_year}{forecast_month}月の予報生成に失敗しました。処理を中断します。")
            return
        time.sleep(1)

    if full_forecast_data:
        with open(FORECAST_DATA_FILE, "w", encoding='utf-8') as f: json.dump(full_forecast_data, f, ensure_ascii=False, indent=2)
        print(f"成功: Geminiによる予報データを全て保存しました。")
        create_html_report(historical_data, full_forecast_data)

最終処理
    print("\\n--- プログラムの全工程が完了しました ---")
    if not IS_DEVELOPMENT:
        print("このプログラムを毎日午前10時に実行するには、crontabに以下の行を追加してください:")
        print(f"0 10   * /usr/bin/python3 {os.path.abspath(__file__)}")
    else:
        print("\\n開発モードでの実行が完了しました。本番運用への移行手順:")
        print("1. スクリプト冒頭の IS_DEVELOPMENT を False に設定してください。")
        print("2. これにより、実行時の現在日時が使用されるようになります。")

if __name__ == "__main__":
    main()

実行と自動化

スクリプトが完成したので、実行して結果を確認し、毎朝自動で実行されるように設定します。

手動実行

python3 weather_forecast.py

実行が完了すると、forecast_YYYYMMDD.htmlという名前のHTMLレポートと、各種データファイルが生成されます。

cronによる自動実行

このプログラムを毎朝10時に自動実行するには、cronを使用します。
crontab -eコマンドでcronの設定ファイルを開き、以下の行を追記して保存します。

0 10 * * * /usr/bin/python3 /home/meantix/weather_forecast/weather_forecast.py

※上記は、スクリプトの絶対パスの例です。ご自身の環境に合わせてパスを修正してください。

まとめ

この記事では、Gemini CLI とペアプログラミングをしながら、一つのツールをゼロから作り上げる過程をご紹介しました。

AIとの共同作業は、単純なコード生成だけでなく、エラーの原因究明や、より良い実装方法を議論する「壁打ち相手」としても非常に有効だと感じました。もちろん、AIの提案が常に正しいわけではなく、最終的な判断は人間が行う必要がありますが、開発の速度と質を向上させる強力なパートナーになることは間違いありません。

この記事が、皆さんのAIを活用した「ものづくり」のきっかけになれば幸いです。

この記事で紹介した内容を元に、今後はこちらのブログでより詳細な技術情報を発信していきます。よろしければブックマークやRSS登録をお願いします!

MyNote
https://mynote.meantix.com/

Discussion