%%{init: {'theme': 'base', 'themeVariables': {'primaryColor': '#2C3E50', 'primaryTextColor': '#fff', 'primaryBorderColor': '#16A085', 'lineColor': '#16A085', 'secondaryColor': '#E67E22', 'tertiaryColor': '#7F8C8D'}}}%%
flowchart LR
subgraph Zigbee[Zigbee Network]
ZC[Zigbee Coordinator] --> ZR1[Router 1]
ZC --> ZR2[Router 2]
ZR1 --> ZE1[End Device 1]
ZR1 --> ZE2[End Device 2]
ZR2 --> ZE3[End Device 3]
end
subgraph Bridge[Zigbee2MQTT]
Z2M[zigbee2mqtt Service]
end
subgraph MQTT[MQTT Broker]
MQ[Mosquitto]
end
subgraph Analyzer[Python Analyzer]
PY[Network Analyzer Script]
DB[(Device State DB)]
ALERT[Alert System]
end
ZC --> Z2M
Z2M --> MQ
MQ --> PY
PY --> DB
PY --> ALERT
style Zigbee fill:#16A085,stroke:#2C3E50
style Bridge fill:#E67E22,stroke:#2C3E50
style MQTT fill:#2C3E50,stroke:#16A085
style Analyzer fill:#7F8C8D,stroke:#2C3E50
998 Zigbee Lab: Python Network Analyzer
998.1 Learning Objectives
By the end of this chapter, you will be able to:
- Connect to Zigbee2MQTT: Interface Python with Zigbee networks via MQTT broker
- Monitor Device Health: Track battery levels, link quality, and online status
- Visualize Network Topology: Map mesh connections between coordinator, routers, and end devices
- Implement Health Alerts: Create automated warnings for low battery and offline devices
- Aggregate Network Statistics: Count messages, calculate uptime, and monitor traffic
TipFor Beginners: Network Analysis Tools
What is this lab? A Python-based monitoring tool that connects to Zigbee2MQTT to provide real-time visibility into your Zigbee network health.
When to use: - For production network monitoring and troubleshooting - When you need battery and signal strength alerts - To visualize mesh topology and routing paths
Key Topics:
| Topic | Focus |
|---|---|
| Zigbee2MQTT | Bridge between Zigbee and MQTT |
| MQTT Protocol | Pub/sub messaging for device data |
| Health Monitoring | Battery, LQI, online status |
| Alerting | Automated notifications |
Prerequisites: - Zigbee Fundamentals - Basic Python programming - MQTT broker running
998.2 Prerequisites
This lab requires:
- Raspberry Pi with Zigbee2MQTT installed
- MQTT broker running (Mosquitto recommended)
- Zigbee coordinator (CC2531 or similar)
- Paired Zigbee devices for monitoring
- Python 3.8+ with paho-mqtt library
NoteZigbee2MQTT Overview
Zigbee2MQTT is an open-source project that bridges Zigbee devices to MQTT, allowing you to monitor and control devices without proprietary hubs. It publishes device data to MQTT topics like zigbee2mqtt/device_name.
998.3 Architecture Overview
998.4 Python Network Analyzer Implementation
998.4.1 Installation
# Install required packages
pip install paho-mqtt
# Verify Zigbee2MQTT is running
sudo systemctl status zigbee2mqtt998.4.2 Complete Analyzer Code
#!/usr/bin/env python3
"""
Zigbee Network Analyzer
-----------------------
Monitors Zigbee network health via Zigbee2MQTT.
Tracks device status, battery levels, and link quality.
"""
import json
import time
from datetime import datetime
from dataclasses import dataclass, field
from typing import Dict, Optional
import paho.mqtt.client as mqtt
@dataclass
class ZigbeeDevice:
"""Represents a Zigbee device in the network."""
friendly_name: str
device_type: str = "Unknown"
battery: Optional[int] = None
link_quality: Optional[int] = None
last_seen: Optional[datetime] = None
message_count: int = 0
is_online: bool = True
def health_status(self) -> str:
"""Determine device health based on battery and connectivity."""
if not self.is_online:
return "offline"
if self.battery is not None:
if self.battery < 10:
return "critical"
if self.battery < 20:
return "warning"
return "healthy"
class ZigbeeNetworkAnalyzer:
"""Analyzes and monitors a Zigbee network via MQTT."""
def __init__(self, mqtt_host: str = "localhost", mqtt_port: int = 1883):
self.mqtt_host = mqtt_host
self.mqtt_port = mqtt_port
self.devices: Dict[str, ZigbeeDevice] = {}
self.start_time = datetime.now()
self.total_messages = 0
self.network_map = {}
# MQTT client setup
self.client = mqtt.Client()
self.client.on_connect = self._on_connect
self.client.on_message = self._on_message
def _on_connect(self, client, userdata, flags, rc):
"""Handle MQTT connection."""
if rc == 0:
print("[OK] Connected to MQTT broker")
# Subscribe to all Zigbee2MQTT topics
client.subscribe("zigbee2mqtt/#")
print("[OK] Subscribed to zigbee2mqtt/#")
else:
print(f"[ERROR] Connection failed with code {rc}")
def _on_message(self, client, userdata, msg):
"""Process incoming MQTT messages."""
topic = msg.topic
try:
payload = json.loads(msg.payload.decode())
except json.JSONDecodeError:
return
# Handle network map updates
if topic == "zigbee2mqtt/bridge/networkmap/raw":
self._process_network_map(payload)
return
# Handle device state updates
if topic.startswith("zigbee2mqtt/") and "/bridge" not in topic:
device_name = topic.replace("zigbee2mqtt/", "")
self._update_device(device_name, payload)
def _process_network_map(self, payload):
"""Process network topology map."""
self.network_map = payload
print("\n[OK] Network map received")
if "nodes" in payload:
for node in payload["nodes"]:
name = node.get("friendly_name", node.get("ieeeAddr"))
if name not in self.devices:
self.devices[name] = ZigbeeDevice(
friendly_name=name,
device_type=node.get("type", "Unknown")
)
else:
self.devices[name].device_type = node.get("type", "Unknown")
def _update_device(self, name: str, payload: dict):
"""Update device state from payload."""
if name not in self.devices:
self.devices[name] = ZigbeeDevice(friendly_name=name)
device = self.devices[name]
device.last_seen = datetime.now()
device.message_count += 1
device.is_online = True
self.total_messages += 1
# Extract common fields
if "battery" in payload:
device.battery = payload["battery"]
if "linkquality" in payload:
device.link_quality = payload["linkquality"]
if "device_type" in payload:
device.device_type = payload["device_type"]
def get_health_alerts(self) -> list:
"""Generate health alerts for problematic devices."""
alerts = []
for name, device in self.devices.items():
status = device.health_status()
if status == "critical":
alerts.append(f"[CRITICAL] {name} - Battery {device.battery}%")
elif status == "warning":
alerts.append(f"[WARNING] {name} - Battery {device.battery}%")
elif status == "offline":
alerts.append(f"[OFFLINE] {name}")
return alerts
def print_status(self):
"""Print current network status."""
uptime = (datetime.now() - self.start_time).seconds
print("\n" + "=" * 80)
print("ZIGBEE NETWORK STATUS")
print("=" * 80)
print(f"\nDevices: {len(self.devices)} total, "
f"{sum(1 for d in self.devices.values() if d.is_online)} online")
print(f"Uptime: {uptime} seconds")
print(f"Messages: {self.total_messages}")
print("\n" + "-" * 80)
print(f"{'Device':<25} {'Type':<12} {'Battery':<10} "
f"{'Link':<8} {'Status':<12} {'Messages':<10}")
print("-" * 80)
for name, device in sorted(self.devices.items()):
battery = f"{device.battery}%" if device.battery else "N/A"
link = str(device.link_quality) if device.link_quality else "N/A"
status_icons = {
"healthy": "[OK] healthy",
"warning": "[!] warning",
"critical": "[X] critical",
"offline": "[ ] offline"
}
status = status_icons.get(device.health_status(), "unknown")
print(f"{name:<25} {device.device_type:<12} {battery:<10} "
f"{link:<8} {status:<12} {device.message_count:<10}")
print("=" * 80)
# Print alerts
alerts = self.get_health_alerts()
if alerts:
print("\n" + "=" * 80)
print("HEALTH ALERTS")
print("=" * 80)
for alert in alerts:
print(alert)
print("=" * 80)
def run(self, status_interval: int = 30):
"""Start the network analyzer."""
print("=== Zigbee Network Analyzer ===")
print(f"Connecting to MQTT broker: {self.mqtt_host}:{self.mqtt_port}")
try:
self.client.connect(self.mqtt_host, self.mqtt_port, 60)
self.client.loop_start()
print("Monitoring network... (Press Ctrl+C to stop)\n")
while True:
time.sleep(status_interval)
self.print_status()
except KeyboardInterrupt:
print("\nShutting down...")
finally:
self.client.loop_stop()
self.client.disconnect()
if __name__ == "__main__":
analyzer = ZigbeeNetworkAnalyzer(
mqtt_host="localhost",
mqtt_port=1883
)
analyzer.run(status_interval=30)998.5 Running the Analyzer
# Start the analyzer
python3 zigbee_analyzer.py
# Or specify custom MQTT broker
python3 zigbee_analyzer.py --host 192.168.1.100 --port 1883998.6 Expected Output
=== Zigbee Network Analyzer ===
Connecting to MQTT broker: localhost:1883
[OK] Connected to MQTT broker
[OK] Subscribed to zigbee2mqtt/#
Monitoring network... (Press Ctrl+C to stop)
[OK] Network map received
================================================================================
ZIGBEE NETWORK STATUS
================================================================================
Devices: 8 total, 7 online
Uptime: 30 seconds
Messages: 127
--------------------------------------------------------------------------------
Device Type Battery Link Status Messages
--------------------------------------------------------------------------------
bedroom/motion EndDevice 85% 145 [OK] healthy 12
bedroom/temp EndDevice 92% 168 [OK] healthy 15
coordinator Coordinator N/A N/A [OK] healthy 0
kitchen/light Router N/A 255 [OK] healthy 8
living_room/door EndDevice 18% 98 [!] warning 9
living_room/light Router N/A 255 [OK] healthy 11
office/temp EndDevice 5% 142 [X] critical 7
patio/motion EndDevice N/A N/A [ ] offline 0
================================================================================
================================================================================
HEALTH ALERTS
================================================================================
[CRITICAL] office/temp - Battery 5%
[WARNING] living_room/door - Battery 18%
[OFFLINE] patio/motion
================================================================================998.7 Extending the Analyzer
998.7.1 Adding Slack Notifications
import requests
def send_slack_alert(webhook_url: str, message: str):
"""Send alert to Slack channel."""
payload = {"text": message}
requests.post(webhook_url, json=payload)
# In the analyzer class:
def check_and_alert(self):
alerts = self.get_health_alerts()
for alert in alerts:
if "[CRITICAL]" in alert:
send_slack_alert(WEBHOOK_URL, f":warning: {alert}")998.7.2 Logging to InfluxDB
from influxdb import InfluxDBClient
def log_to_influx(self, device: ZigbeeDevice):
"""Log device metrics to InfluxDB."""
point = {
"measurement": "zigbee_device",
"tags": {
"device": device.friendly_name,
"type": device.device_type
},
"fields": {
"battery": device.battery or 0,
"link_quality": device.link_quality or 0,
"online": int(device.is_online)
}
}
self.influx_client.write_points([point])998.8 Troubleshooting
WarningCommon Issues
| Problem | Solution |
|---|---|
| Connection refused | Verify MQTT broker is running |
| No messages received | Check Zigbee2MQTT is publishing |
| Devices show offline | May be sleepy end devices |
| Missing battery data | Not all devices report battery |
998.9 Summary
This lab demonstrated building a Python network analyzer for Zigbee:
- MQTT Integration: Connected to Zigbee2MQTT via paho-mqtt library
- Device Tracking: Monitored battery, link quality, and online status
- Health Alerting: Implemented critical/warning/offline detection
- Network Statistics: Tracked message counts and uptime
- Extensibility: Added hooks for Slack and InfluxDB integration
998.10 Whatβs Next
Continue with Zigbee Lab: Mesh Simulator for hands-on experiments with mesh routing, self-healing, and hop counting using an interactive Wokwi simulation.
NoteRelated Chapters
Prerequisites: - Zigbee Fundamentals and Architecture - Protocol concepts - Zigbee Lab: Temperature Network - Hardware setup
Next Steps: - Zigbee Lab: Mesh Simulator - Interactive simulation
Reference: - Zigbee Comprehensive Review - Complete protocol guide