#!/usr/bin/env python3 """ Sismik Sınıflandırma - HuggingFace Space (Gradio) =================================================== RandomForest modeliyle MSEED dosyalarını DEPREM / NOISE olarak sınıflandırır. HF Spaces kurulumu: 1. huggingface.co/new-space -> SDK: Gradio, Hardware: CPU basic (ücretsiz) 2. Bu dosyayı app.py olarak yükle 3. requirements.txt'i yükle 4. rf_model.pkl dosyasını da Space'e yükle (aynı dizine) """ import os import json import tempfile import warnings warnings.filterwarnings("ignore") import numpy as np import joblib import gradio as gr from obspy import read from obspy.signal.trigger import classic_sta_lta, trigger_onset from obspy.signal.filter import envelope as obspy_envelope from scipy import signal as scipy_signal from scipy.stats import kurtosis, skew import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt # --------------------------------------------------------------------------- # Özellik çıkarımı — train_rf2.py ile BİREBİR AYNI # --------------------------------------------------------------------------- BANDPASS = (0.5, 15.0) STA_S = 2.0 LTA_S = 20.0 MIN_NPTS = 200 MODEL_PATH = "rf_model.pkl" def extract_features(tr) -> dict: data = tr.data.astype(np.float64) n = len(data) df = tr.stats.sampling_rate if n < MIN_NPTS: return None data -= np.mean(data) sos = scipy_signal.butter(4, BANDPASS, btype="bandpass", fs=df, output="sos") fdata = scipy_signal.sosfiltfilt(sos, data) rms = np.sqrt(np.mean(fdata**2)) if rms < 1e-10: return None max_amp = np.max(np.abs(fdata)) peak_rms = max_amp / rms zcr = ((fdata[:-1] * fdata[1:]) < 0).sum() / n env = obspy_envelope(fdata) env_smooth = scipy_signal.savgol_filter(env, min(51, n // 4 * 2 + 1), 3) env_smoothness = np.std(np.diff(env_smooth)) / (np.mean(env_smooth) + 1e-10) cum_e = np.cumsum(env ** 2) idx90 = np.searchsorted(cum_e, 0.9 * cum_e[-1]) dur90 = idx90 / df kurt_val = kurtosis(fdata) skew_val = skew(fdata) try: cft = classic_sta_lta(fdata, int(STA_S * df), int(LTA_S * df)) sta_lta_max = np.max(cft) sta_lta_mean = np.mean(cft) triggers = trigger_onset(cft, 4.0, 1.5) num_triggers = len(triggers) except Exception: cft = np.zeros_like(fdata) sta_lta_max, sta_lta_mean, num_triggers = 0.0, 0.0, 0 nperseg = min(256, n // 2) freqs, psd = scipy_signal.welch(fdata, df, nperseg=nperseg) total_power = np.sum(psd) + 1e-30 def band_ratio(f_low, f_high): mask = (freqs >= f_low) & (freqs < f_high) return np.sum(psd[mask]) / total_power low_ratio = band_ratio(0.5, 2.0) mid1_ratio = band_ratio(2.0, 5.0) mid2_ratio = band_ratio(5.0, 10.0) high_ratio = band_ratio(10.0, 25.0) dom_freq_idx = np.argmax(psd[freqs <= 25]) dom_freq = freqs[dom_freq_idx] psd_norm = psd / total_power spec_entropy = -np.sum(psd_norm * np.log(psd_norm + 1e-30)) spec_mean_freq = np.sum(freqs * psd_norm) spec_bandwidth = np.sqrt(np.sum(((freqs - spec_mean_freq) ** 2) * psd_norm)) is_rf9f7 = 1 if "RF9F7" in (tr.stats.station or "").upper() else 0 feats = { "peak_rms": peak_rms, "zcr": zcr, "kurt": kurt_val, "skewness": skew_val, "env_smoothness": env_smoothness, "dur90_s": dur90, "sta_lta_max": sta_lta_max, "sta_lta_mean": sta_lta_mean, "num_triggers": num_triggers, "dom_freq": dom_freq, "low_ratio": low_ratio, "mid1_ratio": mid1_ratio, "mid2_ratio": mid2_ratio, "high_ratio": high_ratio, "spec_entropy": spec_entropy, "spec_bandwidth": spec_bandwidth, "is_rf9f7": is_rf9f7, } # Grafik için ek veriler (feature dict'ine karışmaması için ayrı döndürülür) return feats, fdata, cft, df # --------------------------------------------------------------------------- # Model yükleme (bir kere, global) # --------------------------------------------------------------------------- _bundle = None def get_model(): global _bundle if _bundle is None: _bundle = joblib.load(MODEL_PATH) return _bundle def make_plot(fdata, cft, df, label, deprem_proba): t = np.arange(len(fdata)) / df fig, axes = plt.subplots(2, 1, figsize=(9, 5), sharex=True) color = "#d64545" if label == "DEPREM" else "#3b6fd6" axes[0].plot(t, fdata, color=color, linewidth=0.6) axes[0].set_ylabel("Genlik (filtrelenmiş)") axes[0].set_title(f"Tahmin: {label} (DEPREM olasılığı: %{deprem_proba*100:.1f})") axes[1].plot(t, cft, color="purple", linewidth=0.8) axes[1].axhline(4.0, color="black", linestyle=":", linewidth=1, label="Tetik eşiği") axes[1].set_ylabel("STA/LTA") axes[1].set_xlabel("Zaman (s)") axes[1].legend(loc="upper right", fontsize=8) plt.tight_layout() return fig def predict_mseed(file_obj, threshold): if file_obj is None: return "Lütfen bir MSEED dosyası yükleyin.", None, None # Gradio 5.x'te gr.File genellikle doğrudan dosya yolu (str) döndürür, # eski sürümlerde .name attribute'lu bir obje döndürüyordu — ikisini de destekle filepath = file_obj if isinstance(file_obj, str) else getattr(file_obj, "name", None) if not filepath: return "Dosya yolu okunamadı.", None, None try: st = read(filepath) except Exception as e: return f"Dosya okunamadı: {e}", None, None tr = None for cha in ["EHZ", "HHZ", "BHZ", "EHN"]: sel = st.select(channel=cha) if len(sel) > 0: tr = sel[0] break if tr is None: tr = st[0] result = extract_features(tr) if result is None: return f"Özellik çıkarılamadı (dosya çok kısa, min {MIN_NPTS} örnek gerekli).", None, None feats, fdata, cft, df = result bundle = get_model() model, le, feature_cols = bundle["model"], bundle["le"], bundle["meta"]["feature_cols"] X = np.array([[feats.get(c, 0.0) for c in feature_cols]]) X = np.where(~np.isfinite(X), 0.0, X) proba = model.predict_proba(X)[0] deprem_idx = list(le.classes_).index("DEPREM") deprem_p = float(proba[deprem_idx]) label = "DEPREM" if deprem_p >= threshold else "NOISE" emoji = "🔴 DEPREM" if label == "DEPREM" else "🔵 NOISE" summary = ( f"## {emoji}\n\n" f"**DEPREM olasılığı:** %{deprem_p*100:.1f}\n\n" f"**NOISE olasılığı:** %{(1-deprem_p)*100:.1f}\n\n" f"**Kanal:** {tr.stats.network}.{tr.stats.station}.{tr.stats.location}.{tr.stats.channel}\n\n" f"**Başlangıç:** {tr.stats.starttime}\n\n" f"**Süre:** {tr.stats.npts/tr.stats.sampling_rate:.1f} sn" ) feat_table = "| Özellik | Değer |\n|---|---|\n" for k, v in feats.items(): feat_table += f"| {k} | {v:.4f} |\n" fig = make_plot(fdata, cft, df, label, deprem_p) return summary, feat_table, fig # --------------------------------------------------------------------------- # Gradio arayüzü # --------------------------------------------------------------------------- with gr.Blocks(title="Sismik Sınıflandırma") as demo: gr.Markdown( "# 🌍 Sismik Sinyal Sınıflandırma (Marmara / RaspberryShake)\n" "MSEED dosyası yükleyin, RandomForest modeli DEPREM veya NOISE olarak sınıflandırsın.\n\n" "*Not: Model şu an sınırlı (Marmara bölgesi, RaspberryShake istasyonları) veriyle eğitildi, " "gelişim aşamasındadır.*" ) with gr.Row(): with gr.Column(scale=1): file_input = gr.File(label="MSEED Dosyası (.mseed)", file_types=[".mseed", ".miniseed", ".seed"]) threshold_slider = gr.Slider( minimum=0.1, maximum=0.9, value=0.5, step=0.05, label="DEPREM karar eşiği", info="Düşürürsen daha hassas (recall↑), yükseltirsen daha seçici (precision↑)" ) submit_btn = gr.Button("Analiz Et", variant="primary") with gr.Column(scale=1): result_md = gr.Markdown(label="Sonuç") with gr.Row(): plot_output = gr.Plot(label="Dalga Formu + STA/LTA") with gr.Accordion("Detaylı özellik değerleri", open=False): feat_output = gr.Markdown() submit_btn.click( fn=predict_mseed, inputs=[file_input, threshold_slider], outputs=[result_md, feat_output, plot_output], api_name="predict", ) gr.Markdown( "---\n" "**API kullanımı:** Bu Space'e programatik erişim için `gradio_client` kütüphanesini kullanabilirsiniz:\n" "```python\n" "from gradio_client import Client, file\n" "client = Client(\"KULLANICI_ADI/SPACE_ADI\")\n" "result = client.predict(file(\"ornek.mseed\"), 0.5, api_name=\"/predict\")\n" "```" ) if __name__ == "__main__": demo.queue() demo.launch(show_api=False)