如何结合asciimatics烟花效果与winsound功能实现同步触发运行

核心逻辑 #

• 放弃subprocess多进程方案，改用threading多线程实现烟花渲染、音频播放两个任务的并发，多线程模式下启停控制和状态同步的实现成本远低于多进程
• 引入线程安全的threading.Event作为全局停止信号，每次需要重启时先向两个任务发送停止信号，等待旧线程完全退出后再启动新的线程实例，保证二者完全同步
• 复用asciimatics自带的键盘事件监听能力捕获空格键操作，触发同步重启逻辑
• 音频函数调整为可中断版本，每次播放音符前先检测停止信号，避免重启时需要等完整段音频播放完毕

完整实现代码 #

import threading
import winsound
import os
from asciimatics.screen import Screen
from asciimatics.scene import Scene
from asciimatics.effects import Stars
from asciimatics.particles import Firework
from asciimatics.exceptions import ResizeScreenError, StopApplication

# 音频播放函数 支持中途停止
def anthem(stop_event: threading.Event):
    note_list = [
        (391, 300),
        (329, 300),
        (261, 600),
        (329, 600),
        (391, 600),
        (523, 900)
    ]
    for freq, duration in note_list:
        if stop_event.is_set():
            return
        winsound.Beep(freq, duration)

# 烟花效果主逻辑
def run_fireworks(screen: Screen, stop_event: threading.Event):
    effects = [
        Stars(screen, 200),
    ]
    for _ in range(20):
        effects.append(Firework(screen, screen.height, start_frame=_*10))
    
    # 自定义键盘事件处理
    def handle_key(event):
        if event.key_code == ord(" "):
            stop_event.set()
            raise StopApplication("Restart triggered")
        if event.key_code == ord("q"):
            with open(".quit_flag", "w") as f:
                f.write("1")
            raise StopApplication("Quit triggered")
    
    scene = Scene(effects, -1, unhandled_input=handle_key)
    screen.play([scene], stop_on_resize=True)

# 主控制逻辑
def main():
    while True:
        stop_event = threading.Event()
        # 启动音频线程
        audio_thread = threading.Thread(target=anthem, args=(stop_event,))
        audio_thread.start()
        # 启动烟花渲染
        try:
            Screen.wrapper(run_fireworks, arguments=(stop_event,))
        except ResizeScreenError:
            pass
        # 等待两个线程完全退出
        stop_event.set()
        audio_thread.join()
        # 检测是否是主动退出
        with open(".quit_flag", "r") as f:
            if f.read().strip() == "1":
                break
        # 清理退出标记
        open(".quit_flag", "w").write("0")

if __name__ == "__main__":
    # 初始化退出标记
    open(".quit_flag", "w").write("0")
    try:
        main()
    except KeyboardInterrupt:
        pass
    finally:
        if os.path.exists(".quit_flag"):
            os.remove(".quit_flag")

注意事项 #

• 代码仅支持Windows系统，winsound为Windows专属模块
• 启动后按下空格键即可同步重启烟花和音频，按下q键退出程序
• 若屏幕尺寸发生变化，程序会自动重启效果适配新尺寸

内容的提问来源于stack exchange，提问作者internguy