import html import json import re from functools import lru_cache from pathlib import Path import folium import gradio as gr from huggingface_hub import hf_hub_download, list_repo_files HERE = Path(__file__).resolve().parent FIXTURES = HERE / "fixtures" STATUS_DATASET_ID = "oceanicdayi/eew_status" EVENT_DATASET_ID = "oceanicdayi/eew_hermes_dashboard" WAVEFORM_DATASET_ID = "oceanicdayi/eew_hermes_dashboard" DEFAULT_STATUS = "status/eew_status_report.json" DEFAULT_STATUS_ALT = "eew_status_report.json" EVENT_STATUS_PREFIX = "status/" EVENT_STATUS_PATH = "status/eew_status_report.json" REP_SUMMARY_PATH = "status/rep_summary_detailed.md" WAVEFORM_PREFIX = "tsmip/waveform/" WAVEFORM_IMAGE_EXTS = (".png", ".jpg", ".jpeg", ".webp") MAX_WAVEFORM_IMAGES = 48 LAT_KEYS = { "lat", "latitude", "event_lat", "event_latitude", "epicenter_lat", "epicenter_latitude", "origin_lat", "origin_latitude", "hypo_lat", "hypocenter_lat", "eq_lat", "eqlat", "evlat", } LON_KEYS = { "lon", "lng", "longitude", "event_lon", "event_lng", "event_longitude", "epicenter_lon", "epicenter_lng", "epicenter_longitude", "origin_lon", "origin_lng", "hypo_lon", "hypocenter_lon", "eq_lon", "eqlon", "evlon", } MAG_KEYS = {"magnitude", "mag", "ml", "mw", "m", "mall", "mpd", "mtc", "M", "Mpd", "Mtc", "Mall"} DEPTH_KEYS = {"depth", "depth_km", "dep", "hypo_depth", "hypocenter_depth"} TIME_KEYS = {"time", "origin_time", "event_time", "timestamp", "datetime", "report_time", "trigger_time"} LOCATION_KEYS = {"location", "area", "epicenter", "place", "region", "description"} SOURCE_KEYS = {"source_file", "file", "filename", "source"} def esc(value): return html.escape("" if value is None else str(value)) def to_float(value): try: if value is None: return None text = str(value).strip().replace(",", "") match = re.search(r"-?\d+(?:\.\d+)?", text) if not match: return None return float(match.group(0)) except Exception: return None def first_value(obj, keys): if not isinstance(obj, dict): return None lower_map = {str(k).lower(): k for k in obj.keys()} for key in keys: actual = lower_map.get(str(key).lower()) if actual is not None: value = obj.get(actual) if value not in (None, ""): return value return None @lru_cache(maxsize=1) def status_files(): try: return list_repo_files(STATUS_DATASET_ID, repo_type="dataset") except Exception: return [] @lru_cache(maxsize=1) def event_status_files(): try: return list_repo_files(EVENT_DATASET_ID, repo_type="dataset") except Exception: return [] @lru_cache(maxsize=1) def waveform_files(): try: return list_repo_files(WAVEFORM_DATASET_ID, repo_type="dataset") except Exception: return [] def status_choices(): files = status_files() choices = [] for name in [DEFAULT_STATUS, DEFAULT_STATUS_ALT]: if name in files and name not in choices: choices.append(name) if not choices: choices.append(DEFAULT_STATUS) choices += [f for f in sorted(files, reverse=True) if f.endswith(".json") and f not in choices] return choices or ["fixtures://normal_event.json"] def event_status_choices(): files = event_status_files() choices = [f for f in sorted(files, reverse=True) if f.startswith(EVENT_STATUS_PREFIX) and f.lower().endswith(".json")] if EVENT_STATUS_PATH in choices: choices.remove(EVENT_STATUS_PATH) choices.insert(0, EVENT_STATUS_PATH) return choices or [EVENT_STATUS_PATH] def waveform_image_files(): files = waveform_files() return [ f for f in sorted(files) if f.startswith(WAVEFORM_PREFIX) and f.lower().endswith(WAVEFORM_IMAGE_EXTS) ] def refresh_status(): status_files.cache_clear() choices = status_choices() return gr.update(choices=choices, value=choices[0]), f"已重新讀取 {STATUS_DATASET_ID}：{len(choices)} 筆" def refresh_all_event_files(): event_status_files.cache_clear() choices = event_status_choices() return f"已重新讀取 {EVENT_DATASET_ID}/{EVENT_STATUS_PREFIX}：{len(choices)} 個 JSON 檔，會以表格同時顯示所有可解析事件。" def refresh_waveform_files(): waveform_files.cache_clear() files = waveform_image_files() return f"已重新讀取 {WAVEFORM_DATASET_ID}/{WAVEFORM_PREFIX}：找到 {len(files)} 張波形圖片。" def load_system_status(source): if source and source.startswith("fixtures://"): with open(FIXTURES / source.split("://", 1)[1], "r", encoding="utf-8") as f: return json.load(f), source errors = [] for name in [source, DEFAULT_STATUS, DEFAULT_STATUS_ALT]: if not name: continue try: path = hf_hub_download(repo_id=STATUS_DATASET_ID, filename=name, repo_type="dataset") with open(path, "r", encoding="utf-8") as f: return json.load(f), name except Exception as exc: errors.append(str(exc)) raise RuntimeError("; ".join(errors[-2:])) def load_event_status(source): path = hf_hub_download(repo_id=EVENT_DATASET_ID, filename=source, repo_type="dataset") with open(path, "r", encoding="utf-8") as f: return json.load(f), source def load_rep_summary_markdown(): try: path = hf_hub_download(repo_id=EVENT_DATASET_ID, filename=REP_SUMMARY_PATH, repo_type="dataset") with open(path, "r", encoding="utf-8") as f: text = f.read() return f"### 解析文字資訊\n\n來源：`{EVENT_DATASET_ID}/{REP_SUMMARY_PATH}`\n\n" + text except Exception as exc: return f"### 解析文字資訊\n\n⚠️ 無法讀取 `{EVENT_DATASET_ID}/{REP_SUMMARY_PATH}`：{exc}" def status_level(text): t = str(text).lower() if any(k in t for k in ["alive", "up", "ok", "running", "healthy", "success", "available", "exists", "運行", "正常"]): return "ok" if any(k in t for k in ["warn", "warning", "partial", "degraded", "警告"]): return "warn" if any(k in t for k in ["down", "fail", "error", "exited", "stopped", "missing", "停止"]): return "bad" return "neutral" def normalize_status(info): if isinstance(info, dict): if "alive" in info or "is_alive" in info: value = info.get("alive", info.get("is_alive")) return "alive" if str(value).strip().lower() in {"true", "1", "yes", "alive"} else "not alive" return info.get("status") or info.get("state") or info.get("health") or info.get("ok") or "unknown" return info def module_items(payload): rows = [] for key in ["containers", "container_status", "container", "docker", "modules", "module_status", "services", "checks"]: value = payload.get(key) if isinstance(value, dict): for name, info in value.items(): rows.append((str(name), str(normalize_status(info)))) elif isinstance(value, list): for i, info in enumerate(value): if isinstance(info, dict): name = info.get("name") or info.get("module") or info.get("container") or f"module_{i + 1}" rows.append((str(name), str(normalize_status(info)))) else: rows.append((f"module_{i + 1}", str(info))) header = payload.get("latest_rep_header") or {} if isinstance(header, dict) and header.get("file"): rows.append(("Earthworm EEW header", "available")) rep_files = payload.get("latest_rep_files") if isinstance(rep_files, list): rows.append((".rep files", f"{len(rep_files)} files" if rep_files else "missing")) warning = payload.get("earthworm_log_warning") rows.append(("Earthworm log", "warning" if warning else "ok")) return rows or [("status payload", "loaded")] def render_status(source): try: payload, label = load_system_status(source) except Exception as exc: with open(FIXTURES / "normal_event.json", "r", encoding="utf-8") as f: payload = json.load(f) label = f"fallback: {exc}" alive_cards = [] info_cards = [] for name, status in module_items(payload): if str(status).strip().lower() == "alive": alive_cards.append(f"
{esc(name)}{esc(status)}
") else: cls = status_level(status) info_cards.append(f"
{esc(name)}{esc(status)}
{esc(status)}
") alive_html = "".join(alive_cards) or "
目前沒有 alive 型式狀態。
" info_html = "".join(info_cards) or "
沒有其他狀態資訊。
" source_text = f"{STATUS_DATASET_ID} / {label}" return f"""
EEW STATUS

系統狀態

來源{esc(source_text)}
⌁
Alive 模組綠燈 僅 alive 使用燈號
{alive_html}
其他狀態資訊 非 alive 不使用燈號
{info_html}
""" def parse_rep_event(payload): header = payload.get("latest_rep_header") or {} lines = header.get("head") or [] if not isinstance(lines, list): return None event = {"source_file": header.get("file"), "stations": []} for i, line in enumerate(lines): text = str(line) if text.strip().lower().startswith("year") and i + 1 < len(lines): parts = str(lines[i + 1]).split() if len(parts) >= 14: year, month, day, hour, minute = parts[0:5] sec = parts[5] event.update({ "time": f"{int(year):04d}/{int(month):02d}/{int(day):02d} {int(hour):02d}:{int(minute):02d}:{float(sec):05.2f}", "lat": to_float(parts[6]), "lon": to_float(parts[7]), "depth_km": to_float(parts[8]), "magnitude": to_float(parts[9]), "mpd": to_float(parts[12]) if len(parts) > 12 else None, "mtc": to_float(parts[13]) if len(parts) > 13 else None, "process_time": to_float(parts[14]) if len(parts) > 14 else None, }) if text.strip().lower().startswith("sta"): for row in lines[i + 1:]: cols = str(row).split() if len(cols) >= 6: sta_lat = to_float(cols[4]) sta_lon = to_float(cols[5]) if sta_lat is not None and sta_lon is not None: event["stations"].append({"station": cols[0], "lat": sta_lat, "lon": sta_lon}) return event if event.get("lat") is not None and event.get("lon") is not None else None def score_event_candidate(obj, path, lat, lon): path_text = ".".join(path).lower() score = 0 keys = {str(k).lower() for k in obj.keys()} if keys.intersection({k.lower() for k in MAG_KEYS}): score += 6 if keys.intersection({k.lower() for k in DEPTH_KEYS}): score += 5 if keys.intersection({k.lower() for k in TIME_KEYS}): score += 3 if any(word in path_text for word in ["event", "eew", "earthquake", "latest", "report", "hypo"]): score += 3 if any(word in path_text for word in ["station", "stations", "sta"]): score -= 5 if 18 <= lat <= 28 and 118 <= lon <= 124: score += 2 return score def collect_station_candidates(obj, path=None, out=None): if path is None: path = [] if out is None: out = [] if isinstance(obj, dict): lat = to_float(first_value(obj, LAT_KEYS)) lon = to_float(first_value(obj, LON_KEYS)) path_text = ".".join(path).lower() if lat is not None and lon is not None and any(word in path_text for word in ["station", "stations", "sta"]): name = first_value(obj, {"station", "sta", "name", "code", "id"}) or f"STA{len(out) + 1}" out.append({"station": str(name), "lat": lat, "lon": lon}) for key, value in obj.items(): collect_station_candidates(value, path + [str(key)], out) elif isinstance(obj, list): for idx, value in enumerate(obj[:300]): collect_station_candidates(value, path + [str(idx)], out) return out def recursive_event_candidates(obj, path=None, out=None): if path is None: path = [] if out is None: out = [] if isinstance(obj, dict): lat = to_float(first_value(obj, LAT_KEYS)) lon = to_float(first_value(obj, LON_KEYS)) if lat is not None and lon is not None: score = score_event_candidate(obj, path, lat, lon) out.append({ "time": first_value(obj, TIME_KEYS), "lat": lat, "lon": lon, "depth_km": to_float(first_value(obj, DEPTH_KEYS)), "magnitude": to_float(first_value(obj, MAG_KEYS)), "location": first_value(obj, LOCATION_KEYS), "source_file": first_value(obj, SOURCE_KEYS), "stations": [], "_score": score, }) for key, value in obj.items(): recursive_event_candidates(value, path + [str(key)], out) elif isinstance(obj, list): for idx, value in enumerate(obj[:300]): recursive_event_candidates(value, path + [str(idx)], out) return out def events_from_payload(payload): parsed = parse_rep_event(payload) if parsed: return [parsed] candidates = recursive_event_candidates(payload) filtered = [item for item in candidates if item.get("_score", 0) >= 0] if not filtered and candidates: filtered = sorted(candidates, key=lambda item: item.get("_score", 0), reverse=True)[:1] stations = collect_station_candidates(payload) events = [] seen = set() for item in sorted(filtered, key=lambda e: e.get("_score", 0), reverse=True): key = ( round(item.get("lat") or 0, 4), round(item.get("lon") or 0, 4), str(item.get("time") or ""), str(item.get("magnitude") or ""), ) if key in seen: continue seen.add(key) item.pop("_score", None) item["stations"] = stations events.append(item) return events def load_all_events(): files = event_status_choices() events = [] unresolved = [] for filename in files: try: payload, label = load_event_status(filename) parsed_events = events_from_payload(payload) if parsed_events: for idx, event in enumerate(parsed_events, start=1): event["source_json"] = label event["event_index"] = idx events.append(event) else: unresolved.append({"file": label, "reason": "無可解析座標"}) except Exception as exc: unresolved.append({"file": filename, "reason": str(exc)}) return events, unresolved, files def event_table_html(events, unresolved, files): rows = [] for idx, event in enumerate(events, start=1): mag = event.get("magnitude") dep = event.get("depth_km") lat = event.get("lat") lon = event.get("lon") rows.append( "" f"{idx}" f"{esc(event.get('source_json') or event.get('source_file') or '—')}" f"{esc(event.get('time') or '—')}" f"{esc(f'M {mag:.2f}' if mag is not None else '—')}" f"{esc(f'{dep:.1f}' if dep is not None else '—')}" f"{esc(f'{lat:.4f}' if lat is not None else '—')}" f"{esc(f'{lon:.4f}' if lon is not None else '—')}" f"{esc(event.get('location') or '—')}" f"{len(event.get('stations') or [])}" "" ) body = "".join(rows) or "目前 status 資料夾內沒有可解析的地震事件經緯度資料。" unresolved_html = "" if unresolved: items = "".join(f"
  • {esc(item['file'])}：{esc(item['reason'])}
    • " for item in unresolved) unresolved_html = f"
    未解析檔案 {len(unresolved)} 筆
    " return f"""

    全部地震事件表格

    已掃描 {len(files)} 個 status JSON；成功解析 {len(events)} 筆事件。

    {body}
    #來源檔案發震時間規模深度 km緯度經度位置 / 描述測站數
    {unresolved_html}
    """ def all_events_folium_map(events): valid = [event for event in events if event.get("lat") is not None and event.get("lon") is not None] if not valid: return "
    無可繪製的地震事件座標
    " center_lat = sum(event["lat"] for event in valid) / len(valid) center_lon = sum(event["lon"] for event in valid) / len(valid) fmap = folium.Map(location=[center_lat, center_lon], zoom_start=7, tiles="OpenStreetMap", control_scale=True) all_station_keys = set() for idx, event in enumerate(valid, start=1): lat = event["lat"] lon = event["lon"] mag = event.get("magnitude") dep = event.get("depth_km") mag_text = f"{mag:.2f}" if isinstance(mag, (int, float)) else "—" dep_text = f"{dep:.1f}" if isinstance(dep, (int, float)) else "—" radius = max(8, min(24, 6 + (mag or 0) * 2.5)) popup = folium.Popup( f"事件 {idx}
    來源：{esc(event.get('source_json') or '—')}
    時間：{esc(event.get('time') or '—')}
    規模：M {mag_text}
    深度：{dep_text} km
    座標：{lat:.4f}, {lon:.4f}", max_width=360, ) folium.CircleMarker( location=[lat, lon], radius=radius, color="#b91c1c", weight=3, fill=True, fill_color="#ef4444", fill_opacity=0.72, tooltip=f"地震事件 {idx}", popup=popup, ).add_to(fmap) folium.Marker(location=[lat, lon], icon=folium.Icon(color="red", icon="exclamation-sign"), tooltip=f"地震事件 {idx}").add_to(fmap) for station in event.get("stations") or []: sta_lat = station.get("lat") sta_lon = station.get("lon") if sta_lat is None or sta_lon is None: continue station_key = (round(sta_lat, 4), round(sta_lon, 4), str(station.get("station", ""))) if station_key in all_station_keys: continue all_station_keys.add(station_key) folium.CircleMarker( location=[sta_lat, sta_lon], radius=4, color="#1d4ed8", weight=2, fill=True, fill_color="#3b82f6", fill_opacity=0.65, tooltip=f"測站：{esc(station.get('station', 'station'))}", ).add_to(fmap) folium.LayerControl().add_to(fmap) map_html = fmap.get_root().render() return f"" def render_all_events(): events, unresolved, files = load_all_events() return event_table_html(events, unresolved, files), all_events_folium_map(events), load_rep_summary_markdown() def waveform_url(filename): return f"https://huggingface.co/datasets/{WAVEFORM_DATASET_ID}/resolve/main/{filename}" def waveform_thumb(filename, idx): title = Path(filename).name return f""" {esc(title)} {idx:02d} """ def waveform_card(filename, idx): title = Path(filename).name url = waveform_url(filename) return f"""
    {esc(title)}
    {idx:02d}. {esc(title)} {esc(filename)} 開啟原圖
    """ def render_waveform_gallery(): waveform_files.cache_clear() files = waveform_image_files() shown = files[:MAX_WAVEFORM_IMAGES] featured = shown[0] if shown else None feature_html = "" if featured: feature_html = f"""
    FEATURED

    {esc(Path(featured).name)}

    {esc(featured)}

    開啟重點波形原圖
    {esc(Path(featured).name)}
    """ else: feature_html = "
    目前資料夾內沒有可展示的波形圖片。
    " thumbs = "".join(waveform_thumb(filename, idx) for idx, filename in enumerate(shown, start=1)) cards = "".join(waveform_card(filename, idx) for idx, filename in enumerate(shown, start=1)) more_note = "" if len(files) <= len(shown) else f"
    另有 {len(files) - len(shown)} 張未顯示，可至 Hugging Face 資料夾查看。
    " html_block = f"""

    TSMIP 波形展示

    每次載入皆重新讀取：{esc(WAVEFORM_DATASET_ID)}/{esc(WAVEFORM_PREFIX)}

    共 {len(files)} 張顯示 {len(shown)} 張上限 {MAX_WAVEFORM_IMAGES}
    {feature_html}
    快速檢視 {len(shown)} 張
    {thumbs}
    完整圖庫 Gallery
    {cards}
    {more_note}
    """ return f"✅ 已重新讀取並載入波形展示：{WAVEFORM_DATASET_ID}/{WAVEFORM_PREFIX}，共 {len(files)} 張圖片。", html_block status_opts = status_choices() with gr.Blocks(title="EEW Dashboard") as demo: gr.Markdown("# EEW Dashboard\n系統狀態、全部地震事件與 TSMIP 波形展示。波形展示會每次重新讀取 `tsmip/waveform/`，並以重點圖、縮圖列與圖庫排版呈現。") with gr.Tab("系統狀態"): with gr.Row(): s = gr.Dropdown(choices=status_opts, value=status_opts[0], label="Status file") sr = gr.Button("重新讀取狀態清單") sl = gr.Button("載入狀態") sm = gr.Markdown() lights = gr.HTML() sr.click(refresh_status, outputs=[s, sm]) sl.click(render_status, inputs=s, outputs=lights) s.change(render_status, inputs=s, outputs=lights) demo.load(render_status, inputs=s, outputs=lights) with gr.Tab("地震事件"): with gr.Row(): er = gr.Button("重新讀取事件清單") el = gr.Button("載入全部事件") em = gr.Markdown(f"事件資料來源：`{EVENT_DATASET_ID}/{EVENT_STATUS_PREFIX}`；解析文字資訊：`{EVENT_DATASET_ID}/{REP_SUMMARY_PATH}`") event_table = gr.HTML() event_map = gr.HTML() event_summary = gr.Markdown() er.click(refresh_all_event_files, outputs=em) el.click(render_all_events, outputs=[event_table, event_map, event_summary]) demo.load(render_all_events, outputs=[event_table, event_map, event_summary]) with gr.Tab("靜態波形"): wm = gr.Markdown(f"波形資料夾來源：`{WAVEFORM_DATASET_ID}/{WAVEFORM_PREFIX}`") with gr.Row(): wr = gr.Button("重新讀取波形資料夾") wp = gr.Button("載入波形展示") wave_msg = gr.Markdown() wave_gallery = gr.HTML() wr.click(refresh_waveform_files, outputs=wave_msg) wp.click(render_waveform_gallery, outputs=[wave_msg, wave_gallery]) demo.load(render_waveform_gallery, outputs=[wave_msg, wave_gallery]) if __name__ == "__main__": demo.launch()