Zigbee (ZHA, Zigbee2MQTT) и AI-агент: как подключить умный дом к ASI Biont без программирования

Проблема: Z-стеки, координаторы и тысяча датчиков — кто всем этим управляет?

Умный дом на Zigbee — это мощно, но сложно. У вас может быть 20 датчиков температуры, 10 реле, 5 выключателей и один координатор (Conbee II, Sonoff Zigbee 3.0 USB). Вы используете Home Assistant с Zigbee2MQTT или ZHA. Данные льются рекой: температура, влажность, освещённость, движение. Но что с этим делать? Настроить автоматизацию «если температура > 25°C, включить кондиционер» — легко. А вот проанализировать недельный тренд, предсказать, когда батарейка в датчике сядет, или отправить Telegram-уведомление при аномалии — уже задача для AI.

Обычный подход: вы пишете Python-скрипт, подключаетесь к MQTT-брокеру, вручную парсите JSON-сообщения, логируете данные, реализуете логику. Это требует времени, отладки и понимания протоколов. ASI Biont решает эту задачу иначе: вы просто описываете в чате, что нужно сделать, и AI-агент сам пишет код, подключается к вашему MQTT-брокеру (или ZHA через SSH на Raspberry Pi) и начинает работать.

Решение: MQTT + execute_python — AI сам пишет код под ваш Zigbee-координатор

ASI Biont использует execute_python — универсальный механизм, где AI генерирует Python-скрипт и выполняет его в облачном sandbox-окружении. Для работы с Zigbee2MQTT или ZHA достаточно одного: MQTT-брокера (обычно Mosquitto на Raspberry Pi, где запущен Home Assistant). AI подключается к нему через библиотеку paho-mqtt (доступна в sandbox).

Пользователь пишет в чат:

«Подключись к моему MQTT-брокеру по адресу 192.168.1.100:1883, подпишись на топик zigbee2mqtt/+/+/+ (все датчики), собери данные за последний час, найди аномалии температуры (отклонение > 3°C от среднего) и отправь отчёт в Telegram».

AI генерирует код, выполняет его, анализирует данные и возвращает результат. Никаких панелей управления, кнопок «добавить устройство» или ручного конфигурирования — всё через диалог.

Сценарий: мониторинг температуры в теплице с Zigbee-датчиками

Задача

У вас теплица с 10 датчиками Aqara Temperature/Humidity Sensor (WSDCGQ11LM), подключенными через Zigbee2MQTT к Raspberry Pi. Нужно:
1. Собирать данные с каждого датчика (температура, влажность).
2. Вычислять среднюю температуру за день.
3. Если температура в каком-то секторе превышает 35°C — отправлять уведомление в Telegram.
4. Раз в сутки генерировать отчёт с трендами.

Как это выглядит в ASI Biont

Пользователь описывает задачу в чате:

«У меня Zigbee2MQTT на 192.168.1.100:1883, датчики публикуют в топик zigbee2mqtt/sensor_01, sensor_02 и т.д. Напиши скрипт, который подпишется на все топики zigbee2mqtt/+, будет парсить JSON с полями temperature и humidity, записывать в CSV-файл, и если temperature > 35 — отправлять сообщение в Telegram через мой бот (токен: 123456:ABC-DEF). Выполни скрипт разово на 30 секунд, собери статистику».

AI генерирует код и выполняет его. Вот пример того, что получится:

import paho.mqtt.client as mqtt
import json
import csv
import time
from datetime import datetime
import requests

# Конфигурация
BROKER = "192.168.1.100"
PORT = 1883
TOPIC = "zigbee2mqtt/+"
TELEGRAM_TOKEN = "123456:ABC-DEF"
CHAT_ID = "123456789"

# Хранилище данных
data = []
anomalies = []

