Audio Stream - Process and Buffer a Real-Time Audio Data Stream
Question Details
Round 1 Coding
Problem
You are building an audio processing pipeline. Audio arrives as a continuous stream of integer samples. Implement an AudioBuffer that:
1. Accepts incoming samples one at a time.
2. When the internal buffer reaches a fixed frame_size, processes the frame (apply a simple gain factor) and emits it.
3. Allows querying how many complete frames have been emitted.
python
class AudioBuffer:
def __init__(self, frame_size: int, gain: float):
pass
def write(self,
**sample**: int) -> list[int] | None:
**Returns** the processed frame when buffer is full; None otherwise
pass
def frames_emitted(self) -> int:
pass
def flush(self) -> list[int] | None:
# Emit a partial frame padded with zeros if any samples remain
pass
Example
buf = AudioBuffer(frame_size=4, gain=2.0)
buf.write(10) # -> None
buf.write(20) # -> None
buf.write(30) # -> None
buf.write(40) # -> [20, 40, 60, 80] (gain applied)
buf.frames_emitted() # -> 1
buf.write(50)
buf.flush() # -> [100, 0, 0, 0]
Follow-ups
- How do you handle overflow when gain causes sample values to exceed int16 range?
- How would you implement a sliding window (overlapping frames) instead of non-overlapping?
- If samples arrive from multiple concurrent threads, how do you make
writethread-safe? - How do you design this for a real-time system where dropped frames cause audible glitches?
Full Details
Round 1 Coding
Problem
You are building an audio processing pipeline. Audio arrives as a continuous stream of integer samples. Implement an AudioBuffer that:
1. Accepts incoming samples one at a time.
2. When the internal buffer reaches a fixed frame_size, processes the frame (apply a simple gain factor) and emits it.
3. Allows querying how many complete frames have been emitted.
python
class AudioBuffer:
def __init__(self, frame_size: int, gain: float):
pass
def write(self,
**sample**: int) -> list[int] | None:
**Returns** the processed frame when buffer is full; None otherwise
pass
def frames_emitted(self) -> int:
pass
def flush(self) -> list[int] | None:
# Emit a partial frame padded with zeros if any samples remain
pass
Example
buf = AudioBuffer(frame_size=4, gain=2.0)
buf.write(10) # -> None
buf.write(20) # -> None
buf.write(30) # -> None
buf.write(40) # -> [20, 40, 60, 80] (gain applied)
buf.frames_emitted() # -> 1
buf.write(50)
buf.flush() # -> [100, 0, 0, 0]
Follow-ups
- How do you handle overflow when gain causes sample values to exceed int16 range?
- How would you implement a sliding window (overlapping frames) instead of non-overlapping?
- If samples arrive from multiple concurrent threads, how do you make
writethread-safe? - How do you design this for a real-time system where dropped frames cause audible glitches?
About This Question
This is a reported interview question from a toma interview during the phone round.
It covers the following topics: Coding, Phone, Sliding Window .