[Python] Erstellt eine Klasse, um mit pyaudio Sinuswellen im Hintergrund zu spielen

(Bearbeitet am 20/03/07 11:52) Als Reaktion auf Ihre Angabe haben wir die Deklaration der Variablen geändert. Außerdem wurde die Fehlermeldung so geändert, dass sie nur einmal in der Funktion createData angezeigt wird. (Bearbeitet am 20/03/07 18:52) Schließfunktion hinzugefügt.

Einführung

Ich habe versucht, Sinuswellen zu spielen, während ich andere Aufgaben in Python ausgeführt habe, aber ich konnte eine solche Bibliothek nicht finden, also habe ich sie mithilfe von Threading erstellt. Eine Funktion, die den Mikrofoneingang zurückgibt, ist ebenfalls als Bonus enthalten

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):  #Wiedergabefunktion, Stream- und Wellenformdaten als Argumente
        sp = 0  #Wiedergabepositionszeiger
        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):  #Oszillator
        datal = []
        datar = []

        end_pos = start_pos + 0.05 * 44100
        for n in np.arange(start_pos, end_pos):
            sl = 0.0  #Löschen Sie die Wellenformdaten auf Null
            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 bei großer Amplitude
                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)  #Zum Ende hinzufügen
            datar.append(sr)
        datal = [int(x * 32767.0) for x in datal]  #Wert von 32767-Zwischen 32767
        datar = [int(x * 32767.0) for x in datar]
        s_data = np.array([datal, datar]).T.flatten()
        data = s_data.tolist()
        #In Binär konvertieren
        data = struct.pack("h" * len(data), *data)  #auflisten*Argumente werden erweitert, wenn

        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):
        #Fadenstopp
        self.stream_state = False
        self.thread.join()
        #Ende der Übertragung
        self.out_stream.stop_stream()
        self.out_stream.close()
        if self.INPUT:
            self.in_stream.stop_stream()
            self.in_stream.close()
        self.pa.terminate()

Wie benutzt man

Importieren Sie diese Klasse aus einer anderen Python-Datei im selben Verzeichnis.

main.py

from sinwave import SinWave
import numpy as np

sw = SinWave()

main.py

#Sound abspielen
sw.play(Frequenz,Volumen(0~1),Phase(0~2π),LR Bezeichnung)
#Beispiel:440Hz,Maximale Lautstärke,Phase 0,Nur für das linke Ohr
sw.play(440,1,0,sw.L)
#Beispiel:880Hz,Band 80%,Phase π,Für beide Ohren
sw.play(880,0.8,np.pi,sw.LR)

#Den Ton stoppen
sw.stop(Frequenz)
#Beispiel:Stoppen Sie den Ton von 880Hz
sw.stop(880)#Rückgabewert ist 0
#Beispiel:Stoppen Sie den Ton von 440Hz
sw.stop(440)#Rückgabewert ist 0

Wenn ich versuche, den 440-Hz-Sound zu stoppen, der beim nächsten Akkord nicht mehr klingelt

main.py

sw.stop(440)

Anzeige Diese Frequenz wird nicht gespielt! In der Befehlszeile. Der Rückgabewert der Funktion ist -1. Beim Beenden des Programms

main.py

sw.close()

Dieser Code stoppt Threads und Streams. Ich denke, dass es ein Anfänger ist, aber ich hoffe, es wird hilfreich sein.

Verweise

Ich möchte Wellenformen in Echtzeit mit teratail - python erzeugen und abspielen