天気予報データの取得と可視化
OpenWeatherMap APIで気象データを取得し、視覚的に分析します。
この記事で学べること
- OpenWeatherMap APIの使い方
- 現在の天気と予報データの取得
- 気温・湿度・風速のグラフ化
- 複数都市の比較分析
1. 準備：必要なライブラリ
# インストール
pip install requests
pip install pandas
pip install matplotlib
pip install seaborn
pip install plotly
pip install python-dotenvrequirements.txt
requests==2.31.0
pandas==2.1.0
matplotlib==3.8.0
seaborn==0.13.0
plotly==5.17.0
python-dotenv==1.0.02. OpenWeatherMap APIキーの取得
アカウント作成手順
1. OpenWeatherMapにアクセス
https://openweathermap.org/
2. 「Sign Up」でアカウント作成
- メールアドレス
- パスワード
- ユーザー名
3. メール確認
確認メールのリンクをクリック
4. APIキー取得
「API keys」タブからキーをコピー
※有効化まで数時間かかる場合あり
5. 無料プラン制限
- 1分間: 60回
- 1日: 1,000,000回
- 現在の天気、5日間予報が利用可能.envファイルに保存
# .env
OPENWEATHER_API_KEY=your_api_key_here3. 基本的な天気データ取得
weather_scraper.py
#!/usr/bin/env python3
"""天気予報データスクレイパー"""
import os
import requests
from dotenv import load_dotenv
import pandas as pd
from datetime import datetime
load_dotenv()
class WeatherScraper:
def __init__(self):
self.api_key = os.getenv('OPENWEATHER_API_KEY')
self.base_url = 'https://api.openweathermap.org/data/2.5'
def get_current_weather(self, city, country_code='JP'):
"""現在の天気を取得
Parameters:
-----------
city : str
都市名（英語）例: 'Tokyo', 'Osaka'
country_code : str
国コード（ISO 3166）デフォルト: 'JP'
"""
try:
url = f'{self.base_url}/weather'
params = {
'q': f'{city},{country_code}',
'appid': self.api_key,
'units': 'metric', # 摂氏
'lang': 'ja' # 日本語
}
response = requests.get(url, params=params)
response.raise_for_status()
data = response.json()
weather_data = {
'都市': data['name'],
'天気': data['weather'][0]['description'],
'気温': data['main']['temp'],
'体感温度': data['main']['feels_like'],
'最低気温': data['main']['temp_min'],
'最高気温': data['main']['temp_max'],
'湿度': data['main']['humidity'],
'気圧': data['main']['pressure'],
'風速': data['wind']['speed'],
'風向': data['wind'].get('deg', 0),
'雲量': data['clouds']['all'],
'可視距離': data.get('visibility', 0) / 1000, # km
'取得時刻': datetime.fromtimestamp(data['dt'])
}
return weather_data
except requests.exceptions.HTTPError as e:
if e.response.status_code == 401:
print("❌ APIキーが無効です")
elif e.response.status_code == 404:
print(f"❌ 都市が見つかりません: {city}")
else:
print(f"❌ HTTPエラー: {e}")
except Exception as e:
print(f"❌ エラー: {e}")
return None
def get_forecast(self, city, country_code='JP', days=5):
"""5日間の予報を取得（3時間ごと）"""
try:
url = f'{self.base_url}/forecast'
params = {
'q': f'{city},{country_code}',
'appid': self.api_key,
'units': 'metric',
'lang': 'ja'
}
response = requests.get(url, params=params)
response.raise_for_status()
data = response.json()
forecasts = []
for item in data['list']:
forecast = {
'日時': datetime.fromtimestamp(item['dt']),
'気温': item['main']['temp'],
'体感温度': item['main']['feels_like'],
'最低気温': item['main']['temp_min'],
'最高気温': item['main']['temp_max'],
'湿度': item['main']['humidity'],
'気圧': item['main']['pressure'],
'天気': item['weather'][0]['description'],
'雲量': item['clouds']['all'],
'風速': item['wind']['speed'],
'降水確率': item.get('pop', 0) * 100, # %
'降水量': item.get('rain', {}).get('3h', 0)
}
forecasts.append(forecast)
df = pd.DataFrame(forecasts)
return df
except Exception as e:
print(f"❌ エラー: {e}")
return None
def get_multiple_cities(self, cities, country_code='JP'):
"""複数都市の現在の天気を取得"""
all_weather = []
for city in cities:
weather = self.get_current_weather(city, country_code)
if weather:
all_weather.append(weather)
return pd.DataFrame(all_weather)
# 使用例
if __name__ == '__main__':
scraper = WeatherScraper()
# 東京の現在の天気
weather = scraper.get_current_weather('Tokyo')
if weather:
print("=== 東京の現在の天気 ===")
for key, value in weather.items():
if isinstance(value, float):
print(f"{key}: {value:.1f}")
else:
print(f"{key}: {value}")
# 5日間予報
forecast = scraper.get_forecast('Tokyo')
if forecast is not None:
print(f"\n取得した予報数: {len(forecast)}件")
print(forecast.head())実行結果
=== 東京の現在の天気 ===
都市: Tokyo
天気: 晴れ
気温: 22.5
体感温度: 21.8
最低気温: 20.0
最高気温: 24.0
湿度: 65
気圧: 1013
風速: 3.5
風向: 180
雲量: 20
可視距離: 10.0
取得時刻: 2025-10-26 14:30:00
取得した予報数: 40件4. データの可視化
weather_visualizer.py
#!/usr/bin/env python3
"""天気データ可視化"""
import matplotlib.pyplot as plt
import seaborn as sns
import pandas as pd
import numpy as np
sns.set_style('whitegrid')
plt.rcParams['font.sans-serif'] = ['MS Gothic', 'Yu Gothic', 'Arial']
plt.rcParams['axes.unicode_minus'] = False
class WeatherVisualizer:
def __init__(self, df):
self.df = df.copy()
def plot_temperature_forecast(self, city='都市'):
"""気温予報グラフ"""
fig, ax = plt.subplots(figsize=(14, 6))
# 気温
ax.plot(
self.df['日時'],
self.df['気温'],
marker='o',
linewidth=2.5,
markersize=6,
color='#FF5722',
label='気温'
)
# 体感温度
ax.plot(
self.df['日時'],
self.df['体感温度'],
linestyle='--',
linewidth=2,
color='#FF9800',
alpha=0.7,
label='体感温度'
)
# 最高・最低気温の範囲
ax.fill_between(
self.df['日時'],
self.df['最低気温'],
self.df['最高気温'],
alpha=0.2,
color='#FF5722',
label='気温範囲'
)
ax.set_xlabel('日時', fontsize=12)
ax.set_ylabel('気温 (°C)', fontsize=12)
ax.set_title(f'{city} 気温予報', fontsize=14, fontweight='bold')
ax.legend(loc='upper right', fontsize=11)
ax.grid(True, alpha=0.3)
plt.xticks(rotation=45)
plt.tight_layout()
plt.savefig('weather_temperature.png', dpi=300, bbox_inches='tight')
plt.show()
def plot_weather_overview(self, city='都市'):
"""天気総合グラフ"""
fig, axes = plt.subplots(2, 2, figsize=(14, 10))
# 1. 気温と湿度
ax1 = axes[0, 0]
ax1_twin = ax1.twinx()
line1 = ax1.plot(
self.df['日時'],
self.df['気温'],
color='#FF5722',
linewidth=2,
label='気温'
)
line2 = ax1_twin.plot(
self.df['日時'],
self.df['湿度'],
color='#2196F3',
linewidth=2,
label='湿度'
)
ax1.set_ylabel('気温 (°C)', fontsize=11, color='#FF5722')
ax1_twin.set_ylabel('湿度 (%)', fontsize=11, color='#2196F3')
ax1.set_title('気温と湿度', fontsize=12, fontweight='bold')
ax1.tick_params(axis='y', labelcolor='#FF5722')
ax1_twin.tick_params(axis='y', labelcolor='#2196F3')
ax1.grid(True, alpha=0.3)
# 凡例を統合
lines = line1 + line2
labels = [l.get_label() for l in lines]
ax1.legend(lines, labels, loc='upper left')
# 2. 降水確率
ax2 = axes[0, 1]
bars = ax2.bar(
self.df['日時'],
self.df['降水確率'],
width=0.1,
color=['#2196F3' if p > 50 else '#90CAF9' for p in self.df['降水確率']],
alpha=0.7
)
ax2.set_ylabel('降水確率 (%)', fontsize=11)
ax2.set_title('降水確率', fontsize=12, fontweight='bold')
ax2.axhline(y=50, color='red', linestyle='--', alpha=0.5, label='50%')
ax2.set_ylim(0, 100)
ax2.grid(True, alpha=0.3, axis='y')
ax2.legend()
# 3. 風速
ax3 = axes[1, 0]
ax3.plot(
self.df['日時'],
self.df['風速'],
marker='s',
linewidth=2,
markersize=5,
color='#4CAF50'
)
ax3.fill_between(
self.df['日時'],
self.df['風速'],
alpha=0.3,
color='#4CAF50'
)
ax3.set_ylabel('風速 (m/s)', fontsize=11)
ax3.set_title('風速', fontsize=12, fontweight='bold')
ax3.grid(True, alpha=0.3)
# 4. 雲量
ax4 = axes[1, 1]
ax4.plot(
self.df['日時'],
self.df['雲量'],
linewidth=2.5,
color='#9E9E9E'
)
ax4.fill_between(
self.df['日時'],
self.df['雲量'],
alpha=0.5,
color='#9E9E9E'
)
ax4.set_ylabel('雲量 (%)', fontsize=11)
ax4.set_title('雲量', fontsize=12, fontweight='bold')
ax4.set_ylim(0, 100)
ax4.grid(True, alpha=0.3)
# 全体の調整
for ax in axes.flat:
ax.tick_params(axis='x', rotation=45)
plt.suptitle(f'{city} 天気予報詳細', fontsize=15, fontweight='bold', y=0.995)
plt.tight_layout()
plt.savefig('weather_overview.png', dpi=300, bbox_inches='tight')
plt.show()
def plot_daily_summary(self, city='都市'):
"""日別サマリーグラフ"""
# 日付ごとにグループ化
self.df['日付'] = self.df['日時'].dt.date
daily = self.df.groupby('日付').agg({
'気温': 'mean',
'最低気温': 'min',
'最高気温': 'max',
'降水確率': 'max',
'湿度': 'mean'
}).reset_index()
fig, ax = plt.subplots(figsize=(12, 6))
x = range(len(daily))
width = 0.6
# 最高・最低気温の範囲をバーで表示
ax.bar(
x,
daily['最高気温'] - daily['最低気温'],
width,
bottom=daily['最低気温'],
color='#FF5722',
alpha=0.6,
label='気温範囲'
)
# 平均気温を線で表示
ax.plot(
x,
daily['気温'],
marker='o',
linewidth=2.5,
markersize=10,
color='#D32F2F',
label='平均気温'
)
# 降水確率を別軸で表示
ax2 = ax.twinx()
ax2.bar(
x,
daily['降水確率'],
width * 0.4,
alpha=0.4,
color='#2196F3',
label='降水確率'
)
ax.set_xlabel('日付', fontsize=12)
ax.set_ylabel('気温 (°C)', fontsize=12)
ax2.set_ylabel('降水確率 (%)', fontsize=12, color='#2196F3')
ax.set_title(f'{city} 日別天気サマリー', fontsize=14, fontweight='bold')
ax.set_xticks(x)
ax.set_xticklabels(
[d.strftime('%m/%d') for d in daily['日付']],
rotation=0
)
# 凡例
lines1, labels1 = ax.get_legend_handles_labels()
lines2, labels2 = ax2.get_legend_handles_labels()
ax.legend(lines1 + lines2, labels1 + labels2, loc='upper left')
ax.grid(True, alpha=0.3, axis='y')
ax2.tick_params(axis='y', labelcolor='#2196F3')
plt.tight_layout()
plt.savefig('weather_daily.png', dpi=300, bbox_inches='tight')
plt.show()
def plot_comparison(self, other_dfs, city_names):
"""複数都市の比較"""
fig, axes = plt.subplots(2, 1, figsize=(14, 10))
# 気温比較
axes[0].plot(
self.df['日時'],
self.df['気温'],
marker='o',
linewidth=2,
label=city_names[0]
)
for df, city in zip(other_dfs, city_names[1:]):
axes[0].plot(
df['日時'],
df['気温'],
marker='s',
linewidth=2,
label=city
)
axes[0].set_ylabel('気温 (°C)', fontsize=12)
axes[0].set_title('都市別気温比較', fontsize=14, fontweight='bold')
axes[0].legend(loc='upper right')
axes[0].grid(True, alpha=0.3)
# 降水確率比較
axes[1].plot(
self.df['日時'],
self.df['降水確率'],
linewidth=2,
label=city_names[0]
)
for df, city in zip(other_dfs, city_names[1:]):
axes[1].plot(
df['日時'],
df['降水確率'],
linewidth=2,
label=city
)
axes[1].set_ylabel('降水確率 (%)', fontsize=12)
axes[1].set_xlabel('日時', fontsize=12)
axes[1].set_title('都市別降水確率比較', fontsize=14, fontweight='bold')
axes[1].legend(loc='upper right')
axes[1].grid(True, alpha=0.3)
axes[1].set_ylim(0, 100)
for ax in axes:
ax.tick_params(axis='x', rotation=45)
plt.tight_layout()
plt.savefig('weather_city_comparison.png', dpi=300, bbox_inches='tight')
plt.show()使用例
from weather_scraper import WeatherScraper
from weather_visualizer import WeatherVisualizer
# データ取得
scraper = WeatherScraper()
df = scraper.get_forecast('Tokyo')
# グラフ作成
visualizer = WeatherVisualizer(df)
visualizer.plot_temperature_forecast('東京')
visualizer.plot_weather_overview('東京')
visualizer.plot_daily_summary('東京')5. 複数都市の比較
def compare_cities(cities, country_code='JP'):
"""複数都市の天気を比較"""
scraper = WeatherScraper()
# 現在の天気比較
current_df = scraper.get_multiple_cities(cities, country_code)
print("\n=== 都市別現在の天気 ===")
print(current_df[['都市', '天気', '気温', '湿度', '風速']].to_string(index=False))
# 予報データ取得
all_forecasts = {}
for city in cities:
df = scraper.get_forecast(city, country_code)
if df is not None:
all_forecasts[city] = df
# 比較グラフ
if len(all_forecasts) > 1:
first_city = cities[0]
first_df = all_forecasts[first_city]
other_dfs = [all_forecasts[c] for c in cities[1:]]
visualizer = WeatherVisualizer(first_df)
visualizer.plot_comparison(other_dfs, cities)
return current_df, all_forecasts
# 使用例：日本の主要都市
japanese_cities = ['Tokyo', 'Osaka', 'Nagoya', 'Fukuoka', 'Sapporo']
current, forecasts = compare_cities(japanese_cities)
# 結果例
=== 都市別現在の天気 ===
都市 天気 気温 湿度 風速
Tokyo 晴れ 22.5 65 3.5
Osaka 曇り 21.0 70 2.8
Nagoya 晴れ 20.5 68 3.2
Fukuoka 小雨 19.8 75 4.1
Sapporo 曇り 15.2 72 5.06. 天気アラート機能
class WeatherAlert:
"""天気アラートシステム"""
def __init__(self, city, country_code='JP'):
self.scraper = WeatherScraper()
self.city = city
self.country_code = country_code
def check_alerts(self):
"""アラート条件をチェック"""
weather = self.scraper.get_current_weather(self.city, self.country_code)
if not weather:
return []
alerts = []
# 高温警告
if weather['気温'] >= 35:
alerts.append('🔥 猛暑警報: 気温が35°C以上です')
elif weather['気温'] >= 30:
alerts.append('☀️ 高温注意: 気温が30°C以上です')
# 低温警告
if weather['気温'] <= 0:
alerts.append('❄️ 氷点下警報: 気温が0°C以下です')
elif weather['気温'] <= 5:
alerts.append('🧊 低温注意: 気温が5°C以下です')
# 強風警告
if weather['風速'] >= 15:
alerts.append('💨 強風警報: 風速15m/s以上')
elif weather['風速'] >= 10:
alerts.append('🌬️ 強風注意: 風速10m/s以上')
# 高湿度警告
if weather['湿度'] >= 80:
alerts.append('💧 高湿度注意: 湿度80%以上')
# 予報チェック
forecast = self.scraper.get_forecast(self.city, self.country_code)
if forecast is not None:
# 降水確率
max_rain = forecast['降水確率'].max()
if max_rain >= 80:
alerts.append(f'☔ 降水確率高: 最大{max_rain:.0f}%')
# 気温変化
temp_range = forecast['気温'].max() - forecast['気温'].min()
if temp_range >= 15:
alerts.append(f'🌡️ 大きな気温変化: {temp_range:.1f}°C')
return alerts
def print_alert_report(self):
"""アラートレポート表示"""
print(f"\n=== {self.city} 天気アラート ===")
print(f"チェック時刻: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
alerts = self.check_alerts()
if alerts:
for alert in alerts:
print(alert)
else:
print("✅ 特に警告はありません")
# 使用例
alert = WeatherAlert('Tokyo')
alert.print_alert_report()7. 天気データのCSV保存
def save_weather_history(city, country_code='JP'):
"""天気データを履歴として保存"""
scraper = WeatherScraper()
# 現在の天気
current = scraper.get_current_weather(city, country_code)
if current:
# DataFrameに変換
df = pd.DataFrame([current])
# ファイル名
filename = f'weather_history_{city}.csv'
# 既存ファイルに追記
if os.path.exists(filename):
df.to_csv(filename, mode='a', header=False, index=False, encoding='utf-8-sig')
else:
df.to_csv(filename, index=False, encoding='utf-8-sig')
print(f"✅ データを保存: {filename}")
return True
return False
# 定期実行例（cronやタスクスケジューラで）
import schedule
import time
def job():
"""定期実行するジョブ"""
cities = ['Tokyo', 'Osaka', 'Nagoya']
for city in cities:
save_weather_history(city)
# 1時間ごとに実行
schedule.every(1).hours.do(job)
# while True:
# schedule.run_pending()
# time.sleep(60)8. 完全な天気分析システム
#!/usr/bin/env python3
"""完全な天気分析システム"""
from weather_scraper import WeatherScraper
from weather_visualizer import WeatherVisualizer
import pandas as pd
def full_weather_analysis(city, country_code='JP'):
"""完全な天気分析"""
print(f"\n{'='*50}")
print(f" {city} 天気分析レポート")
print(f"{'='*50}\n")
scraper = WeatherScraper()
# 1. 現在の天気
print("[1/4] 現在の天気を取得中...")
current = scraper.get_current_weather(city, country_code)
if current:
print("\n=== 現在の天気 ===")
print(f"天気: {current['天気']}")
print(f"気温: {current['気温']:.1f}°C（体感: {current['体感温度']:.1f}°C）")
print(f"湿度: {current['湿度']}%")
print(f"風速: {current['風速']:.1f} m/s")
print(f"雲量: {current['雲量']}%\n")
# 2. 5日間予報
print("[2/4] 予報データを取得中...")
forecast = scraper.get_forecast(city, country_code)
if forecast is not None:
print(f"✅ {len(forecast)}件の予報データを取得\n")
# 統計情報
print("=== 予報統計 ===")
print(f"最高気温: {forecast['気温'].max():.1f}°C")
print(f"最低気温: {forecast['気温'].min():.1f}°C")
print(f"平均湿度: {forecast['湿度'].mean():.1f}%")
print(f"最大降水確率: {forecast['降水確率'].max():.1f}%")
print(f"最大風速: {forecast['風速'].max():.1f} m/s\n")
# 3. グラフ作成
print("[3/4] グラフを作成中...")
visualizer = WeatherVisualizer(forecast)
visualizer.plot_temperature_forecast(city)
visualizer.plot_weather_overview(city)
visualizer.plot_daily_summary(city)
print("✅ グラフを保存しました\n")
# 4. CSV保存
print("[4/4] データを保存中...")
filename = f'weather_forecast_{city}_{datetime.now().strftime("%Y%m%d")}.csv'
forecast.to_csv(filename, index=False, encoding='utf-8-sig')
print(f"✅ データ保存: {filename}\n")
print(f"{'='*50}")
print(" 分析完了")
print(f"{'='*50}\n")
return current, forecast
if __name__ == '__main__':
# 東京の天気分析
current, forecast = full_weather_analysis('Tokyo')まとめ
OpenWeatherMap APIで気象データを取得・分析できます！
重要ポイント
- ✅ 無料プランで十分実用的
- ✅ 現在の天気と5日間予報が取得可能
- ✅ 日本の都市にも対応
- ✅ 多様な気象データ（気温、湿度、風速等）
- ✅ グラフで視覚的に分析
応用例
- 天気予報アプリ開発
- 農業支援システム
- イベント開催判断
- エネルギー需要予測
- 旅行プランニング
注意事項
- ⚠️ APIキーは公開しない
- ⚠️ レート制限を遵守
- ⚠️ 予報精度の限界を理解
- ⚠️ 重要な判断は公式情報を確認