Мониторинг домашних устройств на python

Не так давно решил написать небольшой домашний пэт-проект для мониторинга своих устройств в сети. Устройств много, и они все разные, от ПК, планшета, умных часов, до устройств на базе rasberry pi и esp8266. По началу это был обычный скрипт, который просто пинговал, но затем я решил добавить сбор статистики, сохранение её в базу и добавил возможность строить графики и смотреть историю

Сначала настроим простенькое окружение и нам понадобятся всего 3 внешние библиотеки, а библиотека sqlite3 встроена в сам python

SHELL
pip install ping3 matplotlib schedule

Теперь сам скрипт, который мониторит, раз в сутки сохраняет график доступности устройства в PNG-файл в папку graphs/, и PNG открываются в браузере в простом HTML-дашборде (graphs/index.html)

SHELL

import sqlite3
import time
import datetime
import schedule
import matplotlib.pyplot as plt
import os
from ping3 import ping

# -------------------
# Настройки
# -------------------
DB_NAME = "network_log.db"
GRAPH_DIR = "graphs"
HTML_FILE = os.path.join(GRAPH_DIR, "index.html")

DEVICES = {
    "router": "192.168.0.1",
    "laptop": "192.168.0.101",
    # добавь свои устройства сюда
}

# -------------------
# Инициализация базы
# -------------------
def init_db():
    conn = sqlite3.connect(DB_NAME)
    cursor = conn.cursor()
    cursor.execute("""
    CREATE TABLE IF NOT EXISTS logs (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        device TEXT,
        ip TEXT,
        status TEXT,
        latency REAL,
        timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
    )
    """)
    conn.commit()
    conn.close()

    # создаём папку для графиков
    os.makedirs(GRAPH_DIR, exist_ok=True)

# -------------------
# Проверка устройств
# -------------------
def check_devices():
    print(" Проверяю устройства...")
    conn = sqlite3.connect(DB_NAME)
    cursor = conn.cursor()
    for name, ip in DEVICES.items():
        result = ping(ip, timeout=2)
        cursor.execute(
            "INSERT INTO logs (device, ip, status, latency) VALUES (?, ?, ?, ?)",
            (name, ip, "up" if result else "down", result if result else None),
        )
        print(f" - {name} ({ip}): {'UP' if result else 'DOWN'} {f'{result:.3f}s' if result else ''}")
    conn.commit()
    conn.close()

# -------------------
# Построение и сохранение графика
# -------------------
def save_graph(device_name, hours=24):
    conn = sqlite3.connect(DB_NAME)
    cursor = conn.cursor()
    since = datetime.datetime.now() - datetime.timedelta(hours=hours)
    cursor.execute("SELECT timestamp, latency FROM logs WHERE device=? AND timestamp >= ?",
                   (device_name, since))
    data = cursor.fetchall()
    conn.close()

    if not data:
        print(f"Нет данных для {device_name}, график не построен.")
        return None

    times = [row[0] for row in data]
    latency = [row[1] if row[1] else 0 for row in data]

    plt.figure(figsize=(10, 5))
    plt.plot(times, latency, marker="o")
    plt.xticks(rotation=45)
    plt.ylabel("Latency (s)")
    plt.title(f"{device_name} availability (last {hours}h)")
    plt.tight_layout()

    filename = os.path.join(GRAPH_DIR, f"{device_name}_{datetime.date.today()}.png")
    plt.savefig(filename)
    plt.close()

    print(f"График для {device_name} сохранён: {filename}")
    return os.path.basename(filename)

# -------------------
# Генерация HTML-дэшборда
# -------------------
def generate_html(image_files):
    with open(HTML_FILE, "w", encoding="utf-8") as f:
        f.write("<html><head><meta charset='utf-8'>")
        f.write("<title>Network Monitor</title>")
        f.write("<style>body{font-family:sans-serif;background:#f8f9fa;padding:20px;} img{max-width:100%;margin-bottom:30px;box-shadow:0 0 10px rgba(0,0,0,0.3);border-radius:8px;}</style>")
        f.write("</head><body>")
        f.write("<h1>📡 Network Availability Dashboard</h1>")
        f.write(f"<p>Сгенерировано: {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</p>")
        for img in image_files:
            f.write(f"<h2>{img.split('_')[0]}</h2>")
            f.write(f"<img src='{img}' alt='{img}'>")
        f.write("</body></html>")
    print(f"Дашборд обновлён: {HTML_FILE}")

# -------------------
# Сохранение всех графиков + HTML
# -------------------
def save_all_graphs():
    print(" Генерирую графики для всех устройств...")
    images = []
    for device in DEVICES:
        img = save_graph(device, hours=24)
        if img:
            images.append(img)
    if images:
        generate_html(images)

# -------------------
# Основной цикл
# -------------------
if __name__ == "__main__":
    init_db()
    check_devices()  # первая проверка сразу

    # проверка каждые 5 минут
    schedule.every(5).minutes.do(check_devices)

    # генерация графиков раз в день (в полночь)
    schedule.every().day.at("00:05").do(save_all_graphs)

    print("Логгер сети запущен.")
    print(" - Проверка устройств каждые 5 минут")
    print(" - Автогенерация графиков + HTML раз в сутки (00:05)")
    print("Открой graphs/index.html в браузере для просмотра дашборда.\n")
    print("Нажми Ctrl+C для выхода.\n")

    try:
        while True:
            schedule.run_pending()
            time.sleep(1)
    except KeyboardInterrupt:
        print("\n Завершено пользователем.")

Комментарии (0)

Оставить комментарий

Пока нет комментариев. Будьте первым!