103 lines
3.7 KiB
Python
103 lines
3.7 KiB
Python
"""Capture audio et collage du texte transcrit.
|
|
|
|
Aucune dépendance UI : importable par le contrôleur et l'app. La transcription
|
|
et les appels système propres à une plateforme vivent désormais ailleurs
|
|
(`moteur/`, `systeme/`) — ce module ne garde que ce qui n'a pas de frontière
|
|
de plateforme : la capture du micro et le collage du résultat.
|
|
"""
|
|
import time
|
|
|
|
import numpy as np
|
|
import pyperclip
|
|
import sounddevice as sd
|
|
from pynput.keyboard import Controller, Key
|
|
|
|
from core import CONFIG, attendre
|
|
|
|
_kbd = Controller()
|
|
|
|
|
|
def inject_text(text, systeme):
|
|
"""Copie le texte dans le presse-papier puis colle avec Cmd+V dans l'app active.
|
|
|
|
On attend d'abord que l'utilisateur ait relâché ses touches : s'il tient encore
|
|
la hotkey quand on envoie Cmd+V, le système reçoit une combinaison différente et
|
|
l'application cible ne colle rien — sans la moindre erreur de notre côté
|
|
(constaté le 2026-08-21).
|
|
"""
|
|
if not text:
|
|
return
|
|
pyperclip.copy(text)
|
|
libre = attendre(
|
|
lambda: not systeme.modificateurs_enfonces(),
|
|
CONFIG["attente_modificateurs_s"],
|
|
0.03,
|
|
)
|
|
if not libre:
|
|
print("[collage] modificateurs encore enfoncés — collage tenté quand même")
|
|
time.sleep(CONFIG["delai_presse_papier_s"]) # laisser le presse-papier se mettre à jour
|
|
with _kbd.pressed(Key.cmd):
|
|
_kbd.press("v")
|
|
_kbd.release("v")
|
|
print("[collage] ok")
|
|
|
|
|
|
class Recorder:
|
|
"""Capture le micro en mémoire (16 kHz mono float32). Start/stop par le contrôleur.
|
|
|
|
on_block : callback optionnel appelé à chaque bloc audio avec le tableau numpy du bloc
|
|
(utilisé pour alimenter l'animation en temps réel).
|
|
"""
|
|
|
|
def __init__(self, sample_rate, on_block=None, ouvrir_flux=None):
|
|
self.sample_rate = sample_rate
|
|
self.on_block = on_block
|
|
self._frames = []
|
|
self._stream = None
|
|
# ouvrir_flux : injectable pour les tests (aucun peripherique requis).
|
|
self._ouvrir_flux = ouvrir_flux or (lambda **kw: sd.InputStream(**kw))
|
|
|
|
def _callback(self, indata, frames, time_info, status):
|
|
block = indata.copy()
|
|
self._frames.append(block)
|
|
if self.on_block is not None:
|
|
self.on_block(block.flatten())
|
|
|
|
def _liberer(self):
|
|
"""Libère le flux courant. Ne lève jamais, ne bloque pas durablement.
|
|
|
|
On DÉTACHE la référence avant d'agir : même si la libération échoue, le
|
|
prochain démarrage ne doit pas retomber sur un flux mort. Et on utilise
|
|
`abort()` plutôt que `stop()` — ce dernier draine les tampons et peut se
|
|
bloquer sur un périphérique accaparé, ce qui laissait le micro tenu pour
|
|
toujours (2026-08-22 : trois dictées muettes d'affilée).
|
|
"""
|
|
flux, self._stream = self._stream, None
|
|
if flux is None:
|
|
return
|
|
for action in ("abort", "close"):
|
|
try:
|
|
getattr(flux, action)()
|
|
except Exception as exc:
|
|
print(f"[audio] {action}() du flux précédent a échoué : {exc}")
|
|
|
|
def start(self):
|
|
# un flux orphelin d'un essai precedent tiendrait le micro : on libere d'abord
|
|
self._liberer()
|
|
self._frames = []
|
|
self._stream = self._ouvrir_flux(
|
|
samplerate=self.sample_rate,
|
|
channels=1,
|
|
dtype="float32",
|
|
callback=self._callback,
|
|
)
|
|
self._stream.start()
|
|
|
|
def stop(self):
|
|
"""Arrête le flux et renvoie l'audio concaténé (numpy 1-D float32)."""
|
|
frames, self._frames = self._frames, []
|
|
self._liberer()
|
|
if not frames:
|
|
return np.zeros(0, dtype="float32")
|
|
return np.concatenate(frames, axis=0).flatten()
|