Das deutsche Python-Forum

Code: Alles auswählen

 
Testrecording_Dauer = int(Rate / Chunk)*Dauer
t1=time.time()
for i in range(0, Testrecording_Dauer): #Eine Sekunde aufnehmen und Schauen was abgeht...
	data = stream.read(CHUNK, exception_on_overflow = False)
	aframe = np.asarray(data)
	npframes = np.vstack((npframes,aframe))
t2=time.time()
print ("finished testrecording. Dauer: ", t2-t1)

Code: Alles auswählen

# 10 Zufallswerte als Beispiel
values = np.random.rand(10)
print(values)
print(values[values >= 0.5])

Code: Alles auswählen

import pyaudio
import wave
import time
import sys
import numpy as np
import utility
from shutil import copyfile

### Aufnahmevariablen ###
FORMAT = pyaudio.paInt16
CHANNELS = 1
RATE = 44100
CHUNK = 1024
RECORD_SECONDS = 1
WAVE_OUTPUT_FILENAME = "record.wav"
go=0
audio = pyaudio.PyAudio()
###-----###

### stream definieren ###
stream = audio.open(format=FORMAT, channels=CHANNELS,
	rate=RATE, input=True, input_device_index=2,
	frames_per_buffer=CHUNK)
###-----### 

##### Messvariablen ###########
Grenzwert=0.002
Testrecording_Dauer = 1 #Sekunden
Messdauer = 10 #Sekunden
###-----###


while go<2:

### Testrecording ###
	npframes=np.array([0]) # np.Array für die Aufnahme generieren
	print("Testrecording...")
	t1=time.time()
	for i in range(0, int(RATE / CHUNK * Testrecording_Dauer)): #Eine Sekunde aufnehmen und Schauen was abgeht...
		data = stream.read(CHUNK, exception_on_overflow = False)
		#frames.append(data)
		aframe = np.asarray(data)
		npframes = np.vstack((npframes,aframe))
	t2=time.time()
	print ("finished testrecording. Dauer: ", t2-t1)
### Daten umwandeln ###
	sig = np.frombuffer(npframes, dtype='<i2').reshape(-1, CHANNELS) #Das Signal wird konvertiert
	normalized = utility.pcm2float(sig, np.float32) #und normalisiert
### Überprüfen ob der Grenzwert überschritten wurde ###
	for i in range(0, len(normalized)):
		if normalized[i] > Grenzwert:
			go = 1 #Falls ja(go=1), wird das Flag go=1 gesetzt und die Schleife abgebrochen
			break
### Messung - Wird gestartet wenn ein Geräusch detektiert wurde ###
	if go == 1: #Falls go=1 dann soll eine lange Messung durchgeführt werden. Falls nicht werden die Testmessungen weitergeführt
		print("Messung starten")
		go=0 #Zurücksetzen des Go-Flags = nach dem Messdurchgang wieder mit Testrecording beginnen
		t1=time.time()
		frames=[]
		for i in range(0, int(RATE / CHUNK * Messdauer)): #Messdauer Sekunden lang messen
			data = stream.read(CHUNK, exception_on_overflow = False)
			aframe = np.asarray(data)
			npframes = np.vstack((npframes,aframe))
		t2=time.time()
		print ("finished recording. Dauer: ", t2-t1)
		frameswav=npframes #Variable für das Abspeichern der wav-Datei
### Daten umwanden ###
		sig = np.frombuffer(npframes, dtype='<i2').reshape(-1, CHANNELS)
		normalized = utility.pcm2float(sig, np.float32) 
		go=2 # go=2 Messung nur einmal durchführen und anschliessend Programm beenden
		t8=time.time()
### Grenzwerte prüfen ###
		overgrenzwert=normalized[normalized >= Grenzwert] # Array mit allen Werten über dem Grenzwert
		overgrenzwert = np.transpose(overgrenzwert)
		indexes=np.nonzero(normalized >= Grenzwert) # Indexes der Werten über dem Grenzwert
		indexes = np.transpose(indexes)
		indexes = np.delete(indexes, 1,1)
		compressed = np.stack((indexes,overgrenzwert[:, None]),axis=1) # Zusammenführen der Arrays
### Datei speichern
		np.savetxt('compressed.dat',compressed,delimiter=';') #Speichern der Datei "Compressed.dat"
		copyfile('compressed.dat', '/home/shares/test/compressed.dat') # Kopieren in Share-Ordner
		print("Das hat gedauert, und zwar: ", time.time()-t1," Sekunden")

#stop Recording
stream.stop_stream()
stream.close()
audio.terminate()
 
### Wave Datei zur Überprüfung speichern 
waveFile = wave.open(WAVE_OUTPUT_FILENAME, 'wb')
waveFile.setnchannels(CHANNELS)
waveFile.setsampwidth(audio.get_sample_size(FORMAT))
waveFile.setframerate(RATE)
frameswav.tolist()
waveFile.writeframes(b''.join(frameswav))
waveFile.close()
copyfile(WAVE_OUTPUT_FILENAME, "/home/shares/test/wavaufnahme.wav")

Code: Alles auswählen

data = np.hstack(
    stream.read(chunk_size) for _ in range(num_chunks)
)

Code: Alles auswählen

def get_chunks(stream, chunk_size, num_chunks):
    for _ in range(num_chunks):
        yield np.asarray(stream.read(chunk_size))

def filter_chunks(stream):
    for chunk in get_chunks(...):
        if is_interesting(chunk):
            yield chunk

Code: Alles auswählen

chunks = filter(is_interesting, get_chunks(stream, chunk_size, num_chunks))

Code: Alles auswählen

audio = pyaudio.PyAudio()
stream = [audio.open(input_device_index=1),audio.open(input_device_index=2)]
data =np.hstack(stream.read(CHUNK) for in range (num_CHUNK)