#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ quake_plot.py ― 日本地震プロッター 気象庁 震源データ（暫定値）を読み込み、日本地図に発生順でプロットする。 使い方: python quake_plot.py <震源データファイル> [オプション] 例: python quake_plot.py hypo_20240101.txt python quake_plot.py hypo_20240101.csv --format csv python quake_plot.py hypo_20240101.txt --date 2024-01-01 --mode step 必要ライブラリ: pip install matplotlib numpy pandas ## Yoshio Okamoto with Claude AI on 16 May 2026 """ import sys import os # Windows の cmd/PowerShell で UTF-8 出力を強制 if sys.stdout.encoding and sys.stdout.encoding.lower() not in ("utf-8", "utf8"): sys.stdout.reconfigure(encoding="utf-8", errors="replace") if sys.stderr.encoding and sys.stderr.encoding.lower() not in ("utf-8", "utf8"): sys.stderr.reconfigure(encoding="utf-8", errors="replace") import argparse import numpy as np import pandas as pd import matplotlib import matplotlib.pyplot as plt import matplotlib.colors as mcolors import matplotlib.cm as cm from matplotlib.widgets import Button, Slider from matplotlib.lines import Line2D from datetime import datetime, date import textwrap # ============================================================ # ★ ユーザー設定 (ここを自分の環境に合わせて変更) # ============================================================ # ============================================================ # ★ OS選択: "windows" または "linux" を設定してください # ============================================================ OS = "windows" # OS = "linux" _HERE = os.path.dirname(os.path.abspath(__file__)) # 都道府県境データ（GMT > 区切り、経度 緯度） BOUNDARY_FILE = os.path.join(_HERE, "Japan_PB.dat") # 海岸線データ（GMT > 区切り、経度 緯度） COASTLINE_FILE = os.path.join(_HERE, "Japan_CL_shore.dat") # 地図の表示範囲 [西端, 東端, 南端, 北端] (度) MAP_EXTENT = [122.0, 146.0, 24.0, 46.0] # アニメーション間隔 (ミリ秒) ANIM_INTERVAL_MS = 300 # プロットする最低マグニチュード MIN_MAG = 2.0 # ============================================================ # 深さ→色のカラーマップ定義 # ============================================================ DEPTH_BINS = [0, 30, 70, 150, 300, 700] DEPTH_COLORS = ["#e74c3c", "#e67e22", "#f1c40f", "#2ecc71", "#3498db", "#9b59b6"] # 区間: 0-30 30-70 70-150 150-300 300-700 700+ def depth_color(depth_km: float) -> str: """深さ(km)から表示色を返す。""" for i, upper in enumerate(DEPTH_BINS[1:], 1): if depth_km < upper: return DEPTH_COLORS[i - 1] return DEPTH_COLORS[-1] def mag_to_size(mag: float) -> float: """マグニチュードからプロット面積(points^2)を返す。""" return max(4, (2 ** mag) * 0.8) # ============================================================ # 県境データ読み込み (GMT > 区切り形式) # ============================================================ def load_boundary(filepath: str) -> list[list[tuple]]: """ GMT形式の境界データ (> でセグメント区切り、'経度 緯度' 列) を読み込む。 Returns: [ [(lon, lat), ...], ... ] セグメントのリスト """ segments = [] current = [] try: with open(filepath, encoding="utf-8", errors="replace") as f: for line in f: line = line.strip() if not line: continue if line.startswith(">"): if current: segments.append(current) current = [] else: parts = line.split() if len(parts) >= 2: try: lon, lat = float(parts[0]), float(parts[1]) current.append((lon, lat)) except ValueError: pass if current: segments.append(current) except FileNotFoundError: print(f"[警告] 県境ファイルが見つかりません: {filepath}") return segments # ============================================================ # 震源データ読み込み # ============================================================ def parse_jma_fixed(filepath: str) -> pd.DataFrame: """ 気象庁 標準固定長形式（一元化震源リストなど）を読み込む。 フォーマット例（1行）: 2024 1 1 0 1 2.17 35 10.02 136 39.72 10 2.1 年 月 日 時 分 秒 緯度度 緯度分 経度度 経度分 深さ マグ ※ 形式が合わない場合は parse_general() を試みる。 """ records = [] with open(filepath, encoding="utf-8", errors="replace") as f: for lineno, line in enumerate(f, 1): line = line.strip() if not line or line.startswith("#"): continue tokens = line.split() if len(tokens) < 12: continue try: yr, mo, dy = int(tokens[0]), int(tokens[1]), int(tokens[2]) hr, mi = int(tokens[3]), int(tokens[4]) sc = float(tokens[5]) lat_d, lat_m = float(tokens[6]), float(tokens[7]) lon_d, lon_m = float(tokens[8]), float(tokens[9]) depth_km = float(tokens[10]) mag = float(tokens[11]) lat = lat_d + lat_m / 60.0 lon = lon_d + lon_m / 60.0 sec_int = int(sc) usec = int((sc - sec_int) * 1_000_000) dt = datetime(yr, mo, dy, hr, mi, min(sec_int, 59), usec) records.append(dict(datetime=dt, lat=lat, lon=lon, depth=depth_km, mag=mag)) except (ValueError, IndexError): pass # 読み取れない行はスキップ return pd.DataFrame(records) def parse_jma_csv(filepath: str) -> pd.DataFrame: """ CSV 形式の震源データを読み込む。 フォーマット例（ヘッダー行は自動スキップ）: 2024/01/01,00:01:02.2,35.167,136.662,10,2.1,岐阜県美濃中西部 列順: 日付, 時刻, 緯度, 経度, 深さ(km), マグニチュード[, 震源地] """ records = [] with open(filepath, encoding="utf-8", errors="replace") as f: for line in f: line = line.strip() if not line or line.startswith("#"): continue parts = line.split(",") if len(parts) < 6: continue try: dt_str = parts[0].strip() + " " + parts[1].strip() # 秒の小数点を丸める dt = pd.to_datetime(dt_str, errors="coerce") if pd.isnull(dt): continue lat = float(parts[2]) lon = float(parts[3]) depth = float(parts[4]) mag = float(parts[5]) records.append(dict(datetime=dt.to_pydatetime(), lat=lat, lon=lon, depth=depth, mag=mag)) except (ValueError, IndexError): pass return pd.DataFrame(records) def parse_general(filepath: str) -> pd.DataFrame: """ 汎用パーサー: 空白区切りで YYYY MM DD HH MM SS lat lon depth mag の順に並んでいると仮定する（秒は整数でも小数でも可）。 """ records = [] with open(filepath, encoding="utf-8", errors="replace") as f: for line in f: line = line.strip() if not line or line.startswith("#"): continue tokens = line.split() if len(tokens) < 10: continue try: yr, mo, dy = int(tokens[0]), int(tokens[1]), int(tokens[2]) hr, mi = int(tokens[3]), int(tokens[4]) sc = float(tokens[5]) lat = float(tokens[6]) lon = float(tokens[7]) depth = float(tokens[8]) mag = float(tokens[9]) sec_int = int(sc) usec = int((sc - sec_int) * 1_000_000) dt = datetime(yr, mo, dy, hr, mi, min(sec_int, 59), usec) records.append(dict(datetime=dt, lat=lat, lon=lon, depth=depth, mag=mag)) except (ValueError, IndexError): pass return pd.DataFrame(records) def load_earthquakes(filepath: str, fmt: str, target_date: date | None) -> pd.DataFrame: """ 指定フォーマットで震源データを読み込み、日付フィルタリングして返す。 fmt: 'fixed' | 'csv' | 'general' | 'auto' """ parsers = { "fixed": parse_jma_fixed, "csv": parse_jma_csv, "general": parse_general, } if fmt == "auto": # 拡張子で推定 ext = os.path.splitext(filepath)[1].lower() fmt = "csv" if ext == ".csv" else "fixed" df = parsers[fmt](filepath) if df.empty: fmt = "general" df = parsers[fmt](filepath) else: df = parsers[fmt](filepath) if df.empty: print("[エラー] 震源データを読み込めませんでした。") print(" --format オプションで 'fixed' / 'csv' / 'general' を指定してみてください。") sys.exit(1) df = df.sort_values("datetime").reset_index(drop=True) if target_date is not None: mask = df["datetime"].apply(lambda dt: dt.date()) == target_date df = df[mask].reset_index(drop=True) if df.empty: print(f"[警告] {target_date} の地震データがありません。") sys.exit(1) # MIN_MAG 未満を除外 before = len(df) df = df[df["mag"] >= MIN_MAG].reset_index(drop=True) print(f"[情報] 読み込み件数: {before} 件 → M{MIN_MAG}以上: {len(df)} 件") return df # ============================================================ # 地図描画ベース # ============================================================ def draw_base_map(ax: plt.Axes, coast_segs: list, border_segs: list) -> None: """海岸線・県境セグメントを描画する。""" # 県境（薄いグレー点線、細線） for seg in border_segs: if len(seg) < 2: continue lons = [p[0] for p in seg] lats = [p[1] for p in seg] ax.plot(lons, lats, color="#666666", linewidth=0.4, linestyle="--", zorder=2) # 海岸線（明るいグレー実線、やや太め） for seg in coast_segs: if len(seg) < 2: continue lons = [p[0] for p in seg] lats = [p[1] for p in seg] ax.plot(lons, lats, color="#cccccc", linewidth=0.7, zorder=3) def build_legend(ax: plt.Axes) -> None: """深さの凡例とマグニチュードの凡例を追加する。""" depth_labels = ["0–30 km", "30–70 km", "70–150 km", "150–300 km", "300–700 km", "700+ km"] depth_handles = [ plt.scatter([], [], s=30, c=DEPTH_COLORS[i], label=depth_labels[i], zorder=5) for i in range(len(DEPTH_COLORS)) ] mag_sizes = [1, 3, 5, 7] mag_handles = [ plt.scatter([], [], s=mag_to_size(m), c="gray", alpha=0.7, label=f"M {m}", zorder=5) for m in mag_sizes ] leg1 = ax.legend(handles=depth_handles, title="深さ", loc="lower left", fontsize=7, title_fontsize=8, framealpha=0.85, markerscale=1) ax.add_artist(leg1) ax.legend(handles=mag_handles, title="マグニチュード", loc="lower right", fontsize=7, title_fontsize=8, framealpha=0.85, markerscale=1) # ============================================================ # メインアプリ: アニメーション + ステップ両対応 # ============================================================ class QuakePlotter: def __init__(self, df: pd.DataFrame, coast_segs: list, border_segs: list, mode: str = "anim"): self.df = df self.coast_segs = coast_segs self.border_segs = border_segs self.mode = mode # "anim" or "step" self.idx = 0 # 現在のプロット済み件数 self.anim_obj = None self.paused = False self._setup_figure() # ---------------------------------------------------------- def _setup_figure(self): # OS変数に応じてフォントを切り替え _fonts = { "windows": ["Yu Gothic", "Meiryo", "MS Gothic"], "linux": ["IPAexGothic", "Noto Sans CJK JP"], } matplotlib.rcParams["font.family"] = _fonts.get(OS, _fonts["windows"]) matplotlib.rcParams["axes.unicode_minus"] = False # マイナス記号化け防止 fig = plt.figure(figsize=(11, 9)) fig.patch.set_facecolor("#1a1a2e") # メイン地図エリア ax_map = fig.add_axes([0.05, 0.12, 0.88, 0.82]) ax_map.set_facecolor("#0d1b2a") ax_map.set_xlim(MAP_EXTENT[0], MAP_EXTENT[1]) ax_map.set_ylim(MAP_EXTENT[2], MAP_EXTENT[3]) ax_map.set_aspect("equal") ax_map.tick_params(colors="white", labelsize=8) for spine in ax_map.spines.values(): spine.set_edgecolor("#444444") # グリッド ax_map.set_xticks(np.arange(124, 146, 4)) ax_map.set_yticks(np.arange(26, 46, 4)) ax_map.grid(color="#333366", linewidth=0.4, linestyle="--", zorder=1) ax_map.set_xlabel("経度 (°E)", color="white", fontsize=9) ax_map.set_ylabel("緯度 (°N)", color="white", fontsize=9) # 海岸線・県境描画 draw_base_map(ax_map, self.coast_segs, self.border_segs) # タイトル・情報テキスト date_str = self.df["datetime"].iloc[0].strftime("%Y年%m月%d日") ax_map.set_title(f"日本 地震分布 {date_str} (気象庁震源データ 暫定値)", color="white", fontsize=11, pad=8) self.info_text = ax_map.text( 0.02, 0.97, "", transform=ax_map.transAxes, color="white", fontsize=8, va="top", ha="left", bbox=dict(boxstyle="round,pad=0.3", facecolor="#000033", alpha=0.7), zorder=10) build_legend(ax_map) # 散布プロット用コンテナ（1件ずつ追加するためリスト管理） self.sc_list = [] self.ax_map = ax_map self.fig = fig # --- コントロールボタン --- btn_color = "#2c3e50" btn_hcolor = "#3d5a80" if self.mode == "step": ax_prev = fig.add_axes([0.30, 0.02, 0.10, 0.05]) ax_next = fig.add_axes([0.41, 0.02, 0.10, 0.05]) ax_jump = fig.add_axes([0.52, 0.02, 0.10, 0.05]) ax_rst = fig.add_axes([0.63, 0.02, 0.10, 0.05]) self.btn_prev = Button(ax_prev, "<< 前へ", color=btn_color, hovercolor=btn_hcolor) self.btn_next = Button(ax_next, "次へ >>", color=btn_color, hovercolor=btn_hcolor) self.btn_jump = Button(ax_jump, "+10件", color=btn_color, hovercolor=btn_hcolor) self.btn_rst = Button(ax_rst, "リセット", color=btn_color, hovercolor=btn_hcolor) for btn in [self.btn_prev, self.btn_next, self.btn_jump, self.btn_rst]: btn.label.set_color("white") self.btn_prev.on_clicked(self._on_prev) self.btn_next.on_clicked(self._on_next) self.btn_jump.on_clicked(self._on_jump) self.btn_rst.on_clicked(self._on_reset) # モード切替ボタン ax_sw = fig.add_axes([0.74, 0.02, 0.12, 0.05]) self.btn_sw = Button(ax_sw, "Anim", color="#1a5276", hovercolor="#2471a3") self.btn_sw.label.set_color("white") self.btn_sw.on_clicked(self._switch_to_anim) self._plot_up_to(0) else: # anim mode ax_pp = fig.add_axes([0.30, 0.02, 0.12, 0.05]) ax_rst = fig.add_axes([0.43, 0.02, 0.10, 0.05]) self.btn_pp = Button(ax_pp, "[停止]", color=btn_color, hovercolor=btn_hcolor) self.btn_rst = Button(ax_rst, "リセット", color=btn_color, hovercolor=btn_hcolor) for btn in [self.btn_pp, self.btn_rst]: btn.label.set_color("white") self.btn_pp.on_clicked(self._toggle_pause) self.btn_rst.on_clicked(self._on_reset) # モード切替ボタン ax_sw = fig.add_axes([0.74, 0.02, 0.12, 0.05]) self.btn_sw = Button(ax_sw, "Step", color="#1a5276", hovercolor="#2471a3") self.btn_sw.label.set_color("white") self.btn_sw.on_clicked(self._switch_to_step) self._start_animation() # スピードスライダー (アニメ用) if self.mode == "anim": ax_sl = fig.add_axes([0.15, 0.03, 0.12, 0.02]) self.slider = Slider(ax_sl, "速度", 50, 2000, valinit=ANIM_INTERVAL_MS, valstep=50, color="#3d5a80") self.slider.label.set_color("white") self.slider.valtext.set_color("white") self.slider.on_changed(self._on_speed_change) # ---------------------------------------------------------- # プロット関連 # ---------------------------------------------------------- def _plot_one(self, i: int): """i 番目の地震を1点プロットする。""" row = self.df.iloc[i] color = depth_color(row["depth"]) size = mag_to_size(row["mag"]) sc = self.ax_map.scatter( row["lon"], row["lat"], s=size, c=color, alpha=0.75, edgecolors="white", linewidths=0.3, zorder=5) self.sc_list.append(sc) self._update_info(i) def _plot_up_to(self, n: int): """0 から n-1 番目まで一括プロット（ステップモードのリセット後など）。""" for sc in self.sc_list: sc.remove() self.sc_list.clear() for i in range(n): row = self.df.iloc[i] color = depth_color(row["depth"]) size = mag_to_size(row["mag"]) sc = self.ax_map.scatter( row["lon"], row["lat"], s=size, c=color, alpha=0.75, edgecolors="white", linewidths=0.3, zorder=5) self.sc_list.append(sc) self._update_info(n - 1 if n > 0 else 0, n) def _update_info(self, i: int, shown: int | None = None): """情報テキストを更新する。""" if shown is None: shown = i + 1 row = self.df.iloc[i] total = len(self.df) txt = (f"No. {shown:>4} / {total}\n" f"日時: {row['datetime'].strftime('%H:%M:%S')}\n" f"緯度: {row['lat']:.3f}°N 経度: {row['lon']:.3f}°E\n" f"深さ: {row['depth']:.1f} km M {row['mag']:.1f}") self.info_text.set_text(txt) # ---------------------------------------------------------- # アニメーション # ---------------------------------------------------------- def _start_animation(self): self.idx = 0 def update(frame): if self.paused or self.idx >= len(self.df): return self.sc_list self._plot_one(self.idx) self.idx += 1 self.fig.canvas.draw_idle() return self.sc_list self.anim_obj = matplotlib.animation.FuncAnimation( self.fig, update, interval=ANIM_INTERVAL_MS, blit=False, cache_frame_data=False) def _toggle_pause(self, event): self.paused = not self.paused self.btn_pp.label.set_text("[再生]" if self.paused else "[停止]") self.fig.canvas.draw_idle() def _on_speed_change(self, val): if self.anim_obj is not None: self.anim_obj.event_source.interval = int(val) # ---------------------------------------------------------- # ステップ実行 # ---------------------------------------------------------- def _on_next(self, event): if self.idx < len(self.df): self._plot_one(self.idx) self.idx += 1 self.fig.canvas.draw_idle() def _on_prev(self, event): if self.idx > 1: self.idx -= 1 self._plot_up_to(self.idx) self.fig.canvas.draw_idle() def _on_jump(self, event): n = min(self.idx + 10, len(self.df)) self._plot_up_to(n) self.idx = n self.fig.canvas.draw_idle() def _on_reset(self, event): for sc in self.sc_list: sc.remove() self.sc_list.clear() self.idx = 0 self.info_text.set_text("") self.fig.canvas.draw_idle() # ---------------------------------------------------------- # モード切替（簡易: ウィンドウを閉じて再起動を促す） # ---------------------------------------------------------- def _switch_to_step(self, event): if self.anim_obj: self.anim_obj.event_source.stop() plt.close(self.fig) app = QuakePlotter(self.df, self.coast_segs, self.border_segs, mode="step") plt.show() def _switch_to_anim(self, event): plt.close(self.fig) app = QuakePlotter(self.df, self.coast_segs, self.border_segs, mode="anim") plt.show() def show(self): plt.show() # ============================================================ # サンプルデータ生成 (データファイルなしで動作確認) # ============================================================ def generate_sample_data(n: int = 80) -> pd.DataFrame: """テスト用ランダム震源データを生成する。""" rng = np.random.default_rng(42) base = datetime(2024, 1, 1, 0, 0, 0) times = pd.to_timedelta(np.sort(rng.uniform(0, 86400, n)), unit="s") lats = rng.uniform(30, 44, n) lons = rng.uniform(129, 145, n) depths = rng.choice([10, 30, 50, 100, 200, 400], n, p=[0.30, 0.25, 0.20, 0.15, 0.07, 0.03]) mags = rng.uniform(1.0, 6.5, n) dts = [base + td for td in times] return pd.DataFrame(dict(datetime=dts, lat=lats, lon=lons, depth=depths.astype(float), mag=mags)) # ============================================================ # エントリポイント # ============================================================ def main(): parser = argparse.ArgumentParser( description="気象庁震源データ（暫定値）を日本地図にプロットする", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=textwrap.dedent(""" フォーマット指定 (--format): fixed : 気象庁標準固定長 例) 2024 1 1 0 1 2.17 35 10.02 136 39.72 10 2.1 csv : CSV形式 例) 2024/01/01,00:01:02.2,35.167,136.662,10,2.1,震源地名 general : 汎用空白区切り (YYYY MM DD HH MM SS lat lon dep mag) auto : 拡張子から自動判定 (デフォルト) データ入手先 (気象庁): https://www.data.jma.go.jp/svd/eqev/data/bulletin/hypo.html """)) parser.add_argument("datafile", nargs="?", help="震源データファイル (省略するとサンプルデータで起動)") parser.add_argument("--format", choices=["fixed","csv","general","auto"], default="auto", help="データフォーマット (デフォルト: auto)") parser.add_argument("--date", default=None, help="抽出する日付 YYYY-MM-DD (省略するとファイル全体)") parser.add_argument("--mode", choices=["anim","step"], default="anim", help="起動モード: anim=アニメーション, step=ステップ実行 (デフォルト: anim)") parser.add_argument("--coastline", default=COASTLINE_FILE, help="海岸線データファイルパス (デフォルト: Japan_CL_shore.dat)") parser.add_argument("--boundary", default=BOUNDARY_FILE, help="県境データファイルパス (デフォルト: Japan_PB.dat)") args = parser.parse_args() # 地図データ読み込み print("[情報] 海岸線データ読み込み中...") coast_segs = load_boundary(args.coastline) if not coast_segs: print("[警告] 海岸線データなしで続行します。") print("[情報] 県境データ読み込み中...") border_segs = load_boundary(args.boundary) if not border_segs: print("[警告] 県境データなしで続行します。") # 震源データ if args.datafile is None: print("[情報] データファイル未指定 → サンプルデータで起動します。") df = generate_sample_data() else: target_date = (date.fromisoformat(args.date) if args.date else None) df = load_earthquakes(args.datafile, args.format, target_date) # 起動 app = QuakePlotter(df, coast_segs, border_segs, mode=args.mode) app.show() if __name__ == "__main__": import matplotlib.animation main()