[Python] A créé une classe pour jouer des vagues de péché en arrière-plan avec pyaudio

(Modifié le 20/03/07 11:52) En réponse à votre indication, nous avons modifié la déclaration des variables. De plus, le message d'erreur a été modifié pour ne s'afficher qu'une seule fois dans la fonction createData. (Modifié le 20/03/07 à 18h52) Ajout de la fonction de fermeture.

introduction

J'ai essayé de jouer des vagues de péché tout en effectuant d'autres tâches en Python, mais je n'ai pas pu trouver une telle bibliothèque, alors je l'ai créée en utilisant le threading. Une fonction qui renvoie l'entrée micro est également incluse en bonus lol

Code Python

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):  #Fonction de lecture, flux et données de forme d'onde comme arguments
        sp = 0  #Pointeur de position de lecture
        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):  #Oscillateur
        datal = []
        datar = []

        end_pos = start_pos + 0.05 * 44100
        for n in np.arange(start_pos, end_pos):
            sl = 0.0  #Effacer les données de forme d'onde à zéro
            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 lorsque l'amplitude est grande
                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)  #Ajouter à la fin
            datar.append(sr)
        datal = [int(x * 32767.0) for x in datal]  #Valeur de 32767-Entre 32767
        datar = [int(x * 32767.0) for x in datar]
        s_data = np.array([datal, datar]).T.flatten()
        data = s_data.tolist()
        #Convertir en binaire
        data = struct.pack("h" * len(data), *data)  #lister*Les arguments sont développés lorsque

        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):
        #Arrêt de fil
        self.stream_state = False
        self.thread.join()
        #Fin du flux
        self.out_stream.stop_stream()
        self.out_stream.close()
        if self.INPUT:
            self.in_stream.stop_stream()
            self.in_stream.close()
        self.pa.terminate()

Comment utiliser

Importez cette classe à partir d'un autre fichier python dans le même répertoire.

main.py

from sinwave import SinWave
import numpy as np

sw = SinWave()

main.py

#Jouer du son
sw.play(la fréquence,Le volume(0~1),phase(0~2π),Désignation LR)
#Exemple:440Hz,Volume maximum,Phase 0,Pour l'oreille gauche uniquement
sw.play(440,1,0,sw.L)
#Exemple:880Hz,Volume 80%,Phase π,Pour les deux oreilles
sw.play(880,0.8,np.pi,sw.LR)

#Arrêter le son
sw.stop(la fréquence)
#Exemple:Arrêtez le son de 880Hz
sw.stop(880)#La valeur de retour est 0
#Exemple:Arrêtez le son de 440Hz
sw.stop(440)#La valeur de retour est 0

Quand j'essaye d'arrêter le son à 440Hz qui ne sonne plus à l'accord suivant

main.py

sw.stop(440)

Afficher Cette fréquence n'est pas lue! Sur la ligne de commande. La valeur de retour de la fonction est -1. Lorsque vous quittez le programme

main.py

sw.close()

Ce code arrêtera les threads et les flux. Je pense que c'est le codage d'un débutant, mais j'espère que cela vous sera utile.

Les références

Je veux générer et lire des formes d'onde en temps réel avec teratail - python