Refine config and runtime flow
This commit is contained in:
parent
85e082dd46
commit
b3be444625
16 changed files with 642 additions and 137 deletions
|
|
@ -1,8 +1,7 @@
|
|||
from dataclasses import dataclass, field
|
||||
from typing import Iterable
|
||||
from typing import Any, Iterable
|
||||
|
||||
import numpy as np
|
||||
import sounddevice as sd # type: ignore[import-not-found]
|
||||
|
||||
|
||||
@dataclass
|
||||
|
|
@ -14,6 +13,7 @@ class RecordResult:
|
|||
|
||||
|
||||
def list_input_devices() -> list[dict]:
|
||||
sd = _sounddevice()
|
||||
devices = []
|
||||
for idx, info in enumerate(sd.query_devices()):
|
||||
if info.get("max_input_channels", 0) > 0:
|
||||
|
|
@ -22,6 +22,7 @@ def list_input_devices() -> list[dict]:
|
|||
|
||||
|
||||
def default_input_device() -> int | None:
|
||||
sd = _sounddevice()
|
||||
default = sd.default.device
|
||||
if isinstance(default, (tuple, list)) and default:
|
||||
return default[0]
|
||||
|
|
@ -48,7 +49,8 @@ def resolve_input_device(spec: str | int | None) -> int | None:
|
|||
return None
|
||||
|
||||
|
||||
def start_recording(input_spec: str | int | None) -> tuple[sd.InputStream, RecordResult]:
|
||||
def start_recording(input_spec: str | int | None) -> tuple[Any, RecordResult]:
|
||||
sd = _sounddevice()
|
||||
record = RecordResult()
|
||||
device = resolve_input_device(input_spec)
|
||||
|
||||
|
|
@ -66,13 +68,23 @@ def start_recording(input_spec: str | int | None) -> tuple[sd.InputStream, Recor
|
|||
return stream, record
|
||||
|
||||
|
||||
def stop_recording(stream: sd.InputStream, record: RecordResult) -> np.ndarray:
|
||||
def stop_recording(stream: Any, record: RecordResult) -> np.ndarray:
|
||||
if stream:
|
||||
stream.stop()
|
||||
stream.close()
|
||||
return _flatten_frames(record.frames)
|
||||
|
||||
|
||||
def _sounddevice():
|
||||
try:
|
||||
import sounddevice as sd # type: ignore[import-not-found]
|
||||
except ModuleNotFoundError as exc:
|
||||
raise RuntimeError(
|
||||
"sounddevice is not installed; install dependencies with `uv sync --extra x11`"
|
||||
) from exc
|
||||
return sd
|
||||
|
||||
|
||||
def _flatten_frames(frames: Iterable[np.ndarray]) -> np.ndarray:
|
||||
frames = list(frames)
|
||||
if not frames:
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue