79 lines
2.4 KiB
Python
79 lines
2.4 KiB
Python
"""Liberation du flux audio.
|
|
|
|
Bug du 2026-08-22 : trois dictees d'affilee sans le moindre son. Cause :
|
|
`Recorder.start()` ecrasait `self._stream` sans fermer le precedent. Quand un
|
|
`stop()` se bloquait (peripherique accapare), le flux restait ouvert POUR TOUJOURS
|
|
— et l'essai suivant en empilait un autre. Zonza finissait par tenir elle-meme le
|
|
micro qu'elle tentait d'ouvrir.
|
|
|
|
On libere donc AVANT d'ouvrir, avec `abort()` (immediat) plutot que `stop()` (qui
|
|
draine les tampons et peut bloquer), et sans jamais lever.
|
|
"""
|
|
import numpy as np
|
|
|
|
from capture import Recorder
|
|
|
|
|
|
class FauxFlux:
|
|
def __init__(self, casse_sur=()):
|
|
self.actions = []
|
|
self.casse_sur = casse_sur
|
|
|
|
def start(self): self.actions.append("start")
|
|
def abort(self):
|
|
self.actions.append("abort")
|
|
if "abort" in self.casse_sur: raise RuntimeError("peripherique occupe")
|
|
def close(self):
|
|
self.actions.append("close")
|
|
if "close" in self.casse_sur: raise RuntimeError("close impossible")
|
|
|
|
|
|
def _rec(flux_successifs):
|
|
it = iter(flux_successifs)
|
|
return Recorder(16000, ouvrir_flux=lambda **kw: next(it))
|
|
|
|
|
|
def test_demarrer_libere_le_flux_precedent():
|
|
a, b = FauxFlux(), FauxFlux()
|
|
r = _rec([a, b])
|
|
r.start()
|
|
r.start() # second demarrage sans stop entre les deux
|
|
assert "abort" in a.actions and "close" in a.actions, "l'ancien flux doit etre libere"
|
|
assert b.actions == ["start"]
|
|
|
|
|
|
def test_on_prefere_abort_a_stop():
|
|
a = FauxFlux()
|
|
r = _rec([a])
|
|
r.start(); r.stop()
|
|
assert "abort" in a.actions
|
|
assert "stop" not in a.actions, "stop() draine les tampons et peut bloquer"
|
|
|
|
|
|
def test_un_flux_recalcitrant_ne_fait_pas_echouer_le_demarrage():
|
|
a, b = FauxFlux(casse_sur=("abort", "close")), FauxFlux()
|
|
r = _rec([a, b])
|
|
r.start()
|
|
r.start() # ne doit rien lever malgre les deux echecs
|
|
assert b.actions == ["start"]
|
|
|
|
|
|
def test_le_flux_est_detache_meme_si_la_liberation_echoue():
|
|
a = FauxFlux(casse_sur=("abort", "close"))
|
|
r = _rec([a])
|
|
r.start(); r.stop()
|
|
assert r._stream is None, "une reference morte empecherait le prochain demarrage"
|
|
|
|
|
|
def test_stop_rend_l_audio_capture():
|
|
a = FauxFlux()
|
|
r = _rec([a])
|
|
r.start()
|
|
r._callback(np.ones((512, 1), dtype="float32"), 512, None, None)
|
|
audio = r.stop()
|
|
assert audio.shape == (512,)
|
|
|
|
|
|
def test_stop_sans_demarrage_ne_leve_pas():
|
|
assert _rec([]).stop().size == 0
|