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
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
|
"""Private, streaming Opus clip storage."""
from __future__ import annotations
import os
import subprocess
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
import numpy as np
SAMPLE_RATE = 16_000
@dataclass
class OpenClip:
path: Path
process: subprocess.Popen[bytes]
closed: bool = False
def write(self, samples: np.ndarray) -> None:
if self.closed:
raise RuntimeError("cannot write a closed clip")
assert self.process.stdin is not None
audio = np.asarray(samples, dtype="<f4")
self.process.stdin.write(audio.tobytes())
def close(self) -> None:
if self.closed:
return
self.closed = True
assert self.process.stdin is not None
self.process.stdin.close()
assert self.process.stderr is not None
error_output = self.process.stderr.read().decode("utf-8", errors="replace").strip()
return_code = self.process.wait()
if return_code:
self.path.unlink(missing_ok=True)
raise RuntimeError(f"ffmpeg Opus encoding failed ({return_code}): {error_output}")
os.chmod(self.path, 0o600)
def abort(self) -> None:
if self.closed:
return
self.closed = True
self.process.kill()
self.process.wait()
self.path.unlink(missing_ok=True)
class ClipWriter:
"""Open one private Ogg/Opus file only after the segmenter emits a start."""
def __init__(self, root: Path) -> None:
self.root = Path(root).expanduser()
def open(self, started_at: datetime) -> OpenClip:
self.root.mkdir(mode=0o700, parents=True, exist_ok=True)
os.chmod(self.root, 0o700)
destination = self.root / started_at.date().isoformat()
destination.mkdir(mode=0o700, exist_ok=True)
os.chmod(destination, 0o700)
path = self._reserve_path(destination, started_at)
process = subprocess.Popen(
[
"ffmpeg",
"-nostdin",
"-v",
"error",
"-f",
"f32le",
"-ar",
str(SAMPLE_RATE),
"-ac",
"1",
"-i",
"pipe:0",
"-c:a",
"libopus",
"-b:a",
"24k",
"-application",
"voip",
"-vbr",
"constrained",
"-f",
"ogg",
"-y",
str(path),
],
stdin=subprocess.PIPE,
stdout=subprocess.DEVNULL,
stderr=subprocess.PIPE,
)
return OpenClip(path, process)
@staticmethod
def _reserve_path(destination: Path, started_at: datetime) -> Path:
base_name = started_at.strftime("%H-%M-%S")
for suffix in range(10_000):
name = f"{base_name}.opus" if suffix == 0 else f"{base_name}-{suffix:02d}.opus"
path = destination / name
try:
descriptor = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
except FileExistsError:
continue
os.close(descriptor)
return path
raise RuntimeError("unable to reserve a unique clip filename")
|
