Spaces:
Running on Zero
Running on Zero
| # Hugging Face Spaces app ported from sed_ensemble_v2.ipynb | |
| # Active models: AST x3 and PaSST (AudioSet-527); optional CLAP and Silero VAD. | |
| # This intentionally does not use the separate DCASE 2024 CRNN app/checkpoint. | |
| # Required by Hugging Face ZeroGPU. Keep this before importing torch. | |
| import spaces | |
| import os | |
| import sys | |
| import glob | |
| import math | |
| import difflib | |
| import traceback | |
| import numpy as np | |
| import torch | |
| import librosa | |
| import matplotlib | |
| matplotlib.use("Agg") | |
| import matplotlib.pyplot as plt | |
| import gradio as gr | |
| DEVICE = "cuda" if torch.cuda.is_available() else "cpu" | |
| print(f"ZeroGPU-aware inference device: {DEVICE}") | |
| MAX_SEC = int(os.getenv("MAX_SEC", "60")) | |
| MAX_CELLS = int(os.getenv("MAX_CELLS", "1500")) | |
| # Limit CPU thread oversubscription on small hosted workers. | |
| try: | |
| torch.set_num_threads(max(1, min(4, os.cpu_count() or 2))) | |
| except Exception: | |
| pass | |
| # -------------------- Model loading -------------------- | |
| MODELS = {} # name -> (predict_fn(batch[B, samples]) -> probs[B, 527], sample_rate, family) | |
| from transformers import AutoFeatureExtractor, ASTForAudioClassification | |
| def load_ast(model_id): | |
| fe_ = AutoFeatureExtractor.from_pretrained(model_id) | |
| m = ASTForAudioClassification.from_pretrained(model_id).to(DEVICE).eval() | |
| def predict(batch): | |
| x = fe_(list(batch), sampling_rate=16000, return_tensors="pt")["input_values"].to(DEVICE) | |
| with torch.autocast("cuda", dtype=torch.float16, enabled=(DEVICE == "cuda")): | |
| lg = m(input_values=x).logits | |
| return torch.sigmoid(lg.float()).cpu().numpy() | |
| return predict, [m.config.id2label[i] for i in range(len(m.config.id2label))] | |
| AST_MAIN = "MIT/ast-finetuned-audioset-10-10-0.4593" | |
| AST_EXTRA = ["MIT/ast-finetuned-audioset-12-12-0.447", "MIT/ast-finetuned-audioset-14-14-0.443"] # [] to disable | |
| fn, LABELS = load_ast(AST_MAIN) | |
| assert len(LABELS) == 527 | |
| MODELS["AST"] = (fn, 16000, "AST") | |
| for mid in AST_EXTRA: | |
| try: | |
| fn, labs = load_ast(mid) | |
| assert labs == LABELS, "label order differs" | |
| MODELS[mid.split("/")[-1]] = (fn, 16000, "AST") | |
| except Exception as e: | |
| print(f"{mid} skipped:", repr(e)[:200]) | |
| # --- PaSST (hear21passt): AudioSet-527, 32 kHz, <= 10 s --- | |
| try: | |
| from hear21passt.base import get_basic_model | |
| passt = get_basic_model(mode="logits").to(DEVICE).eval() | |
| def passt_predict(batch): | |
| x = np.stack([np.pad(b, (0, 320000 - len(b))) for b in batch]).astype(np.float32) # zero-pad to 10 s | |
| return torch.sigmoid(passt(torch.from_numpy(x).to(DEVICE))).cpu().numpy() | |
| MODELS["PaSST"] = (passt_predict, 32000, "PaSST") | |
| except Exception as e: | |
| print("PaSST skipped:", repr(e)[:300]) | |
| # BEATs is intentionally not enabled here: the working notebook did not load a fine-tuned AudioSet BEATs checkpoint. | |
| # Do NOT run model forward passes at module scope on ZeroGPU. | |
| # ZeroGPU only attaches the real GPU inside a function decorated with @spaces.GPU. | |
| # The first real request to detect() exercises the loaded models. | |
| if not MODELS: | |
| raise RuntimeError("No AudioSet classifier could be loaded. Check model downloads and startup logs.") | |
| # --- CLAP: open-vocabulary categories from free text (experimental, not used for voting) --- | |
| CLAP = None | |
| NEG_PROMPTS = ["Sound of silence", "Sound of background noise", "Sound of people talking", "Sound of music", "Sound of wind"] | |
| try: | |
| from transformers import ClapModel, AutoProcessor | |
| clap_model = ClapModel.from_pretrained("laion/clap-htsat-unfused").to(DEVICE).eval() | |
| clap_proc = AutoProcessor.from_pretrained("laion/clap-htsat-unfused") | |
| def clap_logits(batch48, prompts): | |
| kw = dict(text=prompts, return_tensors="pt", padding=True) | |
| try: | |
| inp = clap_proc(audio=list(batch48), sampling_rate=48000, **kw) # recent transformers | |
| except TypeError: | |
| inp = clap_proc(audios=list(batch48), sampling_rate=48000, **kw) # compatibility fallback | |
| inp = {k: v.to(DEVICE) for k, v in inp.items()} | |
| return clap_model(**inp).logits_per_audio.float().cpu().numpy() # scaled audio-text similarity (B, K) | |
| # Do not execute CLAP inference during module import; inference belongs inside detect(). | |
| CLAP = clap_logits | |
| except Exception as e: | |
| print("CLAP skipped:", repr(e)[:300]) | |
| # --- Silero VAD: exact speech boundaries (speech only) --- | |
| VAD = None | |
| try: | |
| from silero_vad import load_silero_vad, get_speech_timestamps | |
| vad_model = load_silero_vad() | |
| def vad_segments(wav16, thr=0.5): | |
| ts = get_speech_timestamps(torch.from_numpy(np.asarray(wav16, np.float32)), vad_model, sampling_rate=16000, | |
| threshold=thr, min_speech_duration_ms=100, min_silence_duration_ms=150, | |
| return_seconds=True) | |
| return [(float(t["start"]), float(t["end"])) for t in ts] | |
| VAD = vad_segments | |
| except Exception as e: | |
| print("Silero VAD skipped:", repr(e)[:300]) | |
| if not MODELS: | |
| raise RuntimeError("No AudioSet classifier loaded. Check Space build logs and model-download access.") | |
| print("Tagger models:", list(MODELS), "| families:", sorted({v[2] for v in MODELS.values()}), | |
| "| CLAP:", CLAP is not None, "| VAD:", VAD is not None) | |
| # -------------------- Category mapping and event logic -------------------- | |
| GROUPS = { | |
| # --- humans --- | |
| "speech": ["Speech", "Conversation", "Narration, monologue", "Male speech, man speaking", | |
| "Female speech, woman speaking", "Child speech, kid speaking"], | |
| "singing": ["Singing", "Choir", "Yodeling", "Chant", "Humming", "Rapping"], | |
| "shout_scream": ["Shout", "Yell", "Screaming", "Bellow", "Whoop"], | |
| "crying": ["Crying, sobbing", "Baby cry, infant cry", "Whimper"], | |
| "laughter": ["Laughter", "Giggle", "Chuckle, chortle", "Belly laugh", "Baby laughter", "Snicker"], | |
| "cough_sneeze": ["Cough", "Throat clearing", "Sneeze", "Sniff"], | |
| "breathing_snore": ["Breathing", "Gasp", "Snoring", "Wheeze", "Pant", "Respiratory sounds", "Snort", "Sigh"], | |
| "whistling": ["Whistling"], | |
| "clapping_cheer": ["Clapping", "Applause", "Cheering"], | |
| "footsteps": ["Walk, footsteps", "Run", "Shuffle"], | |
| "crowd": ["Crowd", "Hubbub, speech noise, speech babble", "Chatter", "Children shouting", "Children playing"], | |
| # --- animals --- | |
| "dog": ["Dog", "Bark", "Bow-wow", "Yip", "Howl", "Growling"], | |
| "cat": ["Cat", "Meow", "Purr", "Caterwaul"], | |
| "cattle_moo": ["Cattle, bovinae", "Moo", "Cowbell"], | |
| "pig": ["Pig", "Oink"], | |
| "sheep_goat": ["Sheep", "Bleat", "Goat"], | |
| "horse": ["Horse", "Neigh, whinny", "Clip-clop"], | |
| "chicken_rooster": ["Chicken, rooster", "Cluck", "Crowing, cock-a-doodle-doo", "Fowl"], | |
| "duck_goose": ["Duck", "Quack", "Goose", "Honk", "Turkey", "Gobble"], | |
| "bird": ["Bird", "Bird vocalization, bird call, bird song", "Chirp, tweet", "Pigeon, dove", "Coo", "Crow", "Caw", | |
| "Owl", "Hoot", "Squawk", "Bird flight, flapping wings"], | |
| "insect": ["Insect", "Fly, housefly", "Mosquito", "Buzz", "Bee, wasp, etc.", "Cricket"], | |
| "frog": ["Frog", "Croak"], | |
| "wild_animal": ["Roar", "Roaring cats (lions, tigers)", "Wild animals"], | |
| # --- music --- | |
| "music": ["Music", "Musical instrument", "Background music"], | |
| "guitar_strings": ["Guitar", "Plucked string instrument", "Acoustic guitar", "Electric guitar", "Bass guitar", "Banjo", | |
| "Violin, fiddle", "Bowed string instrument", "Cello", "Harp", "Sitar", "Mandolin", "Ukulele"], | |
| "keyboard_instr": ["Piano", "Keyboard (musical)", "Organ", "Synthesizer", "Electric piano", "Harpsichord"], | |
| "drums_percussion": ["Drum", "Drum kit", "Percussion", "Snare drum", "Bass drum", "Cymbal", "Hi-hat", "Tabla", | |
| "Drum machine", "Tambourine", "Maraca"], | |
| "wind_brass": ["Flute", "Trumpet", "Saxophone", "Brass instrument", "Wind instrument, woodwind instrument", | |
| "Harmonica", "Accordion", "Bagpipes", "Clarinet", "Trombone", "French horn", "Didgeridoo"], | |
| "bells_chimes": ["Bell", "Church bell", "Chime", "Wind chime", "Jingle bell", "Bicycle bell", "Tubular bells", "Ding", | |
| "Ding-dong", "Tinkle"], | |
| # --- home / DESED --- | |
| "alarm_bell": ["Alarm", "Alarm clock", "Buzzer", "Smoke detector, smoke alarm", "Beep, bleep"], | |
| "phone_ring": ["Telephone", "Telephone bell ringing", "Ringtone", "Cellphone buzz, vibrating alert", "Dial tone"], | |
| "blender": ["Blender"], | |
| "dishes": ["Dishes, pots, and pans", "Cutlery, silverware", "Chink, clink", "Clink", "Glass", "Shatter", "Clang"], | |
| "shaver_toothbrush": ["Electric shaver, electric razor", "Toothbrush", "Electric toothbrush"], | |
| "frying": ["Frying (food)", "Sizzle"], | |
| "running_water": ["Water", "Pour", "Water tap, faucet", "Splash, splatter", "Trickle, dribble", "Gurgling", | |
| "Sink (filling or washing)", "Drip", "Liquid"], | |
| "toilet_flush": ["Toilet flush"], | |
| "vacuum_cleaner": ["Vacuum cleaner"], | |
| "appliance": ["Washing machine", "Mechanical fan", "Hair dryer", "Microwave oven", "Air conditioning", "Hum"], | |
| "door_knock": ["Door", "Knock", "Slam", "Doorbell", "Sliding door", "Tap"], | |
| "typing_click": ["Typing", "Computer keyboard", "Typewriter", "Click", "Clicking"], | |
| "clock_tick": ["Tick-tock", "Tick", "Clock"], | |
| "rustle_paper": ["Rustle", "Crumpling, crinkling", "Tearing", "Zipper (clothing)"], | |
| # --- environment --- | |
| "rain": ["Rain", "Raindrop", "Rain on surface"], | |
| "wind": ["Wind", "Wind noise (microphone)", "Rustling leaves"], | |
| "thunder": ["Thunderstorm", "Thunder"], | |
| "water_nature": ["Stream", "Waterfall", "Ocean", "Waves, surf"], | |
| "fire": ["Fire", "Crackle", "Wildfire"], | |
| # --- vehicles --- | |
| "car": ["Car", "Car passing by", "Motor vehicle (road)", "Skidding", "Tire squeal"], | |
| "engine": ["Engine", "Idling", "Engine starting", "Light engine (high frequency)", "Medium engine (mid frequency)", | |
| "Heavy engine (low frequency)", "Accelerating, revving, vroom"], | |
| "truck_bus": ["Truck", "Bus", "Air brake", "Reversing beeps"], | |
| "motorcycle": ["Motorcycle", "Moped"], | |
| "train": ["Train", "Rail transport", "Railroad car, train wagon", "Train horn", "Train wheels squealing", | |
| "Subway, metro, underground"], | |
| "aircraft": ["Aircraft", "Fixed-wing aircraft, airplane", "Helicopter", "Jet engine", "Propeller, airscrew", | |
| "Aircraft engine"], | |
| "boat": ["Boat, Water vehicle", "Motorboat, speedboat", "Ship"], | |
| "vehicle_horn": ["Vehicle horn, car horn, honking", "Car alarm", "Toot"], | |
| "emergency_siren": ["Siren", "Emergency vehicle", "Police car (siren)", "Ambulance (siren)", | |
| "Fire engine, fire truck (siren)", "Civil defense siren"], | |
| # --- impacts / tools --- | |
| "gunshot_explosion": ["Gunshot, gunfire", "Explosion", "Fireworks", "Firecracker", "Burst, pop", "Boom", | |
| "Machine gun", "Cap gun"], | |
| "tools": ["Tools", "Hammer", "Drill", "Power tool", "Chainsaw", "Sawing", "Jackhammer", "Sanding", | |
| "Filing (rasp)", "Electric drill", "Lawn mower"], | |
| "impact": ["Thump, thud", "Bang", "Slap, smack", "Smash, crash", "Crack"], | |
| } | |
| CLASSES = list(GROUPS) | |
| TRANSIENT = {"dog", "cat", "cattle_moo", "pig", "sheep_goat", "horse", "chicken_rooster", "duck_goose", "bird", "frog", | |
| "shout_scream", "laughter", "cough_sneeze", "clapping_cheer", "footsteps", "alarm_bell", "phone_ring", | |
| "dishes", "door_knock", "typing_click", "bells_chimes", "vehicle_horn", "gunshot_explosion", "impact", | |
| "clock_tick", "toilet_flush"} | |
| TH_DEFAULT = 0.30 # score (mean over model families) needed to TRIGGER an event | |
| TH = {"speech": 0.40, "music": 0.40, "crowd": 0.40, "engine": 0.40, "car": 0.35, "singing": 0.35, "wind": 0.35, | |
| "cough_sneeze": 0.15, "laughter": 0.18} | |
| for _c in TRANSIENT: # short sounds only fill part of a window, so their scores are lower | |
| TH.setdefault(_c, 0.20) | |
| LOW_FRAC = 0.6 # hysteresis: edges extend while score >= 60% of the trigger threshold | |
| GAP_S = {c: (0.25 if c in TRANSIENT else 1.0) for c in CLASSES} # merge gaps shorter than this | |
| MIN_S = {c: (0.25 if c in TRANSIENT else 1.0) for c in CLASSES} # drop events shorter than this | |
| TRIM = 0.6 # transient events: keep cells >= 60% of the event's peak score | |
| DISCOVER_TH, DISCOVER_MAX = 0.40, 6 # automatic discovery of labels outside the groups above | |
| DISCOVER_SKIP = {"Silence", "Noise", "Environmental noise", "Static", "Mains hum", "Distortion", "White noise", | |
| "Pink noise", "Throbbing", "Vibration", "Inside, small room", "Inside, large room or hall", | |
| "Inside, public space", "Outside, urban or manmade", "Outside, rural or natural", "Sound effect", | |
| "Reverberation", "Echo", "Field recording", "Animal", "Domestic animals, pets", | |
| "Livestock, farm animals, working animals", "Human voice", "Human sounds", "Human group actions", | |
| "Domestic sounds, home sounds", "Vehicle", "Mechanisms", "Tool", "Liquid"} | |
| CLAP_TH = 0.5 # CLAP: softmax mass on a prompt vs. background prompts | |
| # -------------------- Windowing and timeline helpers -------------------- | |
| def build_groups(labels, groups): | |
| low = {l.lower(): i for i, l in enumerate(labels)} | |
| ids = {g: sorted({low[n.lower()] for n in names if n.lower() in low}) for g, names in groups.items()} | |
| missing = {g: [n for n in names if n.lower() not in low] for g, names in groups.items()} | |
| return ids, missing | |
| def window_batches(wav, sr, win_s, hop_s, n_cells, bs=4): | |
| """Cell i covers [i*hop, (i+1)*hop); its window is centred on the cell (audio is zero-padded at both ends).""" | |
| win, hop = int(round(win_s * sr)), int(round(hop_s * sr)) | |
| pad = (win - hop) // 2 | |
| padded = np.zeros(n_cells * hop + (win - hop), np.float32) | |
| n = min(len(wav), len(padded) - pad) | |
| padded[pad:pad + n] = wav[:n] | |
| for i0 in range(0, n_cells, bs): | |
| yield np.stack([padded[i * hop:i * hop + win] for i in range(i0, min(n_cells, i0 + bs))]) | |
| def fuse(P, group_ids, classes, th, vote_rel=0.6): | |
| """P: {family: (cells, 527)} -> S (cells, C) mean group score, votes (cells, C) families in favour, per-family scores.""" | |
| ms = {m: np.stack([p[:, group_ids[c]].max(1) if group_ids[c] else np.zeros(len(p)) for c in classes], 1) | |
| for m, p in P.items()} | |
| thr = np.array([th[c] for c in classes]) | |
| return np.mean(list(ms.values()), 0), np.sum([v >= vote_rel * thr for v in ms.values()], 0), ms | |
| def _runs(a): | |
| out, i = [], None | |
| for k, v in enumerate(np.append(a, False)): | |
| if v and i is None: | |
| i = k | |
| elif not v and i is not None: | |
| out.append([i, k - 1]); i = None | |
| return out | |
| def extract_events(S, votes, names, th, min_votes, hop, dur, low_frac=LOW_FRAC): | |
| """Hysteresis per class: a run needs one TRIGGER cell (score >= th and enough models agreeing); its edges extend | |
| while the score stays >= low_frac*th. Classes are independent, so events of different classes overlap freely.""" | |
| ev = [] | |
| for c, name in enumerate(names): | |
| hi = th[name] | |
| act = S[:, c] >= low_frac * hi | |
| trig = (S[:, c] >= hi) & (votes[:, c] >= min_votes) | |
| gap = math.ceil(GAP_S.get(name, 0.5) / hop - 1e-9) - 1 | |
| merged = [] | |
| for r in _runs(act): | |
| if merged and r[0] - merged[-1][1] - 1 <= gap: | |
| merged[-1][1] = r[1] | |
| else: | |
| merged.append(r) | |
| for i, j in merged: | |
| if not trig[i:j + 1].any() or j - i + 1 < math.ceil(MIN_S.get(name, 0.5) / hop - 1e-9): | |
| continue | |
| if name in TRANSIENT: # window-level tags are smeared; keep the peak region | |
| seg = S[i:j + 1, c]; k = int(seg.argmax()); keep = seg >= TRIM * seg[k] | |
| l = r_ = k | |
| while l > 0 and keep[l - 1]: l -= 1 | |
| while r_ < len(seg) - 1 and keep[r_ + 1]: r_ += 1 | |
| i, j = i + l, i + r_ | |
| ev.append((i * hop, min((j + 1) * hop, dur), name, float(S[i:j + 1, c].mean()), int(votes[i:j + 1, c].max()))) | |
| return sorted(ev) | |
| def refine_with_vad(events, vad, S_speech, votes_speech, hop, th_speech, dur, cover=0.3): | |
| """Replace coarse speech events by VAD segments (exact boundaries) that the taggers also support.""" | |
| out = [e for e in events if e[2] != "speech"] | |
| for a, b in vad: | |
| i0 = int(a / hop); i1 = max(int(math.ceil(b / hop)), i0 + 1) | |
| seg = S_speech[i0:i1] | |
| if len(seg) and np.mean(seg >= 0.5 * th_speech) >= cover: | |
| out.append((float(a), float(min(b, dur)), "speech", float(seg.mean()), int(votes_speech[i0:i1].max()))) | |
| return sorted(out) | |
| def overlaps(events, min_classes=2): | |
| pts = sorted({e[0] for e in events} | {e[1] for e in events}) | |
| segs = [] | |
| for a, b in zip(pts[:-1], pts[1:]): | |
| act = sorted({e[2] for e in events if e[0] <= (a + b) / 2 < e[1]}) | |
| if len(act) >= min_classes: | |
| if segs and segs[-1][2] == act and abs(segs[-1][1] - a) < 1e-9: | |
| segs[-1][1] = b | |
| else: | |
| segs.append([a, b, act]) | |
| return segs | |
| def piano_roll(events, dur, ov): | |
| rows = [] | |
| for e in sorted(events): # rows ordered by first onset | |
| if e[2] not in rows: | |
| rows.append(e[2]) | |
| fig, ax = plt.subplots(figsize=(11, max(2.6, 0.45 * len(rows) + 1.3))) | |
| for a, b, _ in ov: | |
| ax.axvspan(a, b, color="orange", alpha=0.15, lw=0) | |
| for r, c in enumerate(rows): | |
| for s, e, _, conf, _v in [x for x in events if x[2] == c]: | |
| ax.barh(r, e - s, left=s, height=0.7, color=plt.cm.viridis(min(max(conf, 0.0), 1.0))) | |
| ax.set_yticks(range(len(rows))); ax.set_yticklabels(rows, fontsize=9); ax.invert_yaxis() | |
| ax.set_xlim(0, dur); ax.set_xlabel("time (s) [orange = 2+ sounds at once, bar colour = confidence]") | |
| if not rows: | |
| ax.text(0.5, 0.5, "no events", ha="center", va="center", transform=ax.transAxes) | |
| fig.tight_layout() | |
| return fig | |
| def softmax(x, axis=1): | |
| x = x - x.max(axis, keepdims=True); e = np.exp(x) | |
| return e / e.sum(axis, keepdims=True) | |
| def suggest(labels, name): | |
| return difflib.get_close_matches(name, labels, n=2, cutoff=0.6) | |
| # -------------------- Detection function and Gradio UI -------------------- | |
| try: | |
| import noisereduce as nr | |
| except Exception: | |
| nr = None | |
| GROUP_IDS, MISSING = build_groups(LABELS, GROUPS) | |
| bad = [(g, n) for g in CLASSES for n in MISSING[g]] | |
| print(f"{len(CLASSES)} categories | {sum(len(v) for v in GROUP_IDS.values())} AudioSet labels matched | " | |
| f"{len(bad)} names unknown to the model (ignored)") | |
| for g, n in bad[:40]: | |
| print(f" {g}: '{n}' -> did you mean {suggest(LABELS, n) or 'nothing close'}") | |
| for g in CLASSES: | |
| if not GROUP_IDS[g]: | |
| print(f"WARNING: no AudioSet label matched for '{g}' -> it will never fire") | |
| _used = {i for ids in GROUP_IDS.values() for i in ids} | |
| DISCOVER_IDS = [i for i, l in enumerate(LABELS) | |
| if i not in _used and l not in DISCOVER_SKIP and "music" not in l.lower()] | |
| def tag_windows(wavs, win_s, hop_s, dur): | |
| n_cells = int(math.ceil(dur / hop_s)) | |
| per_model = {name: np.concatenate([fn(b) for b in window_batches(wavs[sr], sr, win_s, hop_s, n_cells)]) | |
| for name, (fn, sr, fam) in MODELS.items()} | |
| fam = {} | |
| for name, (_, _, f) in MODELS.items(): | |
| fam.setdefault(f, []).append(per_model[name]) | |
| return {f: np.mean(v, 0) for f, v in fam.items()} | |
| def detect(audio_path, sens=0.0, win_s=2.0, hop_s=0.25, min_votes=2, use_vad=True, denoise=False, | |
| discover=True, extra=""): | |
| if audio_path is None: | |
| return "Upload or record audio.", None | |
| wav48, _ = librosa.load(audio_path, sr=48000, mono=True, duration=MAX_SEC) | |
| if wav48.size == 0: | |
| return "The uploaded audio is empty or could not be decoded.", None | |
| if denoise and nr is not None: # optional: can hurt tagging, so default off | |
| wav48 = nr.reduce_noise(y=wav48, sr=48000, stationary=True, prop_decrease=0.6).astype(np.float32) | |
| wav48 = wav48 / (np.abs(wav48).max() + 1e-6) * 0.9 | |
| dur = len(wav48) / 48000 | |
| hop_s = float(hop_s) | |
| if dur / hop_s > MAX_CELLS: | |
| hop_s = math.ceil(dur / MAX_CELLS / 0.05) * 0.05 | |
| wavs = {48000: wav48} | |
| for sr in {s for _, s, _ in MODELS.values()} | {16000}: | |
| wavs[sr] = wav48 if sr == 48000 else librosa.resample(wav48, orig_sr=48000, target_sr=sr) | |
| P = tag_windows(wavs, float(win_s), hop_s, dur) # {family: (cells, 527)} | |
| nf, mv = len(P), int(min(min_votes, len(P))) | |
| th = {c: float(np.clip(TH.get(c, TH_DEFAULT) + sens, 0.05, 0.9)) for c in CLASSES} | |
| S, votes, ms = fuse(P, GROUP_IDS, CLASSES, th) | |
| names, S_all, V_all = list(CLASSES), S, votes | |
| if discover and DISCOVER_IDS: # any other AudioSet label that is clearly present | |
| Sd = np.mean(list(P.values()), 0)[:, DISCOVER_IDS] | |
| top = np.argsort(Sd.max(0))[::-1][:DISCOVER_MAX] | |
| top = [k for k in top if Sd[:, k].max() >= DISCOVER_TH] | |
| if top: | |
| dn = [f"auto: {LABELS[DISCOVER_IDS[k]]}" for k in top] | |
| Vd = np.sum([p[:, [DISCOVER_IDS[k] for k in top]] >= 0.6 * DISCOVER_TH for p in P.values()], 0) | |
| th.update({n: DISCOVER_TH for n in dn}) | |
| names, S_all, V_all = names + dn, np.hstack([S, Sd[:, top]]), np.hstack([votes, Vd]) | |
| ev = extract_events(S_all, V_all, names, th, mv, hop_s, dur) | |
| note_vad = "" | |
| if use_vad and VAD is not None: | |
| k = CLASSES.index("speech") | |
| ev = refine_with_vad(ev, VAD(wavs[16000]), S[:, k], votes[:, k], hop_s, th["speech"], dur) | |
| note_vad = " | speech boundaries from Silero VAD" | |
| prompts = [p.strip() for p in (extra or "").split(",") if p.strip()] | |
| if prompts and CLAP is not None: # open-vocabulary categories (experimental) | |
| allp = [f"Sound of {p}" for p in prompts] + NEG_PROMPTS | |
| n_cells = int(math.ceil(dur / hop_s)) | |
| Pc = softmax(np.concatenate([CLAP(b, allp) for b in window_batches(wavs[48000], 48000, float(win_s), hop_s, | |
| n_cells)]))[:, :len(prompts)] | |
| cn = [f"clap: {p}" for p in prompts] | |
| ev += extract_events(Pc, (Pc >= CLAP_TH).astype(int), cn, {n: CLAP_TH for n in cn}, 1, hop_s, dur) | |
| elif prompts: | |
| note_vad += " | CLAP not loaded, extra categories ignored" | |
| ev = sorted(ev) | |
| ov = overlaps(ev) | |
| hit = {e[2] for e in ev} # classes that came close but were not reported | |
| near = [] | |
| for c, name in enumerate(CLASSES): | |
| k = int(S[:, c].argmax()) | |
| if name not in hit and S[k, c] >= 0.4 * th[name]: | |
| per = ", ".join(f"{m} {ms[m][k, c]:.2f}" for m in ms) | |
| near.append((S[k, c] / th[name], f" {k * hop_s:6.2f}s {name:<16} fused {S[k, c]:.2f} (needs {th[name]:.2f}, " | |
| f"{int(votes[k, c])}/{nf} agree) [{per}]")) | |
| near = [t for _, t in sorted(near, reverse=True)[:8]] | |
| out = [f"Model families: {', '.join(P)} ({len(MODELS)} networks) | window {win_s}s hop {hop_s:g}s | " | |
| f"{mv}/{nf} families must agree{note_vad}", ""] | |
| for m, p in P.items(): | |
| top = np.argsort(p.max(0))[::-1][:6] | |
| out.append(f"{m} hears: " + ", ".join(f"{LABELS[i]} {p[:, i].max():.2f}" for i in top)) | |
| out += ["", "Start - End | Sound | Conf | Agree", "-" * 66] | |
| out += [f"{s:6.2f}s - {e:6.2f}s | {n:<26} | {c*100:4.1f}% | {'CLAP' if n.startswith('clap:') else f'{v}/{nf}'}" | |
| for s, e, n, c, v in ev] or ["(no events)"] | |
| out += ["", "Overlapping sounds:"] + ([f" {a:6.2f}s - {b:6.2f}s : {' + '.join(cl)}" for a, b, cl in ov] or [" (none)"]) | |
| if near: | |
| out += ["", "Close but NOT reported (lower 'Sensitivity offset' or 'Families that must agree'):"] + near | |
| return "\n".join(out), piano_roll(ev, dur, ov) | |
| demo = gr.Interface( | |
| fn=detect, | |
| inputs=[gr.Audio(type="filepath", label="Audio"), | |
| gr.Slider(-0.2, 0.2, value=0.0, step=0.05, label="Sensitivity offset (lower = more events)"), | |
| gr.Dropdown([1.0, 2.0, 3.0], value=2.0, label="Window (s): 1 = tighter timing, 3 = better tags"), | |
| gr.Dropdown([0.1, 0.25, 0.5], value=0.25, label="Hop (s): smaller = finer but slower"), | |
| gr.Slider(1, 3, value=2, step=1, label="Model families that must agree"), | |
| gr.Checkbox(value=VAD is not None, label="Exact speech boundaries (Silero VAD)"), | |
| gr.Checkbox(value=False, label="Denoise first (experimental, can hurt)"), | |
| gr.Checkbox(value=True, label="Auto-discover other AudioSet sounds"), | |
| gr.Textbox(value="", label="Extra categories, comma separated (CLAP, experimental)", | |
| placeholder="tractor, drone buzzing, cash register")], | |
| outputs=[gr.Textbox(label="Detected events", lines=30), gr.Plot(label="Timeline (overlaps shaded)")], | |
| title="Overlap-aware sound event detection: AST x3 + PaSST + CLAP + VAD", | |
| ) | |
| demo.queue(max_size=16).launch( | |
| ssr_mode=False, | |
| show_error=True, | |
| debug=True | |
| ) |