Python e MQTT para IoT na Pratica
Conecte dispositivos IoT com Python e MQTT. Aprenda a publicar, assinar topicos, processar sensores e criar dashboards em tempo real.
A Internet das Coisas (IoT) esta transformando industrias inteiras, e o protocolo MQTT e o coracao da comunicacao entre dispositivos. Leve, eficiente e perfeito para conexoes instáveis, o MQTT combinado com Python cria um ecossistema poderoso para coletar, processar e visualizar dados de sensores. Neste artigo, vamos construir um sistema IoT completo.
O Que e MQTT
MQTT (Message Queuing Telemetry Transport) e um protocolo de mensagens publish/subscribe projetado para dispositivos com recursos limitados. Ele funciona com tres componentes: o publisher (quem envia mensagens), o subscriber (quem recebe) e o broker (servidor que roteia as mensagens).
As principais vantagens do MQTT sao: baixo consumo de banda, suporte a conexoes instáveis, qualidade de servico configuravel (QoS 0, 1 ou 2) e padroes de topico flexiveis.
Configurando o Ambiente
Instale a biblioteca paho-mqtt e configure um broker local com Mosquitto:
pip install paho-mqtt
# Instalar Mosquitto (macOS)
brew install mosquitto
brew services start mosquitto
# Ou usar Docker
docker run -d --name mosquitto -p 1883:1883 -p 9001:9001 eclipse-mosquitto
Publicando Mensagens (Publisher)
Vamos criar um simulador de sensor que publica dados de temperatura e umidade:
import paho.mqtt.client as mqtt
import json
import time
import random
from datetime import datetime
# Configuracao do broker
BROKER = "localhost"
PORTA = 1883
TOPICO_BASE = "casa/sensores"
def criar_cliente_publisher() -> mqtt.Client:
"""Cria e configura o cliente MQTT."""
cliente = mqtt.Client(client_id="sensor-sala-01")
def on_connect(client, userdata, flags, rc):
status = {0: "Conectado", 1: "Protocolo recusado", 2: "ID rejeitado"}
print(f"Publisher: {status.get(rc, 'Erro desconhecido')}")
def on_publish(client, userdata, mid):
print(f" Mensagem {mid} publicada")
cliente.on_connect = on_connect
cliente.on_publish = on_publish
return cliente
def simular_sensor():
"""Simula leituras de sensor e publica via MQTT."""
cliente = criar_cliente_publisher()
cliente.connect(BROKER, PORTA, keepalive=60)
cliente.loop_start()
temperatura_base = 22.0
umidade_base = 65.0
try:
while True:
# Simular variacao realista
temperatura = temperatura_base + random.uniform(-2.0, 2.0)
umidade = umidade_base + random.uniform(-5.0, 5.0)
payload = {
"sensor_id": "sala-01",
"temperatura": round(temperatura, 1),
"umidade": round(umidade, 1),
"timestamp": datetime.now().isoformat(),
}
# Publicar em topicos especificos
cliente.publish(
f"{TOPICO_BASE}/sala/temperatura",
json.dumps(payload),
qos=1,
)
print(f"Temp: {temperatura:.1f}C | Umid: {umidade:.1f}%")
time.sleep(5)
except KeyboardInterrupt:
print("\nSensor desligado")
finally:
cliente.loop_stop()
cliente.disconnect()
simular_sensor()
Assinando Topicos (Subscriber)
O subscriber recebe e processa as mensagens dos sensores:
import paho.mqtt.client as mqtt
import json
from datetime import datetime
from collections import deque
# Historico de leituras
historico = deque(maxlen=100)
def on_connect(client, userdata, flags, rc):
print(f"Subscriber conectado (codigo: {rc})")
# Assinar topicos com wildcard
client.subscribe("casa/sensores/#", qos=1)
print("Inscrito em: casa/sensores/#")
def on_message(client, userdata, msg):
"""Callback executado ao receber mensagem."""
try:
payload = json.loads(msg.payload.decode())
topico = msg.topic
print(f"\n[{topico}]")
print(f" Sensor: {payload.get('sensor_id')}")
print(f" Temperatura: {payload.get('temperatura')}C")
print(f" Umidade: {payload.get('umidade')}%")
print(f" Horario: {payload.get('timestamp')}")
# Armazenar no historico
historico.append({
"topico": topico,
"dados": payload,
"recebido_em": datetime.now().isoformat(),
})
# Verificar alertas
verificar_alertas(payload, client)
except json.JSONDecodeError:
print(f"Erro ao decodificar: {msg.payload}")
def verificar_alertas(payload: dict, client: mqtt.Client):
"""Verifica condicoes de alerta e publica notificacao."""
temperatura = payload.get("temperatura", 0)
umidade = payload.get("umidade", 0)
sensor_id = payload.get("sensor_id", "desconhecido")
alertas = []
if temperatura > 30:
alertas.append(f"Temperatura alta: {temperatura}C")
if temperatura < 10:
alertas.append(f"Temperatura baixa: {temperatura}C")
if umidade > 80:
alertas.append(f"Umidade alta: {umidade}%")
if umidade < 30:
alertas.append(f"Umidade baixa: {umidade}%")
for alerta in alertas:
mensagem = json.dumps({
"sensor_id": sensor_id,
"alerta": alerta,
"timestamp": datetime.now().isoformat(),
})
client.publish("casa/alertas", mensagem, qos=2)
print(f" ALERTA: {alerta}")
def iniciar_subscriber():
"""Inicia o subscriber MQTT."""
cliente = mqtt.Client(client_id="central-monitoramento")
cliente.on_connect = on_connect
cliente.on_message = on_message
cliente.connect(BROKER, PORTA, keepalive=60)
print("Aguardando mensagens... (Ctrl+C para sair)")
try:
cliente.loop_forever()
except KeyboardInterrupt:
print("\nMonitoramento encerrado")
cliente.disconnect()
iniciar_subscriber()
Sistema Completo com Persistencia
Vamos criar um sistema que armazena leituras em SQLite:
import sqlite3
import json
import paho.mqtt.client as mqtt
from datetime import datetime, timedelta
class MonitorIoT:
"""Sistema de monitoramento IoT com persistencia."""
def __init__(self, broker: str, porta: int, banco: str = "iot_dados.db"):
self.broker = broker
self.porta = porta
self.banco = banco
self.cliente = mqtt.Client(client_id="monitor-central")
self._criar_tabelas()
self._configurar_mqtt()
def _criar_tabelas(self):
conn = sqlite3.connect(self.banco)
conn.execute("""
CREATE TABLE IF NOT EXISTS leituras (
id INTEGER PRIMARY KEY AUTOINCREMENT,
sensor_id TEXT NOT NULL,
temperatura REAL,
umidade REAL,
timestamp TEXT NOT NULL,
recebido_em TEXT NOT NULL
)
""")
conn.execute("""
CREATE TABLE IF NOT EXISTS alertas (
id INTEGER PRIMARY KEY AUTOINCREMENT,
sensor_id TEXT NOT NULL,
tipo TEXT NOT NULL,
mensagem TEXT NOT NULL,
timestamp TEXT NOT NULL
)
""")
conn.commit()
conn.close()
def _configurar_mqtt(self):
self.cliente.on_connect = self._on_connect
self.cliente.on_message = self._on_message
def _on_connect(self, client, userdata, flags, rc):
print(f"Monitor conectado ao broker")
client.subscribe("casa/sensores/#", qos=1)
client.subscribe("casa/alertas", qos=2)
def _on_message(self, client, userdata, msg):
payload = json.loads(msg.payload.decode())
if "alertas" in msg.topic:
self._salvar_alerta(payload)
else:
self._salvar_leitura(payload)
def _salvar_leitura(self, dados: dict):
conn = sqlite3.connect(self.banco)
conn.execute(
"INSERT INTO leituras (sensor_id, temperatura, umidade, timestamp, recebido_em) VALUES (?, ?, ?, ?, ?)",
(dados["sensor_id"], dados["temperatura"], dados["umidade"],
dados["timestamp"], datetime.now().isoformat()),
)
conn.commit()
conn.close()
def _salvar_alerta(self, dados: dict):
conn = sqlite3.connect(self.banco)
conn.execute(
"INSERT INTO alertas (sensor_id, tipo, mensagem, timestamp) VALUES (?, ?, ?, ?)",
(dados["sensor_id"], "temperatura", dados["alerta"], dados["timestamp"]),
)
conn.commit()
conn.close()
def relatorio(self, horas: int = 24) -> dict:
"""Gera relatorio das ultimas N horas."""
conn = sqlite3.connect(self.banco)
limite = (datetime.now() - timedelta(hours=horas)).isoformat()
cursor = conn.execute(
"SELECT AVG(temperatura), MIN(temperatura), MAX(temperatura), AVG(umidade), COUNT(*) FROM leituras WHERE recebido_em > ?",
(limite,),
)
row = cursor.fetchone()
conn.close()
return {
"periodo_horas": horas,
"temp_media": round(row[0] or 0, 1),
"temp_min": round(row[1] or 0, 1),
"temp_max": round(row[2] or 0, 1),
"umid_media": round(row[3] or 0, 1),
"total_leituras": row[4],
}
def iniciar(self):
self.cliente.connect(self.broker, self.porta)
print("Monitor IoT iniciado")
self.cliente.loop_forever()
# Uso
monitor = MonitorIoT("localhost", 1883)
monitor.iniciar()
Topicos e Wildcards MQTT
Entender a hierarquia de topicos e fundamental para um sistema IoT bem organizado:
# Hierarquia de topicos recomendada
# casa/andar/comodo/tipo_sensor
TOPICOS = {
"casa/terreo/sala/temperatura": "Sensor de temperatura da sala",
"casa/terreo/cozinha/gas": "Sensor de gas da cozinha",
"casa/primeiro/quarto1/umidade": "Sensor de umidade do quarto",
"casa/jardim/irrigacao/status": "Status do sistema de irrigacao",
}
# Wildcards
# + substitui um nivel: casa/+/sala/+ (qualquer andar, qualquer sensor da sala)
# # substitui multiplos niveis: casa/# (tudo da casa)
Boas Praticas para IoT com MQTT
Use QoS 0 para dados frequentes e nao criticos, como leituras periodicas. Use QoS 1 para dados importantes que precisam ser entregues pelo menos uma vez. Reserve QoS 2 para comandos criticos onde duplicatas nao sao aceitaveis. Implemente reconexao automatica com backoff exponencial. Mantenha payloads pequenos para economizar banda. E use TLS para criptografar a comunicacao em ambientes de producao.
# Reconexao automatica
def on_disconnect(client, userdata, rc):
if rc != 0:
print(f"Desconectado inesperadamente (codigo: {rc}). Reconectando...")
while True:
try:
client.reconnect()
break
except Exception:
time.sleep(5)
Conclusao
Python e MQTT formam a base perfeita para projetos IoT. Com o paho-mqtt, voce controla todo o ciclo de vida das mensagens, desde a publicacao de dados de sensores ate o processamento e armazenamento em um sistema central. Comece com um sensor simples, evolua para um sistema com alertas e persistencia, e voce tera a infraestrutura para monitorar qualquer ambiente de forma inteligente.
Equipe Python Brasil
Contribuidor do Python Brasil — Aprenda Python em Português