diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml new file mode 100644 index 0000000..9da7759 --- /dev/null +++ b/.pre-commit-config.yaml @@ -0,0 +1,28 @@ +default_language_version: + python: python3.11 + +repos: + # mypy checks type annotations. + - repo: https://github.com/pre-commit/mirrors-mypy + rev: v1.10.0 + hooks: + - id: mypy + args: [--config-file=pyproject.toml] + + # pre-commit-hooks is a collection of Git hooks for code quality, formatting, and analysis. + - repo: https://github.com/pre-commit/pre-commit-hooks + rev: v4.6.0 + hooks: + - id: check-added-large-files + args: ['--maxkb=1000'] + - id: check-toml + - id: debug-statements + + # An extremely fast Python linter and code formatter, written in Rust. + - repo: https://github.com/charliermarsh/ruff-pre-commit + rev: v0.4.5 + hooks: + - id: ruff + args: + - --fix + - id: ruff-format diff --git a/README.md b/README.md index f6db3c1..37fda29 100644 --- a/README.md +++ b/README.md @@ -1,4 +1,5 @@ # Whisper Mic + This repo is based on the work done [here](https://github.com/openai/whisper) by OpenAI. This repo allows you use use a mic as demo. This repo copies some of the README from the original project. ## Video Tutorial @@ -20,8 +21,7 @@ Now a pip package! ## Available models and languages -There are five model sizes, four with English-only versions, offering speed and accuracy tradeoffs. Below are the names of the available models and their approximate memory requirements and relative speed. - +There are five model sizes, four with English-only versions, offering speed and accuracy tradeoffs. Below are the names of the available models and their approximate memory requirements and relative speed. | Size | Parameters | English-only model | Multilingual model | Required VRAM | Relative speed | |:------:|:----------:|:------------------:|:------------------:|:-------------:|:--------------:| @@ -35,13 +35,15 @@ For English-only applications, the `.en` models tend to perform better, especial ## Microphone Demo -You can use the model with a microphone using the ```whisper_mic``` program. Use ```-h``` to see flag options. +You can use the model with a microphone using the ```whisper_mic``` program. Use ```--help``` to see flag options. + +> Example: ```python -m whisper_mic --help``` Some of the more important flags are the ```--model``` and ```--english``` flags. ## Transcribing To A File -Using the command: ```whisper_mic --loop --dictate``` will type the words you say on your active cursor. +Using the command: ```python -m whisper_mic --loop --dictate``` will type the words you say on your active cursor. ## Usage In Other Projects @@ -60,17 +62,21 @@ Check out what the possible arguments are by looking at the ```cli.py``` file ## Troubleshooting If you are having issues, try the following: -``` + +```bash sudo apt install portaudio19-dev python3-pyaudio ``` ## Contributing Some ideas that you can add are: + 1. Supporting different implementations of Whisper 2. Adding additional optional functionality. 3. Add tests +Be sure to run ```pre-commit install``` before making any changes and running it before committing with ```pre-commit run --all-files```. + ## License The model weights of Whisper are released under the MIT License. See their repo for more information. @@ -78,4 +84,5 @@ The model weights of Whisper are released under the MIT License. See their repo This code under this repo is under the MIT license. See [LICENSE](LICENSE) for further details. ## Thanks + Until recently, access to high performing speech to text models was only available through paid serviecs. With this release, I am excited for the many applications that will come. diff --git a/pyproject.toml b/pyproject.toml index 77e81d3..bc342ef 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,40 +1,139 @@ -[build-system] -requires = ["setuptools>=61.0"] -build-backend = "setuptools.build_meta" - -[project] +[tool.poetry] name = "whisper_mic" version = "1.4.2" authors = [ - { name="Blake Mallory", email="blakecmallory@gmail.com" }, + "Blake Mallory ", ] description = "Whisper for your microphone" readme = "README.md" -requires-python = ">=3.9" classifiers = [ "Programming Language :: Python :: 3", "License :: OSI Approved :: MIT License", "Operating System :: OS Independent", ] -dependencies = [ - 'importlib-metadata; python_version>"3.9"', - "attrs", - "click", - "ffmpeg-python", - "more-itertools", - "numpy", - "openai-whisper", - "pyaudio", - "pydantic", - "pydub", - "pynput", - "requests", - "rich", - "speechrecognition", - "tdqm", - "torch", - "transformers", -] -[project.scripts] +[tool.poetry.dependencies] +python = "^3.11.0" + +click = "^8.1.7" +ffmpeg-python= "0.2.0" +more-itertools = "^10.2.0" +numpy = "^1.26.4" +openai-whisper = "^20231117" +pyaudio = "^0.2.14" +pydub = "^0.25.1" +pynput = "^1.7.6" +rich = "^13.7.1" +speechrecognition = "^3.10.4" +tqdm = "^4.66.4" +torch = "^2.3.0" +transformers = "^4.40.2" + +[tool.poetry.group.dev.dependencies] +mypy = "1.10.0" +pre-commit = "3.7.1" +pre-commit-hooks = "4.6.0" +ruff = "0.4.5" + +[tool.poetry.scripts] whisper_mic = "whisper_mic.cli:main" + +[build-system] +requires = ["setuptools>=61.0"] +build-backend = "setuptools.build_meta" + +[tool.mypy] +python_version = "3.11" + +check_untyped_defs = true +disallow_untyped_defs = true +incremental = false +ignore_errors = false +pretty = true +show_error_context = true +show_traceback = true +strict_optional = true +warn_incomplete_stub = true +warn_no_return = true +warn_redundant_casts = true +warn_return_any = true +warn_unreachable = true +warn_unused_configs = true +warn_unused_ignores = true + +[[tool.mypy.overrides]] +module = [ + "click.*", + "faster_whisper.*", + "numpy.*", + "pynput.*", + "rich.*", + "speech_recognition.*", + "torch.*", + "whisper.*", +] +ignore_missing_imports = true + +[tool.ruff] +target-version = "py311" + +exclude = ["alembic"] +indent-width = 4 +line-length = 110 + +[tool.ruff.lint] +# rules from: https://docs.astral.sh/ruff/rules/ +select = [ + "ANN", # flake8-annotations + "ASYNC", # flake8-async + "S", # flake8-bandit + "B", # flake8-bugbear + "A", # flake8-builtins + "COM", # flake8-commas + "C4", # flake8-comprehensions + "DTZ", # flake8-datetimez + "EM", # flake8-errmsg + "LOG", # flake8-logging + "G", # flake8-logging-format + "PIE", # flake8-pie + "T20", # flake8-print + "PT", # flake8-pytest-style + "Q", # flake8-quotes + "RSE", # flake8-raise + "RET", # flake8-return + "SIM", # flake8-simplify + "TID", # flake8-tidy-imports + "ARG", # flake8-unused-arguments + "I", # isort + "N", # pep8-naming + "E", # pycodestyle errors + "W", # pycodestyle warnings + "F", # pyflakes + "UP", # pyupgrade + "RUF", # Ruff-specific rules + "TRY", # tryceratops +] +ignore = [ + "ANN101", # Missing type annotation for `self` in method + "B008", # Do not perform function calls in argument defaults + "E501", # Line too long, handled by black + "G004", # Logging statements should not use f"..." + "W191", # Indentation contains tabs +] +fixable = ["ALL"] +unfixable = [] +# logger-objects = ["logging_setup.logger"] # TODO: test first + +[tool.ruff.format] +line-ending = "auto" +quote-style = "double" +skip-magic-trailing-comma = false + +[tool.ruff.lint.isort] +# force-sort-within-sections = true # TODO: test first +lines-after-imports = 2 +lines-between-types = 1 + +[tool.ruff.lint.pyupgrade] +# Preserve types, even if a file imports `from __future__ import annotations`. +keep-runtime-typing = true diff --git a/requirements.txt b/requirements.txt deleted file mode 100644 index 9deb1e9..0000000 --- a/requirements.txt +++ /dev/null @@ -1,13 +0,0 @@ -numpy -tqdm -more-itertools -transformers>=4.19.0 -ffmpeg-python==0.2.0 -click -pyaudio -SpeechRecognition -pydub -git+https://github.com/openai/whisper.git ---extra-index-url https://download.pytorch.org/whl/cu113 -torch -pynput diff --git a/whisper_mic/__init__.py b/whisper_mic/__init__.py index f2be3d5..e69de29 100755 --- a/whisper_mic/__init__.py +++ b/whisper_mic/__init__.py @@ -1,2 +0,0 @@ -from .whisper_mic import * -from .utils import * \ No newline at end of file diff --git a/whisper_mic/__main__.py b/whisper_mic/__main__.py new file mode 100644 index 0000000..1021f12 --- /dev/null +++ b/whisper_mic/__main__.py @@ -0,0 +1,5 @@ +from whisper_mic.cli import main + + +if __name__ == "__main__": + main() diff --git a/whisper_mic/cli.py b/whisper_mic/cli.py index ef6a337..3c7917c 100755 --- a/whisper_mic/cli.py +++ b/whisper_mic/cli.py @@ -1,50 +1,160 @@ -#!/usr/bin/env python3 - import click -import torch import speech_recognition as sr -from typing import Optional +import torch + +from whisper_mic.utils import get_logger +from whisper_mic.whisper_mic import WhisperMic -from whisper_mic import WhisperMic @click.command() -@click.option("--model", default="base", help="Model to use", type=click.Choice(["tiny","base", "small","medium","large","large-v2","large-v3"])) -@click.option("--device", default=("cuda" if torch.cuda.is_available() else "cpu"), help="Device to use", type=click.Choice(["cpu","cuda","mps"])) -@click.option("--english", default=False, help="Whether to use English model",is_flag=True, type=bool) -@click.option("--verbose", default=False, help="Whether to print verbose output", is_flag=True,type=bool) -@click.option("--energy", default=300, help="Energy level for mic to detect", type=int) -@click.option("--dynamic_energy", default=False,is_flag=True, help="Flag to enable dynamic energy", type=bool) -@click.option("--pause", default=0.8, help="Pause time before entry ends", type=float) -@click.option("--save_file",default=False, help="Flag to save file", is_flag=True,type=bool) -@click.option("--loop", default=False, help="Flag to loop", is_flag=True,type=bool) -@click.option("--dictate", default=False, help="Flag to dictate (implies loop)", is_flag=True,type=bool) -@click.option("--mic_index", default=None, help="Mic index to use", type=int) -@click.option("--list_devices",default=False, help="Flag to list devices", is_flag=True,type=bool) -@click.option("--faster",default=False, help="Use faster_whisper implementation", is_flag=True,type=bool) -@click.option("--hallucinate_threshold",default=400, help="Raise this to reduce hallucinations. Lower this to activate more often.", is_flag=True,type=int) -def main(model: str, english: bool, verbose: bool, energy: int, pause: float, dynamic_energy: bool, save_file: bool, device: str, loop: bool, dictate: bool,mic_index:Optional[int],list_devices: bool,faster: bool,hallucinate_threshold:int) -> None: +@click.option( + "--model", + default="base", + help="Model to use", + type=click.Choice( + ["tiny", "base", "small", "medium", "large", "large-v2", "large-v3"], + ), +) +@click.option( + "--device", + default=("cuda" if torch.cuda.is_available() else "cpu"), + help="Device to use", + type=click.Choice(["cpu", "cuda", "mps"]), +) +@click.option( + "--english", + default=False, + help="Whether to use English model", + is_flag=True, + type=bool, +) +@click.option( + "--verbose", + default=False, + help="Whether to print verbose output", + is_flag=True, + type=bool, +) +@click.option( + "--energy", + default=300, + help="Energy level for mic to detect", + type=int, +) +@click.option( + "--dynamic_energy", + default=False, + is_flag=True, + help="Flag to enable dynamic energy", + type=bool, +) +@click.option( + "--pause", + default=0.8, + help="Pause time before entry ends", + type=float, +) +@click.option( + "--save_file", + default=False, + help="Flag to save file", + is_flag=True, + type=bool, +) +@click.option( + "--loop", + default=False, + help="Flag to loop", + is_flag=True, + type=bool, +) +@click.option( + "--dictate", + default=False, + help="Flag to dictate (implies loop)", + is_flag=True, + type=bool, +) +@click.option( + "--mic_index", + default=None, + help="Mic index to use", + type=int, +) +@click.option( + "--list_devices", + default=False, + help="Flag to list devices", + is_flag=True, + type=bool, +) +@click.option( + "--faster", + default=False, + help="Use faster_whisper implementation", + is_flag=True, + type=bool, +) +@click.option( + "--hallucinate_threshold", + default=400, + help="Raise this to reduce hallucinations. Lower this to activate more often.", + is_flag=True, + type=int, +) +def main( + model: str, + english: bool, + verbose: bool, + energy: int, + pause: float, + dynamic_energy: bool, + save_file: bool, + device: str, + loop: bool, + dictate: bool, + mic_index: int | None, + list_devices: bool, + faster: bool, + hallucinate_threshold: int, +) -> None: + logger = get_logger("cli", "debug") + if list_devices: - print("Possible devices: ",sr.Microphone.list_microphone_names()) + logger.debug("Possible devices: ", sr.Microphone.list_microphone_names()) return - mic = WhisperMic(model=model, english=english, verbose=verbose, energy=energy, pause=pause, dynamic_energy=dynamic_energy, save_file=save_file, device=device,mic_index=mic_index,implementation=("faster_whisper" if faster else "whisper"),hallucinate_threshold=hallucinate_threshold) + mic = WhisperMic( + model=model, + english=english, + verbose=verbose, + energy=energy, + pause=pause, + dynamic_energy=dynamic_energy, + save_file=save_file, + device=device, + mic_index=mic_index, + implementation=("faster_whisper" if faster else "whisper"), + hallucinate_threshold=hallucinate_threshold, + ) if not loop: try: result = mic.listen() - print("You said: " + result) + logger.info(f"You said: {result}") except KeyboardInterrupt: - print("Operation interrupted successfully") + logger.info("Operation interrupted successfully") finally: if save_file: mic.file.close() else: try: - mic.listen_loop(dictate=dictate,phrase_time_limit=2) + mic.listen_loop(dictate=dictate, phrase_time_limit=3) except KeyboardInterrupt: - print("Operation interrupted successfully") + logger.info("Operation interrupted successfully") finally: if save_file: mic.file.close() + if __name__ == "__main__": main() diff --git a/whisper_mic/utils.py b/whisper_mic/utils.py index 3f4015c..800166c 100755 --- a/whisper_mic/utils.py +++ b/whisper_mic/utils.py @@ -1,7 +1,8 @@ import logging -from typing_extensions import Literal -from rich.logging import RichHandler +from typing import Literal + +from rich.logging import RichHandler def get_logger(name: str, level: Literal["info", "warning", "debug"]) -> logging.Logger: @@ -15,4 +16,4 @@ def get_logger(name: str, level: Literal["info", "warning", "debug"]) -> logging logger.propagate = False - return logger \ No newline at end of file + return logger diff --git a/whisper_mic/whisper_mic.py b/whisper_mic/whisper_mic.py index 3c27030..2ffb715 100755 --- a/whisper_mic/whisper_mic.py +++ b/whisper_mic/whisper_mic.py @@ -1,18 +1,24 @@ -import torch +import os +import platform import queue -import speech_recognition as sr +import tempfile import threading -import numpy as np -import os import time -import tempfile -import platform + +from collections.abc import AsyncGenerator, Generator +from typing import cast + +import numpy as np import pynput.keyboard -# from ctypes import * +import speech_recognition as sr +import torch + +from numpy.typing import NDArray from whisper_mic.utils import get_logger -#TODO: This is a linux only fix and needs to be testd. Have one for mac and windows too. + +# TODO: This is a linux only fix and needs to be testd. Have one for mac and windows too. # Define a null error handler for libasound to silence the error message spam # def py_error_handler(filename, line, function, err, fmt): # None @@ -20,11 +26,25 @@ # ERROR_HANDLER_FUNC = CFUNCTYPE(None, c_char_p, c_int, c_char_p, c_int, c_char_p) # c_error_handler = ERROR_HANDLER_FUNC(py_error_handler) + # asound = cdll.LoadLibrary('libasound.so') # asound.snd_lib_error_set_handler(c_error_handler) class WhisperMic: - def __init__(self,model="base",device=("cuda" if torch.cuda.is_available() else "cpu"),english=False,verbose=False,energy=300,pause=2,dynamic_energy=False,save_file=False, model_root="~/.cache/whisper",mic_index=None,implementation="whisper",hallucinate_threshold=300): - + def __init__( + self, + model: str = "base", + device: str = ("cuda" if torch.cuda.is_available() else "cpu"), + english: bool = False, + verbose: bool = False, + energy: int = 300, + pause: float = 2, + dynamic_energy: bool = False, + save_file: bool = False, + model_root: str = "~/.cache/whisper", + mic_index: int | None = None, + implementation: str = "whisper", + hallucinate_threshold: int = 300, + ) -> None: self.logger = get_logger("whisper_mic", "info") self.energy = energy self.hallucinate_threshold = hallucinate_threshold @@ -37,11 +57,12 @@ def __init__(self,model="base",device=("cuda" if torch.cuda.is_available() else self.platform = platform.system() - if self.platform == "darwin": - if device == "mps": - self.logger.warning("Using MPS for Mac, this does not work but may in the future") - device = "mps" - device = torch.device(device) + if self.platform == "darwin" and device == "mps": + self.logger.warning( + "Using MPS for Mac, this does not work but may in the future", + ) + device = "mps" + device = torch.device(device) if (model != "large" and model != "large-v2") and self.english: model = model + ".en" @@ -49,37 +70,54 @@ def __init__(self,model="base",device=("cuda" if torch.cuda.is_available() else model_root = os.path.expanduser(model_root) self.faster = False - if (implementation == "faster_whisper"): + if implementation == "faster_whisper": try: from faster_whisper import WhisperModel - self.audio_model = WhisperModel(model, download_root=model_root, device="auto", compute_type="int8") - self.faster = True # Only set the flag if we succesfully imported the library and opened the model. + + self.audio_model = WhisperModel( + model, + download_root=model_root, + device="auto", + compute_type="int8", + ) + self.faster = ( + True # Only set the flag if we succesfully imported the library and opened the model. + ) except ImportError: - self.logger.error("faster_whisper not installed, falling back to whisper") + self.logger.exception( + "faster_whisper not installed, falling back to whisper", + ) import whisper - self.audio_model = whisper.load_model(model, download_root=model_root).to(device) + + self.audio_model = whisper.load_model( + model, + download_root=model_root, + ).to(device) else: import whisper - self.audio_model = whisper.load_model(model, download_root=model_root).to(device) - + + self.audio_model = whisper.load_model(model, download_root=model_root).to( + device, + ) + self.temp_dir = tempfile.mkdtemp() if save_file else None - self.audio_queue = queue.Queue() - self.result_queue: "queue.Queue[str]" = queue.Queue() - + self.audio_queue: queue.Queue[str] = queue.Queue() + self.result_queue: queue.Queue[str] = queue.Queue() + self.break_threads = False self.mic_active = False - self.banned_results = [""," ","\n",None] + self.banned_results = ["", " ", "\n", None] if save_file: - self.file = open("transcribed_text.txt", "w+", encoding="utf-8") + with open("transcribed_text.txt", "w+", encoding="utf-8") as file: + self.file = file self.__setup_mic(mic_index) - - def __setup_mic(self, mic_index): + def __setup_mic(self, mic_index: int | None) -> None: if mic_index is None: self.logger.info("No mic index provided, using default") self.source = sr.Microphone(sample_rate=16000, device_index=mic_index) @@ -95,89 +133,98 @@ def __setup_mic(self, mic_index): self.logger.info("Mic setup complete") # Whisper takes a Tensor while faster_whisper only wants an NDArray - def __preprocess(self, data): + def __preprocess(self, data: bytes | memoryview) -> tuple[torch.Tensor | NDArray[np.float32], bool]: is_audio_loud_enough = self.is_audio_loud_enough(data) if self.faster: - return np.frombuffer(data, np.int16).flatten().astype(np.float32) / 32768.0,is_audio_loud_enough - else: - return torch.from_numpy(np.frombuffer(data, np.int16).flatten().astype(np.float32) / 32768.0),is_audio_loud_enough - - def is_audio_loud_enough(self, frame): + return np.frombuffer(data, np.int16).flatten().astype(np.float32) / 32768.0, is_audio_loud_enough + + return torch.from_numpy( + np.frombuffer(data, np.int16).flatten().astype(np.float32) / 32768.0, + ), is_audio_loud_enough + + def is_audio_loud_enough(self, frame: bytes | memoryview) -> bool: audio_frame = np.frombuffer(frame, dtype=np.int16) amplitude = np.mean(np.abs(audio_frame)) - return amplitude > self.hallucinate_threshold + return cast(bool, amplitude > self.hallucinate_threshold) - - def __get_all_audio(self, min_time: float = -1.): - audio = bytes() + def __get_all_audio(self, min_time: float = -1.0) -> bytes: got_audio = False time_start = time.time() while not got_audio or time.time() - time_start < min_time: while not self.audio_queue.empty(): - audio += self.audio_queue.get() + data_audio = cast(bytes, self.audio_queue.get()) got_audio = True - data = sr.AudioData(audio,16000,2) - data = data.get_raw_data() - return data - + data = sr.AudioData(data_audio, 16000, 2) + return cast(bytes, data.get_raw_data()) # Handles the task of getting the audio input via microphone. This method has been used for listen() method - def __listen_handler(self, timeout, phrase_time_limit): + def __listen_handler(self, timeout: int | None, phrase_time_limit: int | None) -> None: try: with self.source as microphone: - audio = self.recorder.listen(source=microphone, timeout=timeout, phrase_time_limit=phrase_time_limit) + audio = self.recorder.listen( + source=microphone, + timeout=timeout, + phrase_time_limit=phrase_time_limit, + ) self.__record_load(0, audio) audio_data = self.__get_all_audio() self.__transcribe(data=audio_data) except sr.WaitTimeoutError: - self.result_queue.put_nowait("Timeout: No speech detected within the specified time.") + self.result_queue.put_nowait( + "Timeout: No speech detected within the specified time.", + ) except sr.UnknownValueError: - self.result_queue.put_nowait("Speech recognition could not understand audio.") - + self.result_queue.put_nowait( + "Speech recognition could not understand audio.", + ) # This method is similar to the __listen_handler() method but it has the added ability for recording the audio for a specified duration of time - def __record_handler(self, duration, offset): + def __record_handler(self, duration: int | None, offset: int | None) -> None: with self.source as microphone: - audio = self.recorder.record(source=microphone, duration=duration, offset=offset) - + audio = self.recorder.record( + source=microphone, + duration=duration, + offset=offset, + ) + self.__record_load(0, audio) audio_data = self.__get_all_audio() self.__transcribe(data=audio_data) - - # This method takes the recorded audio data, converts it into raw format and stores it in a queue. - def __record_load(self,_, audio: sr.AudioData) -> None: + # This method takes the recorded audio data, converts it into raw format and stores it in a queue. + def __record_load(self, _: int, audio: sr.AudioData) -> None: data = audio.get_raw_data() self.audio_queue.put_nowait(data) - def __transcribe_forever(self) -> None: while True: if self.break_threads: break self.__transcribe() - - def __transcribe(self,data=None, realtime: bool = False) -> None: - if data is None: - audio_data = self.__get_all_audio() - else: - audio_data = data - audio_data,is_audio_loud_enough = self.__preprocess(audio_data) + # TODO: Allow realtime transcription + # def __transcribe(self, data: bytes | memoryview | None = None, realtime: bool = False) -> None: + def __transcribe(self, data: bytes | memoryview | None = None) -> None: + audio_data = self.__get_all_audio() if data is None else data + audio_data, is_audio_loud_enough = self.__preprocess(audio_data) if is_audio_loud_enough: - predicted_text = '' + predicted_text = "" # faster_whisper returns an iterable object rather than a string if self.faster: - segments, info = self.audio_model.transcribe(audio_data) + segments, _ = self.audio_model.transcribe(audio_data) for segment in segments: predicted_text += segment.text else: if self.english: - result = self.audio_model.transcribe(audio_data,language='english',suppress_tokens="") + result = self.audio_model.transcribe( + audio_data, + language="english", + suppress_tokens="", + ) else: - result = self.audio_model.transcribe(audio_data,suppress_tokens="") + result = self.audio_model.transcribe(audio_data, suppress_tokens="") predicted_text = result["text"] if not self.verbose: @@ -187,59 +234,61 @@ def __transcribe(self,data=None, realtime: bool = False) -> None: if predicted_text not in self.banned_results: self.result_queue.put_nowait(result) - if self.save_file: # os.remove(audio_data) self.file.write(predicted_text) - async def listen_loop_async(self, dictate: bool = False, phrase_time_limit=None) -> None: + async def listen_loop_async( + self, + dictate: bool = False, + phrase_time_limit: int | None = None, + ) -> AsyncGenerator[str, None]: for result in self.listen_continuously(phrase_time_limit=phrase_time_limit): if dictate: self.keyboard.type(result) else: yield result - - def listen_loop(self, dictate: bool = False, phrase_time_limit=None) -> None: + def listen_loop(self, dictate: bool = False, phrase_time_limit: int | None = None) -> None: for result in self.listen_continuously(phrase_time_limit=phrase_time_limit): if dictate: self.keyboard.type(result) else: - print(result) - - - def listen_continuously(self, phrase_time_limit=None): - self.recorder.listen_in_background(self.source, self.__record_load, phrase_time_limit=phrase_time_limit) + self.logger.info(result) + + def listen_continuously(self, phrase_time_limit: int | None = None) -> Generator[str, str, None]: + self.recorder.listen_in_background( + self.source, + self.__record_load, + phrase_time_limit=phrase_time_limit, + ) self.logger.info("Listening...") threading.Thread(target=self.__transcribe_forever, daemon=True).start() while True: yield self.result_queue.get() - - def listen(self, timeout = None, phrase_time_limit=None): + def listen(self, timeout: int | None = None, phrase_time_limit: int | None = None) -> str: self.logger.info("Listening...") self.__listen_handler(timeout, phrase_time_limit) while True: if not self.result_queue.empty(): return self.result_queue.get() - # This method is similar to the listen() method, but it has the ability to listen for a specified duration, mentioned in the "duration" parameter. - def record(self, duration=None, offset=None): + def record(self, duration: int | None = None, offset: int | None = None) -> str: self.logger.info("Listening...") self.__record_handler(duration, offset) while True: if not self.result_queue.empty(): return self.result_queue.get() - def toggle_microphone(self) -> None: - #TO DO: make this work + # TODO: make this work self.mic_active = not self.mic_active if self.mic_active: - print("Mic on") + self.logger.debug("Mic on") else: - print("turning off mic") - self.mic_thread.join() - print("Mic off") + self.logger.debug("turning off mic") + # self.mic_thread.join() # FIXME: doesn't exist + self.logger.debug("Mic off")