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 = Nonewith 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)
                ifnot audio:
                    breakexcept (IOError, ValueError):
                breakelse:
                sample = Sample.from_raw_frames(audio, 2, 44100, 2)
                if sample_array isNone:
                    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 = Noneimport pyaudio

import ctypes

import sys 

CHUNK = 4096

p = pyaudio.PyAudio()

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

    



classRadioThread(threading.Thread):
    defrun(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)


    defaudio_playback(self,pcm_stream):
        global stop_peradio_thread
        while stop_peradio_thread==False:
            try:
                audio = pcm_stream.read(CHUNK)
                stream.write(audio.tobytes())
            except:
                passclassMiniaudioDecoderPcmStream(miniaudio.StreamableSource):
    def__init__(self, fmt, stream):
        self.pcm_stream = miniaudio.stream_any(stream, fmt, dither=miniaudio.DitherMode.TRIANGLE)

    defread(self, size):
        try:
            return self.pcm_stream.send(size)
        except StopIteration:
            returnb""defmain():
    global stop_peradio_thread
    stop_peradio_thread = False
    t1 = RadioThread()
    t1.start()
    whileTrue:
        try:
            time.sleep(1)
        except KeyboardInterrupt:
            stop_peradio_thread = True
            t1.join()
            sys.exit()
            
main()