future/smart-speaker
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Фикс произношения дат и температуры + wheather -лишние файлы

Browse Source
This commit is contained in:
nvfuture
2026-01-10 23:50:02 +03:00
parent 3818f0ad22
commit 8f44fd2460
10 changed files with 295 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files

96
app/audio/tts.py
View File

@@ -20,6 +20,8 @@ from ..core.config import TTS_SPEAKER, TTS_EN_SPEAKER, TTS_SAMPLE_RATE
# Подавляем предупреждения Silero о длинном тексте (мы сами его режем)
warnings.filterwarnings("ignore", message="Text string is longer than 1000 symbols")
_EN_WORD_RE = re.compile(r"[A-Za-z][A-Za-z0-9'-]*")
class TextToSpeech:
"""Класс синтеза речи с поддержкой прерывания."""
@@ -109,19 +111,52 @@ class TextToSpeech:
return [c for c in chunks if c]
def speak(self, text: str, check_interrupt=None, language: str = "ru") -> bool:
def _split_mixed_language(self, text: str) -> list[tuple[str, str]]:
"""
Основная функция: генерирует аудио и воспроизводит его.
Args:
text: Текст для озвучки.
check_interrupt: Функция, возвращающая True, если надо прерваться (например, check_wakeword_once).
language: "ru" или "en".
Returns:
True, если договорил до конца.
False, если был прерван.
Разбивает текст на сегменты русского и английского текста.
Английские слова (латиница) будут озвучены английской моделью.
"""
matches = list(_EN_WORD_RE.finditer(text))
if not matches:
return [(text, "ru")]
segments = []
idx = 0
for match in matches:
if match.start() > idx:
segments.append((text[idx : match.start()], "ru"))
segments.append((match.group(0), "en"))
idx = match.end()
if idx < len(text):
segments.append((text[idx:], "ru"))
# Склеиваем соседние сегменты и прикрепляем чистую пунктуацию к предыдущему.
merged = []
for segment, lang in segments:
if not segment:
continue
if not any(ch.isalnum() for ch in segment):
if merged:
merged[-1] = (merged[-1][0] + segment, merged[-1][1])
else:
merged.append((segment, lang))
continue
if merged and merged[-1][1] == lang:
merged[-1] = (merged[-1][0] + segment, lang)
else:
merged.append((segment, lang))
if merged and not any(ch.isalnum() for ch in merged[0][0]) and len(merged) > 1:
merged[1] = (merged[0][0] + merged[1][0], merged[1][1])
merged = merged[1:]
return merged
def _speak_single_language(
self, text: str, check_interrupt=None, language: str = "ru"
) -> bool:
"""Озвучивание текста одной моделью языка."""
if not text.strip():
return True
@@ -197,6 +232,45 @@ class TextToSpeech:
else:
return False
def _speak_mixed(
self, segments: list[tuple[str, str]], check_interrupt=None
) -> bool:
"""Озвучивание текста с переключением RU/EN по сегментам."""
for segment, lang in segments:
if not segment.strip():
continue
completed = self._speak_single_language(
segment, check_interrupt=check_interrupt, language=lang
)
if not completed:
return False
return True
def speak(self, text: str, check_interrupt=None, language: str = "ru") -> bool:
"""
Основная функция: генерирует аудио и воспроизводит его.
Args:
text: Текст для озвучки.
check_interrupt: Функция, возвращающая True, если надо прерваться (например, check_wakeword_once).
language: "ru" или "en".
Returns:
True, если договорил до конца.
False, если был прерван.
"""
if not text.strip():
return True
if language == "ru":
segments = self._split_mixed_language(text)
if any(lang == "en" for _, lang in segments):
return self._speak_mixed(segments, check_interrupt=check_interrupt)
return self._speak_single_language(
text, check_interrupt=check_interrupt, language=language
)
def _check_interrupt_worker(self, check_interrupt):
"""
Фоновая функция для потока: постоянно опрашивает check_interrupt.

12
app/audio/wakeword.py
View File

@@ -22,7 +22,6 @@ class WakeWordDetector:
self.pa = None
self._stream_closed = True # Флаг состояния потока (закрыт/открыт)
self._last_hit_ts = 0.0
self._hit_streak = 0
def initialize(self):
"""Инициализация Porcupine и PyAudio."""
@@ -136,15 +135,10 @@ class WakeWordDetector:
keyword_index = self.porcupine.process(pcm)
if keyword_index >= 0:
now = time.time()
if now - self._last_hit_ts < 0.6:
self._hit_streak += 1
else:
self._hit_streak = 1
if now - self._last_hit_ts < 0.4:
return False
self._last_hit_ts = now
if self._hit_streak >= 2:
self._hit_streak = 0
print("🛑 Wake word подтвержден во время ответа!")
print("🛑 Wake word обнаружен во время ответа!")
return True
return False
except Exception:

2
app/core/cleaner.py
View File

@@ -237,7 +237,7 @@ def numbers_to_words(text: str) -> str:
preps_list = "|".join(map(re.escape, PREPOSITION_CASES.keys()))
text = re.sub(
rf"(?i)\b((?:{preps_list})\s+)?(\d+(?:[.,]\d+)?)(?=(\s+[а-яА-ЯёЁ]+))?\b",
rf"(?i)(?<!\w)((?:{preps_list})\s+)?([+-]?\d+(?:[.,]\d+)?)(?=(\s+[а-яА-ЯёЁ]+))?\b",
replace_cardinal_match,
text,
)

5
app/core/config.py
View File

@@ -52,3 +52,8 @@ TTS_SPEAKER = "eugene" # Доступные (ru): aidar, baya, kseniya, xenia,
TTS_EN_SPEAKER = os.getenv("TTS_EN_SPEAKER", "en_0")
# Частота дискретизации для воспроизведения (качество звука)
TTS_SAMPLE_RATE = 48000
# --- Настройки погоды ---
WEATHER_LAT = os.getenv("WEATHER_LAT")
WEATHER_LON = os.getenv("WEATHER_LON")
WEATHER_CITY = os.getenv("WEATHER_CITY", "Ухта")

144
app/features/weather.py Normal file
View File

@@ -0,0 +1,144 @@
"""
Weather feature module.
Fetches weather data from Open-Meteo API.
"""
import requests
from datetime import datetime
from ..core.config import WEATHER_LAT, WEATHER_LON, WEATHER_CITY
def get_wmo_description(code: int) -> str:
"""Decodes WMO weather code to Russian description."""
codes = {
0: "ясно",
1: "преимущественно ясно",
2: "переменная облачность",
3: "пасмурно",
45: "туман",
48: "изморозь",
51: "легкая морось",
53: "умеренная морось",
55: "плотная морось",
56: "ледяная морось",
57: "плотная ледяная морось",
61: "слабый дождь",
63: "умеренный дождь",
65: "сильный дождь",
66: "ледяной дождь",
67: "сильный ледяной дождь",
71: "слабый снег",
73: "снегопад",
75: "сильный снегопад",
77: "снежные зерна",
80: "слабый ливень",
81: "умеренный ливень",
82: "сильный ливень",
85: "слабый снегопад",
86: "сильный снегопад",
95: "гроза",
96: "гроза с градом",
99: "сильная гроза с градом"
}
return codes.get(code, "осадки")
def get_weather_report() -> str:
"""
Fetches detailed weather report.
Structure:
1. Current temp and precipitation.
2. Today's min/max temp.
3. Next 4 hours forecast (temp + precipitation).
"""
if not all([WEATHER_LAT, WEATHER_LON, WEATHER_CITY]):
return "Настройки погоды не найдены. Проверьте конфигурацию."
url = "https://api.open-meteo.com/v1/forecast"
params = {
"latitude": WEATHER_LAT,
"longitude": WEATHER_LON,
"current": "temperature_2m,precipitation,weather_code",
"hourly": "temperature_2m,precipitation,weather_code",
"daily": "temperature_2m_max,temperature_2m_min",
"timezone": "auto",
"forecast_days": 2
}
try:
response = requests.get(url, params=params, timeout=5)
response.raise_for_status()
data = response.json()
# --- 1. Current Weather ---
curr = data["current"]
temp_now = round(curr["temperature_2m"])
precip_now = curr["precipitation"]
code_now = curr["weather_code"]
desc_now = get_wmo_description(code_now)
report = f"Сейчас в городе {WEATHER_CITY} {temp_now} градусов, {desc_now}."
if precip_now > 0:
report += f" Выпало {precip_now} миллиметров осадков."
# --- 2. Today's Range ---
# daily arrays usually start from today [0]
daily = data["daily"]
t_max = round(daily["temperature_2m_max"][0])
t_min = round(daily["temperature_2m_min"][0])
report += f" Сегодня температура будет от {t_min} до {t_max} градусов."
# --- 3. Forecast Next 4 Hours ---
current_hour = datetime.now().hour
hourly_temps = data["hourly"]["temperature_2m"]
hourly_precip = data["hourly"]["precipitation"]
hourly_codes = data["hourly"]["weather_code"]
# Start from next hour
start_idx = current_hour + 1
end_idx = min(start_idx + 4, len(hourly_temps))
next_temps = hourly_temps[start_idx:end_idx]
next_precip = hourly_precip[start_idx:end_idx]
next_codes = hourly_codes[start_idx:end_idx]
if next_temps:
report += " Прогноз на ближайшие 4 часа: "
# Group by roughly similar weather to avoid repetition?
# Or just list them simply.
# "В 14:00 -5, ясно. В 15:00 -5, снег." -> a bit verbose.
# Simplified: "Температура около -5, возможен слабый снег."
# Let's verify if weather changes significantly.
# If consistent, summarize. If not, list.
# Simple approach for TTS:
avg_temp = round(sum(next_temps) / len(next_temps))
# Check if any precipitation is expected
will_precip = any(p > 0 for p in next_precip)
unique_codes = set(next_codes)
# Determine dominant weather description
if len(unique_codes) == 1:
weather_desc = get_wmo_description(list(unique_codes)[0])
else:
# Priority to precipitation codes
precip_codes = [c for c in unique_codes if c > 3] # >3 implies not clear/cloudy
if precip_codes:
weather_desc = get_wmo_description(max(precip_codes)) # Take the most severe
else:
weather_desc = "переменная облачность"
report += f"температура около {avg_temp} градусов, {weather_desc}."
if will_precip:
report += " Ожидаются осадки."
else:
report += " Без существенных осадков."
return report
except Exception as e:
print(f"❌ Ошибка получения погоды: {e}")
return "Не удалось получить полные данные о погоде."

58
app/main.py
View File

@@ -23,7 +23,13 @@ import os
from collections import deque
# Для воспроизведения звуков (mp3)
try:
from pygame import mixer
except Exception as exc:
mixer = None
_MIXER_IMPORT_ERROR = exc
else:
_MIXER_IMPORT_ERROR = None
# Импорт наших модулей
from .audio.wakeword import (
@@ -38,6 +44,7 @@ from .core.cleaner import clean_response
from .audio.tts import speak, initialize as init_tts
from .audio.sound_level import set_volume, parse_volume_text
from .features.alarm import get_alarm_clock
from .features.weather import get_weather_report
# Список стоп-слов, чтобы прервать диалог или остановить ассистента
STOP_WORDS = {
@@ -134,10 +141,20 @@ def main():
# Предварительная инициализация моделей
print("⏳ Инициализация моделей...")
# Инициализация звуковой системы для эффектов
mixer.init()
ding_sound_path = "assets/sounds/ding.wav"
# Инициализация звуковой системы для эффектов (опционально)
ding_sound = None
if mixer is None:
print(
"Warning: pygame mixer not available; sound effects disabled."
f" ({_MIXER_IMPORT_ERROR})"
)
else:
try:
mixer.init()
except Exception as exc:
print(f"Warning: pygame mixer init failed; sound effects disabled. ({exc})")
else:
ding_sound_path = "assets/sounds/ding.wav"
if os.path.exists(ding_sound_path):
ding_sound = mixer.Sound(ding_sound_path)
ding_sound.set_volume(0.3)
@@ -242,8 +259,10 @@ def main():
# Проверка команд будильника ("поставь будильник на 7")
alarm_response = alarm_clock.parse_command(user_text)
if alarm_response:
speak(alarm_response)
last_response = alarm_response
clean_alarm_response = clean_response(alarm_response, language="ru")
speak(clean_alarm_response)
last_response = clean_alarm_response
skip_wakeword = False
continue
# Проверка команды громкости ("громкость 5")
@@ -256,8 +275,9 @@ def main():
if level is not None:
if set_volume(level):
msg = f"Громкость установлена на {level}"
speak(msg)
last_response = msg
clean_msg = clean_response(msg, language="ru")
speak(clean_msg)
last_response = clean_msg
else:
speak("Не удалось установить громкость.")
else:
@@ -265,10 +285,34 @@ def main():
"Я не понял число громкости. Скажите число от одного до десяти."
)
skip_wakeword = True
continue
except Exception as e:
print(f"❌ Ошибка громкости: {e}")
speak("Не удалось изменить громкость.")
skip_wakeword = True
continue
# Проверка команды "Погода"
weather_triggers = [
"погода",
"погоду",
"что на улице",
"какая температура",
"сколько градусов",
"холодно ли",
"жарко ли",
"нужен ли зонт",
"брать ли зонт",
"прогноз погоды",
]
if any(trigger in user_text.lower() for trigger in weather_triggers):
weather_report = get_weather_report()
clean_report = clean_response(weather_report, language="ru")
speak(clean_report)
last_response = clean_report
skip_wakeword = True
continue
# Проверка запроса на перевод

BIN
assets/models/Alexandr_en_linux_v4_0_0/Alexandr_en_linux_v4_0_0.ppn
View File

Binary file not shown.

BIN
ding.wav
View File

Binary file not shown.

28
scripts/generate_ding.py
View File

@@ -1,28 +0,0 @@
import wave
import math
import struct
def generate_ding(filename="assets/sounds/ding.wav", frequency=800, duration=0.15):
sample_rate = 44100
n_frames = int(sample_rate * duration)
with wave.open(filename, "w") as wav_file:
wav_file.setnchannels(1)
wav_file.setsampwidth(2)
wav_file.setframerate(sample_rate)
data = []
for i in range(n_frames):
# Затухающая синусоида
t = i / sample_rate
value = int(
32767.0 * math.sin(2 * math.pi * frequency * t) * (1 - t / duration)
)
data.append(struct.pack("<h", value))
wav_file.writeframes(b"".join(data))
if __name__ == "__main__":
generate_ding()

20
tests/test_cleaner.py
View File

@@ -1,20 +0,0 @@
# Простые проверки для модуля cleaner (запуск вручную).
from app.core import cleaner
import traceback
try:
print("Testing cleaner...")
text = "В 1999 году."
res = cleaner.clean_response(text)
print(f"Result: {res}")
text = "![image](http://example.com)"
res = cleaner.clean_response(text)
print(f"Result: {res}")
text = "[link](http://example.com)"
res = cleaner.clean_response(text)
print(f"Result: {res}")
except Exception:
traceback.print_exc()