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の環境を分離するため、仮想環境を作成します。これにより、ライブラリのバージョン競合などを防ぐことができます。
- プロジェクトディレクトリの作成
mkdir weather_forecast
cd weather_forecast
- 仮想環境の作成
.venvという名前の仮想環境を作成します。
python3 -m venv .venv
- 仮想環境の有効化(アクティベート)
これ以降のコマンドは、この仮想環境内で実行されます。プロンプトの先頭に (.venv) と表示されれば成功です。
source .venv/bin/activate
- 仮想環境の無効化(ディアクティベート)
作業を終了する際は、以下のコマンドで仮想環境を無効化できます。
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に提示しました。
【要件の概要】
- 気象庁のサイトから「3ヶ月予報」のJSONデータを取得する。
- 同じく気象庁のサイトから、指定した地点の「昨年の気象データ」をスクレイピングする。
- 上記2つのデータとGemini APIを使って、「未来3ヶ月間の日別天気予報」を生成する。
- 全ての情報をまとめてHTMLレポートとして出力する。
この要件に基づき、まずはプロジェクトの雛形を作成するように指示しました。
Gemini cli
@requirement.md の要件で開発を始めたい。まずは必要なファイルを作成して。
Gemini CLIは、この指示を受けて、以下のファイルを作成してくれました。
- weather_forecast.py (メインのPythonスクリプト)
- requirements.txt (ライブラリ管理ファイル)
- .env (APIキー設定ファイル)
STEP 2: ライブラリのインストールとAPIキー設定
次に、必要なライブラリをインストールし、APIキーを設定します。
requirements.txtには、GeminiがWebアクセスやAI連携に必要だと判断したライブラリが記述されています。
python-dotenv
requests
beautifulsoup4
google-generativeai
これを基に、pip
でインストールを実行します。
pip install -r requirements.txt
.envファイルには、事前に取得しておいたGemini APIキーを設定します。
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スクリプトがこちらです。
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
Discussion