aboutsummaryrefslogtreecommitdiff
path: root/.venv/lib/python3.12/site-packages/openai/helpers/microphone.py
diff options
context:
space:
mode:
Diffstat (limited to '.venv/lib/python3.12/site-packages/openai/helpers/microphone.py')
-rw-r--r--.venv/lib/python3.12/site-packages/openai/helpers/microphone.py100
1 files changed, 100 insertions, 0 deletions
diff --git a/.venv/lib/python3.12/site-packages/openai/helpers/microphone.py b/.venv/lib/python3.12/site-packages/openai/helpers/microphone.py
new file mode 100644
index 00000000..62a6d8d8
--- /dev/null
+++ b/.venv/lib/python3.12/site-packages/openai/helpers/microphone.py
@@ -0,0 +1,100 @@
+# mypy: ignore-errors
+from __future__ import annotations
+
+import io
+import time
+import wave
+import asyncio
+from typing import Any, Type, Union, Generic, TypeVar, Callable, overload
+from typing_extensions import TYPE_CHECKING, Literal
+
+from .._types import FileTypes, FileContent
+from .._extras import numpy as np, sounddevice as sd
+
+if TYPE_CHECKING:
+ import numpy.typing as npt
+
+SAMPLE_RATE = 24000
+
+DType = TypeVar("DType", bound=np.generic)
+
+
+class Microphone(Generic[DType]):
+ def __init__(
+ self,
+ channels: int = 1,
+ dtype: Type[DType] = np.int16,
+ should_record: Union[Callable[[], bool], None] = None,
+ timeout: Union[float, None] = None,
+ ):
+ self.channels = channels
+ self.dtype = dtype
+ self.should_record = should_record
+ self.buffer_chunks = []
+ self.timeout = timeout
+ self.has_record_function = callable(should_record)
+
+ def _ndarray_to_wav(self, audio_data: npt.NDArray[DType]) -> FileTypes:
+ buffer: FileContent = io.BytesIO()
+ with wave.open(buffer, "w") as wav_file:
+ wav_file.setnchannels(self.channels)
+ wav_file.setsampwidth(np.dtype(self.dtype).itemsize)
+ wav_file.setframerate(SAMPLE_RATE)
+ wav_file.writeframes(audio_data.tobytes())
+ buffer.seek(0)
+ return ("audio.wav", buffer, "audio/wav")
+
+ @overload
+ async def record(self, return_ndarray: Literal[True]) -> npt.NDArray[DType]: ...
+
+ @overload
+ async def record(self, return_ndarray: Literal[False]) -> FileTypes: ...
+
+ @overload
+ async def record(self, return_ndarray: None = ...) -> FileTypes: ...
+
+ async def record(self, return_ndarray: Union[bool, None] = False) -> Union[npt.NDArray[DType], FileTypes]:
+ loop = asyncio.get_event_loop()
+ event = asyncio.Event()
+ self.buffer_chunks: list[npt.NDArray[DType]] = []
+ start_time = time.perf_counter()
+
+ def callback(
+ indata: npt.NDArray[DType],
+ _frame_count: int,
+ _time_info: Any,
+ _status: Any,
+ ):
+ execution_time = time.perf_counter() - start_time
+ reached_recording_timeout = execution_time > self.timeout if self.timeout is not None else False
+ if reached_recording_timeout:
+ loop.call_soon_threadsafe(event.set)
+ raise sd.CallbackStop
+
+ should_be_recording = self.should_record() if callable(self.should_record) else True
+ if not should_be_recording:
+ loop.call_soon_threadsafe(event.set)
+ raise sd.CallbackStop
+
+ self.buffer_chunks.append(indata.copy())
+
+ stream = sd.InputStream(
+ callback=callback,
+ dtype=self.dtype,
+ samplerate=SAMPLE_RATE,
+ channels=self.channels,
+ )
+ with stream:
+ await event.wait()
+
+ # Concatenate all chunks into a single buffer, handle empty case
+ concatenated_chunks: npt.NDArray[DType] = (
+ np.concatenate(self.buffer_chunks, axis=0)
+ if len(self.buffer_chunks) > 0
+ else np.array([], dtype=self.dtype)
+ )
+
+ if return_ndarray:
+ return concatenated_chunks
+ else:
+ return self._ndarray_to_wav(concatenated_chunks)