Assert RMS stays above `min_rms` for `duration` seconds. Raises `AssertionError` if RMS drops below threshold at any sample.
(
self, *, min_rms: float = 0.01, duration: float = 1.0, sample_interval: float = 0.05
)
| 179 | await asyncio.sleep(0.05) |
| 180 | |
| 181 | async def assert_audio_continuous( |
| 182 | self, *, min_rms: float = 0.01, duration: float = 1.0, sample_interval: float = 0.05 |
| 183 | ) -> None: |
| 184 | """Assert RMS stays above `min_rms` for `duration` seconds. |
| 185 | |
| 186 | Raises `AssertionError` if RMS drops below threshold at any sample. |
| 187 | """ |
| 188 | end = asyncio.get_event_loop().time() + duration |
| 189 | seen_above_threshold = False |
| 190 | while asyncio.get_event_loop().time() < end: |
| 191 | if self._latest_rms >= min_rms: |
| 192 | seen_above_threshold = True |
| 193 | else: |
| 194 | if seen_above_threshold: |
| 195 | raise AssertionError( |
| 196 | f"audio energy dropped to {self._latest_rms} below {min_rms}" |
| 197 | ) |
| 198 | await asyncio.sleep(sample_interval) |
| 199 | if not seen_above_threshold: |
| 200 | raise AssertionError( |
| 201 | f"audio energy never reached {min_rms} during the {duration}s window" |
| 202 | ) |
| 203 | |
| 204 | async def _run(self) -> None: |
| 205 | async for ev in self._stream: |
no test coverage detected