TL;DR
我想用threading
创建类似于daemon
的东西。我用的是Windows机器。所以,python-daemon
在我的系统上不起作用。你知道吗
问题
我正在开发一个应用程序,它使用threading
在检测到事件时生成音调。在应用程序中,我正在流式传输一些数据、处理和检测某些事件。当检测到事件时,会生成一个音调。我正在用daemon = True
创建一个thread
,并调用一个音调生成函数generate_tone
。但是,我不认为我需要生成多个线程来运行这个任务。你知道吗
请注意,下面的代码是另一个在后台运行的thread
调用的子模块。后台线程正在传输来自传感器的数据。在下面的代码中,generate_tone
函数生成0.2秒的正弦音调。如果我在不使用线程的情况下调用这个函数,它将占用当前线程0.2秒的时间,从而引入0.2秒的延迟。但是,我不想发生这种事。我也不想在调用线程之后使用time.sleep(0.2)
,即在thread.start()
之后,因为这样做会再次在循环中引入0.2秒。你知道吗
另外,请注意下面的代码不会产生任何声音,因为在执行前一个线程之前调用了一个新线程。但是,当调用threading.enumerate()
时,它将打印所有线程。你知道吗
import pyaudio
import wave
import numpy as np
import math
import threading
def play_tone(stream, frequency=440, length=1, rate=44100):
chunks = []
chunks.append(sine(frequency, length, rate))
chunk = np.concatenate(chunks) * 0.25
stream.write(chunk.astype(np.float32).tostring())
def generate_tone():
p = pyaudio.PyAudio()
stream = p.open(format=pyaudio.paFloat32,
channels=1, rate=44100, output=1)
play_tone(stream,length=0.20)
stream.close()
p.terminate()
def sine(frequency, length, rate):
length = int(length * rate)
factor = float(frequency) * (math.pi * 2) / rate
return np.sin(np.arange(length) * factor)
def detect_event():
T = 0.1
iter = 0
iter_max = 10
while iter < iter_max:
# some random vector x
x = np.random.rand(10,)
# check if any element of x is greater than T
if len(x[ np.where( x >= T ) ]) != 0:
beep_thread = threading.Thread(target=generate_tone,args=(),daemon=True)
beep_thread.start()
print(threading.enumerate())
iter = iter + 1
if __name__ == '__main__':
detect_event()
目前没有回答
相关问题 更多 >
编程相关推荐