Raspberry Pi e IoT com Python: 100 exemplos do zero ao profissional
Um Raspberry Pi é um pequeno computador que consegue ler sensores, acionar relés e enviar dados pela rede, por isso é uma base comum para projetos de IoT (Internet das Coisas). Este artigo leva você do primeiro programa em Python até um pipeline completo de sensores, em 100 exemplos. Você vai aprender os fundamentos do Python, processar dados de sensores, controlar pinos com GPIO, ler sensores por I2C, SPI e 1-Wire, enviar dados com HTTP e MQTT, criar um pequeno painel web, guardar leituras no SQLite e executar tudo como um serviço que inicia com o sistema. Nota de transparência: os exemplos que rodam em qualquer computador foram executados e a saída real aparece depois do código. Os demais precisam de um Raspberry Pi, de um serviço de rede ou de um serviço do sistema, então foram apenas verificados quanto à sintaxe e a saída não é mostrada.
O que você precisa
Um Raspberry Pi 3, 4 ou 5 com Raspberry Pi OS, um cartão microSD, uma fonte de alimentação e uma conexão de rede. O Python 3.10 ou mais recente já vem no Raspberry Pi OS. Para os exemplos de hardware, você vai precisar de um LED, um resistor, um botão, fios jumper e alguns sensores. A maioria dos exemplos executáveis funciona em qualquer computador com Python 3.10 ou mais recente, e foi assim que eles foram testados.
Como executar um exemplo
Salve um exemplo como main.py e execute com python3 main.py. Os exemplos marcados como executáveis mostram a saída real de uma execução em um computador. Os exemplos bash são arquivos de configuração do Pi e não foram executados aqui.
Tópico 1: Começando com Python no Pi
O Python é a linguagem usual dos projetos com Raspberry Pi porque é fácil de ler e tem bibliotecas para quase todos os sensores. Os primeiros exemplos cobrem saída, variáveis, tipos e cálculos.
Executando o primeiro programa
Salve um arquivo com a extensão .py e execute-o com python3 no Pi. A função print escreve texto no terminal, e as f-strings colocam variáveis dentro do texto com {nome}. O Python lê o arquivo de cima para baixo, por isso a ordem das linhas importa.
Tipos e o que significam
Números podem ser int (inteiros) ou float (decimais), o texto é str, True e False são bool, e None significa ausência de valor. Use type() para verificar o tipo de um valor. As leituras de sensores costumam ser float, enquanto os números de pinos costumam ser int.
Aritmética para cálculos de sensores
A divisão / sempre produz float, // divide e descarta a parte fracionária, e % dá o resto. round() limita as casas decimais exibidas. Use esses operadores para converter leituras brutas e agrupar valores.
Ilustrações
Exemplo 1: Olá do Pi
Seu primeiro programa Python no Raspberry Pi OS.
print("Hello from my Raspberry Pi!")
Saída:
Hello from my Raspberry Pi!
Exemplo 2: Variáveis e f-strings
Guarde valores e mostre-os dentro de textos.
sensor_name = "DHT22"
temperature = 27.5
print(f"{sensor_name} reads {temperature} °C")
Saída:
DHT22 reads 27.5 °C
Exemplo 3: Tipos de dados
Os tipos básicos do Python e seus nomes.
for value in (42, 3.14, "pi", True, None):
print(repr(value), type(value).__name__)
Saída:
42 int
3.14 float
'pi' str
True bool
None NoneType
Exemplo 4: Aritmética, divisão inteira e arredondamento
Divisão, resto e arredondamento para cálculos de sensores.
readings = 17
per_group = 5
print(readings / per_group)
print(readings // per_group, readings % per_group)
print(round(3.14159, 2))
Saída:
3.4
3 2
3.14
Exemplo 5: Trabalhando com texto
Limpe nomes de dispositivos e verifique prefixos.
status = " ONLINE "
clean = status.strip().lower()
print(clean, clean.startswith("on"), len(clean))
Saída:
online True 6
Exemplo 6: Conversão de unidades
Converta Celsius em Fahrenheit com uma função.
def celsius_to_fahrenheit(c):
return c * 9 / 5 + 32
for c in (0, 25, 37.5):
print(f"{c} C = {celsius_to_fahrenheit(c):.1f} F")
Saída:
0 C = 32.0 F
25 C = 77.0 F
37.5 C = 99.5 F
Tópico 2: Decisões e Repetições
Um projeto com Pi reage às suas entradas. As condições decidem o que acontece, e os laços repetem o trabalho, como coletar uma amostra do sensor a cada poucos segundos.
if, elif e else
As condições são testadas de cima para baixo, e o primeiro ramo verdadeiro é executado. Coloque a verificação mais específica primeiro. Uma comparação encadeada como 20 <= t <= 25 verifica uma faixa inteira em uma linha.
for, while, break e continue
O for repete sobre um intervalo ou uma sequência. O while repete até que uma condição mude. O break sai do laço, e o continue pula a volta atual, o que é útil para ignorar uma leitura ruim e seguir em frente.
Valores verdadeiros e a instrução match
Valores vazios, como uma string ou uma lista vazia, contam como falsos. A instrução match, disponível desde o Python 3.10, compara um comando com vários padrões, uma forma limpa de tratar comandos de texto vindos de um aplicativo de celular.
Exemplo 7: if, elif e else
Escolha uma ação a partir da temperatura.
temperature = 31
if temperature >= 35:
print("Too hot: stop the fan test")
elif temperature >= 28:
print("Warm: run the fan")
else:
print("Comfortable")
Saída:
Warm: run the fan
Exemplo 8: O laço for com range
Repita uma tarefa um número fixo de vezes.
for minute in range(0, 15, 5):
print(f"minute {minute}")
Saída:
minute 0
minute 5
minute 10
Exemplo 9: O laço while
Repita até que uma condição mude.
countdown = 3
while countdown > 0:
print("sampling in", countdown)
countdown -= 1
print("start")
Saída:
sampling in 3
sampling in 2
sampling in 1
start
Exemplo 10: break e continue
Pule leituras ruins e pare em caso de falha.
for reading in [21, None, 23, 99, 24]:
if reading is None:
continue
if reading > 50:
print("sensor fault, stop")
break
print("ok", reading)
Saída:
ok 21
ok 23
sensor fault, stop
Exemplo 11: Valores verdadeiros e falsos
Quais valores contam como verdadeiros ou falsos.
for value in ("", "Pi", 0, 5, [], [1]):
print(repr(value), "is true" if value else "is false")
Saída:
'' is false
'Pi' is true
0 is false
5 is true
[] is false
[1] is true
Exemplo 12: Comparações encadeadas
Verifique se um valor está dentro de um intervalo.
humidity = 55
print(40 <= humidity <= 60)
print(humidity > 60 or humidity < 40)
Saída:
True
False
Exemplo 13: Expressão condicional
Um if-else em uma linha que produz um valor.
reading = 12
state = "ON" if reading < 15 else "OFF"
print(state)
Saída:
ON
Exemplo 14: A instrução match
Associe comandos de texto a ações.
def describe(command):
match command:
case "on" | "1":
return "turn relay on"
case "off" | "0":
return "turn relay off"
case _:
return "unknown command"
for c in ("on", "0", "blink"):
print(c, "->", describe(c))
Saída:
on -> turn relay on
0 -> turn relay off
blink -> unknown command
Tópico 3: Funções e a Biblioteca Padrão
Uma função dá nome a uma tarefa que você repete. A biblioteca padrão já tem funções de matemática, estatística e tempo, então você raramente precisa escrevê-las.
Definindo funções
def cria uma função, e return devolve um valor. Argumentos padrão dão valores sensatos quando quem chama os omite, e argumentos nomeados deixam a chamada legível, como sample(4, unit='volts').
Devolvendo e recebendo vários valores
Uma função pode devolver vários valores como tupla, que quem chama desempacota. *args aceita qualquer quantidade de valores posicionais, e **kwargs aceita qualquer quantidade de valores nomeados, o que combina com funções auxiliares de registro.
lambda, map, filter e recursão
Uma lambda é uma pequena função sem nome, útil com map e filter para transformar uma lista. Uma função recursiva chama a si mesma e precisa de uma condição de parada; o fatorial é o exemplo clássico.
Exemplo 15: Funções com argumentos padrão
Dê valores padrão sensatos a uma função.
def sample(pin, samples=3, unit="raw"):
return f"read pin {pin} x{samples} ({unit})"
print(sample(4))
print(sample(4, 10, unit="volts"))
Saída:
read pin 4 x3 (raw)
read pin 4 x10 (volts)
Exemplo 16: Retornando vários valores
Devolva o mínimo e o máximo juntos.
def min_max(values):
return min(values), max(values)
low, high = min_max([19.5, 22.0, 18.2])
print(f"low {low}, high {high}")
Saída:
low 18.2, high 22.0
Exemplo 17: *args e **kwargs
Aceite qualquer quantidade de valores posicionais e nomeados.
def log_event(*parts, **fields):
text = " ".join(str(p) for p in parts)
extra = ", ".join(f"{k}={v}" for k, v in fields.items())
print(text, "|", extra)
log_event("door", "opened", sensor="reed-1", room="kitchen")
Saída:
door opened | sensor=reed-1, room=kitchen
Exemplo 18: lambda, map e filter
Converta valores brutos do ADC em uma linha.
raw = [512, 1023, 0, 768]
volts = list(map(lambda x: round(x * 3.3 / 1023, 2), raw))
nonzero = list(filter(lambda v: v > 0, volts))
print(volts)
print(nonzero)
Saída:
[1.65, 3.3, 0.0, 2.48]
[1.65, 3.3, 2.48]
Exemplo 19: Biblioteca padrão: math e statistics
Use os módulos embutidos de matemática e estatística.
import math
import statistics
samples = [2.0, 4.0, 4.5, 3.5]
print(round(math.sqrt(16), 2), math.pi > 3.14)
print(statistics.mean(samples), statistics.median(samples))
Saída:
4.0 True
3.5 3.75
Exemplo 20: Recursão
Uma função que chama a si mesma, com uma condição de parada.
def factorial(n):
if n <= 1:
return 1
return n * factorial(n - 1)
print(factorial(5), factorial(10))
Saída:
120 3628800
Tópico 4: Estruturas de Dados para Dispositivos e Leituras
Um projeto com Pi guarda listas de leituras, dicionários de configurações de dispositivos e conjuntos de nomes únicos. Escolher a estrutura certa deixa o código mais curto e mais rápido.
Listas e compreensões
Uma lista guarda uma sequência ordenada que você pode alterar. Uma compreensão de lista cria uma nova lista a partir de outra em uma só linha, como converter uma lista de valores em Celsius para Fahrenheit.
Dicionários para valores nomeados
Um dicionário associa uma chave a um valor, ideal para as configurações de um dispositivo ou para uma leitura com seus campos. Use get() para ler uma chave que pode não existir e sorted() para exibir as chaves em ordem.
Tuplas, conjuntos e fatiamento
Uma tupla é um grupo fixo de valores. Um conjunto guarda itens únicos e aceita união e interseção. Fatias como valores[-2:] pegam parte de uma lista, e [::-1] a inverte.
Exemplo 21: Listas
Adicione, insira e remova itens.
readings = [20.1, 20.4]
readings.append(20.9)
readings.insert(0, 19.8)
readings.remove(20.4)
print(readings, len(readings), readings[-1])
Saída:
[19.8, 20.1, 20.9] 3 20.9
Exemplo 22: Compreensões de lista
Crie uma lista nova a partir de uma antiga.
celsius = [0, 10, 20, 30]
fahrenheit = [c * 9 / 5 + 32 for c in celsius if c != 10]
print(fahrenheit)
Saída:
[32.0, 68.0, 86.0]
Exemplo 23: Tuplas e desempacotamento
Grupos fixos de valores que não podem mudar.
location = ("kitchen", 2, 3.5)
room, floor, height = location
print(room, floor, height)
try:
location[0] = "garage"
except TypeError as e:
print("tuples cannot change:", type(e).__name__)
Saída:
kitchen 2 3.5
tuples cannot change: TypeError
Exemplo 24: Dicionários
Guarde valores nomeados de um dispositivo.
device = {"name": "node-1", "pin": 17, "online": True}
device["pin"] = 22
device["room"] = "garage"
print(device.get("name"), device.get("temp", "n/a"), len(device))
Saída:
node-1 n/a 4
Exemplo 25: Saída de dicionário ordenada
Imprima a contagem de cada evento em ordem.
counts = {"motion": 4, "door": 1, "temp": 9}
for name in sorted(counts):
print(f"{name:<8}{counts[name]:>3}")
Saída:
door 1
motion 4
temp 9
Exemplo 26: Conjuntos
Valores únicos e operações de conjunto.
seen = {"kitchen", "garage"}
seen.add("kitchen")
new = {"garage", "attic"}
print(len(seen), sorted(seen | new), sorted(seen & new))
Saída:
2 ['attic', 'garage', 'kitchen'] ['garage']
Exemplo 27: Fatiamento
Pegue partes de uma lista e inverta-a.
samples = [1, 2, 3, 4, 5, 6, 7]
print(samples[:3], samples[-2:], samples[::2])
print(samples[::-1])
Saída:
[1, 2, 3] [6, 7] [1, 3, 5, 7]
[7, 6, 5, 4, 3, 2, 1]
Exemplo 28: Listas de dicionários
Filtre a lista de dispositivos que estão online.
devices = [
{"id": "a1", "online": True},
{"id": "b2", "online": False},
{"id": "c3", "online": True},
]
online = [d["id"] for d in devices if d["online"]]
print("online:", online)
Saída:
online: ['a1', 'c3']
Tópico 5: Texto, Arquivos e Formatos de Dados
Os dados dos sensores precisam ser guardados e compartilhados. Arquivos de texto, CSV e JSON são os formatos que a maioria dos projetos de IoT usa, e a biblioteca padrão lê e escreve os três.
Lendo e escrevendo arquivos
Abra um arquivo com um bloco with para que ele feche sozinho, e use encoding='utf-8' para texto. Os objetos Path do pathlib funcionam em qualquer sistema operacional e facilitam juntar nomes de arquivos.
CSV e JSON
O módulo csv escreve linhas com cabeçalho, que planilhas e ferramentas de dados entendem. O módulo json transforma dicionários em texto para serviços web e mensagens MQTT, e faz o caminho inverso.
Formatando tabelas e datas
Larguras de formato como {nome:<10} alinham colunas no terminal. O datetime formata carimbos de tempo com strftime e soma durações com timedelta, então as leituras podem ser agrupadas por horário.
Exemplo 29: Formatando a saída em tabela
Alinhe colunas com larguras de formatação.
rows = [("Temp", 27.456), ("Humidity", 55.0), ("Pressure", 1013.2)]
for name, value in rows:
print(f"{name:<10}{value:>10.1f}")
Saída:
Temp 27.5
Humidity 55.0
Pressure 1013.2
Exemplo 30: Lendo e escrevendo arquivos de texto
Salve e carregue valores de sensores em um arquivo.
from pathlib import Path
log = Path("sensor_log.txt")
with log.open("w", encoding="utf-8") as f:
f.write("21.5\n22.0\n22.4\n")
with log.open("r", encoding="utf-8") as f:
values = [float(line) for line in f]
log.unlink()
print("read", len(values), "values, last", values[-1])
Saída:
read 3 values, last 22.4
Exemplo 31: Arquivos CSV
Escreva e leia uma tabela de medições.
import csv
from pathlib import Path
path = Path("readings.csv")
with path.open("w", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
writer.writerow(["time", "temp_c"])
writer.writerows([["08:00", 21.5], ["08:05", 21.9]])
with path.open(newline="", encoding="utf-8") as f:
for row in csv.DictReader(f):
print(row["time"], float(row["temp_c"]) + 1)
path.unlink()
Saída:
08:00 22.5
08:05 22.9
Exemplo 32: JSON
Converta dados para texto JSON e de volta.
import json
payload = {"device": "node-1", "readings": [21.5, 21.9], "ok": True}
text = json.dumps(payload, indent=2)
print(text)
print(json.loads(text)["readings"][1])
Saída:
{
"device": "node-1",
"readings": [
21.5,
21.9
],
"ok": true
}
21.9
Exemplo 33: Caminhos
Trabalhe com nomes de arquivos e pastas com segurança.
from pathlib import PurePosixPath
path = PurePosixPath("/home/pi/projects/iot/config.json")
print(path.name, path.suffix, path.parent)
print(path.parent / "logs")
Saída:
config.json .json /home/pi/projects/iot
/home/pi/projects/iot/logs
Exemplo 34: Datas e horas
Formate e some valores de data e hora.
from datetime import datetime, timedelta
start = datetime(2026, 1, 15, 8, 0)
later = start + timedelta(minutes=90)
print(start.strftime("%Y-%m-%d %H:%M"), "->", later.strftime("%H:%M"))
Saída:
2026-01-15 08:00 -> 09:30
Tópico 6: Erros, Registro de Logs e Medição de Tempo
Um dispositivo funciona por meses sem que ninguém o observe. Ele precisa sobreviver a leituras ruins, registrar o que aconteceu e medir o próprio tempo de execução.
Tratando erros com try e except
Coloque o código que pode falhar dentro de try, capture a exceção específica e continue executando. O finally roda com ou sem erro, por isso é o lugar certo para desligar uma saída ou fechar um arquivo.
Registro de logs no lugar do print
O módulo logging registra mensagens com um nível, como INFO ou WARNING, e você escolhe quais níveis aparecem. Os logs podem ir para um arquivo, o que é necessário quando o Pi funciona sem tela.
Verificações defensivas e medição de tempo
O assert rejeita valores impossíveis logo no início, durante o desenvolvimento. O time.perf_counter mede o tempo decorrido com precisão, o que ajuda a encontrar código lento em um computador pequeno.
Exemplo 35: try e except
Trate uma leitura ruim sem parar o programa.
def parse_temperature(text):
try:
return float(text)
except ValueError:
print(f"bad reading: {text!r}")
return None
print(parse_temperature("23.4"))
print(parse_temperature("23,4"))
Saída:
23.4
bad reading: '23,4'
None
Exemplo 36: finally e limpeza
Faça a limpeza sempre, mesmo após um erro.
from pathlib import Path
target = Path("demo_state.txt")
try:
with target.open("w", encoding="utf-8") as f:
f.write("on")
print("saved")
finally:
target.unlink(missing_ok=True)
print("cleaned up")
Saída:
saved
cleaned up
Exemplo 37: Exceções personalizadas
Lance um erro claro para um valor de sensor impossível.
class SensorError(Exception):
"""Raised when a sensor gives an impossible value."""
def check(value):
if value < -40 or value > 85:
raise SensorError(f"value {value} is outside the sensor range")
return value
try:
check(120)
except SensorError as e:
print("sensor error:", e)
Saída:
sensor error: value 120 is outside the sensor range
Exemplo 38: Registro de logs
Registre mensagens com níveis, em vez de usar print.
import logging
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
log = logging.getLogger("node-1")
log.info("started")
log.warning("battery at 18%")
log.debug("this line is hidden at INFO level")
Saída:
Exemplo 39: Verificações defensivas com assert
Rejeite entradas inválidas logo no início.
def set_pwm(duty):
assert 0 <= duty <= 100, "duty must be 0 to 100"
return duty
print(set_pwm(50))
try:
set_pwm(150)
except AssertionError as e:
print("rejected:", e)
Saída:
50
rejected: duty must be 0 to 100
Exemplo 40: Medindo o tempo decorrido
Cronometre um trecho de código com um relógio preciso.
import time
start = time.perf_counter()
total = sum(i * i for i in range(200_000))
elapsed = time.perf_counter() - start
print("sum of squares:", total)
print("took measurable time:", elapsed >= 0)
Saída:
sum of squares: 2666646666700000
took measurable time: True
Tópico 7: Processando Dados de Sensores
Os valores brutos dos sensores são ruidosos e muitas vezes estão em unidades erradas. Estes exemplos transformam números brutos em leituras úteis: calibrados, suavizados, verificados quanto a valores atípicos e comparados com limites.
Calibração e conversão de unidades
Um ADC devolve contagens brutas. Dois pontos de referência conhecidos dão uma calibração em linha reta, permitindo converter contagens em volts. Guarde a calibração junto com o dispositivo, porque cada sensor é diferente.
Suavização e histerese
Uma média móvel das últimas leituras remove picos repentinos. A histerese usa dois limites, um para ligar e outro para desligar, de modo que um relé não fique oscilando quando o valor está perto de um dos limites.
Estatística e valores atípicos
A média, a mediana e o desvio padrão descrevem uma janela de leituras. A mediana ignora valores extremos, por isso é uma boa forma de remover um pico errado antes de calcular a média.
Exemplo 41: Calibrando valores brutos do ADC
Converta números brutos em volts com dois pontos de referência.
RAW_MIN, RAW_MAX = 102, 921 # readings at 0 V and at full scale
VOLT_MAX = 3.3
def to_volts(raw):
return (raw - RAW_MIN) * VOLT_MAX / (RAW_MAX - RAW_MIN)
for raw in (102, 512, 921):
print(raw, "->", round(to_volts(raw), 3), "V")
Saída:
102 -> 0.0 V
512 -> 1.652 V
921 -> 3.3 V
Exemplo 42: Filtro de média móvel
Suavize leituras ruidosas em uma janela.
from collections import deque
window = deque(maxlen=3)
def smooth(value):
window.append(value)
return sum(window) / len(window)
for reading in [20.0, 25.0, 21.0, 40.0, 22.0]:
print(f"raw {reading:5.1f} smoothed {smooth(reading):6.2f}")
Saída:
raw 20.0 smoothed 20.00
raw 25.0 smoothed 22.50
raw 21.0 smoothed 22.00
raw 40.0 smoothed 28.67
raw 22.0 smoothed 27.67
Exemplo 43: Histerese para um relé de ventilador
Impeça que o relé ligue e desligue rápido demais.
def relay_state(temp, currently_on, on_at=24.0, off_at=22.0):
if not currently_on and temp >= on_at:
return True
if currently_on and temp <= off_at:
return False
return currently_on
state = False
for t in [21.0, 23.0, 24.5, 23.5, 22.5, 21.9]:
state = relay_state(t, state)
print(t, "fan", "ON" if state else "OFF")
Saída:
21.0 fan OFF
23.0 fan OFF
24.5 fan ON
23.5 fan ON
22.5 fan ON
21.9 fan OFF
Exemplo 44: Unidades de pressão e arredondamento
Converta hPa em inHg e arredonde corretamente.
def hpa_to_inhg(hpa):
return hpa * 0.02953
pressure = 1013.25
print(f"{pressure} hPa = {hpa_to_inhg(pressure):.2f} inHg")
print(round(2.5), round(3.5))
Saída:
1013.25 hPa = 29.92 inHg
2 4
Exemplo 45: Estatística em uma janela
Média, dispersão e amplitude das leituras recentes.
import statistics
window = [21.4, 21.6, 21.5, 21.9, 21.7]
print("mean", round(statistics.mean(window), 2))
print("stdev", round(statistics.pstdev(window), 3))
print("range", round(max(window) - min(window), 2))
Saída:
mean 21.62
stdev 0.172
range 0.5
Exemplo 46: Remoção de valores atípicos com a mediana
Ignore valores muito distantes da leitura típica.
import statistics
def remove_outliers(values, tolerance=3.0):
med = statistics.median(values)
return [v for v in values if abs(v - med) <= tolerance]
raw = [21.5, 21.7, 85.0, 21.6, -40.0, 21.8]
print(remove_outliers(raw))
Saída:
[21.5, 21.7, 21.6, 21.8]
Exemplo 47: Taxa de variação
Quanto um valor sobe por minuto.
def rate_per_minute(prev_value, prev_minute, value, minute):
if minute == prev_minute:
return 0.0
return (value - prev_value) / (minute - prev_minute)
print(round(rate_per_minute(20.0, 0, 21.2, 4), 2), "C per minute")
Saída:
0.3 C per minute
Exemplo 48: Uma leitura como classe de dados
Descreva uma medição com uma pequena classe.
from dataclasses import dataclass, asdict
@dataclass
class Reading:
device: str
temperature_c: float
humidity_pct: float
r = Reading("node-1", 21.5, 48.0)
print(r)
print(asdict(r))
Saída:
Reading(device='node-1', temperature_c=21.5, humidity_pct=48.0)
{'device': 'node-1', 'temperature_c': 21.5, 'humidity_pct': 48.0}
Tópico 8: GPIO: Controlando LEDs, Botões, Motores e Relés
O conector de 40 pinos permite ao Pi acionar saídas e ler entradas. Os exemplos mostram LEDs, botões, redução de brilho com PWM, servos, um sensor de distância e relés. Confira sempre a fiação com o diagrama de pinos antes de ligar a energia.
Saídas, entradas e resistores pull-up
Configure um pino como saída para acender um LED por meio de um resistor, ou como entrada para ler um botão. Um resistor pull-up mantém a entrada em nível alto até que o botão a ligue ao terra, para que o pino nunca fique flutuando.
PWM e servos
A modulação por largura de pulso liga e desliga o pino rapidamente; a proporção do tempo ligado define o brilho ou a posição de um servo. A biblioteca gpiozero esconde a maior parte do trabalho de temporização.
Relés, sensores e desligamento seguro
Um relé aciona equipamentos de maior potência, como uma bomba, então o Pi só aciona a bobina de controle, pequena. Desligue as saídas sempre em um bloco finally e chame cleanup, para que uma falha não deixe um motor funcionando.
Exemplo 49: Piscar um LED (GPIO)
Acenda e apague um LED por um pino do Raspberry Pi.
import time
import RPi.GPIO as GPIO
LED_PIN = 17
GPIO.setmode(GPIO.BCM)
GPIO.setup(LED_PIN, GPIO.OUT)
try:
for _ in range(5):
GPIO.output(LED_PIN, GPIO.HIGH)
time.sleep(0.5)
GPIO.output(LED_PIN, GPIO.LOW)
time.sleep(0.5)
finally:
GPIO.cleanup()
Este exemplo precisa de um Raspberry Pi, de um serviço de rede ou de um serviço do sistema, então foi verificado apenas quanto à sintaxe. Nenhuma saída é mostrada.
Exemplo 50: Ler um botão com pull-up
Detecte o pressionamento de um botão com segurança e atraso de debounce.
import time
import RPi.GPIO as GPIO
BUTTON_PIN = 27
GPIO.setmode(GPIO.BCM)
GPIO.setup(BUTTON_PIN, GPIO.IN, pull_up_down=GPIO.PUD_UP)
try:
while True:
if GPIO.input(BUTTON_PIN) == GPIO.LOW: # pressed connects the pin to GND
print("button pressed")
time.sleep(0.2) # simple debounce
time.sleep(0.01)
finally:
GPIO.cleanup()
Este exemplo precisa de um Raspberry Pi, de um serviço de rede ou de um serviço do sistema, então foi verificado apenas quanto à sintaxe. Nenhuma saída é mostrada.
Exemplo 51: Escurecendo um LED com PWM
Mude o brilho de um LED com modulação por largura de pulso.
import time
import RPi.GPIO as GPIO
GPIO.setmode(GPIO.BCM)
GPIO.setup(18, GPIO.OUT)
pwm = GPIO.PWM(18, 1000) # 1 kHz
pwm.start(0)
try:
for duty in range(0, 101, 10):
pwm.ChangeDutyCycle(duty)
time.sleep(0.1)
finally:
pwm.stop()
GPIO.cleanup()
Este exemplo precisa de um Raspberry Pi, de um serviço de rede ou de um serviço do sistema, então foi verificado apenas quanto à sintaxe. Nenhuma saída é mostrada.
Exemplo 52: LED com gpiozero
Uma biblioteca GPIO mais simples e segura para LEDs.
from gpiozero import LED
from time import sleep
led = LED(17)
for _ in range(3):
led.on()
sleep(0.5)
led.off()
sleep(0.5)
Este exemplo precisa de um Raspberry Pi, de um serviço de rede ou de um serviço do sistema, então foi verificado apenas quanto à sintaxe. Nenhuma saída é mostrada.
Exemplo 53: Callbacks de botão
Reaja ao pressionamento sem um laço de consulta.
from gpiozero import Button
from signal import pause
button = Button(27)
button.when_pressed = lambda: print("pressed")
button.when_released = lambda: print("released")
pause()
Este exemplo precisa de um Raspberry Pi, de um serviço de rede ou de um serviço do sistema, então foi verificado apenas quanto à sintaxe. Nenhuma saída é mostrada.
Exemplo 54: Controle de servo
Mova um servo para posições definidas.
from gpiozero import Servo
from time import sleep
servo = Servo(12)
for position in (-1, 0, 1, 0): # -1 = left, 0 = centre, 1 = right
servo.value = position
sleep(1)
Este exemplo precisa de um Raspberry Pi, de um serviço de rede ou de um serviço do sistema, então foi verificado apenas quanto à sintaxe. Nenhuma saída é mostrada.
Exemplo 55: Sensor de distância ultrassônico
Meça distância com o sensor HC-SR04.
import time
import RPi.GPIO as GPIO
TRIG, ECHO = 23, 24
GPIO.setmode(GPIO.BCM)
GPIO.setup(TRIG, GPIO.OUT)
GPIO.setup(ECHO, GPIO.IN)
def distance_cm():
GPIO.output(TRIG, False)
time.sleep(0.05)
GPIO.output(TRIG, True)
time.sleep(0.00001)
GPIO.output(TRIG, False)
while GPIO.input(ECHO) == 0:
start = time.time()
while GPIO.input(ECHO) == 1:
end = time.time()
return (end - start) * 34300 / 2 # speed of sound, there and back
print(round(distance_cm(), 1), "cm")
GPIO.cleanup()
Este exemplo precisa de um Raspberry Pi, de um serviço de rede ou de um serviço do sistema, então foi verificado apenas quanto à sintaxe. Nenhuma saída é mostrada.
Exemplo 56: Relé com valor padrão seguro
Acione uma bomba com segurança e sempre desligue-a.
from gpiozero import OutputDevice
relay = OutputDevice(22, active_high=True, initial_value=False)
relay.on() # power the pump
relay.off() # always end with the relay off
Este exemplo precisa de um Raspberry Pi, de um serviço de rede ou de um serviço do sistema, então foi verificado apenas quanto à sintaxe. Nenhuma saída é mostrada.
Exemplo 57: Tons de buzzer
Toque sons simples em um buzzer.
from gpiozero import TonalBuzzer
from time import sleep
buzzer = TonalBuzzer(26)
for note in (0.0, 0.5, 1.0):
buzzer.play(note)
sleep(0.3)
buzzer.stop()
Este exemplo precisa de um Raspberry Pi, de um serviço de rede ou de um serviço do sistema, então foi verificado apenas quanto à sintaxe. Nenhuma saída é mostrada.
Exemplo 58: Desligamento seguro do GPIO
Deixe as saídas seguras antes de o programa terminar.
import RPi.GPIO as GPIO
GPIO.setmode(GPIO.BCM)
GPIO.setup(17, GPIO.OUT)
try:
GPIO.output(17, GPIO.HIGH)
raise RuntimeError("simulated fault")
except RuntimeError as e:
print("error:", e)
finally:
GPIO.output(17, GPIO.LOW) # make the output safe before cleanup
GPIO.cleanup()
Este exemplo precisa de um Raspberry Pi, de um serviço de rede ou de um serviço do sistema, então foi verificado apenas quanto à sintaxe. Nenhuma saída é mostrada.
Tópico 9: Sensores por I2C, SPI, 1-Wire e a Câmera
Muitos sensores conversam com o Pi por um barramento. O I2C usa dois fios e um endereço, o SPI usa quatro fios e um seletor de chip, e o 1-Wire usa um único fio de dados. O módulo de câmera é um caso especial com biblioteca própria.
I2C: varredura e leitura de registradores
Cada dispositivo I2C tem um endereço. Faça uma varredura do barramento primeiro para descobrir o que está conectado. Leia o registrador de ID do chip para confirmar o sensor, depois escreva o registrador de controle e leia os bytes de dados.
SPI e entradas analógicas
O Pi não tem entradas analógicas, então um ADC como o MCP3008 converte tensões pelo SPI. Envie o número do canal e monte o valor de 10 bits a partir dos bytes de resposta.
1-Wire, sensores DHT e a câmera
Um DS18B20 no 1-Wire aparece como um arquivo em /sys/bus/w1; ative a interface no raspi-config antes. Sensores DHT às vezes falham, então tente de novo. O módulo de câmera captura imagens fixas e quadros com o picamera2.
Exemplo 59: Varredura do barramento I2C
Descubra quais sensores estão conectados.
from smbus2 import SMBus
with SMBus(1) as bus:
found = []
for address in range(0x03, 0x78):
try:
bus.read_byte(address)
found.append(hex(address))
except OSError:
pass
print("devices:", found)
Este exemplo precisa de um Raspberry Pi, de um serviço de rede ou de um serviço do sistema, então foi verificado apenas quanto à sintaxe. Nenhuma saída é mostrada.
Exemplo 60: Lendo o sensor de pressão BMP280
Verifique o ID do chip e leia bytes brutos de pressão pelo I2C.
from smbus2 import SMBus
ADDRESS = 0x76
with SMBus(1) as bus:
chip_id = bus.read_byte_data(ADDRESS, 0xD0)
print("chip id", hex(chip_id), "(expect 0x58)")
bus.write_byte_data(ADDRESS, 0xF4, 0b00100111) # normal mode, oversampling x1
raw = bus.read_i2c_block_data(ADDRESS, 0xF7, 6)
print("raw bytes", raw)
Este exemplo precisa de um Raspberry Pi, de um serviço de rede ou de um serviço do sistema, então foi verificado apenas quanto à sintaxe. Nenhuma saída é mostrada.
Exemplo 61: Temperatura e umidade com DHT22
Leia o DHT22 e tente de novo quando não estiver pronto.
import time
import board
import adafruit_dht
dht = adafruit_dht.DHT22(board.D4)
for _ in range(5):
try:
print(f"{dht.temperature:.1f} C {dht.humidity:.0f} %")
except RuntimeError as e:
print("sensor not ready:", e) # DHT sensors fail sometimes; just retry
time.sleep(2)
Este exemplo precisa de um Raspberry Pi, de um serviço de rede ou de um serviço do sistema, então foi verificado apenas quanto à sintaxe. Nenhuma saída é mostrada.
Exemplo 62: Entrada analógica com MCP3008
Leia tensões analógicas pelo SPI.
import spidev
spi = spidev.SpiDev()
spi.open(0, 0) # bus 0, device 0
spi.max_speed_hz = 1_350_000
def read_channel(channel):
cmd = [1, (8 + channel) << 4, 0]
reply = spi.xfer2(cmd)
return ((reply[1] & 3) << 8) | reply[2] # 10-bit value, 0 to 1023
print("channel 0 raw:", read_channel(0))
spi.close()
Este exemplo precisa de um Raspberry Pi, de um serviço de rede ou de um serviço do sistema, então foi verificado apenas quanto à sintaxe. Nenhuma saída é mostrada.
Exemplo 63: Temperatura DS18B20 pelo 1-Wire
Leia um sensor de temperatura à prova d'água pelo kernel.
from pathlib import Path
import glob
device_dirs = glob.glob("/sys/bus/w1/devices/28-*")
if not device_dirs:
raise SystemExit("no DS18B20 found: enable 1-Wire in raspi-config")
lines = (Path(device_dirs[0]) / "w1_slave").read_text().splitlines()
if lines[0].endswith("YES"):
milli = int(lines[1].split("t=")[1])
print(f"{milli / 1000:.3f} C")
Este exemplo precisa de um Raspberry Pi, de um serviço de rede ou de um serviço do sistema, então foi verificado apenas quanto à sintaxe. Nenhuma saída é mostrada.
Exemplo 64: Novas tentativas para sensores instáveis
Tente a leitura de novo antes de desistir.
import time
def read_with_retries(read_fn, attempts=3, delay=1.0):
for attempt in range(1, attempts + 1):
try:
return read_fn()
except OSError as e:
print(f"attempt {attempt} failed: {e}")
time.sleep(delay)
raise RuntimeError("sensor unavailable after retries")
Este exemplo precisa de um Raspberry Pi, de um serviço de rede ou de um serviço do sistema, então foi verificado apenas quanto à sintaxe. Nenhuma saída é mostrada.
Exemplo 65: Calibração salva em JSON
Guarde os valores de calibração em um arquivo e recarregue-os.
import json
from pathlib import Path
CONFIG = Path("calibration.json")
DEFAULTS = {"offset_c": 0.0, "scale": 1.0}
def load():
if CONFIG.exists():
return {**DEFAULTS, **json.loads(CONFIG.read_text(encoding="utf-8"))}
return dict(DEFAULTS)
def save(cfg):
CONFIG.write_text(json.dumps(cfg, indent=2), encoding="utf-8")
cfg = load()
cfg["offset_c"] = -0.4
save(cfg)
print("saved:", load())
CONFIG.unlink()
Saída:
saved: {'offset_c': -0.4, 'scale': 1.0}
Exemplo 66: Tirando uma foto
Capture uma imagem com o módulo de câmera do Pi.
from picamera2 import Picamera2
from time import sleep
cam = Picamera2()
cam.configure(cam.create_still_configuration())
cam.start()
sleep(2) # let the auto-exposure settle
cam.capture_file("photo.jpg")
cam.stop()
Este exemplo precisa de um Raspberry Pi, de um serviço de rede ou de um serviço do sistema, então foi verificado apenas quanto à sintaxe. Nenhuma saída é mostrada.
Exemplo 67: Detecção de movimento a partir de quadros
Compare quadros para perceber movimento.
import numpy as np
from picamera2 import Picamera2
cam = Picamera2()
cam.configure(cam.create_preview_configuration(main={"size": (320, 240), "format": "YUV420"}))
cam.start()
previous = None
for _ in range(20):
frame = cam.capture_array()[:240].astype(np.int16)
if previous is not None and np.mean(np.abs(frame - previous)) > 10:
print("motion")
previous = frame
cam.stop()
Este exemplo precisa de um Raspberry Pi, de um serviço de rede ou de um serviço do sistema, então foi verificado apenas quanto à sintaxe. Nenhuma saída é mostrada.
Tópico 10: Redes com Sockets e HTTP
Um dispositivo de IoT precisa conversar com outros computadores. Os sockets são a base de baixo nível, o HTTP é como funcionam os serviços web e as APIs, e um código robusto trata tempos limite, novas tentativas e períodos sem conexão.
Sockets TCP e UDP
Um socket TCP é uma conexão confiável, e o UDP envia mensagens isoladas sem conexão. Associar à porta 0 faz o sistema operacional escolher uma porta livre, o que evita conflitos entre os exemplos.
Requisições HTTP com tempo limite
O requests faz chamadas HTTP em uma linha. Defina sempre um tempo limite, ou uma rede que não responde vai travar o programa. O raise_for_status transforma uma resposta de erro em uma exceção que você pode tratar.
Novas tentativas, espera exponencial e filas offline
Repita as requisições que falharam com recuo exponencial, esperando mais a cada vez, para não sobrecarregar um servidor ocupado. Quando a rede cai, guarde as leituras em uma fila e envie-as quando a conexão voltar.
Exemplo 68: Servidor e cliente TCP de eco
Envie bytes a um servidor e leia a resposta.
import socket
import threading
def serve(server):
conn, _ = server.accept()
with conn:
data = conn.recv(1024)
conn.sendall(b"echo: " + data)
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind(("127.0.0.1", 0)) # port 0: let the OS choose a free port
server.listen(1)
port = server.getsockname()[1]
thread = threading.Thread(target=serve, args=(server,))
thread.start()
with socket.create_connection(("127.0.0.1", port)) as client:
client.sendall(b"temp=21.5")
print(client.recv(1024).decode())
thread.join()
server.close()
Saída:
echo: temp=21.5
Exemplo 69: Mensagens UDP no loopback
Envie um datagrama sem estabelecer conexão.
import socket
receiver = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
receiver.bind(("127.0.0.1", 0))
addr = receiver.getsockname()
sender = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sender.sendto(b"humidity=48", addr)
data, _ = receiver.recvfrom(1024)
print("received:", data.decode())
sender.close()
receiver.close()
Saída:
received: humidity=48
Exemplo 70: POST de leituras para um serviço web
Envie uma leitura do sensor como JSON por HTTPS.
import requests
reading = {"device": "node-1", "temperature_c": 21.5}
response = requests.post(
"https://api.example.com/readings",
json=reading,
timeout=5,
)
response.raise_for_status()
print("server said:", response.status_code)
Este exemplo precisa de um Raspberry Pi, de um serviço de rede ou de um serviço do sistema, então foi verificado apenas quanto à sintaxe. Nenhuma saída é mostrada.
Exemplo 71: GET de JSON de uma API com tempo limite
Busque dados e trate falhas sem travar.
import requests
try:
r = requests.get("https://api.example.com/forecast", params={"city": "Bangkok"}, timeout=5)
r.raise_for_status()
data = r.json()
print(data.get("temp_c"))
except requests.RequestException as e:
print("request failed:", e)
Este exemplo precisa de um Raspberry Pi, de um serviço de rede ou de um serviço do sistema, então foi verificado apenas quanto à sintaxe. Nenhuma saída é mostrada.
Exemplo 72: Nova tentativa com recuo exponencial
Espere mais a cada tentativa que falhar.
import random
import time
def with_backoff(fn, tries=4, base=0.1):
for n in range(tries):
try:
return fn()
except ConnectionError:
wait = base * (2 ** n)
print(f"try {n + 1} failed, waiting {wait:.1f}s")
time.sleep(wait)
raise ConnectionError("gave up")
random.seed(3)
calls = {"n": 0}
def flaky():
calls["n"] += 1
if calls["n"] < 3:
raise ConnectionError("network down")
return "uploaded"
print(with_backoff(flaky))
Saída:
try 1 failed, waiting 0.1s
try 2 failed, waiting 0.2s
uploaded
Exemplo 73: Enviando JSON com urllib
Envie a requisição usando apenas a biblioteca padrão.
import json
import urllib.request
body = json.dumps({"device": "node-1", "humidity_pct": 48}).encode()
req = urllib.request.Request(
"https://api.example.com/readings",
data=body,
headers={"Content-Type": "application/json"},
method="POST",
)
with urllib.request.urlopen(req, timeout=5) as resp:
print("status", resp.status)
Este exemplo precisa de um Raspberry Pi, de um serviço de rede ou de um serviço do sistema, então foi verificado apenas quanto à sintaxe. Nenhuma saída é mostrada.
Exemplo 74: Um servidor web de status
Sirva uma pequena página de status no Pi.
from http.server import BaseHTTPRequestHandler, HTTPServer
import json
class Status(BaseHTTPRequestHandler):
def do_GET(self):
body = json.dumps({"status": "ok", "uptime_s": 42}).encode()
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(body)
HTTPServer(("0.0.0.0", 8000), Status).serve_forever()
Este exemplo precisa de um Raspberry Pi, de um serviço de rede ou de um serviço do sistema, então foi verificado apenas quanto à sintaxe. Nenhuma saída é mostrada.
Exemplo 75: Verificando a rede antes de enviar
Guarde os dados em fila quando a internet cair.
import socket
def internet_ok(host="1.1.1.1", port=53, timeout=3):
try:
with socket.create_connection((host, port), timeout=timeout):
return True
except OSError:
return False
print("online" if internet_ok() else "offline, queue the data")
Este exemplo precisa de um Raspberry Pi, de um serviço de rede ou de um serviço do sistema, então foi verificado apenas quanto à sintaxe. Nenhuma saída é mostrada.
Tópico 11: MQTT: Publicar e Assinar
O MQTT é o protocolo de mensagens padrão da IoT. Os dispositivos publicam leituras em tópicos, um broker as entrega, e qualquer número de assinantes as recebe. É leve o bastante para dispositivos pequenos e tolerante a redes fracas.
Tópicos, publicação e assinatura
Um tópico é um caminho como home/kitchen/temperature. Publicar envia uma mensagem para um tópico por meio do broker. Assine usando curingas: + corresponde a um nível e # corresponde a tudo abaixo dele.
Qualidade de serviço e mensagens retidas
O QoS 0 é enviar e esquecer, o QoS 1 garante a entrega pelo menos uma vez, e o QoS 2 entrega exatamente uma vez. Uma mensagem retida é guardada pelo broker, então um novo assinante recebe imediatamente o último valor.
Last will, TLS e descoberta
A mensagem last will é publicada pelo broker quando um dispositivo cai sem aviso, o que permite detectar dispositivos offline. Use TLS e uma senha vinda de variáveis de ambiente para um broker na internet. O Home Assistant encontra sensores por uma mensagem de descoberta enviada a um tópico especial.
Exemplo 76: MQTT: publicando telemetria
Envie leituras a um broker em um tópico.
import json
import time
import paho.mqtt.client as mqtt
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="node-1")
client.connect("broker.local", 1883, keepalive=60)
client.loop_start()
payload = {"temp_c": 21.5, "ts": int(time.time())}
client.publish("home/kitchen/temperature", json.dumps(payload), qos=1)
time.sleep(1)
client.loop_stop()
client.disconnect()
Este exemplo precisa de um Raspberry Pi, de um serviço de rede ou de um serviço do sistema, então foi verificado apenas quanto à sintaxe. Nenhuma saída é mostrada.
Exemplo 77: MQTT: assinando mensagens
Receba leituras de muitos dispositivos com curingas.
import paho.mqtt.client as mqtt
def on_connect(client, userdata, flags, reason_code, properties):
client.subscribe("home/+/temperature")
def on_message(client, userdata, msg):
print(msg.topic, "->", msg.payload.decode())
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.local", 1883)
client.loop_forever()
Este exemplo precisa de um Raspberry Pi, de um serviço de rede ou de um serviço do sistema, então foi verificado apenas quanto à sintaxe. Nenhuma saída é mostrada.
Exemplo 78: MQTT: QoS e mensagens retidas
Escolha garantias de entrega e mantenha o último valor.
import paho.mqtt.client as mqtt
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
client.connect("broker.local", 1883)
client.loop_start()
# QoS 0 = at most once, 1 = at least once, 2 = exactly once
client.publish("home/door/state", "closed", qos=1, retain=True) # last value is kept
client.loop_stop()
client.disconnect()
Este exemplo precisa de um Raspberry Pi, de um serviço de rede ou de um serviço do sistema, então foi verificado apenas quanto à sintaxe. Nenhuma saída é mostrada.
Exemplo 79: MQTT: last will para detectar offline
Avise a todos quando um dispositivo cair.
import paho.mqtt.client as mqtt
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="node-1")
client.will_set("home/node-1/status", payload="offline", qos=1, retain=True)
client.publish("home/node-1/status", "online", qos=1, retain=True) # set on connect
client.connect("broker.local", 1883)
client.loop_forever()
Este exemplo precisa de um Raspberry Pi, de um serviço de rede ou de um serviço do sistema, então foi verificado apenas quanto à sintaxe. Nenhuma saída é mostrada.
Exemplo 80: MQTT sobre TLS
Criptografe a conexão com o broker.
import ssl
import paho.mqtt.client as mqtt
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
client.tls_set(ca_certs="/etc/ssl/certs/ca-certificates.crt",
tls_version=ssl.PROTOCOL_TLS_CLIENT)
client.username_pw_set("node-1", "use-a-secret-from-env-not-code")
client.connect("broker.example.com", 8883)
client.loop_forever()
Este exemplo precisa de um Raspberry Pi, de um serviço de rede ou de um serviço do sistema, então foi verificado apenas quanto à sintaxe. Nenhuma saída é mostrada.
Exemplo 81: Mensagem de descoberta do Home Assistant
Publique um sensor para que o Home Assistant o encontre.
import json
config = {
"name": "Kitchen temperature",
"unique_id": "node-1_kitchen_temp",
"state_topic": "home/kitchen/temperature",
"unit_of_measurement": "°C",
"device_class": "temperature",
"value_template": "{{ value_json.temp_c }}",
}
topic = "homeassistant/sensor/node-1_kitchen_temp/config"
print(topic)
print(json.dumps(config, ensure_ascii=False, indent=2))
Saída:
homeassistant/sensor/node-1_kitchen_temp/config
{
"name": "Kitchen temperature",
"unique_id": "node-1_kitchen_temp",
"state_topic": "home/kitchen/temperature",
"unit_of_measurement": "°C",
"device_class": "temperature",
"value_template": "{{ value_json.temp_c }}"
}
Exemplo 82: Curingas de tópicos MQTT
Verifique quais tópicos uma assinatura alcança.
def topic_matches(pattern, topic):
p_parts = pattern.split("/")
t_parts = topic.split("/")
for i, part in enumerate(p_parts):
if part == "#":
return True
if i >= len(t_parts):
return False
if part != "+" and part != t_parts[i]:
return False
return len(p_parts) == len(t_parts)
tests = [
("home/+/temperature", "home/kitchen/temperature", True),
("home/#", "home/kitchen/humidity", True),
("home/+/temperature", "home/kitchen/humidity", False),
]
for pattern, topic, expected in tests:
matched = topic_matches(pattern, topic)
print(pattern, topic, "match" if matched else "no match", "(as expected)" if matched == expected else "(UNEXPECTED)")
Saída:
home/+/temperature home/kitchen/temperature match (as expected)
home/# home/kitchen/humidity match (as expected)
home/+/temperature home/kitchen/humidity no match (as expected)
Tópico 12: Painéis Web e APIs com Flask
O Flask é um pequeno framework web em Python. Com poucas linhas, ele serve uma página de status, devolve JSON ou aceita leituras de outros dispositivos, então você pode ver o seu Pi pelo celular ou pelo navegador.
Rotas e respostas
Um decorador de rota, como @app.get('/'), liga uma URL a uma função. A função devolve texto, uma página ou JSON com jsonify. Use o endereço 0.0.0.0 para aceitar conexões da rede local.
Recebendo dados com validação
Uma rota POST lê JSON da requisição. Valide o tipo e a faixa antes de guardar qualquer coisa, e devolva o status 400 quando os dados estiverem errados, para que quem envia saiba o que corrigir.
Threads em segundo plano e painéis
Uma thread em segundo plano lê o sensor enquanto o servidor web responde às requisições. Mantenha os dados compartilhados simples e seguros. Um pequeno template HTML transforma as leituras mais recentes em uma página que qualquer pessoa na rede pode ver.
Exemplo 83: Endpoint hello com Flask
Sirva a primeira rota web no Pi.
from flask import Flask
app = Flask(__name__)
@app.get("/")
def hello():
return "Raspberry Pi is alive"
if __name__ == "__main__":
app.run(host="0.0.0.0", port=5000)
Este exemplo precisa de um Raspberry Pi, de um serviço de rede ou de um serviço do sistema, então foi verificado apenas quanto à sintaxe. Nenhuma saída é mostrada.
Exemplo 84: Endpoint JSON com Flask
Retorne a leitura mais recente como JSON.
from flask import Flask, jsonify
app = Flask(__name__)
latest = {"temp_c": 21.5, "humidity_pct": 48}
@app.get("/api/latest")
def get_latest():
return jsonify(latest)
app.run(host="0.0.0.0", port=5000)
Este exemplo precisa de um Raspberry Pi, de um serviço de rede ou de um serviço do sistema, então foi verificado apenas quanto à sintaxe. Nenhuma saída é mostrada.
Exemplo 85: POST com Flask e validação
Aceite leituras e rejeite valores ruins com um código de erro.
from flask import Flask, request, jsonify
app = Flask(__name__)
readings = []
@app.post("/api/readings")
def add_reading():
data = request.get_json(silent=True) or {}
temp = data.get("temp_c")
if not isinstance(temp, (int, float)) or not -40 <= temp <= 85:
return jsonify(error="temp_c must be a number from -40 to 85"), 400
readings.append(temp)
return jsonify(saved=len(readings)), 201
app.run(port=5000)
Este exemplo precisa de um Raspberry Pi, de um serviço de rede ou de um serviço do sistema, então foi verificado apenas quanto à sintaxe. Nenhuma saída é mostrada.
Exemplo 86: Um painel HTML pequeno
Mostre uma página com a temperatura mais recente.
from flask import Flask, render_template_string
app = Flask(__name__)
PAGE = """
<h1>Kitchen</h1>
<p>Temperature: {{ temp }} °C</p>
"""
@app.get("/")
def home():
return render_template_string(PAGE, temp=21.5)
app.run(port=5000)
Este exemplo precisa de um Raspberry Pi, de um serviço de rede ou de um serviço do sistema, então foi verificado apenas quanto à sintaxe. Nenhuma saída é mostrada.
Exemplo 87: Flask com uma thread de sensor em segundo plano
Leia o sensor continuamente enquanto o servidor web roda.
import threading
import time
from flask import Flask, jsonify
app = Flask(__name__)
state = {"temp_c": None}
def sensor_loop():
while True:
state["temp_c"] = 21.5 # replace with the real sensor read
time.sleep(5)
threading.Thread(target=sensor_loop, daemon=True).start()
@app.get("/api/temp")
def temp():
return jsonify(state)
app.run(port=5000)
Este exemplo precisa de um Raspberry Pi, de um serviço de rede ou de um serviço do sistema, então foi verificado apenas quanto à sintaxe. Nenhuma saída é mostrada.
Tópico 13: Guardando e Resumindo Leituras
Um dispositivo que registra durante um ano precisa de um armazenamento simples que não encha o cartão SD. O SQLite é um banco de dados em um único arquivo que vem com o Python, e os arquivos CSV servem para um histórico simples.
Tabelas e inserções no SQLite
O SQLite guarda as tabelas em um único arquivo. Use os marcadores ? nos comandos INSERT, que protegem contra injeção de SQL e cuidam das aspas. Chame commit depois de escrever para que os dados sejam salvos.
Agregações e intervalos de tempo
GROUP BY com AVG e COUNT resume as leituras por dispositivo. Agrupar os carimbos de tempo por minuto, por exemplo pegando os cinco primeiros caracteres de um horário ISO, dá médias simples de um minuto.
Retenção e registros somente de acréscimo
Apague linhas antigas em uma programação para manter o armazenamento pequeno. Um arquivo CSV somente de acréscimo adiciona uma linha por leitura e escreve o cabeçalho uma única vez, o que é simples e robusto.
Exemplo 88: SQLite: criar tabela e inserir
Guarde leituras em um arquivo de banco de dados.
import sqlite3
conn = sqlite3.connect(":memory:") # use "readings.db" to keep data on the Pi
conn.execute("""CREATE TABLE readings (
ts TEXT NOT NULL, device TEXT NOT NULL, temp_c REAL NOT NULL)""")
rows = [("2026-01-15T08:00", "node-1", 21.5), ("2026-01-15T08:05", "node-1", 21.9)]
conn.executemany("INSERT INTO readings VALUES (?, ?, ?)", rows)
conn.commit()
print(conn.execute("SELECT COUNT(*) FROM readings").fetchone()[0], "rows stored")
conn.close()
Saída:
2 rows stored
Exemplo 89: SQLite: consultas agregadas
Calcule a média e a quantidade de leituras por dispositivo.
import sqlite3
conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE readings (device TEXT, temp_c REAL)")
conn.executemany("INSERT INTO readings VALUES (?, ?)",
[("node-1", 21.5), ("node-1", 22.5), ("node-2", 19.0)])
for device, avg, n in conn.execute(
"SELECT device, AVG(temp_c), COUNT(*) FROM readings GROUP BY device ORDER BY device"):
print(f"{device}: average {avg:.2f} from {n} readings")
conn.close()
Saída:
node-1: average 22.00 from 2 readings
node-2: average 19.00 from 1 readings
Exemplo 90: Registrador de histórico em CSV
Acrescente leituras a um arquivo com cabeçalho.
import csv
from pathlib import Path
LOG = Path("sensor_history.csv")
FIELDS = ["ts", "temp_c"]
def log_reading(ts, temp_c):
new_file = not LOG.exists()
with LOG.open("a", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=FIELDS)
if new_file:
writer.writeheader()
writer.writerow({"ts": ts, "temp_c": temp_c})
log_reading("08:00", 21.5)
log_reading("08:05", 21.9)
print(LOG.read_text(encoding="utf-8"))
LOG.unlink()
Saída:
ts,temp_c
08:00,21.5
08:05,21.9
Exemplo 91: Agrupando leituras por minuto
Reúna leituras em médias de um minuto.
from collections import defaultdict
from statistics import mean
readings = [("08:00:10", 21.4), ("08:00:40", 21.6), ("08:01:05", 21.9)]
buckets = defaultdict(list)
for ts, value in readings:
buckets[ts[:5]].append(value) # keep HH:MM
for minute in sorted(buckets):
print(minute, round(mean(buckets[minute]), 2))
Saída:
08:00 21.5
08:01 21.9
Exemplo 92: Retenção de dados
Apague linhas antigas para manter pequeno o armazenamento do Pi.
import sqlite3
conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE readings (day INTEGER, temp_c REAL)")
conn.executemany("INSERT INTO readings VALUES (?, ?)", [(d, 20.0 + d / 10) for d in range(1, 11)])
keep_from = 6
deleted = conn.execute("DELETE FROM readings WHERE day < ?", (keep_from,)).rowcount
conn.commit()
remaining = conn.execute("SELECT COUNT(*) FROM readings").fetchone()[0]
print("deleted", deleted, "kept", remaining)
conn.close()
Saída:
deleted 5 kept 5
Tópico 14: Executando como Serviço e Configuração
Um projeto só é útil se iniciar sozinho depois de uma queda de energia e continuar funcionando. O systemd executa serviços na inicialização, o cron executa tarefas em um horário, e as variáveis de ambiente mantêm as configurações fora do código.
Serviços systemd
Um arquivo de unidade descreve como executar o programa, a pasta de trabalho, o usuário e o que fazer em caso de falha com Restart=on-failure. Ative o serviço para que ele inicie na inicialização, e leia os logs com journalctl.
Tarefas agendadas com cron
Uma linha do crontab executa um comando em horários fixos, como a cada cinco minutos. Redirecione a saída para um arquivo de log para saber o que aconteceu enquanto você estava fora.
Configuração pelo ambiente
Leia configurações como o ID do dispositivo, o intervalo de envio e o endereço do broker a partir de variáveis de ambiente com valores padrão seguros. Você pode alterá-las no arquivo de serviço sem editar o código.
Exemplo 93: Serviço systemd na inicialização
Execute o registrador automaticamente quando o Pi ligar.
# /etc/systemd/system/sensor-logger.service
[Unit]
Description=Sensor logger
After=network-online.target
Wants=network-online.target
[Service]
ExecStart=/usr/bin/python3 /home/pi/iot/logger.py
WorkingDirectory=/home/pi/iot
Restart=on-failure
RestartSec=5
User=pi
[Install]
WantedBy=multi-user.target
# enable and start it:
# sudo systemctl daemon-reload
# sudo systemctl enable --now sensor-logger.service
# journalctl -u sensor-logger -f
Este exemplo precisa de um Raspberry Pi, de um serviço de rede ou de um serviço do sistema, então foi verificado apenas quanto à sintaxe. Nenhuma saída é mostrada.
Exemplo 94: Tarefas agendadas com cron
Execute um script a cada cinco minutos.
# edit the schedule with: crontab -e
# every 5 minutes, record the time and the readings file size
*/5 * * * * /usr/bin/python3 /home/pi/iot/snapshot.py >> /home/pi/iot/cron.log 2>&1
# every day at 02:30, remove logs older than 30 days
30 2 * * * find /home/pi/iot/logs -name "*.csv" -mtime +30 -delete
Este exemplo precisa de um Raspberry Pi, de um serviço de rede ou de um serviço do sistema, então foi verificado apenas quanto à sintaxe. Nenhuma saída é mostrada.
Exemplo 95: Arquivo de heartbeat para um watchdog
Grave um carimbo de tempo para o watchdog saber que o programa está vivo.
import time
from pathlib import Path
HEARTBEAT = Path("heartbeat.txt")
def beat(count):
HEARTBEAT.write_text(f"{count} {time.time():.0f}", encoding="utf-8")
for i in range(3):
beat(i)
print("beat", i, "written")
time.sleep(0.05)
print("last:", HEARTBEAT.read_text(encoding="utf-8").split()[0])
HEARTBEAT.unlink()
Saída:
beat 0 written
beat 1 written
beat 2 written
last: 2
Exemplo 96: Configuração por variáveis de ambiente
Leia as configurações com valores padrão seguros.
import os
def get_config():
return {
"device_id": os.environ.get("DEVICE_ID", "node-1"),
"interval_s": int(os.environ.get("INTERVAL_S", "60")),
"broker": os.environ.get("MQTT_HOST", "localhost"),
}
print(get_config())
Saída:
{'device_id': 'node-1', 'interval_s': 60, 'broker': 'localhost'}
Tópico 15: Logs, Segurança e um Pipeline Completo
Os últimos tópicos reúnem tudo: logs com rotação para que o cartão SD não encha, segredos mantidos fora do código, validação dos dados recebidos e um pipeline completo do sensor até o alarme.
Logs com rotação e heartbeats
Um manipulador de arquivo com rotação limita o tamanho de cada arquivo de log e guarda alguns arquivos antigos. Um arquivo de heartbeat com carimbo de tempo permite que um watchdog verifique se o programa continua vivo.
Segredos e validação
Guarde senhas e tokens em variáveis de ambiente e nunca os imprima. Valide cada valor recebido quanto ao tipo e à faixa antes que ele chegue ao armazenamento ou a um alarme, porque sensores e redes podem enviar qualquer coisa.
Montando o pipeline
Um pipeline lê um sensor, suaviza o valor com uma janela, compara com um limite, levanta um sinal de alarme e guarda cada etapa. Cada parte é pequena e testável, e é assim que o código profissional de IoT é organizado.
Exemplo 97: Arquivos de log rotativos
Impeça que os logs encham o cartão SD.
import logging
from logging.handlers import RotatingFileHandler
from pathlib import Path
path = Path("iot.log")
handler = RotatingFileHandler(path, maxBytes=200, backupCount=2, encoding="utf-8")
log = logging.getLogger("iot")
log.setLevel(logging.INFO)
log.addHandler(handler)
for i in range(20):
log.info("reading %d: temp=21.%d", i, i % 10)
handler.close()
print("files:", sorted(p.name for p in Path(".").glob("iot.log*")))
for p in Path(".").glob("iot.log*"):
p.unlink()
Saída:
files: ['iot.log', 'iot.log.1', 'iot.log.2']
Exemplo 98: Segredos vindos do ambiente
Carregue senhas do ambiente e nunca as imprima.
import os
def require_secret(name):
value = os.environ.get(name)
if not value:
raise SystemExit(f"set {name} in the environment (do not write it in the code)")
return value
def mask(secret):
return secret[:2] + "*" * (len(secret) - 2)
os.environ["MQTT_PASSWORD"] = "example-secret" # demo only
token = require_secret("MQTT_PASSWORD")
print("loaded secret:", mask(token))
Saída:
loaded secret: ex************
Exemplo 99: Validando dados de sensor recebidos
Verifique tipo e faixa antes de guardar a leitura.
from dataclasses import dataclass
@dataclass
class Sample:
device: str
temp_c: float
def validate(raw):
errors = []
if not isinstance(raw.get("device"), str) or not raw["device"]:
errors.append("device is required")
temp = raw.get("temp_c")
if not isinstance(temp, (int, float)):
errors.append("temp_c must be a number")
elif not -40 <= temp <= 85:
errors.append("temp_c out of sensor range")
if errors:
return None, errors
return Sample(raw["device"], float(temp)), []
print(validate({"device": "node-1", "temp_c": 21.5}))
print(validate({"device": "", "temp_c": 300}))
Saída:
(Sample(device='node-1', temp_c=21.5), [])
(None, ['device is required', 'temp_c out of sensor range'])
Exemplo 100: Um pipeline pequeno e completo
Leia, suavize, detecte um alarme e armazene os resultados.
import sqlite3
from collections import deque
def fake_sensor(n):
base = [21.0, 21.2, 21.1, 35.0, 21.3, 21.4] # 35.0 is a spike
return base[n % len(base)]
window = deque(maxlen=3)
conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE readings (tick INTEGER, raw REAL, smooth REAL, alarm INTEGER)")
for tick in range(6):
raw = fake_sensor(tick)
window.append(raw)
smooth = sum(window) / len(window)
alarm = 1 if smooth > 25 else 0
conn.execute("INSERT INTO readings VALUES (?, ?, ?, ?)", (tick, raw, smooth, alarm))
print(f"tick {tick}: raw {raw:5.1f} smooth {smooth:5.2f} alarm {alarm}")
alarms = conn.execute("SELECT COUNT(*) FROM readings WHERE alarm = 1").fetchone()[0]
print("alarm ticks:", alarms)
conn.close()
Saída:
tick 0: raw 21.0 smooth 21.00 alarm 0
tick 1: raw 21.2 smooth 21.10 alarm 0
tick 2: raw 21.1 smooth 21.10 alarm 0
tick 3: raw 35.0 smooth 25.77 alarm 1
tick 4: raw 21.3 smooth 25.80 alarm 1
tick 5: raw 21.4 smooth 25.90 alarm 1
alarm ticks: 3
Projeto Prático: Monitor de Ambiente
Construa um monitor que use ideias de cada seção. Escreva você mesmo primeiro e depois compare com os exemplos. Teste as partes em Python puro no seu computador antes de conectar o hardware.
1. Escreva uma função que converta leituras brutas em temperatura usando dois pontos de calibração e teste-a com três valores.
2. Suavize as leituras com uma média móvel dos últimos cinco valores e adicione histerese para um ventilador com limites de 24 e 22.
3. Guarde cada leitura em uma tabela SQLite e imprima a média por minuto depois de dez leituras.
4. Publique a leitura mais recente em um tópico MQTT a cada 60 segundos, e quando o broker estiver offline, coloque as leituras em uma fila.
5. Crie um serviço systemd que inicie o monitor na inicialização e leia os logs com journalctl.
Próximos passos
Em seguida, aprenda uma plataforma de IoT como o Home Assistant ou o Node-RED, adicione identidade segura de dispositivos e atualizações over-the-air, e estude o Raspberry Pi Pico para trabalhar com microcontroladores. Teste cada exemplo de hardware com a sua própria fiação primeiro, e confira a documentação oficial do Raspberry Pi para os números de pinos do seu modelo.
Resumo
Um Raspberry Pi se torna um dispositivo de IoT de verdade quando lê sensores de forma confiável, processa os dados, os guarda, os envia pela rede e funciona sem ajuda. Ao longo de 100 exemplos, este artigo cobriu os fundamentos do Python, decisões e repetições, funções, estruturas de dados, arquivos e formatos, tratamento de erros e registro de logs, processamento de sensores, GPIO, I2C, SPI, 1-Wire e a câmera, sockets e HTTP, MQTT, Flask, SQLite, systemd e cron, e um pipeline completo. Construa o monitor de ambiente do projeto prático e mude uma coisa de cada vez para entender como cada parte se comporta.