Timer
This commit is contained in:
175
app/features/timer.py
Normal file
175
app/features/timer.py
Normal file
@@ -0,0 +1,175 @@
|
||||
"""Timer module."""
|
||||
|
||||
# Модуль таймера.
|
||||
# Отвечает за установку таймеров (в оперативной памяти), их проверку и воспроизведение звука.
|
||||
|
||||
import subprocess
|
||||
import re
|
||||
from datetime import datetime, timedelta
|
||||
from pathlib import Path
|
||||
from ..core.config import BASE_DIR
|
||||
from ..audio.stt import listen
|
||||
|
||||
# Звуковой файл сигнала (используем тот же, что и для будильника)
|
||||
ALARM_SOUND = BASE_DIR / "assets" / "sounds" / "Apex-1.mp3"
|
||||
|
||||
|
||||
class TimerManager:
|
||||
def __init__(self):
|
||||
# Список активных таймеров: {"end_time": datetime, "label": str}
|
||||
self.timers = []
|
||||
|
||||
def add_timer(self, seconds: int, label: str):
|
||||
"""Добавление нового таймера."""
|
||||
end_time = datetime.now() + timedelta(seconds=seconds)
|
||||
self.timers.append({"end_time": end_time, "label": label})
|
||||
# Сортируем, чтобы ближайший был первым
|
||||
self.timers.sort(key=lambda x: x["end_time"])
|
||||
print(f"⏳ Таймер установлен на {label} (до {end_time.strftime('%H:%M:%S')})")
|
||||
|
||||
def cancel_all_timers(self):
|
||||
"""Отмена всех таймеров."""
|
||||
count = len(self.timers)
|
||||
self.timers = []
|
||||
print(f"🔕 Все таймеры ({count}) отменены.")
|
||||
|
||||
def check_timers(self):
|
||||
"""
|
||||
Проверка: не истек ли какой-то таймер?
|
||||
Вызывается в главном цикле.
|
||||
Возвращает True, если таймер сработал (и был обработан).
|
||||
"""
|
||||
if not self.timers:
|
||||
return False
|
||||
|
||||
now = datetime.now()
|
||||
# Смотрим первый (самый ранний) таймер
|
||||
# Используем индекс 0, так как список отсортирован
|
||||
first_timer = self.timers[0]
|
||||
|
||||
if now >= first_timer["end_time"]:
|
||||
# Таймер сработал!
|
||||
# Удаляем его из списка
|
||||
label = first_timer["label"]
|
||||
self.timers.pop(0)
|
||||
|
||||
print(f"⌛ ТАЙМЕР ИСТЕК: {label}")
|
||||
self.trigger_timer(label)
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
def trigger_timer(self, label: str):
|
||||
"""
|
||||
Логика срабатывания таймера.
|
||||
Запускает воспроизведение MP3 и слушает команду "Стоп".
|
||||
"""
|
||||
print(f"🔔 ТАЙМЕР НА {label} СРАБОТАЛ! (Скажите 'Стоп')")
|
||||
|
||||
# Запуск плеера mpg123 в бесконечном цикле
|
||||
cmd = ["mpg123", "-q", "--loop", "-1", str(ALARM_SOUND)]
|
||||
|
||||
try:
|
||||
process = subprocess.Popen(cmd)
|
||||
except FileNotFoundError:
|
||||
print(
|
||||
"❌ Ошибка: mpg123 не найден. Установите его: sudo apt install mpg123"
|
||||
)
|
||||
return
|
||||
|
||||
try:
|
||||
stop_words = [
|
||||
"стоп",
|
||||
"хватит",
|
||||
"тихо",
|
||||
"замолчи",
|
||||
"отмена",
|
||||
"александр стоп",
|
||||
"спасибо",
|
||||
]
|
||||
|
||||
# Цикл ожидания стоп-команды
|
||||
while True:
|
||||
text = listen(timeout_seconds=3.0, detection_timeout=3.0)
|
||||
if text:
|
||||
text_lower = text.lower()
|
||||
if any(word in text_lower for word in stop_words):
|
||||
print(f"🛑 Таймер остановлен по команде: '{text}'")
|
||||
break
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Ошибка во время таймера: {e}")
|
||||
finally:
|
||||
# Обязательно убиваем процесс плеера
|
||||
process.terminate()
|
||||
try:
|
||||
process.wait(timeout=1)
|
||||
except subprocess.TimeoutExpired:
|
||||
process.kill()
|
||||
print("🔕 Таймер выключен.")
|
||||
|
||||
def parse_command(self, text: str) -> str | None:
|
||||
"""
|
||||
Парсинг команды установки таймера.
|
||||
Примеры: "таймер на 5 минут", "засеки 10 секунд".
|
||||
"""
|
||||
text = text.lower()
|
||||
|
||||
# Ключевые слова для таймера
|
||||
if not any(word in text for word in ["таймер", "засеки", "поставь таймер"]):
|
||||
return None
|
||||
|
||||
if "отмени" in text or "удали" in text:
|
||||
self.cancel_all_timers()
|
||||
return "Хорошо, все таймеры отменены."
|
||||
|
||||
# Поиск времени
|
||||
# Ищем комбинации: число + (час/мин/сек)
|
||||
# Пример: "1 час 30 минут", "5 минут", "30 секунд"
|
||||
|
||||
total_seconds = 0
|
||||
found_time = False
|
||||
parts = []
|
||||
|
||||
# Часы
|
||||
match_hours = re.search(r"(\d+)\s*(?:час|часа|часов)", text)
|
||||
if match_hours:
|
||||
h = int(match_hours.group(1))
|
||||
total_seconds += h * 3600
|
||||
parts.append(f"{h} ч")
|
||||
found_time = True
|
||||
|
||||
# Минуты
|
||||
match_minutes = re.search(r"(\d+)\s*(?:мин|минуту|минуты|минут)", text)
|
||||
if match_minutes:
|
||||
m = int(match_minutes.group(1))
|
||||
total_seconds += m * 60
|
||||
parts.append(f"{m} мин")
|
||||
found_time = True
|
||||
|
||||
# Секунды
|
||||
match_seconds = re.search(r"(\d+)\s*(?:сек|секунду|секунды|секунд)", text)
|
||||
if match_seconds:
|
||||
s = int(match_seconds.group(1))
|
||||
total_seconds += s
|
||||
parts.append(f"{s} сек")
|
||||
found_time = True
|
||||
|
||||
if found_time and total_seconds > 0:
|
||||
label = " ".join(parts)
|
||||
self.add_timer(total_seconds, label)
|
||||
return f"Засек {label}."
|
||||
|
||||
# Если сказали "таймер", но не нашли время
|
||||
return "Я не понял, на сколько поставить таймер. Скажите, например, 'таймер на 5 минут'."
|
||||
|
||||
|
||||
# Глобальный экземпляр
|
||||
_timer_manager = None
|
||||
|
||||
|
||||
def get_timer_manager():
|
||||
global _timer_manager
|
||||
if _timer_manager is None:
|
||||
_timer_manager = TimerManager()
|
||||
return _timer_manager
|
||||
Reference in New Issue
Block a user