70 lines
2.3 KiB
Python
70 lines
2.3 KiB
Python
"""Opt-in live cloud contract smoke test using a short raw PCM fixture."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import os
|
|
from pathlib import Path
|
|
import stat
|
|
|
|
from .backend import backend_from_config
|
|
from .config import load_config
|
|
|
|
MAX_SAMPLE_BYTES = 240_000 # five seconds of 24 kHz mono PCM16
|
|
CHUNK_BYTES = 4_800
|
|
|
|
|
|
def _read_sample(path: Path) -> bytes:
|
|
descriptor = os.open(
|
|
path,
|
|
os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0),
|
|
)
|
|
try:
|
|
info = os.fstat(descriptor)
|
|
if not stat.S_ISREG(info.st_mode):
|
|
raise ValueError("sample must be a regular file")
|
|
if not 0 < info.st_size <= MAX_SAMPLE_BYTES or info.st_size % 2:
|
|
raise ValueError("sample must contain at most five seconds of even-length PCM16")
|
|
sample = bytearray()
|
|
while len(sample) <= MAX_SAMPLE_BYTES:
|
|
chunk = os.read(descriptor, min(64 * 1024, MAX_SAMPLE_BYTES + 1 - len(sample)))
|
|
if not chunk:
|
|
break
|
|
sample.extend(chunk)
|
|
if len(sample) != info.st_size:
|
|
raise ValueError("sample changed while it was being read")
|
|
return bytes(sample)
|
|
finally:
|
|
os.close(descriptor)
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(
|
|
description="Send one non-sensitive 24 kHz mono PCM16 sample to the configured cloud backend."
|
|
)
|
|
parser.add_argument("sample", type=Path, help="raw 24 kHz mono signed-16 little-endian PCM")
|
|
parser.add_argument(
|
|
"--config",
|
|
type=Path,
|
|
default=Path("~/.config/pi-dictate/config.json").expanduser(),
|
|
)
|
|
arguments = parser.parse_args()
|
|
|
|
config = load_config(arguments.config)
|
|
if config.backend not in {"meta", "openai"}:
|
|
parser.error("the configured backend must be meta or openai")
|
|
backend = backend_from_config(config)
|
|
sample = _read_sample(arguments.sample)
|
|
turn = backend.start_turn()
|
|
try:
|
|
for offset in range(0, len(sample), CHUNK_BYTES):
|
|
turn.enqueue_pcm(sample[offset : offset + CHUNK_BYTES])
|
|
transcript = turn.finish()
|
|
finally:
|
|
turn.close()
|
|
print(f"{backend.descriptor.backend}/{backend.descriptor.model}: {transcript!r}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|