Этот агент подключается к MQTT брокеру, анализирует топики и датчики, а затем выдаёт умные советы по улучшению системы и экономии энергии.
mosquitto_sub или paho-mqtt)
import paho.mqtt.client as mqtt
import json
import time
from datetime import datetime
# ---------- 1. ПОДКЛЮЧЕНИЕ К MQTT (ваши данные) ----------
BROKER = "localhost" # или IP вашей Raspberry Pi
PORT = 1883
USER = "" # если нужен логин/пароль
PASSWORD = ""
# Хранилище топиков и данных
topics_data = {}
last_seen = {}
# База "хороших практик" (можно позже заменить на запрос к интернету / LLM)
good_practices = {
"temperature": {"suggest": "🌡️ Добавьте датчик влажности рядом с температурным", "priority": 2},
"presence": {"suggest": "📸 Поставьте датчик движения/CO2, чтобы отключать свет и отопление", "priority": 1},
"leak": {"suggest": "💧 Датчик протечки — обязателен под раковиной и стиралкой", "priority": 1},
"heating": {"suggest": "🔥 Настройте нагреватель по расписанию: с 8 до 18 (никого нет) — выключать", "priority": 1},
"light": {"suggest": "💡 Автоматизация света: при движении включать, через 5 мин без движения — выключать", "priority": 2},
"window": {"suggest": "🪟 Датчик открытия окна + отключать отопление при проветривании", "priority": 2}
}
# ---------- 2. ОБРАБОТЧИК ВХОДЯЩИХ СООБЩЕНИЙ ----------
def on_message(client, userdata, msg):
topic = msg.topic
payload = msg.payload.decode()
# Сохраняем последнее значение и время
topics_data[topic] = payload
last_seen[topic] = datetime.now()
print(f"[MQTT] {topic} -> {payload}")
# Здесь можно добавить простой анализ в реальном времени
if "temperature" in topic:
try:
temp = float(payload)
if temp > 25:
print(" ⚠️ Совет: Слишком жарко, можно снизить нагрев (экономия 5-10%)")
except:
pass
def on_connect(client, userdata, flags, rc):
print("Подключено к MQTT брокеру, код:", rc)
# Подписываемся на все топики (для сканирования системы)
client.subscribe("#") # слушаем всё
# Если много топиков, лучше подписаться на "room/+/+" или перечислить
# ---------- 3. АНАЛИЗ СИСТЕМЫ И ГЕНЕРАЦИЯ СОВЕТОВ ----------
def generate_recommendations():
print("\n🔍 Анализ системы...")
found_topics = list(topics_data.keys())
# 3.1 Какие типы датчиков уже есть?
has_temp = any("temp" in t for t in found_topics)
has_presence = any(("presence" in t or "motion" in t or "pir" in t) for t in found_topics)
has_leak = any("leak" in t or "water" in t for t in found_topics)
has_heating = any(("heat" in t or "climate" in t or "thermostat" in t) for t in found_topics)
has_window = any("window" in t for t in found_topics)
system_state = {
"temperature": has_temp,
"presence": has_presence,
"leak": has_leak,
"heating": has_heating,
"window": has_window
}
recommendations = []
# Сравниваем с лучшими практиками
for key, practice in good_practices.items():
if not system_state.get(key, False):
recommendations.append({
"text": practice["suggest"],
"priority": practice["priority"],
"missing": key
})
# Специальные продвинутые советы (анализ трафика/привычек)
# Например, если нет датчика присутствия, но есть нагрев — советуем добавить
if has_heating and not has_presence:
recommendations.append({
"text": "🎯 В комнате есть нагреватель, но нет датчика присутствия! Настройте управление по движению — сэкономите до 30%",
"priority": 1,
"missing": "auto_heating"
})
# Совет по протечке, если вообще нет leak-датчиков
if not has_leak:
recommendations.append({
"text": "⚠️ Критично: Нет датчиков протечки. Установите их под ванной и стиральной машиной — предотвратите потоп.",
"priority": 1,
"missing": "leak"
})
return recommendations
# ---------- 4. ФУНКЦИЯ СВЯЗИ С ИНТЕРНЕТ (сравнение с общими практиками) ----------
def fetch_ideas_from_internet(query="умные советы для автоматизации дома"):
"""
Здесь можно сделать запрос к ChatGPT API или просто вернуть статичные идеи.
Для примера используем локальную базу знаний.
"""
internet_ideas = [
"🔌 Умная розетка + датчик тока: отключайте телевизор/комп в режиме ожидания (экономия 5-10%)",
"🌙 Датчик освещённости + жалюзи: автоматически закрывать от солнца в жару — экономия на кондиционере",
"🚪 Датчик открытия двери + сигнализация: если дверь открыта более 5 минут, уведомление в Telegram",
"📊 Анализ температуры по комнатам: выравнивайте отопление, чтобы не греть пустую спальню",
"💨 Датчик качества воздуха (CO2): при превышении включать бризер/проветривание"
]
return internet_ideas
# ---------- 5. ЗАПУСК И ЦИКЛ ОБНОВЛЕНИЙ ----------
def main():
client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message
if USER:
client.username_pw_set(USER, PASSWORD)
client.connect(BROKER, PORT, 60)
client.loop_start() # асинхронный приём сообщений
print("Агент запущен. Собираем данные о топиках в течение 10 секунд...")
time.sleep(10) # подождать, пока придут первые сообщения
# Генерируем персонализированные советы
recs = generate_recommendations()
print("\n" + "="*60)
print("🤖 AI-АГЕНТ: РЕКОМЕНДАЦИИ ДЛЯ ВАШЕЙ СИСТЕМЫ")
print("="*60)
if recs:
for r in recs:
print(f"\n✅ {r['text']}")
else:
print("✅ Отлично! Ваша система уже содержит основные датчики.")
print("\n🌍 ИДЕИ ИЗ ИНТЕРНЕТА (популярные практики):")
for idea in fetch_ideas_from_internet()[:3]:
print(f" • {idea}")
print("\n💡 Совет дня: подключите этот скрипт к Telegram боту — будет слать рекомендации раз в сутки.")
# Оставляем работу агента (можно добавить бесконечный цикл или переодический анализ)
try:
while True:
time.sleep(60)
# Раз в минуту можно проверять новые топики или пересчитывать советы
# (здесь мог бы быть ваш динамический анализ)
pass
except KeyboardInterrupt:
print("\nАгент остановлен.")
client.loop_stop()
if __name__ == "__main__":
main()
pip install paho-mqtt
sudo systemctl start mosquitto)
python3 ai_agent.py
#), соберёт первые 10 секунд данных, проанализирует и выдаст советы.mosquitto_pub -t "room/living/temperature" -m "22.5"
good_practices — подключите API GPT или локальную LLM (через Ollama). Передавайте туда список found_topics и просите: "Предложи 5 умных автоматизаций для этих MQTT устройств".ai_agent/recommendations, а ваша iot панель их отображает.[MQTT] room/kitchen/temperature -> 24.3 [MQTT] room/bedroom/presence -> 0 ... ============================================================ 🤖 AI-АГЕНТ: РЕКОМЕНДАЦИИ ДЛЯ ВАШЕЙ СИСТЕМЫ ============================================================ ✅ 💧 Датчик протечки — обязателен под раковиной и стиралкой ✅ 🎯 В комнате есть нагреватель, но нет датчика присутствия! Настройте управление по движению — сэкономите до 30% 🌍 ИДЕИ ИЗ ИНТЕРНЕТА: • 🔌 Умная розетка + датчик тока: отключайте телевизор/комп в режиме ожидания • 🌙 Датчик освещённости + жалюзи: автоматически закрывать от солнца в жару
Этот код — база, которую легко расширять. Можешь встроить его в свою IoT панель как отдельный виджет "AI Agent", добавить планировщик (раз в сутки анализ), и через неделю он уже будет давать персонализированные советы под твой дом.
📌 Важно: скрипт подписывается на # — может быть много сообщений. Для продакшена лучше подписываться на префиксы вроде room/+/+ или на известные топики твоей системы.
Комментарии
Пока нет комментариев. Будьте первым!