179 lines
7.3 KiB
Python
179 lines
7.3 KiB
Python
import shutil
|
|
import wave
|
|
try:
|
|
import simpleaudio as sa
|
|
except ImportError:
|
|
sa = None
|
|
sa = None # Force it to not use it cause it fucks stuff up
|
|
import os
|
|
import subprocess
|
|
import time
|
|
import sys
|
|
from PyQt5.QtCore import QThread, pyqtSignal
|
|
|
|
|
|
class AlertWorker(QThread):
|
|
"""Worker thread to play an alert sound safely without blocking UI.
|
|
Uses winsound on Windows, external system players on Unix (afplay/paplay/aplay/ffplay),
|
|
and falls back to simpleaudio if available. Supports cooperative stop.
|
|
"""
|
|
finished = pyqtSignal(bool, str) # success, message
|
|
|
|
def __init__(self, wav_path: str, parent=None):
|
|
super().__init__(parent)
|
|
self.wav_path = wav_path
|
|
self._stop = False
|
|
self._subproc = None
|
|
self._play_obj = None
|
|
|
|
def stop(self):
|
|
"""Request the worker to stop early."""
|
|
try:
|
|
self._stop = True
|
|
if self._play_obj is not None:
|
|
try:
|
|
self._play_obj.stop()
|
|
except Exception:
|
|
pass
|
|
if self._subproc is not None:
|
|
try:
|
|
self._subproc.terminate()
|
|
except Exception:
|
|
pass
|
|
except Exception:
|
|
pass
|
|
|
|
def _find_unix_player(self):
|
|
"""Return (cmd_list, name) for an available player on Unix or (None, None)."""
|
|
try:
|
|
if sys.platform.startswith('darwin'):
|
|
if shutil.which('afplay'):
|
|
return (['afplay'], 'afplay')
|
|
# Linux and others
|
|
if shutil.which('paplay'):
|
|
return (['paplay'], 'paplay')
|
|
if shutil.which('aplay'):
|
|
return (['aplay', '-q'], 'aplay')
|
|
if shutil.which('ffplay'):
|
|
return (['ffplay', '-nodisp', '-autoexit', '-loglevel', 'error'], 'ffplay')
|
|
except Exception:
|
|
pass
|
|
return (None, None)
|
|
|
|
def run(self):
|
|
try:
|
|
if not os.path.exists(self.wav_path):
|
|
self.finished.emit(False, f"File not found: {self.wav_path}")
|
|
return
|
|
|
|
# Windows path: prefer winsound (native, safe)
|
|
if sys.platform.startswith('win'):
|
|
ws_error = "unknown"
|
|
try:
|
|
import winsound as _ws # type: ignore
|
|
# Resolve flags safely even if some attributes are missing
|
|
SND_FILENAME = getattr(_ws, 'SND_FILENAME', 0x00020000)
|
|
SND_SYNC = getattr(_ws, 'SND_SYNC', 0x0000) # 0 is synchronous by default
|
|
flags = SND_FILENAME | SND_SYNC
|
|
# Ensure PlaySound exists
|
|
play_fn = getattr(_ws, 'PlaySound', None)
|
|
if play_fn is None:
|
|
raise RuntimeError('winsound.PlaySound not available')
|
|
for _ in range(4):
|
|
if self._stop:
|
|
break
|
|
try:
|
|
play_fn(self.wav_path, flags)
|
|
except Exception as e:
|
|
# On failure, break to try alternative backends
|
|
ws_error = str(e)
|
|
break
|
|
time.sleep(0.001)
|
|
else:
|
|
# Completed all 4 plays
|
|
self.finished.emit(True, "Alert played")
|
|
return
|
|
# If here, winsound failed at some point; continue to fallbacks
|
|
except Exception as e:
|
|
ws_error = str(e)
|
|
# Try simpleaudio on Windows as fallback
|
|
if sa is not None:
|
|
try:
|
|
with wave.open(self.wav_path, 'rb') as wf:
|
|
n_channels = max(1, wf.getnchannels())
|
|
sampwidth = max(1, wf.getsampwidth())
|
|
framerate = max(8000, wf.getframerate() or 44100)
|
|
frames = wf.readframes(wf.getnframes())
|
|
for _ in range(4):
|
|
if self._stop:
|
|
break
|
|
self._play_obj = sa.play_buffer(frames, n_channels, sampwidth, framerate)
|
|
self._play_obj.wait_done()
|
|
time.sleep(0.002)
|
|
self.finished.emit(True, "Alert played")
|
|
return
|
|
except Exception as e2:
|
|
self.finished.emit(False, f"Playback error (winsound fallback -> simpleaudio): {e2}")
|
|
return
|
|
else:
|
|
self.finished.emit(False, f"Audio backend not available (winsound failed: {ws_error})")
|
|
return
|
|
|
|
# Non-Windows: try external players first
|
|
cmd, name = self._find_unix_player()
|
|
if cmd is not None:
|
|
for _ in range(4):
|
|
if self._stop:
|
|
break
|
|
try:
|
|
self._subproc = subprocess.Popen(cmd + [self.wav_path], stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL)
|
|
# Poll until done or stop requested
|
|
while True:
|
|
if self._stop:
|
|
try:
|
|
self._subproc.terminate()
|
|
except Exception:
|
|
pass
|
|
break
|
|
ret = self._subproc.poll()
|
|
if ret is not None:
|
|
break
|
|
time.sleep(0.01)
|
|
except Exception as e:
|
|
# Try next backend
|
|
cmd = None
|
|
break
|
|
finally:
|
|
self._subproc = None
|
|
time.sleep(0.002)
|
|
if cmd is not None:
|
|
self.finished.emit(True, "Alert played")
|
|
return
|
|
|
|
# Fallback: simpleaudio if available
|
|
if sa is not None:
|
|
try:
|
|
with wave.open(self.wav_path, 'rb') as wf:
|
|
n_channels = max(1, wf.getnchannels())
|
|
sampwidth = max(1, wf.getsampwidth())
|
|
framerate = max(8000, wf.getframerate() or 44100)
|
|
frames = wf.readframes(wf.getnframes())
|
|
for _ in range(4):
|
|
if self._stop:
|
|
break
|
|
self._play_obj = sa.play_buffer(frames, n_channels, sampwidth, framerate)
|
|
self._play_obj.wait_done()
|
|
time.sleep(0.002)
|
|
self.finished.emit(True, "Alert played")
|
|
return
|
|
except Exception as e:
|
|
self.finished.emit(False, f"Playback error (simpleaudio): {e}")
|
|
return
|
|
|
|
self.finished.emit(False, "No audio backend available (afplay/paplay/aplay/ffplay/simpleaudio)")
|
|
except Exception as e:
|
|
try:
|
|
self.finished.emit(False, str(e))
|
|
except Exception:
|
|
pass |