|
14 | 14 | try: |
15 | 15 | import sounddevice as sd |
16 | 16 | except Exception: # pragma: no cover |
17 | | - sd = None # type: ignore |
| 17 | + sd = None # type: ignore # pylint: disable=invalid-name |
18 | 18 |
|
19 | 19 |
|
20 | 20 | class AudioRecorder: |
@@ -179,12 +179,12 @@ def start(self, name: Optional[str] = None) -> str: |
179 | 179 | fd, mic_path = tempfile.mkstemp(suffix=".f32") |
180 | 180 | os.close(fd) |
181 | 181 | self._mic_raw_path = Path(mic_path) |
182 | | - self._mic_fh = open(mic_path, "wb") |
| 182 | + self._mic_fh = open(mic_path, "wb") # pylint: disable=consider-using-with |
183 | 183 |
|
184 | 184 | fd, lb_path = tempfile.mkstemp(suffix=".f32") |
185 | 185 | os.close(fd) |
186 | 186 | self._loopback_raw_path = Path(lb_path) |
187 | | - self._loopback_fh = open(lb_path, "wb") |
| 187 | + self._loopback_fh = open(lb_path, "wb") # pylint: disable=consider-using-with |
188 | 188 |
|
189 | 189 | try: |
190 | 190 | self._audio_interface = sd.InputStream( |
@@ -227,7 +227,7 @@ def _start_macos_loopback(self) -> bool: |
227 | 227 |
|
228 | 228 | self._macos_loopback = MacOSLoopback( |
229 | 229 | sample_rate=self.sample_rate, |
230 | | - callback=lambda indata: self._write_loopback_frame(indata), |
| 230 | + callback=self._write_loopback_frame, |
231 | 231 | ) |
232 | 232 | self._macos_loopback.start() |
233 | 233 | return True |
@@ -290,19 +290,9 @@ def stop(self) -> bool: |
290 | 290 | ch = self._mic_channels_written |
291 | 291 | mic_data = raw.reshape(-1, ch) if ch > 1 else raw.reshape(-1, 1) |
292 | 292 |
|
293 | | - if self._has_loopback and loopback_raw_bytes: |
294 | | - lb_raw = np.frombuffer(loopback_raw_bytes, dtype=np.float32) |
295 | | - lbch = self._loopback_channels_written |
296 | | - lb_data = lb_raw.reshape(-1, lbch) if lbch > 1 else lb_raw.reshape(-1, 1) |
297 | | - if loopback_sr and abs(loopback_sr - self.sample_rate) > 1: |
298 | | - from scipy.signal import resample as scipy_resample |
299 | | - n_out = int(round(len(lb_data) * self.sample_rate / loopback_sr)) |
300 | | - lb_data = scipy_resample(lb_data.squeeze(), n_out).astype(np.float32).reshape(-1, 1) |
301 | | - audio_data = self._mix_stereo(mic_data, lb_data) |
302 | | - out_channels = 2 |
303 | | - else: |
304 | | - audio_data = mic_data |
305 | | - out_channels = ch |
| 293 | + audio_data, out_channels = self._build_audio_data( |
| 294 | + mic_data, ch, loopback_raw_bytes, loopback_sr |
| 295 | + ) |
306 | 296 |
|
307 | 297 | max_val = np.max(np.abs(audio_data)) |
308 | 298 | if max_val > 0: |
@@ -339,6 +329,27 @@ def _cleanup_temp_files(self) -> None: |
339 | 329 | self._loopback_raw_path.unlink(missing_ok=True) |
340 | 330 | self._loopback_raw_path = None |
341 | 331 |
|
| 332 | + def _build_audio_data( |
| 333 | + self, |
| 334 | + mic_data: np.ndarray, |
| 335 | + mic_ch: int, |
| 336 | + loopback_raw_bytes: bytes, |
| 337 | + loopback_sr: Optional[int], |
| 338 | + ) -> tuple[np.ndarray, int]: |
| 339 | + """Mix mic and loopback raw bytes into final audio array.""" |
| 340 | + if not (self._has_loopback and loopback_raw_bytes): |
| 341 | + return mic_data, mic_ch |
| 342 | + |
| 343 | + from scipy.signal import resample as scipy_resample # pylint: disable=import-outside-toplevel |
| 344 | + |
| 345 | + lb_raw = np.frombuffer(loopback_raw_bytes, dtype=np.float32) |
| 346 | + lbch = self._loopback_channels_written |
| 347 | + lb_data = lb_raw.reshape(-1, lbch) if lbch > 1 else lb_raw.reshape(-1, 1) |
| 348 | + if loopback_sr and abs(loopback_sr - self.sample_rate) > 1: |
| 349 | + n_out = int(round(len(lb_data) * self.sample_rate / loopback_sr)) |
| 350 | + lb_data = scipy_resample(lb_data.squeeze(), n_out).astype(np.float32).reshape(-1, 1) |
| 351 | + return self._mix_stereo(mic_data, lb_data), 2 |
| 352 | + |
342 | 353 | def _mix_stereo(self, mic: np.ndarray, loopback: np.ndarray) -> np.ndarray: |
343 | 354 | """Combine mic and loopback into a stereo array [n, 2] (L=mic, R=loopback).""" |
344 | 355 | # Reduce to mono |
|
0 commit comments