0
$\begingroup$

I've made a circular buffer of size 10,000 in which I'm getting real time input from the mic. Now I want to access first 1920 samples every time and do the post processing but I'm only able to access all the elements instead of getting the required number of elements each time. Is there a way to access a chunk of values from a circular buffer in real time? Any kind of help would be appreciated. I'm adding the code which I'm using to get real time input from mic.

import numpy as np
import struct
import collections
import python_speech_features
import threading

FORMATIN = pyaudio.paInt16
# FORMATOUT = pyaudio.paFloat32
FORMATOUT = FORMATIN
CHANNELS = 1
RATE = 48000
PLAYCHUNK = 1920 #40ms
RECORDCHUNK = 960 #20ms

audio = pyaudio.PyAudio()



# start Recording
streamIn = audio.open(format=FORMATIN, channels=CHANNELS,
                      rate=RATE, input=True, input_device_index=0,
                      frames_per_buffer=PLAYCHUNK)
streamOut = audio.open(format=FORMATOUT, channels=CHANNELS,
                       rate=RATE, output=True, input_device_index=0,
                       frames_per_buffer=PLAYCHUNK)
print("recording...")

x = RingBuffer(20000)

while True:
    in_data = streamIn.read(CHUNK)
    in_data = np.frombuffer(in_data, dtype = np.int16)
    for i in in_data:
        x.append(i)
    print(x.get())
    
streamOut.write(in_data)

Furthermore, the code for Ring Buffer which I'm using to store real-time data is:

    """ class that implements a not-yet-full buffer """
    def __init__(self,size_max):
        self.max = size_max
        self.data = []
        
    class __Full:
        """ class that implements a full buffer """
        def append(self, x):
            """ Append an element overwriting the oldest one. """
            self.data[self.cur] = x
            self.cur = (self.cur+1) % self.max
        def get(self):
            """ return list of elements in correct order """
            return self.data[self.cur:]+self.data[:self.cur]

    def append(self,x):
        """append an element at the end of the buffer"""
        self.data.append(x)
        if len(self.data) == self.max:
            self.cur = 0
            # Permanently change self's class from non-full to full
            self.__class__ = self.__Full

    def get(self):
        """ Return a list of elements from the oldest to the newest. """
        return self.data
    
$\endgroup$
3
  • $\begingroup$ Without some actual code it will be hard to help you. $\endgroup$
    – Max
    Jun 8 at 8:50
  • $\begingroup$ Okay lemme edit it and add the code I'm using to get input from mic. $\endgroup$ Jun 8 at 10:10
  • $\begingroup$ @Max, I've added the code I'm using to get real time input from mic and to store data into the buffer. $\endgroup$ Jun 8 at 10:19

0

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.