创建守护进程以在后台运行函数而不使用线程

2024-04-27 05:23:14 发布

您现在位置:Python中文网/ 问答频道 /正文

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()

Tags: importstreamratedefnp线程threadlength