скрипты для сбора данных уровня заряда батареек зигби устройств. Скрипт отправляет список устройств, у которых порог заряда ниже заданного на емаил через смтп сервер локальный.
Система запрещает ставить pip-пакеты глобально, поэтому:
python3 -m venv /home/mazzick/venv /home/mazzick/venv/bin/pip install PyMySQL
Скрипт делает следующее:
zigbee2mqtt/%/battery с значением ≤20.#!/home/mazzick/venv/bin/python3
import pymysql
import smtplib
from email.message import EmailMessage
# === НАСТРОЙКИ ===
DB_CONFIG = {
'host': 'localhost',
'user': 'iot_user',
'password': '1212121212',
'database': 'iot_db',
'charset': 'utf8mb4'
}
THRESHOLD = 11 # порог в процентах
EMAIL_TO = "9447333@gmail.com" # ← ОБЯЗАТЕЛЬНО замени на свой email!
# =================
def get_low_battery_devices():
conn = pymysql.connect(**DB_CONFIG)
try:
with conn.cursor() as cursor:
cursor.execute("""
SELECT
SUBSTRING_INDEX(topic, '/', -2) AS device,
topic,
CAST(value AS DECIMAL(5,2)) AS level
FROM sensor_data
WHERE topic LIKE %s
AND CAST(value AS DECIMAL(5,2)) <= %s
""", ('zigbee2mqtt/%/battery', THRESHOLD))
return cursor.fetchall()
finally:
conn.close()
def send_alert(devices):
if not devices:
print("🔋 Все батарейки в норме")
return
body = "Низкий заряд батареи у следующих Zigbee-устройств:\n\n"
for dev, topic, level in devices:
body += f"• {dev} → {level}%\n"
msg = EmailMessage()
msg.set_content(body)
msg["Subject"] = f"⚠️ Заряд батареи ниже {THRESHOLD}% ({len(devices)} шт.)"
msg["From"] = "zigbee-alert@local"
msg["To"] = EMAIL_TO
# Отправка через локальный sendmail/postfix
with smtplib.SMTP("localhost") as s:
s.send_message(msg)
print(f"📧 Отправлено оповещение о {len(devices)} устройстве(ах)")
if __name__ == "__main__":
send_alert(get_low_battery_devices())
Это тот же скрипт, только через смпт яндекса или гугла (этот скрипт не проверял, но тут только смтп яндекса добавил)
#!/home/mazzick/venv/bin/python3
import pymysql
import smtplib
from email.message import EmailMessage
# === НАСТРОЙКИ ===
DB_CONFIG = {
'host': 'localhost',
'user': 'iot_user',
'password': '12321312321',
'database': 'iot_db',
'charset': 'utf8mb4'
}
THRESHOLD = 20 # порог в процентах
# Настройки почты (YANDEX)
EMAIL_USER = "твой@yandex.ru" # ← ЗАМЕНИ на свою почту
EMAIL_APP_PASSWORD = "ваш_пароль_приложения" # ← ЗАМЕНИ на пароль приложения
EMAIL_TO = "твой@yandex.ru" # ← Куда присылать оповещения
# =================
def get_low_battery_devices():
conn = pymysql.connect(**DB_CONFIG)
try:
with conn.cursor() as cursor:
cursor.execute("""
SELECT
SUBSTRING_INDEX(topic, '/', -2) AS device,
topic,
CAST(value AS DECIMAL(5,2)) AS level
FROM sensor_data
WHERE topic LIKE %s
AND CAST(value AS DECIMAL(5,2)) <= %s
""", ('zigbee2mqtt/%/battery', THRESHOLD))
return cursor.fetchall()
finally:
conn.close()
def send_alert(devices):
if not devices:
print("🔋 Все батарейки в норме")
return
body = "Низкий заряд батареи у следующих Zigbee-устройств:\n\n"
for dev, topic, level in devices:
body += f"• {dev} → {level}%\n"
msg = EmailMessage()
msg.set_content(body)
msg["Subject"] = f"⚠️ Заряд батареи ниже {THRESHOLD}% ({len(devices)} шт.)"
msg["From"] = EMAIL_USER
msg["To"] = EMAIL_TO
# Отправка через Yandex
try:
with smtplib.SMTP_SSL("smtp.yandex.ru", 465) as s:
s.login(EMAIL_USER, EMAIL_APP_PASSWORD)
s.send_message(msg)
print(f"📧 Отправлено оповещение о {len(devices)} устройстве(ах)")
except Exception as e:
print(f"❌ Ошибка отправки email: {e}")
if __name__ == "__main__":
send_alert(get_low_battery_devices())
Ну а этот скрипт просто в файл сохраняет список устройств зигби с севшей батарейкой без емаилов. Потом можно переделать под свои задачи.
#!/home/mazzick/venv/bin/python3
import pymysql
# === НАСТРОЙКИ ===
DB_CONFIG = {
'host': 'localhost',
'user': 'iot_user',
'password': '123456',
'database': 'iot_db',
'charset': 'utf8mb4'
}
THRESHOLD = 50
ALERT_FILE = "/home/mazzick/low_battery_alert.txt"
# =================
def get_low_battery_devices():
conn = pymysql.connect(**DB_CONFIG)
try:
with conn.cursor() as cursor:
cursor.execute("""
SELECT
SUBSTRING_INDEX(topic, '/', -2) AS device,
CAST(value AS DECIMAL(5,2)) AS level
FROM sensor_data
WHERE topic LIKE %s
AND CAST(value AS DECIMAL(5,2)) <= %s
""", ('zigbee2mqtt/%/battery', THRESHOLD))
return cursor.fetchall()
finally:
conn.close()
def main():
devices = get_low_battery_devices()
if not devices:
# Опционально: записать, что всё ок
with open(ALERT_FILE, "w") as f:
f.write("OK: Все батарейки в норме.\n")
print("🔋 Все батарейки в норме")
else:
with open(ALERT_FILE, "w") as f:
f.write(f"Низкий заряд батареи (порог ≤{THRESHOLD}%):\n\n")
for dev, level in devices:
f.write(f"• {dev} → {level}%\n")
print(f"⚠️ Найдено {len(devices)} устройств(а) с низким зарядом. См. {ALERT_FILE}")
if __name__ == "__main__":
main()
Комментарии
Пока нет комментариев. Будьте первым!