root
95bb7c50ee
Publish and subscribe over wss://livekit.uni-wh.de:7800 and refuse cleartext ws://. Conference room is uwh-telhai. Includes the uncommitted encoded H.264 publish path, Rally hairpin, and KMS wall overlay.
92 lines
2.5 KiB
Python
92 lines
2.5 KiB
Python
"""Per-hop noise gate for C920 mics (LiveKit active-speaker needs distinct energy)."""
|
|
from __future__ import annotations
|
|
|
|
import numpy as np
|
|
|
|
FLOOR_DB = -120.0
|
|
|
|
|
|
def rms_dbfs(pcm: np.ndarray) -> float:
|
|
"""RMS of int16 PCM in dBFS. Empty/zero -> FLOOR_DB."""
|
|
x = np.asarray(pcm)
|
|
if x.size == 0:
|
|
return FLOOR_DB
|
|
rms = float(np.sqrt(np.mean(np.square(x.astype(np.float64) / 32768.0))))
|
|
if rms <= 1e-12:
|
|
return FLOOR_DB
|
|
return 20.0 * float(np.log10(rms))
|
|
|
|
|
|
def hop_mono_int16(pcm: np.ndarray) -> np.ndarray:
|
|
x = np.asarray(pcm)
|
|
if x.ndim == 2:
|
|
return np.mean(x.astype(np.float32), axis=1).astype(np.int16)
|
|
return np.asarray(x, dtype=np.int16)
|
|
|
|
|
|
def gate_enabled_for(identity: str, speaker_camera: str, enabled: bool) -> bool:
|
|
"""Dante/rally stays ungated; C920 seats are gated when enabled."""
|
|
if not enabled:
|
|
return False
|
|
ident = (identity or "").strip()
|
|
skip = (speaker_camera or "").strip()
|
|
if ident and skip and ident == skip:
|
|
return False
|
|
return True
|
|
|
|
|
|
def hold_hops(hold_s: float, hop_s: float) -> int:
|
|
hop = max(1e-6, float(hop_s))
|
|
return max(1, int(round(float(hold_s) / hop)))
|
|
|
|
|
|
def make_gate(
|
|
identity: str,
|
|
speaker_camera: str,
|
|
enabled: bool,
|
|
open_db: float,
|
|
close_db: float,
|
|
hold_s: float,
|
|
hop_s: float,
|
|
) -> HopGate | None:
|
|
if not gate_enabled_for(identity, speaker_camera, enabled):
|
|
return None
|
|
return HopGate(
|
|
open_db=open_db,
|
|
close_db=close_db,
|
|
hold_hops=hold_hops(hold_s, hop_s),
|
|
)
|
|
|
|
|
|
class HopGate:
|
|
"""Open on close-talk energy; hangover then mute so neighbors stay silent."""
|
|
|
|
def __init__(
|
|
self,
|
|
open_db: float = -32.0,
|
|
close_db: float = -40.0,
|
|
hold_hops: int = 3,
|
|
) -> None:
|
|
self.open_db = float(open_db)
|
|
self.close_db = min(float(close_db), self.open_db)
|
|
self.hold_hops = max(0, int(hold_hops))
|
|
self.open = False
|
|
self.hold = 0
|
|
|
|
def apply(self, pcm: np.ndarray) -> np.ndarray:
|
|
pcm = np.asarray(pcm, dtype=np.int16)
|
|
db = rms_dbfs(pcm)
|
|
if db >= self.open_db:
|
|
self.open = True
|
|
self.hold = self.hold_hops
|
|
return pcm
|
|
if self.open:
|
|
if db >= self.close_db:
|
|
self.hold = self.hold_hops
|
|
return pcm
|
|
if self.hold > 0:
|
|
self.hold -= 1
|
|
return pcm
|
|
self.open = False
|
|
return np.zeros_like(pcm)
|