1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
|
"""Runtime wiring for capture, VAD, segmentation, and durable clips."""
from __future__ import annotations
import logging
import time
from collections.abc import Callable, Iterable
from datetime import datetime
from pathlib import Path
from typing import Protocol
import numpy as np
from mic_clipper.audio_input import PulseAudioInput
from mic_clipper.segmenter import Audio, End, Segmenter, Start
from mic_clipper.storage import ClipWriter, OpenClip
from mic_clipper.vad import SileroVad
LOGGER = logging.getLogger(__name__)
class Vad(Protocol):
def probability(self, samples: np.ndarray) -> float: ...
class Writer(Protocol):
def open(self, started_at: datetime) -> OpenClip: ...
def record_frames(
frames: Iterable[np.ndarray],
*,
vad: Vad,
writer: Writer,
segmenter: Segmenter,
now: Callable[[], datetime],
threshold: float = 0.5,
) -> None:
"""Record one finite or failing frame stream, closing an active encoder on exit."""
clip: OpenClip | None = None
try:
for samples in frames:
events = segmenter.push(samples, vad.probability(samples) >= threshold, now())
for event in events:
if isinstance(event, Start):
clip = writer.open(event.started_at)
try:
clip.write(event.samples)
except BaseException:
clip.abort()
clip = None
raise
elif isinstance(event, Audio):
if clip is None:
raise RuntimeError("segment audio arrived without an active encoder")
clip.write(event.samples)
elif isinstance(event, End):
if clip is None:
raise RuntimeError("segment end arrived without an active encoder")
clip.close()
clip = None
finally:
if clip is not None:
clip.close()
def run_forever(root: Path, *, threshold: float = 0.5) -> None:
"""Recreate all live components after a transient audio-service failure."""
retry_seconds = 1
while True:
try:
record_frames(
PulseAudioInput().frames(),
vad=SileroVad(),
writer=ClipWriter(root),
segmenter=Segmenter(pre_roll_seconds=0.5, silence_seconds=1.5),
now=lambda: datetime.now().astimezone(),
threshold=threshold,
)
raise RuntimeError("Pulse input ended unexpectedly")
except KeyboardInterrupt:
LOGGER.info("microphone clip recorder stopped")
return
except Exception as error:
LOGGER.warning(
"audio recorder unavailable (%s); retrying in %s seconds",
error,
retry_seconds,
)
time.sleep(retry_seconds)
retry_seconds = min(retry_seconds * 2, 60)
|
