Python: How To Decode A Mp3 Chunk Into PCM Samples?

I'm trying to catch chunks of an mp3 webstream and decoding them into PCM samples for signal processing. I tried to catch the audio via requests and io.BytesIO to save the data as

Solution 1:

You are missing the decoding of mp3 stream. You are just saving mp3 file as wav.

You first need to decode mp3 audio. Which will give you PCM samples + audio info.

Solution 2:

With the help from Irmen and his "miniaudio" and "synthesizer" library, I could solve the problem.

The problem was, that most radio webstreams uses the ICECAST protocol, which includes interleaved metadata information, so you can't decode it directly.

With the example script https://github.com/irmen/synthesizer/blob/master/examples/internetradio.py as template, I could write a script, which records a webstream until KeyboardInterrupt and saves it as a .wav file.

Here's the main part I edited:

 ...

 def _audio_playback(self, pcm_stream):
    sample_array = None

    with Output(mixing="sequential", frames_per_chunk=44100 // 4) as output:
        print("begin recording")
        while self.decode_flag:
            try:
                audio = pcm_stream.read(44100 * 2 * 2 // 20)
                if not audio:
                    break
            except (IOError, ValueError):
                break
            else:
                sample = Sample.from_raw_frames(audio, 2, 44100, 2)
                if sample_array is None:
                    sample_array = sample.get_frames_numpy_float()
                else:
                    sample_array = np.append(sample_array, sample.get_frames_numpy_float(), axis=0)
        print("...saving")
        wavf.write(self.file_location, 44100, sample_array)
        print("saved")

...

Solution 3:

Based on Bendzko answer here is my code:

pip install pyaudio miniaudio

import threading

import urllib.request

import time
try:
    import miniaudio
except ImportError:
    miniaudio = None
    
import pyaudio

import ctypes

import sys 

CHUNK = 4096

p = pyaudio.PyAudio()

stream = p.open(format=pyaudio.paInt16,channels=2,rate=44100,output=True)

    



class RadioThread(threading.Thread):
    def run(self):
        self.url = "https://impradio.bytemasters.gr/8002/stream"
        #run in threading
        client = miniaudio.IceCastClient(self.url)
        pcm_stream = MiniaudioDecoderPcmStream(client.audio_format,client)
        self.audio_playback(pcm_stream)


    def audio_playback(self,pcm_stream):
        global stop_peradio_thread
        while stop_peradio_thread==False:
            try:
                audio = pcm_stream.read(CHUNK)
                stream.write(audio.tobytes())
            except:
                pass
                
                
class MiniaudioDecoderPcmStream(miniaudio.StreamableSource):
    def __init__(self, fmt, stream):
        self.pcm_stream = miniaudio.stream_any(stream, fmt, dither=miniaudio.DitherMode.TRIANGLE)

    def read(self, size):
        try:
            return self.pcm_stream.send(size)
        except StopIteration:
            return b""

def main():
    global stop_peradio_thread
    stop_peradio_thread = False
    t1 = RadioThread()
    t1.start()
    while True:
        try:
            time.sleep(1)
        except KeyboardInterrupt:
            stop_peradio_thread = True
            t1.join()
            sys.exit()
            
main()