Manages chunk-wise audio upsampling with overlap handling. This class processes sequential audio chunks, upsamples them from 24kHz to 48kHz using `scipy.signal.resample_poly`, and manages overlap between chunks to mitigate boundary artifacts. The processed, upsampled audio segments
| 4 | from typing import Optional |
| 5 | |
| 6 | class UpsampleOverlap: |
| 7 | """ |
| 8 | Manages chunk-wise audio upsampling with overlap handling. |
| 9 | |
| 10 | This class processes sequential audio chunks, upsamples them from 24kHz to 48kHz |
| 11 | using `scipy.signal.resample_poly`, and manages overlap between chunks to |
| 12 | mitigate boundary artifacts. The processed, upsampled audio segments are |
| 13 | returned as Base64 encoded strings. It maintains internal state to handle |
| 14 | the overlap correctly across calls. |
| 15 | """ |
| 16 | def __init__(self): |
| 17 | """ |
| 18 | Initializes the UpsampleOverlap processor. |
| 19 | |
| 20 | Sets up the internal state required for tracking previous audio chunks |
| 21 | and their resampled versions to handle overlaps during processing. |
| 22 | """ |
| 23 | self.previous_chunk: Optional[np.ndarray] = None |
| 24 | self.resampled_previous_chunk: Optional[np.ndarray] = None |
| 25 | |
| 26 | def get_base64_chunk(self, chunk: bytes) -> str: |
| 27 | """ |
| 28 | Processes an incoming audio chunk, upsamples it, and returns the relevant segment as Base64. |
| 29 | |
| 30 | Converts the raw PCM bytes (assumed 16-bit signed integer) chunk to a |
| 31 | float32 numpy array, normalizes it, and upsamples from 24kHz to 48kHz. |
| 32 | It uses the previous chunk's data to create an overlap, resamples the |
| 33 | combined audio, and extracts the central portion corresponding primarily |
| 34 | to the current chunk, using overlap to smooth transitions. The state is |
| 35 | updated for the next call. The extracted audio segment is converted back |
| 36 | to 16-bit PCM bytes and returned as a Base64 encoded string. |
| 37 | |
| 38 | Args: |
| 39 | chunk: Raw audio data bytes (PCM 16-bit signed integer format expected). |
| 40 | |
| 41 | Returns: |
| 42 | A Base64 encoded string representing the upsampled audio segment |
| 43 | corresponding to the input chunk, adjusted for overlap. Returns an |
| 44 | empty string if the input chunk is empty. |
| 45 | """ |
| 46 | audio_int16 = np.frombuffer(chunk, dtype=np.int16) |
| 47 | # Handle potential empty chunks gracefully |
| 48 | if audio_int16.size == 0: |
| 49 | return "" # Return empty string for empty input chunk |
| 50 | |
| 51 | audio_float = audio_int16.astype(np.float32) / 32768.0 |
| 52 | |
| 53 | # Upsample the current chunk independently first, needed for state and first chunk logic |
| 54 | upsampled_current_chunk = resample_poly(audio_float, 48000, 24000) |
| 55 | |
| 56 | if self.previous_chunk is None: |
| 57 | # First chunk: Output the first half of its upsampled version |
| 58 | half = len(upsampled_current_chunk) // 2 |
| 59 | part = upsampled_current_chunk[:half] |
| 60 | else: |
| 61 | # Subsequent chunks: Combine previous float chunk with current float chunk |
| 62 | combined = np.concatenate((self.previous_chunk, audio_float)) |
| 63 | # Upsample the combined chunk |