def on_message(client, userdata, msg):
    try:
        payload = json.loads(msg.payload.decode())
        sensor_id = msg.topic.split("/")[-1]
        temp = payload.get("temperature")
        humidity = payload.get("humidity")
        if temp is not None:
            record = {
                "sensor": sensor_id,
                "temperature": temp,
                "humidity": humidity,
                "timestamp": datetime.now().isoformat()
            }
            data.append(record)
            if temp > 35:
                anomalie = f"❗ Аномалия: датчик {sensor_id}, температура {temp}°C"
                anomalie.append(anomalie)
                requests.post(
                    f"https://api.telegram.org/bot{TELEGRAM_TOKEN}/sendMessage",
                    json={"chat_id": CHAT_ID, "text": anomalie}
                )
    except Exception as e:
        print(f"Ошибка парсинга: {e}")

client = mqtt.Client()
client.on_message = on_message
client.connect(BROKER, PORT, 60)
client.subscribe(TOPIC)
client.loop_start()

# Сбор данных 30 секунд
time.sleep(30)
client.loop_stop()

# Сохранение в CSV
with open("sensor_data.csv", "w", newline="") as f:
    writer = csv.DictWriter(f, fieldnames=["sensor", "temperature", "humidity", "timestamp"])
    writer.writeheader()
    writer.writerows(data)

# Анализ
if data:
    temps = [r["temperature"] for r in data if r["temperature"] is not None]
    avg_temp = sum(temps) / len(temps)
    print(f"Средняя температура за период: {avg_temp:.2f}°C")
    print(f"Всего аномалий: {len(anomalies)}")
else:
    print("Данные не получены")

Результат

AI за 30 секунд:
- Подключился к MQTT-брокеру.
- Собрал данные с 10 датчиков.
- Записал их в CSV-файл.
- Обнаружил 2 аномалии (температура > 35°C).
- Отправил Telegram-уведомления.
- Вывел среднюю температуру.

Пользователю не пришлось писать код — только описать задачу на естественном языке.

Альтернативные сценарии интеграции Zigbee с ASI Biont

1. ZHA через SSH на Raspberry Pi

Если вы используете ZHA (Zigbee Home Automation) в Home Assistant, AI может подключиться к Raspberry Pi по SSH (через paramiko) и напрямую читать данные из базы SQLite Home Assistant или выполнять запросы к REST API Home Assistant (через aiohttp). Пример запроса в чате:

«Подключись по SSH к 192.168.1.100 (user: pi, pass: raspberry), выполни curl к localhost:8123/api/states, найди все сенсоры с температурой, вычисли среднюю за последний час».

AI сгенерирует скрипт с paramiko, выполнит команды на Raspberry Pi и вернёт результат.

2. Управление Zigbee-устройствами через MQTT publish

AI может не только читать, но и управлять. Например, включить реле или выключатель по команде:

«Опубликуй в топик zigbee2mqtt/switch_bedroom/set JSON {"state": "ON"}».

AI использует paho-mqtt для publish:

import paho.mqtt.publish as publish
publish.single("zigbee2mqtt/switch_bedroom/set", '{"state": "ON"}', hostname="192.168.1.100")

3. Долгосрочный мониторинг с сохранением в базу данных

AI может настроить скрипт, который запускается по расписанию (через cron на сервере или вручную) и сохраняет данные в PostgreSQL:

«Каждые 10 минут собирай данные с zigbee2mqtt/+/temperature, записывай в таблицу sensor_log в PostgreSQL на 192.168.1.200 (user: admin, db: home)».

AI сгенерирует скрипт с psycopg2 и paho-mqtt, который выполняется в sandbox и пишет в БД.

Почему это выгодно?

Вывод

Zigbee-устройства — это основа умного дома, но без интеллектуальной обработки данных они остаются просто «умными» датчиками. ASI Biont превращает ваш Zigbee-координатор в полноценного AI-ассистента, который анализирует тренды, предсказывает проблемы и отправляет уведомления. Всё, что нужно — описать задачу в чате. AI сам подключается к MQTT-брокеру, пишет код и выполняет его.

Попробуйте интеграцию Zigbee с ASI Biont на asibiont.com — подключите свои датчики и увидьте, как AI управляет умным домом за секунды, а не за часы.