Spaces:
Running on Zero
Running on Zero
| """ | |
| pairs.py — Comprobaciones de parejas de pistas (proyectos de diálogo a dos pistas). | |
| Cada persona graba su voz en su propio fichero. Las dos pistas de una conversación deben: | |
| - durar exactamente lo mismo (mismo número de muestras), y | |
| - contener solo la voz de quien graba (sin que se cuele la del otro por el altavoz). | |
| Emparejado: dos ficheros son pareja si su nombre coincide quitando la última parte (lo que va tras el | |
| último «_», «-» o espacio: A/B, 1/2, el nombre de cada persona…). Si solo se suben dos ficheros, son pareja. | |
| """ | |
| from __future__ import annotations | |
| import re | |
| from dataclasses import dataclass | |
| from pathlib import Path | |
| import numpy as np | |
| import dsp | |
| from audio_io import read_audio | |
| CROSSTALK_RISE_DB = 6 # si una pista sube ≥ 6 dB sobre su ruido mientras habla el otro, se cuela su voz | |
| CROSSTALK_MIN_FRAMES = 50 # ≥ 1 s de tramas útiles para opinar | |
| class PairResult: | |
| a: str | |
| b: str | |
| dur_a: float | |
| dur_b: float | |
| same_length: bool | |
| diff_ms: float | |
| crosstalk: list[str] # avisos de voz cruzada | |
| def pair_key(name: str) -> str: | |
| stem = Path(name).stem.lower() | |
| parts = re.split(r"[\s_\-]+", stem.strip()) | |
| return " ".join(parts[:-1]) if len(parts) > 1 else stem | |
| def find_pairs(paths: list[str]) -> tuple[list[tuple[str, str]], list[str]]: | |
| """Devuelve (parejas, sueltos).""" | |
| if len(paths) == 2: | |
| return [(paths[0], paths[1])], [] | |
| groups: dict[str, list[str]] = {} | |
| for p in paths: | |
| groups.setdefault(pair_key(Path(p).name), []).append(p) | |
| pairs, alone = [], [] | |
| for g in groups.values(): | |
| if len(g) == 2: | |
| pairs.append((g[0], g[1])) | |
| else: | |
| alone += g | |
| return pairs, alone | |
| def _levels(path: str) -> tuple[np.ndarray, int, float]: | |
| data, sr, meta = read_audio(path) | |
| mono = dsp.to_mono(data)[:, 0] | |
| fdb, _ = dsp.frame_levels(mono, sr) | |
| return fdb, len(mono), sr | |
| def _leak(src: np.ndarray, dst: np.ndarray) -> float | None: | |
| """¿Cuánto sube el nivel de `dst` (sobre su propio ruido de fondo) mientras habla `src`? | |
| En una conversación por turnos, mientras uno habla el otro suele callar: su pista se queda en su ruido | |
| de fondo. Si en esos momentos la pista sube, es que se cuela la voz del otro. Devuelve los dB de subida.""" | |
| live_s, live_d = src[src > dsp.DIGITAL_SILENCE_DB], dst[dst > dsp.DIGITAL_SILENCE_DB] | |
| if live_s.size < CROSSTALK_MIN_FRAMES or live_d.size < CROSSTALK_MIN_FRAMES: | |
| return None | |
| talk = src > np.percentile(live_s, 10) + 15 | |
| if talk.sum() < CROSSTALK_MIN_FRAMES: | |
| return None | |
| return float(np.median(dst[talk]) - np.percentile(live_d, 10)) | |
| def check_pair(a: str, b: str) -> PairResult: | |
| fa, na, sra = _levels(a) | |
| fb, nb, srb = _levels(b) | |
| warn = [] | |
| if sra == srb: | |
| n = min(len(fa), len(fb)) | |
| for src, dst, who, other in ((fa[:n], fb[:n], Path(a).name, Path(b).name), | |
| (fb[:n], fa[:n], Path(b).name, Path(a).name)): | |
| r = _leak(src, dst) | |
| if r is not None and r >= CROSSTALK_RISE_DB: | |
| warn.append(f"En «{other}» parece oírse la voz de «{who}» (su pista sube {r:.0f} dB cuando habla " | |
| "la otra persona). Usad auriculares en la llamada.") | |
| da, db = na / sra, nb / srb | |
| return PairResult(Path(a).name, Path(b).name, da, db, na == nb and sra == srb, abs(da - db) * 1000, warn) | |