mirror of
https://github.com/pdxlocations/contact.git
synced 2026-03-28 17:12:35 +01:00
Compare commits
3 Commits
1.4.6
...
close-inte
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
6b18809215 | ||
|
|
b048fe2480 | ||
|
|
600fc61ed7 |
@@ -129,8 +129,10 @@ def start() -> None:
|
||||
|
||||
try:
|
||||
curses.wrapper(main)
|
||||
interface_state.interface.close()
|
||||
except KeyboardInterrupt:
|
||||
logging.info("User exited with Ctrl+C")
|
||||
interface_state.interface.close()
|
||||
sys.exit(0)
|
||||
except Exception as e:
|
||||
logging.critical("Fatal error", exc_info=True)
|
||||
|
||||
@@ -2,7 +2,46 @@ import logging
|
||||
import os
|
||||
import platform
|
||||
import shutil
|
||||
import time
|
||||
import subprocess
|
||||
import threading
|
||||
# Debounce notification sounds so a burst of queued messages only plays once.
|
||||
_SOUND_DEBOUNCE_SECONDS = 0.8
|
||||
_sound_timer: threading.Timer | None = None
|
||||
_sound_timer_lock = threading.Lock()
|
||||
_last_sound_request = 0.0
|
||||
|
||||
|
||||
def schedule_notification_sound(delay: float = _SOUND_DEBOUNCE_SECONDS) -> None:
|
||||
"""Schedule a notification sound after a short quiet period.
|
||||
|
||||
If more messages arrive before the delay elapses, the timer is reset.
|
||||
This prevents playing a sound for each message when a backlog flushes.
|
||||
"""
|
||||
global _sound_timer, _last_sound_request
|
||||
|
||||
now = time.monotonic()
|
||||
with _sound_timer_lock:
|
||||
_last_sound_request = now
|
||||
|
||||
# Cancel any previously scheduled sound.
|
||||
if _sound_timer is not None:
|
||||
try:
|
||||
_sound_timer.cancel()
|
||||
except Exception:
|
||||
pass
|
||||
_sound_timer = None
|
||||
|
||||
def _fire(expected_request_time: float) -> None:
|
||||
# Only play if nothing newer has been scheduled.
|
||||
with _sound_timer_lock:
|
||||
if expected_request_time != _last_sound_request:
|
||||
return
|
||||
play_sound()
|
||||
|
||||
_sound_timer = threading.Timer(delay, _fire, args=(now,))
|
||||
_sound_timer.daemon = True
|
||||
_sound_timer.start()
|
||||
from typing import Any, Dict
|
||||
|
||||
from contact.utilities.utils import (
|
||||
@@ -108,7 +147,7 @@ def on_receive(packet: Dict[str, Any], interface: Any) -> None:
|
||||
|
||||
|
||||
if config.notification_sound == "True":
|
||||
play_sound()
|
||||
schedule_notification_sound()
|
||||
|
||||
message_bytes = packet["decoded"]["payload"]
|
||||
message_string = message_bytes.decode("utf-8")
|
||||
|
||||
Reference in New Issue
Block a user