[Python] Created a class to play sin waves in the background with pyaudio

(Edited at 20/03/07 11:52) In response to your indication, we changed the variable declaration. Another change is to display the error message only once in the createData function. (20/03/07 18:52 edit) Added close function.

Introduction

I tried to play a sine wave while performing other tasks in Python, but I couldn't find such a library, so I made it using threading. A function that returns the microphone input is also included as a bonus lol

Python code

sinwave.py

import threading
import time
import pyaudio
import numpy as np
import struct


class SinWave():
    # settings
    INPUT = True
    OUT_FORMAT = pyaudio.paInt16
    IN_FORMAT = pyaudio.paFloat32
    RATE = 44100
    CHUNK = 1024

    L = 1
    R = 2
    LR = 3

    def __init__(self):
        self.pos = 0
        self.flagl = False
        self.flagr = False
        self.fvpp_list = [[0, 0, 0, 3]]
        self.stream_state = True

        self.pa = pyaudio.PyAudio()
        self.out_stream = self.pa.open(format=self.OUT_FORMAT,
                                       channels=2,
                                       rate=self.RATE,
                                       input=False,
                                       output=True,
                                       frames_per_buffer=self.CHUNK)
        self.thread = threading.Thread(target=self.output)
        self.thread.start()

        if self.INPUT:
            self.in_stream = self.pa.open(format=self.IN_FORMAT,
                                          channels=1,
                                          rate=self.RATE,
                                          input=True,
                                          output=False,
                                          frames_per_buffer=self.CHUNK)

    def output(self):
        while self.stream_state:
            data, self.pos = self.createData(
                self.fvpp_list, start_pos=self.pos)
            self.update(self.out_stream, data)

    def update(self, stream, data):  #Playback function, stream and waveform data as arguments
        sp = 0  #Playback position pointer
        buffer = data[sp:sp + self.CHUNK * 2]
        while buffer:
            stream.write(buffer)
            sp = sp + self.CHUNK * 2
            buffer = data[sp:sp + self.CHUNK * 2]

    def createData(self, fvpp, start_pos=0):  #Oscillator
        datal = []
        datar = []

        end_pos = start_pos + 0.05 * 44100
        for n in np.arange(start_pos, end_pos):
            sl = 0.0  #Clear waveform data to zero
            sr = 0.0
            for f in fvpp:
                sl += np.sin(2 * np.pi * f[0] * n /
                             44100 + f[2]) * f[1] * (f[3] % 2)
                sr += np.sin(2 * np.pi * f[0] * n /
                             44100 + f[2]) * f[1] * (f[3] // 2)
                #Clipping when the amplitude is large
                if sl > 1.0:
                    sl = 1.0
                    if self.flagl:
                        print("WARNING! Left Max Volume!!")
                        self.flagl = False
                if sr > 1.0:
                    sr = 1.0
                    if self.flagr:
                        print("WARNING! Right Max Volume!!")
                        self.flagr = False
                if sl < -1.0:
                    sl = -1.0
                if sr < -1.0:
                    sr = -1.0

            datal.append(sl)  #Add to the end
            datar.append(sr)
        datal = [int(x * 32767.0) for x in datal]  #Value from 32767-Between 32767
        datar = [int(x * 32767.0) for x in datar]
        s_data = np.array([datal, datar]).T.flatten()
        data = s_data.tolist()
        #Convert to binary
        data = struct.pack("h" * len(data), *data)  #on list*If you add, the argument will be expanded

        return data, end_pos

    def play(self, freq, vol, phase, pan):
        self.fvpp_list.append([freq, abs(vol), phase, pan])
        self.flagl = True
        self.flagr = True

    def stop(self, freq):
        if freq in [row[0] for row in self.fvpp_list]:
            del self.fvpp_list[[row[0] for row in self.fvpp_list].index(freq)]
            return 0
        else:
            print("This frequency is not played!")
            return -1

    def input(self):
        ret = self.in_stream.read(self.CHUNK, exception_on_overflow=False)
        ret = np.fromstring(ret, np.float32)

        return ret

    def close(self):
        #Thread stop
        self.stream_state = False
        self.thread.join()
        #End of stream
        self.out_stream.stop_stream()
        self.out_stream.close()
        if self.INPUT:
            self.in_stream.stop_stream()
            self.in_stream.close()
        self.pa.terminate()

How to use

Import this class from another python file in the same directory.

main.py

from sinwave import SinWave
import numpy as np

sw = SinWave()

main.py

#Sound reproduction
sw.play(frequency,Volume(0~1),phase(0~2π),LR designation)
#Example:440Hz,Maximum volume,Phase 0,For left ear only
sw.play(440,1,0,sw.L)
#Example:880Hz,Volume 80%,Phase π,For both ears
sw.play(880,0.8,np.pi,sw.LR)

#Stopping the sound
sw.stop(frequency)
#Example:Stop the sound of 880Hz
sw.stop(880)#Return value is 0
#Example:Stop the 440Hz sound
sw.stop(440)#Return value is 0

When I try to stop a 440Hz note that is no longer ringing at the next chord

main.py

sw.stop(440)

Display This frequency is not played! On the command line. The return value of the function is -1. When quitting the program

main.py

sw.close()

This code will stop threads and streams. I think that it is a beginner coding bare, but I hope it will be helpful.

References

I want to generate and play waveforms in real time with teratail--python