Objetivo y caso de uso

Qué construirás: Un controlador de invernadero utilizando Raspberry Pi 3 B+, Waveshare RS485 HAT y sensor Bosch BME680 para monitoreo y automatización ambiental.

Para qué sirve

• Monitoreo de temperatura y humedad en tiempo real utilizando el BME680.
• Control de sistemas de riego automatizados mediante comandos Modbus RS485.
• Integración con plataformas IoT a través de MQTT para visualización remota.
• Gestión de alertas por condiciones ambientales críticas (temperaturas extremas).

Resultado esperado

• Latencia de respuesta del sistema de riego menor a 200 ms.
• Monitoreo de datos ambientales con una frecuencia de 1 Hz.
• Envío de datos a la nube con un throughput de 5 paquetes/s.
• Alertas generadas en tiempo real en caso de condiciones fuera de rango.

Público objetivo: Ingenieros y desarrolladores de sistemas embebidos; Nivel: Avanzado

Arquitectura/flujo: Comunicación entre Raspberry Pi y dispositivos RS485 mediante el HAT de Waveshare, utilizando el bus I2C para el sensor BME680.

Nivel: Avanzado

Prerrequisitos

• Sistema operativo:
• Raspberry Pi OS Bookworm 64‑bit (release 2024-10-22 o posterior)
• Kernel Linux 6.x incluido en la imagen oficial (no es necesario fijar versión exacta)
• Toolchain y versiones exactas:
• Python 3.11 (intérprete del sistema)
• Módulo de entornos virtuales: python3.11-venv (APT)
• pip 24.2 (actualizado dentro del venv)
• Paquetes APT:
• i2c-tools 4.3-2
• python3-gpiozero 1.6.2-1
• python3-smbus 5.1-1 (opcional si no usas smbus2 vía pip)
• Paquetes pip (versiones fijadas en el entorno virtual):
• pyserial==3.5
• minimalmodbus==2.1.1
• smbus2==0.5.0
• bme680==1.1.1
• gpiozero==1.6.2
• typer==0.12.5 (CLI opcional clara)
• rich==13.9.3 (logging legible opcional)
• Acceso y permisos:
• Usuario perteneciente a los grupos dialout e i2c
• Acceso SSH o consola local

Notas importantes de compatibilidad:
– Para el control RS485 half‑duplex del SP3485 del HAT de Waveshare utilizaremos la línea RTS (GPIO17) como señal DE/RE. La conmutación de dirección se hará en usuario con pyserial.rs485.RS485Settings.
– El BME680 se usará por I2C en el bus 1 (SDA GPIO2, SCL GPIO3).

Materiales

• Computadora principal:
• Raspberry Pi 3 Model B+ (exactamente este modelo)
• Interfaces y sensores:
• Waveshare RS485 HAT (SP3485)
• Bosch BME680 (breakout I2C 3.3 V)
• Otros:
• Fuente 5 V/2.5 A para Raspberry Pi 3 Model B+
• Cables Dupont macho‑hembra para el BME680 (SDA/SCL/3V3/GND)
• Destornillador pequeño para borneras del RS485 HAT
• Resistencias de terminación si no vienen en el HAT (el HAT suele integrar jumper de 120 Ω)
• Un módulo/esclavo Modbus RTU real en bus RS485 (por ejemplo, un relé Modbus de 4 canales o un módulo de 8 relés, ID=1), y opcionalmente un módulo de entradas analógicas (ID=2) para simular humedad de suelo
• PC o la misma Raspberry con cable de red/wi‑fi para instalar paquetes

Objetivo del proyecto:
– Proyecto “rs485-modbus-greenhouse-control”: el Pi mide T/RH/Presión/Calidad de aire con el BME680 y gobierna actuadores en bus RS485‑Modbus (ventilación, riego, calefacción, nebulización) según consignas.

Preparación y conexión

1) Habilitar interfaces en Raspberry Pi OS Bookworm (64‑bit)

Opción A: usando raspi-config (interactivo):
– sudo raspi-config
– Interface Options:
– I2C: Enable
– Serial Port:
– Login shell over serial? No
– Enable serial interface? Yes
– Finish y reboot

Opción B: edición directa de /boot/firmware/config.txt:
– Edita el fichero:
– sudo nano /boot/firmware/config.txt
– Añade/asegura estas líneas (al final del archivo, una por línea):
– enable_uart=1
– dtoverlay=pi3-disable-bt
– dtparam=i2c_arm=on
– dtoverlay=uart0,ctsrts=on
– Guarda y sal; desactiva servicios que usen la UART:
– sudo systemctl disable –now hciuart.service || true
– Asegura pertenencia a grupos:
– sudo usermod -aG dialout,i2c $USER
– Reinicia:
– sudo reboot

Tras el reinicio, comprueba:
– ls -l /dev/serial0 → debe apuntar a /dev/ttyAMA0 en Pi 3 B+ con Bluetooth deshabilitado (PL011 estable)
– i2cdetect -y 1 → debe mostrar 0x76 o 0x77 (BME680)

2) Jumper/DIP del Waveshare RS485 HAT (SP3485)

• Terminación 120 Ω: habilita el jumper de terminación si el HAT es el único en el extremo del bus (ON).
• Polarización (bias): si tu bus no la provee, habilita los jumpers de pull‑up/pull‑down (si los trae).
• Control de dirección: sitúa el jumper DE/RE en la posición “RTS” (esto conecta DE/RE del SP3485 a la línea RTS del PL011, que en el Pi está en GPIO17).
• UART: el HAT usa GPIO14 (TXD) y GPIO15 (RXD) automáticamente al enchufarlo sobre el conector de 40 pines.

Conexión del bus:
– Bornera A(+) y B(−) del HAT a A/B de tu red RS485 de invernadero.
– GND de referencia: conecta GND si la topología lo requiere (recomendado para distancias largas o fuentes distintas).

3) Cableado del Bosch BME680 (I2C)

Conecta el módulo BME680 a la cabecera del Pi (niveles a 3.3 V):
– VCC del BME680 → 3V3 (Pin físico 1)
– GND del BME680 → GND (Pin físico 6 o 9, etc.)
– SDA del BME680 → GPIO2/SDA1 (Pin físico 3)
– SCL del BME680 → GPIO3/SCL1 (Pin físico 5)
– Dirección I2C: por defecto suele ser 0x76. Si el pin SDO está a VCC, será 0x77.

4) Mapa de pines/puertos

Comprobaciones rápidas:
– i2cdetect -y 1 → 0x76 u 0x77
– raspi-gpio get 17 → confirmará que el pin existe (el estado cambiará dinámicamente durante transmisión)

Código completo

A continuación se presenta un script Python 3.11 completo para “rs485-modbus-greenhouse-control”. Integra:
– Lectura periódica de BME680 por I2C (temperatura, humedad, presión y gas).
– Control de actuadores Modbus RTU en bus RS485:
– Esclavo 1 (ID=1): Módulo de relés (coils) para ventilación, bomba de riego, calefacción, nebulización.
– Esclavo 2 (ID=2, opcional): Módulo de registros (holding registers) para humedad de suelo y nivel de tanque.

El control implementa una lógica simple con histéresis basada en consignas definidas por CLI. Se usa minimalmodbus con pyserial.rs485 para conmutar la dirección vía RTS (GPIO17).

Mapa lógico de Modbus utilizado

• Esclavo 1 (ID=1), módulo relés:
• Coil 0: Ventilación (fan)
• Coil 1: Bomba de riego (pump)
• Coil 2: Calefacción (heater)

• Coil 3: Nebulización (mister)

• Esclavo 2 (ID=2), módulo sensores (opcional):

• Holding Register 0 (HR0): Humedad de suelo (%) escalada x10 (p.ej., 735 = 73.5%)
• Holding Register 1 (HR1): Nivel de tanque (%) escalada x10

Ajusta estos índices a tu hardware real si difieren.

Script principal: rs485_modbus_greenhouse.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import time
import math
import logging
from dataclasses import dataclass
from typing import Optional, Tuple

import serial
from serial.rs485 import RS485Settings
import minimalmodbus
import bme680
from smbus2 import SMBus
import typer
from rich.logging import RichHandler

# -------------------------------------------
# Configuración de logging
# -------------------------------------------
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)-8s | %(message)s",
    datefmt="[%H:%M:%S]",
    handlers=[RichHandler(rich_tracebacks=True)]
)
log = logging.getLogger("greenhouse")

# -------------------------------------------
# Dataclasses de consignas y estado
# -------------------------------------------
@dataclass
class Setpoints:
    temp_target_c: float = 26.0
    rh_target_pct: float = 65.0
    vpd_target_kpa: float = 1.0
    hyst_temp_c: float = 1.0
    hyst_rh_pct: float = 5.0
    max_co2_gas_index: int = 200  # pseudoindicador basado en gas resistance

@dataclass
class Actuators:
    fan: bool = False
    pump: bool = False
    heater: bool = False
    mister: bool = False

@dataclass
class Measurements:
    temp_c: float = float("nan")
    rh_pct: float = float("nan")
    pressure_hpa: float = float("nan")
    gas_ohm: float = float("nan")
    vpd_kpa: float = float("nan")
    soil_rh_pct: Optional[float] = None
    tank_level_pct: Optional[float] = None

# -------------------------------------------
# Utilidades: VPD y heurística IAQ simplificada
# -------------------------------------------
def saturation_vapor_pressure_kpa(t_c: float) -> float:
    # Fórmula de Tetens
    return 0.61078 * math.exp((17.27 * t_c) / (t_c + 237.3))

def vpd_kpa(t_c: float, rh_pct: float) -> float:
    svp = saturation_vapor_pressure_kpa(t_c)
    avp = svp * (rh_pct / 100.0)
    return max(0.0, svp - avp)

def gas_to_index(gas_ohm: float) -> int:
    # Heurística simple (no es BSEC IAQ). Escala resistencia a 0-500 aprox.
    if gas_ohm <= 0:
        return 500
    base = max(1.0, min(gas_ohm, 1e6))
    idx = int(500 - 100 * math.log10(base))
    return max(0, min(idx, 500))

# -------------------------------------------
# Inicialización BME680
# -------------------------------------------
def init_bme680(address: int = 0x76, i2c_bus: int = 1) -> bme680.BME680:
    sensor = bme680.BME680(address=address, i2c_device=SMBus(i2c_bus))
    sensor.set_humidity_oversample(bme680.OS_2X)
    sensor.set_pressure_oversample(bme680.OS_4X)
    sensor.set_temperature_oversample(bme680.OS_8X)
    sensor.set_filter(bme680.FILTER_SIZE_3)
    sensor.set_gas_status(bme680.ENABLE_GAS_MEAS)
    sensor.set_gas_heater_temperature(320)
    sensor.set_gas_heater_duration(150)
    sensor.select_gas_heater_profile(0)
    return sensor

def read_bme680(sensor: bme680.BME680) -> Tuple[float, float, float, float]:
    if sensor.get_sensor_data():
        t = sensor.data.temperature
        rh = sensor.data.humidity
        p = sensor.data.pressure
        gas = float(sensor.data.gas_resistance) if sensor.data.heat_stable else float('nan')
        return t, rh, p, gas
    return float('nan'), float('nan'), float('nan'), float('nan')

# -------------------------------------------
# Inicialización Modbus RTU (RS485 via RTS)
# -------------------------------------------
def make_modbus_instrument(port: str, slave_id: int, baud: int, parity: str, timeout: float) -> minimalmodbus.Instrument:
    instr = minimalmodbus.Instrument(port, slave_id)
    instr.serial.baudrate = baud
    instr.serial.bytesize = 8
    instr.serial.parity = {'N': serial.PARITY_NONE, 'E': serial.PARITY_EVEN, 'O': serial.PARITY_ODD}[parity.upper()]
    instr.serial.stopbits = 1
    instr.serial.timeout = timeout
    # Conmutación RS485 en RTS (GPIO17) con SP3485: True=TX, False=RX
    instr.serial.rs485_mode = RS485Settings(
        rts_level_for_tx=True,
        rts_level_for_rx=False,
        loopback=False,
        delay_before_tx=0.0,
        delay_before_rx=0.0001
    )
    instr.mode = minimalmodbus.MODE_RTU
    # Evitar eco local en algunos adaptadores
    instr.clear_buffers_before_each_transaction = True
    return instr

# -------------------------------------------
# Operaciones Modbus de alto nivel
# -------------------------------------------
def write_coil(instr: minimalmodbus.Instrument, coil_addr: int, value: bool, retries: int = 3) -> bool:
    for _ in range(retries):
        try:
            instr.write_bit(coil_addr, int(value), functioncode=5)  # FC5 write single coil
            return True
        except Exception as e:
            log.warning(f"Error escribiendo coil {coil_addr}: {e}")
            time.sleep(0.05)
    return False

def read_hr(instr: minimalmodbus.Instrument, reg_addr: int, retries: int = 3) -> Optional[int]:
    for _ in range(retries):
        try:
            return instr.read_register(reg_addr, number_of_decimals=0, functioncode=3, signed=False)
        except Exception as e:
            log.warning(f"Error leyendo HR{reg_addr}: {e}")
            time.sleep(0.05)
    return None

# -------------------------------------------
# Lógica de control del invernadero
# -------------------------------------------
def control_logic(meas: Measurements, sp: Setpoints, state: Actuators) -> Actuators:
    new_state = Actuators(**vars(state))

    # Reglas de temperatura (calefacción / ventilación)
    if not math.isnan(meas.temp_c):
        if meas.temp_c > sp.temp_target_c + sp.hyst_temp_c:
            new_state.heater = False
            new_state.fan = True
        elif meas.temp_c < sp.temp_target_c - sp.hyst_temp_c:
            new_state.heater = True
            new_state.fan = False

    # Reglas de humedad relativa / VPD (nebulización / ventilación)
    if not math.isnan(meas.rh_pct) and not math.isnan(meas.vpd_kpa):
        if meas.rh_pct < sp.rh_target_pct - sp.hyst_rh_pct or meas.vpd_kpa > sp.vpd_target_kpa + 0.1:
            new_state.mister = True
        elif meas.rh_pct > sp.rh_target_pct + sp.hyst_rh_pct or meas.vpd_kpa < sp.vpd_target_kpa - 0.1:
            new_state.mister = False

    # Gas/IAQ simple: si “índice de gas” alto (peor calidad), fuerza ventilación
    if not math.isnan(meas.gas_ohm):
        idx = gas_to_index(meas.gas_ohm)
        if idx > sp.max_co2_gas_index:
            new_state.fan = True

    # Reglas de riego por humedad de suelo (si está disponible)
    if meas.soil_rh_pct is not None:
        if meas.soil_rh_pct < 35.0:
            new_state.pump = True
        elif meas.soil_rh_pct > 45.0:
            new_state.pump = False

    return new_state

# -------------------------------------------
# Aplicación principal
# -------------------------------------------
def main(
    port: str = typer.Option("/dev/serial0", help="Puerto serie (UART) del RS485 HAT"),
    baud: int = typer.Option(19200, help="Baudios Modbus RTU"),
    parity: str = typer.Option("E", help="Paridad: N/E/O"),
    timeout: float = typer.Option(0.2, help="Timeout Modbus en segundos"),
    slave_relays: int = typer.Option(1, help="ID Modbus del módulo de relés"),
    slave_sensors: Optional[int] = typer.Option(2, help="ID Modbus del módulo de sensores (opcional)"),
    bme_addr: int = typer.Option(0x76, help="Dirección I2C del BME680 (0x76 o 0x77)"),
    i2c_bus: int = typer.Option(1, help="Bus I2C (1 por defecto)"),
    loop_period: float = typer.Option(2.0, help="Periodo de control (s)"),
    temp_target: float = typer.Option(26.0, help="Consigna de temperatura (°C)"),
    rh_target: float = typer.Option(65.0, help="Consigna de RH (%)"),
    vpd_target: float = typer.Option(1.0, help="Consigna de VPD (kPa)")
):
    log.info("Inicializando BME680…")
    sensor = init_bme680(address=bme_addr, i2c_bus=i2c_bus)
    # Warm-up gas
    log.info("Calentando sensor de gas (BME680) 30 s…")
    time.sleep(30)

    log.info(f"Preparando Modbus RTU en {port} @ {baud} {parity} 8E1 timeout={timeout}s")
    relays = make_modbus_instrument(port, slave_relays, baud, parity, timeout)
    sensors = None
    if slave_sensors is not None:
        sensors = make_modbus_instrument(port, slave_sensors, baud, parity, timeout)

    sp = Setpoints(temp_target_c=temp_target, rh_target_pct=rh_target, vpd_target_kpa=vpd_target)
    state = Actuators()

    last_apply = None

    while True:
        # Medición local
        t_c, rh, p_hpa, gas = read_bme680(sensor)
        my_vpd = vpd_kpa(t_c, rh) if (not math.isnan(t_c) and not math.isnan(rh)) else float("nan")

        meas = Measurements(
            temp_c=t_c,
            rh_pct=rh,
            pressure_hpa=p_hpa,
            gas_ohm=gas,
            vpd_kpa=my_vpd,
        )

        # Medición de sensores Modbus opcionales
        if sensors is not None:
            soil_raw = read_hr(sensors, 0)
            tank_raw = read_hr(sensors, 1)
            if soil_raw is not None:
                meas.soil_rh_pct = soil_raw / 10.0
            if tank_raw is not None:
                meas.tank_level_pct = tank_raw / 10.0

        # Lógica de control
        new_state = control_logic(meas, sp, state)

        # Aplicación en hardware (esclavo relés)
        if new_state != state or (last_apply is None) or (time.time() - last_apply > 10.0):
            ok0 = write_coil(relays, 0, new_state.fan)
            ok1 = write_coil(relays, 1, new_state.pump)
            ok2 = write_coil(relays, 2, new_state.heater)
            ok3 = write_coil(relays, 3, new_state.mister)
            last_apply = time.time()
            if not all([ok0, ok1, ok2, ok3]):
                log.warning("No se pudieron aplicar todos los estados de relés (revisa bus/ID/terminación).")
            state = new_state

        # Registro
        log.info(
            f"T={meas.temp_c:.2f}°C RH={meas.rh_pct:.1f}% VPD={meas.vpd_kpa:.2f}kPa "
            f"P={meas.pressure_hpa:.1f}hPa Gas={meas.gas_ohm if not math.isnan(meas.gas_ohm) else float('nan'):.0f}Ω "
            f"Soil={meas.soil_rh_pct if meas.soil_rh_pct is not None else float('nan'):.1f}% "
            f"Tank={meas.tank_level_pct if meas.tank_level_pct is not None else float('nan'):.1f}% | "
            f"Fan={int(state.fan)} Pump={int(state.pump)} Heater={int(state.heater)} Mister={int(state.mister)}"
        )

        time.sleep(loop_period)

if __name__ == "__main__":
    typer.run(main)

Puntos clave del código:
– RS485Settings en pyserial con rts_level_for_tx=True y rts_level_for_rx=False: esto hace que la línea RTS (GPIO17) conduzca DE=1 al transmitir y RE=0 al recibir, conmutando correctamente el SP3485 del HAT de Waveshare.
– Lógica de histéresis para evitar oscilaciones.
– Lectura de registros Modbus opcionales para humedad de suelo y nivel de tanque (si tienes un módulo de sensores).
– bme680: se ajustan oversamplings y el calentador de gas; se añade un warm‑up de 30 s.

Script auxiliar de verificación del BME680 (opcional)

#!/usr/bin/env python3
# bme680_check.py
import time
import bme680
from smbus2 import SMBus

def main(addr=0x76, bus=1):
    sensor = bme680.BME680(address=addr, i2c_device=SMBus(bus))
    sensor.set_humidity_oversample(bme680.OS_2X)
    sensor.set_pressure_oversample(bme680.OS_4X)
    sensor.set_temperature_oversample(bme680.OS_8X)
    sensor.set_filter(bme680.FILTER_SIZE_3)
    sensor.set_gas_status(bme680.ENABLE_GAS_MEAS)
    sensor.set_gas_heater_temperature(320)
    sensor.set_gas_heater_duration(150)
    sensor.select_gas_heater_profile(0)

    print("Calentando sensor de gas 20 s…")
    time.sleep(20)
    for _ in range(10):
        if sensor.get_sensor_data():
            t = sensor.data.temperature
            rh = sensor.data.humidity
            p = sensor.data.pressure
            gas = sensor.data.gas_resistance if sensor.data.heat_stable else float('nan')
            print(f"T={t:.2f}C RH={rh:.1f}% P={p:.1f}hPa GAS={gas:.0f}Ω heat_stable={sensor.data.heat_stable}")
        time.sleep(1.0)

if __name__ == "__main__":
    main()

Compilación/flash/ejecución

No hay compilación; es Python. Sigue estrictamente estos pasos en la Raspberry Pi 3 Model B+ con Raspberry Pi OS Bookworm 64‑bit:

1) Preparar sistema

• Actualiza índices APT y paquetes base:
• sudo apt update
• sudo apt full-upgrade -y
• Instala utilidades e interfaces:
• sudo apt install -y python3.11-venv python3-pip i2c-tools python3-gpiozero

2) Comprobar interfaces

• I2C:
• sudo i2cdetect -y 1
• Debe aparecer 0x76 o 0x77.
• UART:
• ls -l /dev/serial0
• groups $USER → debe incluir dialout e i2c (si no, cerrar sesión y volver a entrar o reboot)

3) Crear entorno de trabajo y venv

• mkdir -p ~/rs485-modbus-greenhouse-control
• cd ~/rs485-modbus-greenhouse-control
• python3 -m venv .venv
• source .venv/bin/activate
• python -m pip install –upgrade pip==24.2
• pip install pyserial==3.5 minimalmodbus==2.1.1 smbus2==0.5.0 bme680==1.1.1 gpiozero==1.6.2 typer==0.12.5 rich==13.9.3

4) Crear los scripts

• Copia/pega el contenido de rs485_modbus_greenhouse.py en:
• nano rs485_modbus_greenhouse.py
• Hazlo ejecutable (opcional):
• chmod +x rs485_modbus_greenhouse.py
• Script de prueba BME680 (opcional):
• nano bme680_check.py
• chmod +x bme680_check.py

5) Ejecutar pruebas iniciales

• BME680:
• ./bme680_check.py
• Debes ver lecturas plausibles de temperatura/humedad/presión y, tras estabilizar, gas_resistance distinto de NaN.
• Modbus (sin lógica de control):
• Si tienes un módulo de relés Modbus en ID=1, prueba a activar un relé manualmente usando minimalmodbus en REPL:
  • python
  • from serial.rs485 import RS485Settings
  • import serial, minimalmodbus
  • instr = minimalmodbus.Instrument(«/dev/serial0», 1)
  • instr.serial.baudrate=19200; instr.serial.parity=serial.PARITY_EVEN; instr.serial.timeout=0.3
  • instr.serial.rs485_mode=RS485Settings(rts_level_for_tx=True, rts_level_for_rx=False)
  • instr.mode=minimalmodbus.MODE_RTU
  • instr.write_bit(0, 1, functioncode=5) # enciende coil 0
  • instr.write_bit(0, 0, functioncode=5) # apaga coil 0
• Si esto funciona, la conmutación DE/RE por RTS está operativa.

6) Ejecutar el controlador

• Ejecución por defecto (IDs 1 y 2):
• ./rs485_modbus_greenhouse.py –port /dev/serial0 –baud 19200 –parity E –timeout 0.2 –slave-relays 1 –slave-sensors 2 –bme-addr 0x76 –i2c-bus 1 –loop-period 2.0 –temp-target 26 –rh-target 65 –vpd-target 1.0
• Ejecución sin módulo de sensores (solo relés):
• ./rs485_modbus_greenhouse.py –slave-relays 1 –slave-sensors None
• Observa el log: deben aparecer valores de T/RH/VPD y el estado de Fan/Pump/Heater/Mister en cada ciclo.

Validación paso a paso

1) Validación eléctrica y de bus:
– Verifica que el HAT RS485 tiene la terminación de 120 Ω activa solo si estás en un extremo del bus. En caso contrario, desactívala para evitar sobrecarga.
– Confirma polarización (bias) de la línea (pull‑up en A y pull‑down en B) en un único punto de la red. Muchos HAT lo ofrecen via jumpers.

2) Validación I2C (BME680):
– sudo i2cdetect -y 1 → aparecerá 0x76 o 0x77.
– Ejecuta ./bme680_check.py y verifica:
– Temperatura entre 15–40 °C (según ambiente).
– RH entre 20–90% (según ambiente).
– Presión 900–1100 hPa.
– gas_resistance > 5 kΩ tras unos 30–60 s (depende del aire ambiente).

3) Validación UART/RS485:
– Conecta el módulo de relés (ID=1) y energízalo.
– Desde el REPL de Python con minimalmodbus, escribe y lee una bobina (coil):
– instr.write_bit(0, 1, functioncode=5) → deberías oír clic o ver LED de relé encendido.
– instr.write_bit(0, 0, functioncode=5) → debe apagarse.
– Si inviertes A/B por error, no funcionará; corrige A↔B.

4) Validación del controlador completo:
– Ejecuta ./rs485_modbus_greenhouse.py con tus parámetros.
– Observa en el log, por ejemplo:
– T alta -> Fan=1, Heater=0.
– RH baja -> Mister=1.
– Soil<35% (si HR0 existe) -> Pump=1.
– Modifica el ambiente para comprobar respuestas:
– Calienta el sensor ligeramente con la mano → sube T; observa Fan.
– Exhala cerca del sensor → sube RH y gas; la lógica puede activar ventilación/nebulización.
– Verifica que los relés cambian consecuentemente (LEDs o salida).

5) Validación de estabilidad:
– Deja el sistema 10–15 minutos y confirma que la histéresis evita oscilaciones (no cambia de estado continuamente cerca de la consigna).

Troubleshooting

1) El puerto serie no responde (/dev/serial0 inexistente o en ttyS0 inestable):
– Asegúrate de haber añadido dtoverlay=pi3-disable-bt y enable_uart=1 en /boot/firmware/config.txt.
– Deshabilita el login por serial en raspi-config (Serial Port → Login shell: No).
– Revisa: ls -l /dev/serial0; debe apuntar a /dev/ttyAMA0 (PL011) en Pi 3 B+.

2) Conflicto de permisos al abrir el puerto:
– Añade tu usuario a dialout: sudo usermod -aG dialout $USER y reinicia sesión (o reboot).

3) A/B del RS485 invertidos:
– Síntoma: timeouts constantes, ninguna respuesta. Solución: invierte los conductores A y B en la bornera.

4) Falta de terminación/bias en el bus:
– Síntoma: lecturas/escrituras erráticas, CRCs incorrectos. Solución: habilita 120 Ω en los extremos, y una única pareja de resistencias de polarización en el bus.

5) RTS sin conmutar DE/RE:
– Verifica el jumper “DE/RE=RTS” en el HAT de Waveshare.
– Asegúrate de configurar instr.serial.rs485_mode con RS485Settings en el código.
– Comprueba dtoverlay=uart0,ctsrts=on para exponer la línea RTS en GPIO17.

6) BME680 no aparece en i2cdetect:
– Revisa cableado SDA/SCL/3V3/GND.
– Cambia la dirección a 0x77 si tu placa usa SDO a VCC, o a 0x76 si SDO a GND.
– Habilita I2C en raspi-config y reinicia.

7) Lectura de gas NaN o poco estable:
– El sensor necesita calentamiento (30–180 s). Asegura sensor.data.heat_stable antes de usar gas_resistance.
– Evita corrientes de aire directas o condensación.

8) Paridad/baudios incorrectos:
– Asegúrate de que todos los dispositivos Modbus usan la misma configuración (p.ej., 19200 8E1). Cambia –baud/–parity en el script según tus esclavos.

Mejoras/variantes

• Integración con BSEC (librería de Bosch) para IAQ real y control avanzado basado en CO2 equivalente y VOC. Requiere SDK adicional y licencia; puede ejecutarse en el Pi, pero implica instalar binarios específicos.
• Persistencia y visualización:
• Exportar métricas a InfluxDB/Prometheus y visualizar en Grafana (temperatura, RH, VPD, estados de relés).
• Registrar a SQLite/CSV con timestamps.
• Supervisión remota:
• Añadir una API REST (FastAPI) para leer estado y forzar modos manual/automático.
• Implementar un servidor Modbus TCP que exponga el estado a un SCADA externo.
• Robustez industrial:
• Watchdog por hardware/software.
• Retries con backoff exponencial y detección de esclavos caídos.
• Separación de alimentación y EMC: uso de transceptores aislados y tierra de referencia dedicada.
• Control más sofisticado:
• Control PID para temperatura/humedad, con anti‑windup.
• Calendarios y riegos por etapas según humedad de suelo y horario solar.

Checklist de verificación

• [ ] Raspberry Pi 3 Model B+ con Raspberry Pi OS Bookworm 64‑bit instalada y actualizada.
• [ ] dtoverlay=pi3-disable-bt, enable_uart=1 y dtparam=i2c_arm=on configurados en /boot/firmware/config.txt.
• [ ] Grupos dialout e i2c asignados al usuario actual.
• [ ] Waveshare RS485 HAT (SP3485) instalado; terminación/bias configurados; DE/RE en posición RTS.
• [ ] BME680 cableado a 3V3, GND, SDA (GPIO2), SCL (GPIO3); detectado en i2cdetect (0x76/0x77).
• [ ] Entorno virtual creado; pip 24.2; paquetes instalados con versiones fijadas (pyserial==3.5, minimalmodbus==2.1.1, smbus2==0.5.0, bme680==1.1.1, gpiozero==1.6.2).
• [ ] Prueba de encendido de un relé Modbus (write_bit coil 0) funciona.
• [ ] ./bme680_check.py muestra lecturas plausibles y gas estable tras calentamiento.
• [ ] ./rs485_modbus_greenhouse.py corre sin errores; log muestra T/RH/VPD y estados de actuadores.
• [ ] Validación funcional: cambios ambientales provocan respuestas esperadas (ventilación, nebulización, etc.).

Con este caso práctico, has configurado un sistema de control de invernadero “rs485-modbus-greenhouse-control” basado en Raspberry Pi 3 Model B+ con HAT RS485 (SP3485) y sensor BME680, utilizando Raspberry Pi OS Bookworm 64‑bit, Python 3.11 y una toolchain reproducible. Has cubierto desde la habilitación de interfaces y cableado, hasta el desarrollo, despliegue y validación del lazo de control sobre Modbus RTU en RS485.

Objetivo y caso de uso

Qué construirás: Un intercomunicador dúplex completo con supresión de ruido utilizando Raspberry Pi Zero 2 W, un micrófono I2S y un amplificador I2S.

Para qué sirve

• Comunicación bidireccional en entornos ruidosos, como fábricas o talleres.
• Proyectos de domótica para intercomunicación entre habitaciones.
• Aplicaciones de asistencia para personas con discapacidad auditiva.
• Desarrollo de sistemas de alerta en vehículos o maquinaria pesada.

Resultado esperado

• Latencia de audio inferior a 50 ms en la transmisión de voz.
• Reducción de ruido de fondo en un 80% gracias a la implementación de RNNoise.
• Capacidad de manejar hasta 10 paquetes de audio por segundo sin pérdida de calidad.
• Consumo de energía inferior a 1 W durante la operación continua.

Público objetivo: Desarrolladores y entusiastas de la electrónica; Nivel: Avanzado

Arquitectura/flujo: Raspberry Pi Zero 2 W -> Micrófono I2S -> Procesamiento de audio -> Amplificador I2S -> Salida de audio.

Nivel: Avanzado

Prerrequisitos

• Sistema operativo:
• Raspberry Pi OS Bookworm 64‑bit (Debian 12), imagen “Lite” o “Full”.
• Kernel Linux 6.6.x (serie LTS en Bookworm para Raspberry Pi).
• Hardware exacto:
• Raspberry Pi Zero 2 W
• Adafruit SPH0645 I2S Mic
• MAX98357A I2S Class‑D Amplifier (Adafruit u otro breakout equivalente; se usa como DAC+amplificador I2S)
• Toolchain y versiones concretas para este caso:
• Python 3.11.2 (preinstalado en Raspberry Pi OS Bookworm de 64 bits)
• pip 24.2 (se actualizará explícitamente)
• gcc (Debian 12.2.0-14) 12.2.0
• g++ (Debian 12.2.0-14) 12.2.0
• CMake 3.25.1
• PortAudio 19.6.0 (vía paquete portaudio19-dev)
• ALSA lib 1.2.8
• Paquetes Python (se fijan versiones exactas):
  • numpy==1.26.4
  • sounddevice==0.4.6
  • rnnoise==0.4.1
  • pyyaml==6.0.1
  • gpiozero==1.6.2 (opcional para PTT por GPIO)
• Herramientas/paquetes del sistema:
• build-essential, python3.11-venv, python3-dev, libasound2-dev, portaudio19-dev, librnnoise0, librnnoise-dev, alsa-utils, git

Notas:
– Raspberry Pi OS Bookworm utiliza por defecto PipeWire/WirePlumber en la edición “Full”. Para audio de baja latencia y control directo de I2S usaremos ALSA y dispositivos hw:…, evitando capas extra. Las instrucciones de este caso práctico abren los dispositivos ALSA en modo “hw” para reducir la latencia y maximizar la precisión.

Materiales

• Kit base:
• Raspberry Pi Zero 2 W
• Tarjeta microSD (≥16 GB, clase A1/A2 recomendada)
• Alimentación 5 V/2.5 A con cable micro‑USB de buena calidad
• Cabecera GPIO de 40 pines soldada en la Pi Zero 2 W (si no viene pre‑soldada)
• Audio I2S:
• Adafruit SPH0645 I2S Microphone (micrófono I2S, 3.3 V)
• MAX98357A I2S Class-D Amplifier (alimentación 5 V recomendada)
• Altavoz 4–8 Ω (3–5 W)
• Conexión:
• Cables dupont macho‑hembra x10–12
• Protoboard (opcional pero recomendable)
• Opcionales (para control de PTT local):
• Pulsador + resistencia 10 kΩ (pull‑down si no usamos internal pull‑up)
• LED + resistencia 330 Ω

Este caso práctico se centra exclusivamente en el modelo “Raspberry Pi Zero 2 W + Adafruit SPH0645 I2S Mic + MAX98357A Amp”. Todo el cableado, configuración, código y validación asume este conjunto.

Preparación y conexión

Habilitar I2S y configurar la tarjeta combinada

Usaremos el overlay del kernel “googlevoicehat-soundcard”, que configura simultáneamente:
– Entrada por micrófono I2S (compatible con SPH0645)
– Salida por DAC/amp I2S (compatible con MAX98357A)

Este overlay configura la interfaz I2S de la Pi (BCLK, LRCLK, DIN, DOUT) en los pines estándar y crea un único dispositivo ALSA full‑duplex, ideal para un intercomunicador con micrófono y altavoz I2S.

Pasos:

1) Edita el fichero de arranque (como root):

sudo nano /boot/firmware/config.txt

2) Añade al final (evitando duplicados):

dtparam=audio=off
dtoverlay=googlevoicehat-soundcard

3) Guarda y reinicia:

sudo reboot

4) Tras reiniciar, verifica los dispositivos ALSA:

arecord -l
aplay -l

Deberías ver una tarjeta similar a “snd-googlevoicehat” o con descripción “Google voiceHAT SoundCard”. Tomaremos esta tarjeta como hw:0,0 en el resto de pasos.

Conexión de pines

La interfaz I2S estándar de Raspberry Pi usa los siguientes pines GPIO:

• BCLK: GPIO18 (PCM_CLK)
• LRCLK: GPIO19 (PCM_FS)
• DIN (entrada a la Pi): GPIO20 (PCM_DIN)
• DOUT (salida desde la Pi): GPIO21 (PCM_DOUT)

El micrófono I2S (SPH0645) entrega datos al pin DIN de la Pi (PCM_DIN). El MAX98357A recibe datos desde el pin DOUT de la Pi (PCM_DOUT). Ambos comparten BCLK y LRCLK.

Tabla de cableado recomendado:

Observaciones:

• Alimenta el SPH0645 estrictamente a 3.3 V. No uses 5 V en el micrófono.
• Alimenta el MAX98357A preferentemente con 5 V; de este modo obtienes potencia de salida adecuada. Conecta un altavoz de 4–8 Ω a las bornas del MAX98357A.
• Conecta BCLK y LRCLK en paralelo al mic y al amp desde la Pi.
• El SPH0645 tiene un pin SEL; a GND entrega datos por canal izquierdo; a 3V3 por canal derecho. Usaremos SEL a GND.
• Mantén cortos los cables I2S y de buena calidad para minimizar jitter y EMI.

Comprobación eléctrica básica

• Mide 3.3 V y 5 V con multímetro antes de alimentar definitivamente.
• Verifica continuidad y que no hay cortos entre 3V3 y GND.
• Enciende la Pi y confirma que el MAX98357A no se calienta en vacío y que el micrófono no muestra síntomas de alimentación incorrecta.

Código completo

A continuación implementaremos un intercomunicador full‑duplex por UDP entre dos nodos, con:
– Captura desde I2S (SPH0645) vía ALSA a 48 kHz mono
– Supresión de ruido en tiempo real con RNNoise
– Reproducción a I2S (MAX98357A) vía ALSA a 48 kHz mono
– Modo semi‑dúplex con PTT (push‑to‑talk) opcional por GPIO, para evitar realimentación acústica local
– Mecanismo VOX simple (activación por voz) opcional si no hay PTT
– Buffering de baja latencia (frames de 10 ms = 480 muestras a 48 kHz)
– Enlace UDP con control de jitter básico

Estructura:
– Una hebra captura->denoise->envía
– Otra hebra recibe->reproduce
– Sin bloqueo entre hebras usando colas y sockets no bloqueantes

Requisitos Python (en venv):
– numpy==1.26.4
– sounddevice==0.4.6
– rnnoise==0.4.1
– gpiozero==1.6.2 (opcional)

Archivo: intercom.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import argparse
import socket
import struct
import threading
import queue
import time
import sys
import signal

import numpy as np
import sounddevice as sd
from rnnoise import RNNoise

try:
    from gpiozero import Button, LED
    GPIO_AVAILABLE = True
except Exception:
    GPIO_AVAILABLE = False
    Button = None
    LED = None

# Parámetros de audio
SAMPLE_RATE = 48000       # Hz
CHANNELS = 1              # Mono (mic I2S suele ser mono)
FRAME_MS = 10             # 10 ms
FRAME_SAMPLES = int(SAMPLE_RATE * FRAME_MS / 1000)  # 480
PCM_FORMAT = 'int16'      # Usaremos S16_LE

# Empaquetado de tramas UDP: cabecera simple (seq, ts)
HEADER_FORMAT = "!II"     # seq (u32), timestamp_ms (u32)

class Intercom:
    def __init__(self, 
                 playback_device=None, 
                 capture_device=None, 
                 rx_port=6000, 
                 tx_ip="192.168.1.100", 
                 tx_port=6000,
                 ptt_gpio=None, 
                 led_gpio=None, 
                 vox_threshold=0.01, 
                 vox_hold_ms=300,
                 denoise=True):
        self.playback_device = playback_device
        self.capture_device = capture_device
        self.rx_port = rx_port
        self.tx_ip = tx_ip
        self.tx_port = tx_port
        self.vox_threshold = vox_threshold
        self.vox_hold_ms = vox_hold_ms
        self.denoise_enabled = denoise

        # Sockets
        self.sock_rx = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.sock_rx.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.sock_rx.bind(("0.0.0.0", self.rx_port))
        self.sock_rx.setblocking(False)

        self.sock_tx = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.sock_tx.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

        # RNNoise
        self.rnnoise = RNNoise() if self.denoise_enabled else None

        # GPIO PTT y LED (opcional)
        self.ptt_button = None
        self.tx_led = None
        self.ptt_active = False
        if ptt_gpio is not None and GPIO_AVAILABLE:
            self.ptt_button = Button(ptt_gpio, pull_up=True)
            self.ptt_button.when_pressed = self._ptt_down
            self.ptt_button.when_released = self._ptt_up
        if led_gpio is not None and GPIO_AVAILABLE:
            self.tx_led = LED(led_gpio)

        # Control de ejecución
        self.running = True
        self.seq = 0
        self.last_vox_ms = 0

        # Buffers y streams
        self.play_q = queue.Queue(maxsize=50)  # Cola de reproducción
        self.tx_lock = threading.Lock()

        # Streams PortAudio/ALSA
        self.in_stream = None
        self.out_stream = None

    def _ptt_down(self):
        self.ptt_active = True
        if self.tx_led:
            self.tx_led.on()

    def _ptt_up(self):
        self.ptt_active = False
        if self.tx_led:
            self.tx_led.off()

    def _should_transmit(self, frame):
        # Si PTT existe, gobierna
        if self.ptt_button is not None:
            return self.ptt_active

        # VOX simple si no hay PTT: energía RMS > umbral
        rms = np.sqrt(np.mean((frame.astype(np.float32) / 32768.0) ** 2))
        now_ms = int(time.time() * 1000)
        if rms > self.vox_threshold:
            self.last_vox_ms = now_ms
            return True
        # Mantener transmisión por hold_ms para no cortar sílabas
        if now_ms - self.last_vox_ms < self.vox_hold_ms:
            return True
        return False

    def _denoise(self, frame):
        if not self.rnnoise:
            return frame
        # RNNoise espera 480 muestras a 48k por trama, int16
        # Devuelve float32 [-1, 1]; convertimos de vuelta a int16
        den = self.rnnoise.process_frame(frame)
        den = np.clip(den, -1.0, 1.0)
        return (den * 32767.0).astype(np.int16)

    def audio_in_cb(self, indata, frames, time_info, status):
        if status:
            # Reporte de underrun/overrun de PortAudio
            print(f"[IN] Status: {status}", file=sys.stderr)

        # Convertir a int16
        data = np.frombuffer(indata, dtype=np.int16)

        # Denoise por tramas de 480
        # sounddevice puede entregar 'frames' múltiplos de FRAME_SAMPLES
        outbuf = []
        for off in range(0, len(data), FRAME_SAMPLES):
            chunk = data[off:off + FRAME_SAMPLES]
            if len(chunk) < FRAME_SAMPLES:
                break
            if self.denoise_enabled:
                chunk = self._denoise(chunk)
            outbuf.append(chunk)

            # Decidir transmisión (PTT/VOX)
            if self._should_transmit(chunk):
                # Construir paquete UDP
                with self.tx_lock:
                    header = struct.pack(HEADER_FORMAT, self.seq, int(time.time() * 1000) & 0xFFFFFFFF)
                    self.seq = (self.seq + 1) & 0xFFFFFFFF
                pkt = header + chunk.tobytes()
                try:
                    self.sock_tx.sendto(pkt, (self.tx_ip, self.tx_port))
                except Exception as e:
                    print(f"[TX] Error enviando: {e}", file=sys.stderr)

        # Semidúplex: cuando transmitimos, silenciamos altavoz local para evitar acople
        if self.tx_led:
            # LED indica TX activo
            if self._should_transmit(data[:FRAME_SAMPLES]):
                self.tx_led.on()
            else:
                self.tx_led.off()

    def audio_out_worker(self):
        # Hilo que drena paquetes recibidos y los escribe en el stream de salida
        while self.running:
            # Recepción no bloqueante
            try:
                pkt, addr = self.sock_rx.recvfrom(1500)
                if len(pkt) >= struct.calcsize(HEADER_FORMAT) + FRAME_SAMPLES * 2:
                    # Extraer cabecera
                    _seq, _ts = struct.unpack(HEADER_FORMAT, pkt[:8])
                    payload = pkt[8:]
                    # Rechazar si estamos en TX (semidúplex)
                    if self.ptt_button is not None and self.ptt_active:
                        continue
                    try:
                        self.out_stream.write(payload)
                    except sd.PortAudioError as e:
                        print(f"[OUT] PortAudioError: {e}", file=sys.stderr)
            except BlockingIOError:
                pass
            except Exception as e:
                print(f"[RX] Error: {e}", file=sys.stderr)
            time.sleep(0.001)

    def start(self):
        # Configurar streams ALSA vía sounddevice/PortAudio
        # Selección explícita de dispositivo (índice o nombre), si se proporcionó
        in_dev = self.capture_device if self.capture_device is not None else None
        out_dev = self.playback_device if self.playback_device is not None else None

        self.in_stream = sd.RawInputStream(
            samplerate=SAMPLE_RATE,
            channels=CHANNELS,
            dtype='int16',
            blocksize=FRAME_SAMPLES,
            device=in_dev,
            callback=self.audio_in_cb,
            latency='low'
        )
        self.out_stream = sd.RawOutputStream(
            samplerate=SAMPLE_RATE,
            channels=CHANNELS,
            dtype='int16',
            blocksize=FRAME_SAMPLES,
            device=out_dev,
            latency='low'
        )

        self.out_stream.start()
        self.in_stream.start()

        t = threading.Thread(target=self.audio_out_worker, daemon=True)
        t.start()

    def stop(self):
        self.running = False
        time.sleep(0.05)
        try:
            if self.in_stream:
                self.in_stream.stop()
                self.in_stream.close()
        except:
            pass
        try:
            if self.out_stream:
                self.out_stream.stop()
                self.out_stream.close()
        except:
            pass
        if self.tx_led:
            self.tx_led.off()

def main():
    parser = argparse.ArgumentParser(description="i2s-noise-suppression-intercom")
    parser.add_argument("--tx-ip", type=str, required=True, help="IP remota a la que enviar audio")
    parser.add_argument("--tx-port", type=int, default=6000, help="Puerto UDP remoto")
    parser.add_argument("--rx-port", type=int, default=6000, help="Puerto UDP local de escucha")
    parser.add_argument("--capture-device", type=str, default=None, help="Dispositivo de captura (índice o nombre)")
    parser.add_argument("--playback-device", type=str, default=None, help="Dispositivo de reproducción (índice o nombre)")
    parser.add_argument("--ptt-gpio", type=int, default=None, help="GPIO BCM para PTT (opcional)")
    parser.add_argument("--led-gpio", type=int, default=None, help="GPIO BCM para LED TX (opcional)")
    parser.add_argument("--vox-threshold", type=float, default=0.01, help="Umbral VOX RMS (0..1)")
    parser.add_argument("--vox-hold-ms", type=int, default=300, help="Tiempo de retención VOX (ms)")
    parser.add_argument("--no-denoise", action="store_true", help="Desactivar RNNoise")
    args = parser.parse_args()

    ic = Intercom(
        playback_device=args.playback_device,
        capture_device=args.capture_device,
        rx_port=args.rx_port,
        tx_ip=args.tx_ip,
        tx_port=args.tx_port,
        ptt_gpio=args.ptt_gpio,
        led_gpio=args.led_gpio,
        vox_threshold=args.vox_threshold,
        vox_hold_ms=args.vox_hold_ms,
        denoise=not args.no_denoise
    )

    def handle_sigint(signum, frame):
        ic.stop()
        sys.exit(0)

    signal.signal(signal.SIGINT, handle_sigint)
    signal.signal(signal.SIGTERM, handle_sigint)

    ic.start()
    print("Intercom en ejecución. Ctrl+C para salir.")
    while True:
        time.sleep(1)

if __name__ == "__main__":
    main()

Breve explicación de partes clave:
– audio_in_cb: callback que recibe bloques de 10 ms desde ALSA (micrófono I2S). Cada bloque se pasa por RNNoise (si está activo) y, en función de PTT/VOX, se envía por UDP como PCM S16_LE con cabecera de secuencia y timestamp.
– audio_out_worker: hilo que recibe por UDP y escribe directamente en el stream de salida (MAX98357A). Si PTT está activo, silenciamos la reproducción para evitar acople en el mismo nodo.
– FRAME_SAMPLES=480 a 48 kHz: elección estándar para RNNoise y baja latencia.
– Dispositivos ALSA: se pueden seleccionar por nombre/índice; si no se especifican, usa el predeterminado del sistema. Recomendación: forzar por nombre la tarjeta del “googlevoicehat-soundcard”.

Compilación/instalación/ejecución

1) Actualiza el sistema e instala dependencias

sudo apt update
sudo apt full-upgrade -y
sudo reboot

Tras reiniciar:

sudo apt install -y \
  build-essential cmake pkg-config git \
  python3.11 python3.11-venv python3-dev \
  libasound2-dev portaudio19-dev \
  librnnoise0 librnnoise-dev \
  alsa-utils

Comprueba versiones (aprox. en Raspberry Pi OS Bookworm 64‑bit):
– gcc 12.2.0
– cmake 3.25.1
– ALSA lib 1.2.8
– PortAudio 19.6.0

gcc --version | head -n1
cmake --version | head -n1
aplay --version

2) Crea y activa un entorno virtual de Python 3.11

python3 -m venv ~/venvs/intercom
source ~/venvs/intercom/bin/activate
python --version

Deberías ver “Python 3.11.x”.

Actualiza pip a la versión fijada:

python -m pip install --upgrade pip==24.2

Instala paquetes Python con versiones exactas:

pip install numpy==1.26.4 sounddevice==0.4.6 rnnoise==0.4.1 pyyaml==6.0.1 gpiozero==1.6.2

Verifica:

python -c "import numpy, sounddevice, rnnoise, gpiozero; print(numpy.__version__, sounddevice.__version__)"

3) Verifica que la tarjeta I2S está disponible

Lista capturas y reproducciones:

arecord -l
aplay -l

Debería aparecer una tarjeta asociada a “Google voiceHAT SoundCard” o similar, normalmente como card 0, device 0 (hw:0,0). Si no es card 0, ajusta índices en los comandos de prueba.

Prueba rápida de captura (5 s a 48 kHz mono):

arecord -D hw:0,0 -f S16_LE -c 1 -r 48000 -d 5 test_mic.wav
aplay test_mic.wav

Si escuchas tu voz por el altavoz vía MAX98357A, la ruta I2S funciona.

Ajusta volumen de reproducción con alsamixer:

alsamixer

Selecciona la tarjeta del voiceHAT y sube el volumen Master o PCM según disponibilidad.

4) Descarga el código y ejecútalo

Crea un directorio de trabajo y guarda el script:

mkdir -p ~/projects/i2s-intercom
cd ~/projects/i2s-intercom
nano intercom.py

Pega el código completo mostrado arriba, guarda y sal.

Lista dispositivos por nombre para usar con sounddevice:

python - << 'PY'
import sounddevice as sd
for i,d in enumerate(sd.query_devices()):
    print(i, d['name'])
PY

Anota el índice o nombre exacto de:
– Dispositivo de captura (mic I2S, suele ser el mismo “voiceHAT”)
– Dispositivo de reproducción (amp I2S, “voiceHAT”)

Supón que ambos son el dispositivo 0 (ajusta si no lo son).

Prepara dos nodos (dos Raspberry Pi Zero 2 W con el mismo montaje), cada uno conoce la IP del otro:
– Nodo A (IP A): enviará a IP B
– Nodo B (IP B): enviará a IP A

En nodo A:

source ~/venvs/intercom/bin/activate
cd ~/projects/i2s-intercom
python intercom.py --tx-ip <IP_B> --tx-port 6000 --rx-port 6000 --capture-device 0 --playback-device 0 --ptt-gpio 17 --led-gpio 27

En nodo B:

source ~/venvs/intercom/bin/activate
cd ~/projects/i2s-intercom
python intercom.py --tx-ip <IP_A> --tx-port 6000 --rx-port 6000 --capture-device 0 --playback-device 0 --ptt-gpio 17 --led-gpio 27

• Si no tienes el pulsador PTT ni el LED, omite –ptt-gpio y –led-gpio. El script usará VOX (activación por voz) con umbral y hold configurables.

Parámetros útiles:
– –vox-threshold 0.008 … 0.02 según ruido ambiente
– –vox-hold-ms 200 … 600 para no “cortar” sílabas
– –no-denoise para desactivar RNNoise (comparativa A/B)

Validación paso a paso

1) Verificación del hardware I2S
– arecord -l y aplay -l muestran una tarjeta “Google voiceHAT SoundCard” (o similar).
– arecord -D hw:0,0 -f S16_LE -c 1 -r 48000 -d 5 test.wav y aplay test.wav reproducen audio nítido por el altavoz.

2) Nivel y ganancia
– Ejecuta alsamixer y selecciona la tarjeta correcta.
– Asegura que el volumen maestro no está en mute y el nivel de salida esté entre 70–90% para pruebas.

3) Latencia y estabilidad
– En el intercom, habla por el micrófono del nodo A y deberías escucharte en <200–300 ms en el nodo B, dependiendo de la red.
– Ajusta el volumen para evitar acople.

4) Supresión de ruido
– Con ruido de fondo (ventilador, calle), compara:
– Modo con RNNoise (por defecto).
– Modo sin RNNoise: añade “–no-denoise”.
– Deberías percibir atenuación del ruido estacionario (≈ 10–20 dB en frecuencias persistentes) y mejora de inteligibilidad.

5) Semidúplex PTT/VOX
– Con PTT por botón: al pulsar, el LED enciende y se transmite; al soltar, se recibe. Verifica que el altavoz local se silencia en TX.
– Con VOX: ajusta –vox-threshold y –vox-hold-ms. El sistema transmite solo en voz. Habla y observa que la transmisión se corta después del hold.

6) Robustez de UDP
– Simula jitter: si hay pequeñas pérdidas, la reproducción debería seguir sin cortes apreciables gracias a los tamaños pequeños de trama (10 ms).

7) Consumo CPU
– Monitoriza con top/htop; en Zero 2 W, RNNoise a 48 kHz y 10 ms por trama suele ser sostenible. Ajusta prioridad o reduce tasa a 16 kHz si fuera necesario (ver “Mejoras/variantes”).

8) Prueba cruzada de routing
– Cambia los puertos y verifica que los sockets se abren correctamente. Usa netstat -anu o ss -lun para ver puertos en uso.

Troubleshooting

1) No aparece la tarjeta I2S tras el reboot
– Causa: Falta “dtoverlay=googlevoicehat-soundcard” o “dtparam=audio=off”.
– Solución:
– Edita /boot/firmware/config.txt y añade:
– dtparam=audio=off
– dtoverlay=googlevoicehat-soundcard
– Revisa que no tengas overlays conflictivos (otros DAC I2S).
– Reboot.

2) arecord/aplay funcionan pero el script no encuentra el dispositivo
– Causa: sounddevice/PortAudio usa índices distintos.
– Solución:
– Lista dispositivos con el snippet de Python mostrado.
– Pasa –capture-device y –playback-device con el índice o nombre correcto (o usa “hw:CARD=…” si corresponde).

3) Audio con chasquidos o underruns/overruns frecuentes
– Causa: Blocksize no óptimo, CPU al 100%, latencia muy baja para la red.
– Solución:
– Aumenta blocksize (p. ej., duplica FRAME_MS a 20 ms y FRAME_SAMPLES a 960).
– Cierra servicios de fondo pesados. Usa la versión “Lite” del OS.
– Asegura buena alimentación 5 V/2.5 A; evita undervoltage (dmesg | grep -i voltage).
– Revisa cableado I2S y longitudes de cables.

4) Realimentación acústica (acople)
– Causa: el mic capta el altavoz local o remoto.
– Solución:
– Aumenta separación física o baja el volumen en alsamixer.
– Usa PTT (semidúplex) o VOX para no reproducir mientras transmites.
– Encapsula el micrófono en una caja con antiviento o pantalla acústica.

5) El micrófono no capta nada (silencio)
– Causa: SEL mal configurado, DOUT no conectado a GPIO20, mala alimentación del mic (5 V por error).
– Solución:
– Verifica que SEL del SPH0645 está a GND (canal izquierdo).
– Revisa DOUT (mic) -> GPIO20 (PCM_DIN).
– Confirma 3.3 V en el mic, no 5 V.

6) El MAX98357A no suena
– Causa: DIN no conectado a GPIO21, falta BCLK/LRCLK, altavoz mal conectado.
– Solución:
– Verifica GPIO21 (PCM_DOUT) -> DIN del MAX98357A.
– BCLK (GPIO18) y LRCLK (GPIO19) conectados y compartidos.
– Revisa el altavoz (4–8 Ω) y las bornas del módulo.

7) PipeWire/PulseAudio interfieren
– Causa: sesión gráfica con PipeWire tomando control del audio.
– Solución:
– Ejecuta el script especificando dispositivos “hw:…” vía sounddevice para saltar capas.
– Si es necesario, detén servicios de usuario de PipeWire para pruebas:
– systemctl –user stop pipewire pipewire-pulse wireplumber

8) Latencia de red demasiado alta o jitter
– Causa: Wi‑Fi saturado, enlace inestable.
– Solución:
– Usa 5 GHz si es posible en otro modelo (Zero 2 W es 2.4 GHz; aproxima el AP).
– Reduce bitrate con compresión (Opus) o aumenta FRAME_MS a 20 ms.
– Conecta por Ethernet usando un adaptador USB OTG 10/100 (opción avanzada).

Mejoras/variantes

• Compresión Opus
• Reduce el ancho de banda UDP de ~768 kbps PCM 48 kHz mono a 16–32 kbps con códec Opus.
• Paquetes: libopus0, opuslib (pip) o pyogg.

• Inserta la codificación/decodificación en el pipeline, manteniendo RNNoise antes del codificador.

• Echo Cancellation (AEC)

• Integra WebRTC Audio Processing (AEC + AGC + NS). En Pi Zero 2 W es posible pero más exigente.
• Paquetes: libwebrtc-audio-processing-dev (compilación), binding Python o C++ embebido.

• Arquitectura: ruta de referencia de altavoz hacia el AEC y mic como ruta primaria.

• Resample a 16 kHz

• RNNoise admite entrada 48 kHz, pero puedes resamplear a 16 kHz para reducir CPU y ancho de banda.

• Usa librosa o soxr (pysoxr) para calidad alta, o implementa un filtro simple para prototipo.

• Buffer adaptativo y jitter buffer

• Añade un pequeño jitter buffer con timestamps y reordenamiento por seq para mayor robustez en redes ruidosas.

• Seguridad y descubrimiento

• Descubrimiento mDNS/Avahi para IPs dinámicas.

• Cifrado (DTLS/SRTP) si el entorno lo requiere.

• Supervisión y métricas

• Exponer métricas (pérdida de paquetes, jitter, latencia) vía Prometheus o logs JSON para diagnóstico.

• Integración con servicios del sistema

• Crear un servicio systemd para iniciar el intercom al arranque y gestionar reinicios.

Checklist de verificación

• [ ] He instalado Raspberry Pi OS Bookworm 64‑bit y actualizado el sistema.
• [ ] He habilitado el overlay en /boot/firmware/config.txt:
• [ ] dtparam=audio=off
• [ ] dtoverlay=googlevoicehat-soundcard
• [ ] He cableado:
• [ ] 3V3 a SPH0645 (no al MAX98357A)
• [ ] 5V a MAX98357A
• [ ] GND común a todos los módulos
• [ ] GPIO18 (BCLK) a BCLK de mic y amp
• [ ] GPIO19 (LRCLK) a LRCLK/LRC de mic y amp
• [ ] GPIO20 (DIN de Pi) a DOUT del SPH0645
• [ ] GPIO21 (DOUT de Pi) a DIN del MAX98357A
• [ ] SEL del SPH0645 a GND
• [ ] arecord -l y aplay -l muestran la tarjeta I2S (voiceHAT)
• [ ] arecord/aplay de prueba funcionan a 48 kHz mono
• [ ] He creado venv con Python 3.11.2 y pip 24.2
• [ ] He instalado numpy==1.26.4, sounddevice==0.4.6, rnnoise==0.4.1, pyyaml==6.0.1, gpiozero==1.6.2
• [ ] He listado dispositivos con sounddevice y anotado índices correctos
• [ ] He ejecutado intercom.py en ambos nodos con IPs cruzadas
• [ ] PTT funciona (si se usa) y LED indica TX; VOX funciona (si se usa)
• [ ] Noto supresión de ruido perceptible con RNNoise activado
• [ ] Latencia aceptable y sin chasquidos; niveles ajustados en alsamixer

---

Con este caso práctico, has implementado un intercomunicador full‑duplex con micrófono I2S Adafruit SPH0645 y amplificador I2S MAX98357A sobre Raspberry Pi Zero 2 W, ejecutando supresión de ruido en tiempo real con RNNoise, usando ALSA/PortAudio a 48 kHz para baja latencia. La arquitectura es extensible a compresión Opus, AEC con WebRTC y despliegue como servicio systemd para un intercom profesional y robusto.

Objetivo y caso de uso

Qué construirás: Una aplicación de detección de personas en tiempo real utilizando Raspberry Pi 4, HQ Camera y Google Coral USB con OpenCV.

Para qué sirve

• Monitoreo de seguridad en tiempo real en espacios públicos.
• Control de acceso automatizado en edificios.
• Estadísticas de afluencia en eventos o tiendas.
• Asistencia en la robótica para la navegación y detección de obstáculos.

Resultado esperado

• Detección de hasta 30 personas por segundo con una latencia menor a 100 ms.
• Precisión de detección superior al 85% en condiciones de luz variable.
• Generación de alertas en tiempo real a través de MQTT al detectar intrusos.
• Visualización de datos en un dashboard con métricas de afluencia por hora.

Público objetivo: Desarrolladores y entusiastas de la IA; Nivel: Avanzado

Arquitectura/flujo: Raspberry Pi 4 con HQ Camera captura imágenes, Google Coral USB procesa la detección con TensorFlow Lite, y los resultados se envían a través de MQTT para su visualización y análisis.

Nivel: Avanzado

Prerrequisitos

• Sistema Operativo:
• Raspberry Pi OS Bookworm 64‑bit (imagen 2024-10-22 o posterior)
• Kernel Linux 6.x (el que trae la imagen oficial)
• Hardware exacto:
• Raspberry Pi 4 Model B (2/4/8 GB)
• Cámara Raspberry Pi HQ Camera (sensor Sony IMX477)
• Acelerador Google Coral USB (Edge TPU)
• Toolchain y versiones utilizadas en este caso práctico:
• Python 3.11 (python3 —versión por defecto en Bookworm; verificación incluida más abajo)
• OpenCV 4.6.0 (paquete apt: python3-opencv en Bookworm)
• libcamera (stack por defecto en Bookworm; usado por Picamera2)
• Picamera2 0.3.x (paquete apt: python3-picamera2)
• NumPy 1.26.x
• TensorFlow Lite Runtime 2.11.0 (tflite-runtime vía pip)
• PyCoral 2.0.0 (pycoral vía pip)
• Edge TPU runtime (libedgetpu1-std 16.0; paquete apt desde repositorio de Coral)
• Conectividad:
• Acceso a Internet en la Raspberry Pi para instalar paquetes y descargar el modelo TFLite.
• Herramientas:
• Terminal/SSH
• Editor de texto (nano, vim o VS Code Remote)

Comandos para verificar versiones instaladas tras la preparación (sección posterior):

python3 -V
python3 -c "import cv2, numpy, platform; print('OpenCV:', cv2.__version__, 'NumPy:', numpy.__version__, 'Arch:', platform.machine())"
python3 -c "import picamera2; import picamera2 as p2; print('Picamera2 ok')"
python3 -c "import tflite_runtime; import tflite_runtime.interpreter as tfl; print('TFLite:', tfl.__version__)"
python3 -c "import pycoral; import pycoral.utils.edgetpu as e; print('PyCoral ok')"
dpkg -l | grep libedgetpu1

Nota: Las versiones concretas pueden variar marginalmente con el tiempo; aquí fijamos las que se han validado en este caso práctico. Si su salida difiere, ajuste el entorno conforme a las instrucciones de instalación ancladas a versiones.

Materiales

• Raspberry Pi 4 Model B + Raspberry Pi HQ Camera (IMX477) + Google Coral USB (Edge TPU)
• Tarjeta microSD (32 GB recomendados, clase A1/A2)
• Fuente de alimentación oficial 5V/3A USB-C para Pi 4
• Cable plano CSI para HQ Camera (incluido con la cámara)
• Disipadores/ventilador (recomendado para cargas sostenidas)
• (Opcional) Trípode u ópticas C/CS para la HQ Camera
• (Opcional) Monitor/teclado/ratón; alternativamente, SSH habilitado

Preparación y conexión

Habilitar interfaces y preparar el sistema

1) Flashear Raspberry Pi OS Bookworm 64-bit:
– Use Raspberry Pi Imager (v1.8.5 o superior).
– Seleccione Raspberry Pi OS (64-bit) – versión Bookworm.
– En “Ajustes” (opcional): configure Wi‑Fi, hostname y SSH.
– Escriba la imagen en la microSD y arranque la Pi.

2) Actualizar sistema:
– Conéctese por terminal y ejecute:
bash
sudo apt update
sudo apt full-upgrade -y
sudo reboot

3) Activar cámara HQ (IMX477) con libcamera:
– En Bookworm no es necesario activar la “Legacy Camera”; usaremos la pila libcamera y Picamera2.
– Asegure que el overlay del sensor IMX477 está presente (ayuda a la detección en algunos casos):
bash
sudo nano /boot/firmware/config.txt
Añada o verifique estas líneas (sin secciones duplicadas):
# HQ Camera IMX477 (libcamera)
dtoverlay=imx477
gpu_mem=128
Guarde, cierre y reinicie:
bash
sudo reboot

4) Instalar paquetes base por apt (OpenCV, Picamera2, utilidades):
bash
sudo apt update
sudo apt install -y \
python3-pip python3-venv python3-opencv python3-picamera2 \
libcamera-apps python3-numpy git wget curl \
python3-gpiozero python3-smbus python3-spidev

5) Añadir el repositorio de Coral y runtime de Edge TPU:
– Importar clave y añadir el repo de Google Coral (método “signed-by”):
«`bash
sudo install -d -m 0755 /usr/share/keyrings
curl -sSL https://packages.cloud.google.com/apt/doc/apt-key.gpg | \
sudo gpg –dearmor -o /usr/share/keyrings/coral-archive-keyring.gpg

 echo "deb [signed-by=/usr/share/keyrings/coral-archive-keyring.gpg] https://packages.cloud.google.com/apt coral-edgetpu-stable main" | \
   sudo tee /etc/apt/sources.list.d/coral-edgetpu.list

 sudo apt update
 sudo apt install -y libedgetpu1-std
 ```

• Nota: libedgetpu1-std (runtime Edge TPU “standard”). Para máxima velocidad (más TDP), podría instalar libedgetpu1-max en su lugar.

6) Crear y usar entorno virtual Python 3.11 con acceso a paquetes del sistema:
bash
python3 -m venv --system-site-packages ~/venvs/pi-coral
source ~/venvs/pi-coral/bin/activate
python -m pip install --upgrade pip wheel
pip install --upgrade \
tflite-runtime==2.11.0 \
pycoral==2.0.0 \
numpy==1.26.4

7) Validar cámara y Coral:
– Cámara:
bash
libcamera-hello -t 3000
Debe ver vista previa 3 segundos sin errores.
– Coral USB:
– Conecte el Coral USB a un puerto USB 3.0 (azul) de la Pi 4.
– Verifique:
bash
lsusb | grep -i google
dmesg | grep -i edgetpu
– Debe ver el dispositivo enumerado y líneas de carga del driver Edge TPU.

Conexión física

Tabla de puertos/elementos de conexión:

Recomendaciones:
– Inserte el cable CSI con orientación correcta: la cara de contactos debe coincidir con la del conector de la Pi (marcado “CAMERA”). Asegure bien la presilla.
– Monte la HQ Camera firmemente para evitar vibraciones.
– Mantenga el Coral USB en un puerto USB 3.0 para rendimiento óptimo.

Código completo

A continuación se presenta un script Python que:
– Captura frames de la HQ Camera (IMX477) mediante Picamera2 y libcamera.
– Ejecuta inferencia en Edge TPU con un modelo TFLite optimizado (SSD MobileNet v2 COCO).
– Filtra la clase “person”.
– Define zonas poligonales en coordenadas normalizadas (0–1).
– Calcula si el centroide de cada persona detectada cae dentro de cada zona.
– Dibuja overlays (cajas, etiquetas, zonas y contadores) con OpenCV.
– Muestra FPS y estado de TPU.

Guarde el archivo como opencv_coral_person_zones.py en su directorio de trabajo.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import argparse
import time
import sys
import os
from collections import defaultdict

import numpy as np
import cv2

from picamera2 import Picamera2
from pycoral.adapters import common, detect
from pycoral.utils.edgetpu import make_interpreter

# ------------------------------------------------------------
# Configuración de zonas (normalizadas 0–1 respecto a (ancho, alto))
# Puede ajustar o añadir zonas aquí. Cada zona es un polígono.
# ------------------------------------------------------------
DEFAULT_ZONES = {
    "Zona A": [(0.05, 0.10), (0.45, 0.10), (0.45, 0.60), (0.05, 0.60)],
    "Zona B": [(0.55, 0.10), (0.95, 0.10), (0.95, 0.60), (0.55, 0.60)],
    "Zona C": [(0.20, 0.65), (0.80, 0.65), (0.95, 0.95), (0.05, 0.95)],
}

# ------------------------------------------------------------
# Utilidades de carga de etiquetas COCO
# ------------------------------------------------------------
def load_labels(path):
    lbls = {}
    with open(path, 'r', encoding='utf-8') as f:
        for line in f:
            pair = line.strip().split(maxsplit=1)
            if len(pair) == 2:
                lbls[int(pair[0])] = pair[1].strip()
    return lbls

# ------------------------------------------------------------
# Conversión de zonas normalizadas a píxeles
# ------------------------------------------------------------
def denorm_zone(zone_points_norm, width, height):
    pts = []
    for (x, y) in zone_points_norm:
        pts.append((int(x * width), int(y * height)))
    return np.array(pts, dtype=np.int32)

# ------------------------------------------------------------
# Dibujo de zonas y contadores
# ------------------------------------------------------------
def draw_zones_and_counts(frame_bgr, zones_px, counts, color=(0, 255, 255)):
    for name, poly in zones_px.items():
        cv2.polylines(frame_bgr, [poly], isClosed=True, color=color, thickness=2)
        # Rótulo con conteo
        moments = cv2.moments(poly)
        if moments["m00"] != 0:
            cx = int(moments["m10"] / moments["m00"])
            cy = int(moments["m01"] / moments["m00"])
        else:
            # Centro aproximado
            rect = cv2.boundingRect(poly)
            cx, cy = rect[0] + rect[2] // 2, rect[1] + rect[3] // 2
        label = f"{name}: {counts.get(name, 0)}"
        cv2.putText(frame_bgr, label, (cx - 40, cy),
                    cv2.FONT_HERSHEY_SIMPLEX, 0.6, color, 2, cv2.LINE_AA)

# ------------------------------------------------------------
# Test punto en polígono (centroide dentro de zona)
# ------------------------------------------------------------
def point_in_zone(point, poly):
    # poly: np.array Nx2 int32; point: (x, y)
    # Usamos cv2.pointPolygonTest: >0 dentro, 0 en borde, <0 fuera
    val = cv2.pointPolygonTest(poly, point, False)
    return val >= 0

# ------------------------------------------------------------
# Dibujo de cajas y etiquetas
# ------------------------------------------------------------
def draw_detection(frame_bgr, bbox, text, color=(0, 255, 0)):
    x0, y0, x1, y1 = bbox
    cv2.rectangle(frame_bgr, (x0, y0), (x1, y1), color, 2)
    cv2.putText(frame_bgr, text, (x0, max(0, y0 - 5)),
                cv2.FONT_HERSHEY_SIMPLEX, 0.6, color, 2, cv2.LINE_AA)

# ------------------------------------------------------------
# Pipeline principal
# ------------------------------------------------------------
def main():
    parser = argparse.ArgumentParser(description="opencv-coral-person-detection-zones (Raspberry Pi 4 + HQ Camera + Coral USB)")
    parser.add_argument("--model", required=False, default="models/ssd_mobilenet_v2_coco_quant_postprocess_edgetpu.tflite",
                        help="Ruta al modelo TFLite compilado para Edge TPU")
    parser.add_argument("--labels", required=False, default="models/coco_labels.txt",
                        help="Ruta al archivo de etiquetas COCO")
    parser.add_argument("--threshold", type=float, default=0.45, help="Umbral de confianza")
    parser.add_argument("--width", type=int, default=1280, help="Ancho de captura")
    parser.add_argument("--height", type=int, default=720, help="Alto de captura")
    parser.add_argument("--display", action="store_true", help="Mostrar ventana con OpenCV")
    parser.add_argument("--max-fps", type=float, default=20.0, help="FPS objetivo")
    args = parser.parse_args()

    # Carga etiquetas
    labels = load_labels(args.labels)
    # Determinar id de "person" en archivo COCO (normalmente 1 en los labels de Coral)
    person_ids = {k for k, v in labels.items() if v.lower() == "person"}
    if not person_ids:
        print("ADVERTENCIA: No se encontró la etiqueta 'person' en el archivo de labels. Continuará, pero filtre manualmente si corresponde.", file=sys.stderr)

    # Intérprete Edge TPU
    print(f"Cargando modelo Edge TPU: {args.model}")
    interpreter = make_interpreter(args.model)
    interpreter.allocate_tensors()
    in_w, in_h = common.input_size(interpreter)
    print(f"Entrada modelo: {in_w}x{in_h}")

    # Configurar cámara con Picamera2
    picam2 = Picamera2()
    config = picam2.create_video_configuration(main={"size": (args.width, args.height), "format": "RGB888"})
    picam2.configure(config)
    picam2.start()
    time.sleep(0.5)  # Calentamiento sensor

    # Preparar zonas (en píxeles)
    zones_px = {name: denorm_zone(pts, args.width, args.height) for name, pts in DEFAULT_ZONES.items()}

    # Control de FPS
    frame_period = 1.0 / max(1e-3, args.max_fps)
    last_time = time.time()
    fps_avg = 0.0
    fps_alpha = 0.9  # EMA

    try:
        while True:
            now = time.time()
            if now - last_time < frame_period:
                time.sleep(0.001)
                continue
            last_time = now

            # Captura frame RGB desde Picamera2
            frame_rgb = picam2.capture_array()
            h, w, _ = frame_rgb.shape

            # Preprocesamiento: redimensionar a tamaño de entrada del modelo
            # Nota: common.set_resized_input devuelve 'scale' para detect.get_objects
            resized = cv2.resize(frame_rgb, (in_w, in_h), interpolation=cv2.INTER_NEAREST)
            common.set_input(interpreter, resized)

            # Inferencia Edge TPU
            t0 = time.time()
            interpreter.invoke()
            # 'scale' si hubiéramos preservado aspecto; aquí redimensionamos exacto, así image_scale = 1.0
            objs = detect.get_objects(interpreter, args.threshold, image_scale=1.0)
            t1 = time.time()
            infer_ms = (t1 - t0) * 1000.0

            # Postprocesado: escalar cajas al tamaño original
            # Las cajas están relativas al tamaño de entrada (in_w, in_h)
            sx, sy = w / in_w, h / in_h
            zone_counts = defaultdict(int)

            # Convertir frame a BGR para OpenCV visual
            frame_bgr = cv2.cvtColor(frame_rgb, cv2.COLOR_RGB2BGR)

            for obj in objs:
                cls_id = obj.id
                score = obj.score
                # Filtrar solo 'person'
                if person_ids and cls_id not in person_ids:
                    continue

                bbox = obj.bbox
                # bbox.xmin, ymin, xmax, ymax en coords del input
                x0 = max(0, min(w - 1, int(bbox.xmin * sx)))
                y0 = max(0, min(h - 1, int(bbox.ymin * sy)))
                x1 = max(0, min(w - 1, int(bbox.xmax * sx)))
                y1 = max(0, min(h - 1, int(bbox.ymax * sy)))
                cx = int((x0 + x1) / 2)
                cy = int((y0 + y1) / 2)

                # Determinar zona por centroide
                hit_zones = []
                for name, poly in zones_px.items():
                    if point_in_zone((cx, cy), poly):
                        zone_counts[name] += 1
                        hit_zones.append(name)

                label = f"person {score:.2f}"
                if hit_zones:
                    label += " [" + ",".join(hit_zones) + "]"
                    color = (0, 165, 255)  # Naranja si dentro de zonas
                else:
                    color = (0, 255, 0)    # Verde si fuera

                draw_detection(frame_bgr, (x0, y0, x1, y1), label, color)

                # Marcar centroide
                cv2.circle(frame_bgr, (cx, cy), 4, (255, 0, 0), -1)

            # Dibujar zonas y contadores
            draw_zones_and_counts(frame_bgr, zones_px, zone_counts, color=(0, 255, 255))

            # FPS estimados
            cur_fps = 1.0 / max(1e-6, (time.time() - now))
            fps_avg = fps_alpha * fps_avg + (1 - fps_alpha) * cur_fps

            # HUD
            hud = f"FPS:{fps_avg:5.1f}  Inference:{infer_ms:4.1f} ms  Model:{os.path.basename(args.model)}"
            cv2.putText(frame_bgr, hud, (10, 20),
                        cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255, 255, 0), 2, cv2.LINE_AA)

            if args.display:
                cv2.imshow("opencv-coral-person-detection-zones", frame_bgr)
                key = cv2.waitKey(1) & 0xFF
                if key == ord('q'):
                    break

    except KeyboardInterrupt:
        pass
    finally:
        picam2.stop()
        if args.display:
            cv2.destroyAllWindows()


if __name__ == "__main__":
    main()

Explicación breve de partes clave:
– Picamera2: acceso directo a frames RGB desde la HQ Camera con la pila libcamera (Bookworm).
– PyCoral + Edge TPU: make_interpreter carga el modelo TFLite compilado para TPU; common.set_input + interpreter.invoke ejecutan inferencia; detect.get_objects extrae objetos de la operación “Detection_PostProcess”.
– Zonas: definidas en coordenadas normalizadas para no depender de resolución. Se convierten a píxeles con denorm_zone. La pertenencia se comprueba con pointPolygonTest sobre el centroide de la caja, lo cual es robusto y eficiente.
– Overlay: OpenCV dibuja polígonos, cajas, centroides y textos con conteos por zona y métricas de rendimiento.

Además, proveemos un script de validación mínima del Coral con una imagen estática, por si desea verificar la inferencia sin cámara:

#!/usr/bin/env python3
# archivo: test_coral_image.py
import sys, cv2
import numpy as np
from pycoral.utils.edgetpu import make_interpreter
from pycoral.adapters import common, detect

model_path = sys.argv[1] if len(sys.argv) > 1 else "models/ssd_mobilenet_v2_coco_quant_postprocess_edgetpu.tflite"
img_path   = sys.argv[2] if len(sys.argv) > 2 else "models/grace_hopper.bmp"  # o cualquier imagen local

interpreter = make_interpreter(model_path)
interpreter.allocate_tensors()
in_w, in_h = common.input_size(interpreter)

img_bgr = cv2.imread(img_path)
if img_bgr is None:
    print("No se pudo leer la imagen:", img_path)
    sys.exit(1)
img_rgb = cv2.cvtColor(img_bgr, cv2.COLOR_BGR2RGB)
resized = cv2.resize(img_rgb, (in_w, in_h), interpolation=cv2.INTER_NEAREST)
common.set_input(interpreter, resized)

interpreter.invoke()
objs = detect.get_objects(interpreter, 0.3, image_scale=1.0)
print("Detecciones:", len(objs))
for o in objs:
    print("id:", o.id, "score:", o.score, "bbox:", o.bbox)

Compilación/flash/ejecución

A continuación, los pasos exactos y ordenados para reproducir el caso:

1) Preparar sistema (si aún no lo hizo):
bash
sudo apt update
sudo apt full-upgrade -y
sudo apt install -y python3-pip python3-venv python3-opencv python3-picamera2 libcamera-apps python3-numpy git wget curl python3-gpiozero python3-smbus python3-spidev

2) Activar overlay de IMX477 y GPU memoria (si aún no):
bash
sudo sed -i '$a dtoverlay=imx477\ngpu_mem=128' /boot/firmware/config.txt
sudo reboot

3) Instalar Edge TPU runtime:
bash
sudo install -d -m 0755 /usr/share/keyrings
curl -sSL https://packages.cloud.google.com/apt/doc/apt-key.gpg | sudo gpg --dearmor -o /usr/share/keyrings/coral-archive-keyring.gpg
echo "deb [signed-by=/usr/share/keyrings/coral-archive-keyring.gpg] https://packages.cloud.google.com/apt coral-edgetpu-stable main" | sudo tee /etc/apt/sources.list.d/coral-edgetpu.list
sudo apt update
sudo apt install -y libedgetpu1-std

4) Crear venv y dependencias Python de usuario:
bash
python3 -m venv --system-site-packages ~/venvs/pi-coral
source ~/venvs/pi-coral/bin/activate
pip install --upgrade pip wheel
pip install tflite-runtime==2.11.0 pycoral==2.0.0 numpy==1.26.4

5) Descarga de modelo y labels (SSD MobileNet v2 COCO para Edge TPU):
bash
mkdir -p ~/opencv-coral-zones/models
cd ~/opencv-coral-zones/models
wget -O ssd_mobilenet_v2_coco_quant_postprocess_edgetpu.tflite \
https://github.com/google-coral/test_data/raw/master/ssd_mobilenet_v2_coco_quant_postprocess_edgetpu.tflite
wget -O coco_labels.txt \
https://github.com/google-coral/test_data/raw/master/coco_labels.txt

6) Obtener el código del proyecto:
bash
cd ~/opencv-coral-zones
wget -O opencv_coral_person_zones.py https://raw.githubusercontent.com/your-org/your-repo/main/opencv_coral_person_zones.py # (si no tiene URL, copie/pegue el código en un archivo)
chmod +x opencv_coral_person_zones.py

Si no usa wget, cree el archivo con nano:
bash
nano opencv_coral_person_zones.py
# pegue el script completo, guarde con Ctrl+O, Enter; salga con Ctrl+X
chmod +x opencv_coral_person_zones.py

7) Prueba rápida de la cámara:
bash
libcamera-hello -t 2000

8) Prueba rápida del Coral con imagen:
bash
python test_coral_image.py models/ssd_mobilenet_v2_coco_quant_postprocess_edgetpu.tflite /usr/share/libcamera/pipeline_handler/data/test.jpg
# o coloque su propia imagen

9) Ejecución principal (con visualización):
bash
cd ~/opencv-coral-zones
source ~/venvs/pi-coral/bin/activate
./opencv_coral_person_zones.py \
--model models/ssd_mobilenet_v2_coco_quant_postprocess_edgetpu.tflite \
--labels models/coco_labels.txt \
--threshold 0.45 \
--width 1280 --height 720 \
--display \
--max-fps 20

10) Ejecución headless (sin ventana; útil si transmite frames a otro proceso):
bash
./opencv_coral_person_zones.py \
--model models/ssd_mobilenet_v2_coco_quant_postprocess_edgetpu.tflite \
--labels models/coco_labels.txt \
--threshold 0.45 \
--width 1280 --height 720

Validación paso a paso

1) Verificar que el sistema ve la cámara HQ:
– Comando:
bash
libcamera-hello -t 2000
– Debe abrirse una ventana breve sin errores. Si está por SSH sin X, puede usar:
bash
libcamera-still -o /tmp/test.jpg
y luego comprobar que /tmp/test.jpg existe y contiene una imagen válida.

2) Verificar el Coral USB:
– Comandos:
bash
lsusb | grep -i google
dmesg | grep -i edgetpu
– Debe listarse el dispositivo Google y mensajes del kernel/udev cargando el Edge TPU.

3) Verificar las bibliotecas y versiones:
– Comandos:
bash
python -V
python -c "import cv2, numpy; print('OpenCV', cv2.__version__, 'NumPy', numpy.__version__)"
python -c "import tflite_runtime.interpreter as tfl; print('TFLite', tfl.__version__)"
python -c "import pycoral; print('PyCoral OK')"
– Debe ver algo como:
– Python 3.11.x
– OpenCV 4.6.0
– NumPy 1.26.x
– TFLite 2.11.0
– PyCoral OK

4) Verificar inferencia en imagen estática:
– Comando:
bash
python test_coral_image.py models/ssd_mobilenet_v2_coco_quant_postprocess_edgetpu.tflite /usr/share/libcamera/pipeline_handler/data/test.jpg
– Debe imprimir un número de detecciones y sus cajas. No requiere cámara funcionando.

5) Ejecutar el pipeline en vivo:
– Comando:
bash
./opencv_coral_person_zones.py --model models/ssd_mobilenet_v2_coco_quant_postprocess_edgetpu.tflite --labels models/coco_labels.txt --display
– Resultado esperado:
– Una ventana con la imagen en vivo.
– Zonas dibujadas (amarillo/cian).
– Detecciones con caja y etiqueta “person X.XX” (verde si fuera de zonas, naranja si dentro).
– Contador por zona (Zona A: n, etc.).
– HUD con FPS y tiempo de inferencia.

6) Validación funcional “por zonas”:
– Mueva una persona frente a la cámara y entre/salga de las regiones dibujadas. Observe:
– El contador de la zona correspondiente incrementa/decrementa según presencia.
– La etiqueta de la detección incluye “[Zona X]” cuando el centroide cae dentro.

7) Pequeñas pruebas de robustez:
– Cambie la iluminación y verifique que la cámara mantiene frames estables.
– Pruebe resoluciones 640×480 o 1920×1080 (siempre respetando el rendimiento) modificando –width y –height.
– Ajuste –threshold a 0.5–0.6 para reducir falsos positivos y observe el impacto.

Troubleshooting

1) No se detecta la cámara (libcamera-hello falla):
– Verifique el cable CSI: orientación de los contactos y que las presillas están firmes.
– Revise /boot/firmware/config.txt e incluya dtoverlay=imx477 y gpu_mem=128. Reinicie.
– Asegure que está usando Raspberry Pi OS Bookworm 64-bit y no habilitó la “Legacy Camera”.
– Compruebe dmesg para errores relacionados a i2c/csi o sensor no encontrado.

2) No aparece el Coral USB:
– Conecte al puerto USB 3.0 (azul) directo, sin hubs. Pruebe otro cable USB.
– Compruebe alimentación adecuada (evite subvoltajes). Revise dmesg para “under-voltage”.
– lsusb debe listar un dispositivo Google; reinstale libedgetpu1-std si es necesario.
– Pruebe otro puerto USB o reinicie.

3) Error al invocar el intérprete o “Edge TPU not found”:
– Verifique que el runtime libedgetpu1-std está instalado correctamente:
bash
dpkg -l | grep libedgetpu1
– Asegure que sólo un proceso use el Coral a la vez.
– Ejecute un ejemplo mínimo de PyCoral para confirmar funcionamiento (test_coral_image.py).

4) Ventana OpenCV no abre por SSH:
– Si está por SSH sin forwarding X, use el modo sin ventana (no pase –display) y guarde frames a disco si necesita validar.
– Alternativamente, utilice un escritorio remoto o VNC.

5) FPS muy bajos:
– Reduzca la resolución (–width/–height), por ejemplo 640×480.
– Limite –max-fps a 10–15 si su CPU/GPU está saturada.
– Verifique que el Coral está en USB 3.0 (y no 2.0).

6) Falsos positivos o detecciones inestables:
– Aumente –threshold (0.55–0.6).
– Iluminación: evite contraluces extremos.
– Use ROI (zonas) más ajustadas para reducir el área de interés.

7) Zonas mal alineadas:
– Recuerde que las zonas están normalizadas 0–1; ajuste DEFAULT_ZONES para su encuadre.
– Compruebe proporción de aspecto; si cambia la resolución, las zonas se adaptan automáticamente, pero quizá desee reubicarlas.

8) Error “TFLite_Detection_PostProcess custom op not supported”:
– Asegúrese de usar tflite-runtime 2.11.0 y pycoral 2.0.0 con modelo “…_edgetpu.tflite” exacto.
– No use un modelo TFLite no compilado para Edge TPU.

Mejoras/variantes

• Persistencia y configuración externa:
• Cargue zonas desde un archivo YAML/JSON y permita edición sin tocar el código.

• Guarde eventos (timestamp, zona, score) en SQLite o CSV para auditorías.

• Estrategia de pertenencia a zona:

• En lugar de “centroide dentro”, calcule intersección de área entre caja y polígono (mask & bitwise AND) para casos donde la persona entra parcialmente.

• Contadores de paso:

• Trace líneas virtuales y detecte cruces del centroide (enter/exit) con estados temporales y direccionalidad.

• Modelo especializado “person-only”:

• Pruebe un modelo Edge TPU entrenado específicamente para personas para ganar precisión/latencia si el resto de clases no es relevante.

• Transmisión/servicios:

• Publique los resultados vía MQTT/HTTP (Flask/FastAPI) con JSON de detecciones por zona.

• Stream de vídeo con overlays por GStreamer RTSP o WebRTC.

• Rendimiento:

• Ajuste exposure/ISO de la HQ Camera, o fije FPS de captura para latencia estable.

• Use libedgetpu1-max si la refrigeración y alimentación lo permiten.

• Seguridad y robustez:

• Systemd service que arranque el script al boot y reinicie ante fallos.
• Logging estructurado (JSON) y métricas (Prometheus node exporter + exporter propio).

Checklist de verificación

• [ ] Raspberry Pi 4 Model B con Raspberry Pi OS Bookworm 64-bit actualizado.
• [ ] HQ Camera (IMX477) conectada al conector CSI con orientación correcta.
• [ ] dtoverlay=imx477 y gpu_mem=128 añadidos a /boot/firmware/config.txt; sistema reiniciado.
• [ ] Coral USB conectado a puerto USB 3.0 (azul); lsusb y dmesg lo reconocen.
• [ ] Paquetes apt instalados: python3-opencv, python3-picamera2, libcamera-apps, python3-numpy.
• [ ] Entorno virtual Python creado con –system-site-packages y activado.
• [ ] Dependencias Python instaladas en venv: tflite-runtime==2.11.0, pycoral==2.0.0, numpy==1.26.4.
• [ ] Modelo Edge TPU y labels descargados en ~/opencv-coral-zones/models.
• [ ] Prueba de cámara OK: libcamera-hello -t 2000.
• [ ] Prueba de Coral con imagen OK: test_coral_image.py reporta detecciones.
• [ ] Script principal ejecuta, muestra zonas, cajas y contadores.
• [ ] Al menos una detección de “person” actualiza correctamente los contadores de zona.
• [ ] FPS y latencia de inferencia razonables para su resolución (p. ej., ~15–25 FPS a 1280×720 con Coral).

Con este caso práctico, ha implementado un pipeline avanzado de “opencv-coral-person-detection-zones” totalmente reproducible y coherente con el hardware Raspberry Pi 4 Model B + Raspberry Pi HQ Camera (IMX477) + Google Coral USB (Edge TPU), utilizando Raspberry Pi OS Bookworm 64‑bit y Python 3.11, con la toolchain fijada en versiones conocidas y estables.

Objetivo y caso de uso

Qué construirás: Un robot que utiliza SLAM 2D para mapear y navegar en interiores con Raspberry Pi 5 y RPLIDAR A1.

Para qué sirve

• Navegación autónoma en entornos interiores utilizando mapeo en tiempo real.
• Generación de mapas 2D precisos para la planificación de rutas en robots móviles.
• Integración de datos de sensores para mejorar la toma de decisiones en robótica.
• Demostración de capacidades de SLAM en proyectos educativos y de investigación.

Resultado esperado

• Mapas generados con precisión de hasta 2 cm en entornos interiores.
• Latencias de procesamiento de datos de sensor de menos de 100 ms.
• Velocidad de navegación del robot de hasta 0.5 m/s con control de motores eficiente.
• Capacidad de manejar hasta 10 paquetes de datos por segundo desde el RPLIDAR A1.

Público objetivo: Desarrolladores y estudiantes avanzados de robótica; Nivel: Avanzado

Arquitectura/flujo: Raspberry Pi 5 + RPLIDAR A1 + TB6612FNG + Python para control y procesamiento de datos.

Nivel: Avanzado

Prerrequisitos

• Sistema operativo:
• Raspberry Pi OS Bookworm 64-bit (Debian 12), kernel 6.6.x o superior. Verificación rápida:
• cat /etc/os-release debe mostrar VERSION_CODENAME=bookworm
• uname -r debe mostrar 6.6.* (o superior)
• Modelo exacto del hardware:
• Raspberry Pi 5
• Slamtec RPLIDAR A1 (con adaptador USB oficial de Slamtec)
• Driver de motores TB6612FNG (para dos motores DC en configuración diferencial)
• Toolchain y versiones exactas utilizadas en este caso:
• Python 3.11.2 (Bookworm) — ver: python3 --version
• pip 24.2
• setuptools 70.0.0
• wheel 0.43.0
• gcc 12.2.0 — ver: gcc --version
• cmake 3.25.1 — ver: cmake --version (opcional, solo si compilas librerías nativas)
• Daemon pigpio (apt) v79 — ver: pigpiod -v y pigs -v
• Librerías Python (instaladas en venv):
• rplidar 0.9.3
• breezyslam 0.0.8
• numpy 1.26.4
• pigpio 1.78
• pillow 10.4.0
• gpiozero 2.0 (para utilidades GPIO; el control principal será con pigpio)
• Conocimientos previos:
• Linux básico, consola, edición de archivos.
• Conceptos de PWM y control H-bridge (TB6612FNG).
• Fundamentos de SLAM 2D y navegación diferencial.

Materiales

• Raspberry Pi 5 (4GB u 8GB; ambas funcionan para este caso).
• Slamtec RPLIDAR A1 con su base y adaptador USB oficial de Slamtec.
• TB6612FNG motor driver (placa con terminales VM, VCC, STBY, AIN1/AIN2, BIN1/BIN2, PWMA/PWMB).
• Dos motores DC de 6–12 V (una base robótica diferencial típica).
• Batería para motores (ej. 7.4 V LiPo 2S) y cableado correspondiente.
• Fuente 5V/3A (o superior) para Raspberry Pi 5 mediante USB-C.
• Cables Dupont (hembra-hembra y hembra-macho según corresponda).
• Cable USB-A a Micro-USB (o USB-C si tu base usa adaptador distinto) para conectar el RPLIDAR A1 al puerto USB 3.0 de la Raspberry Pi 5.
• Estructura/chasis para montaje (plataforma móvil con soporte para lidar).

Nota: Usa siempre GND común entre la Raspberry Pi 5 y el TB6612FNG/baterías de motor.

Preparación y conexión

Habilitar interfaces y servicios

• Actualizar el sistema:
• sudo apt update
• sudo apt full-upgrade -y

• sudo reboot

• Habilitar opciones y preparar permisos:

• Interfaces (opcional pero recomendado):
• sudo raspi-config
• Interface Options:
  • SSH: Enable (si gestionarás remotamente)
  • Serial Port: Disable login shell, y puedes dejar el puerto serial deshabilitado porque el LIDAR irá por USB.
  • I2C/SPI: No necesarios en este proyecto (puedes dejarlos deshabilitados).
• Finish y reboot si lo pide.
• Arrancar el daemon de pigpio:
• sudo systemctl enable –now pigpiod
• Añadir el usuario al grupo dialout para acceso a /dev/ttyUSB0 (LIDAR):
• sudo usermod -aG dialout $USER

• Cierra sesión y vuelve a entrar (o newgrp dialout).

• Comprobar LIDAR y pigpio:

• Conectar el RPLIDAR A1 al puerto USB 3.0 (azul) de la Raspberry Pi 5.
• Verificar dispositivo:
• ls -l /dev/ttyUSB*
• dmesg | tail
• Verificar pigpio:
• pigs -v (debe mostrar versión 79)
• pigpiod -v (debe coincidir)

Cableado entre Raspberry Pi 5 y TB6612FNG

• Conexiones de potencia:
• VM (TB6612FNG): al positivo de la batería de motores (p. ej., 7.4 V LiPo).
• VCC (TB6612FNG): a 3.3 V de la Raspberry Pi (pin físico 1 o 17).

• GND (TB6612FNG): a GND de la Raspberry Pi (p. ej., pin físico 6). GND de la batería y de la Pi deben estar comunes.

• Conexiones de control (asignación propuesta coherente con PWM hardware):

• Motores:
• A01/A02 a las dos terminales del motor izquierdo (Motor A).
• B01/B02 a las dos terminales del motor derecho (Motor B).

• Si ves que un motor gira al revés, intercambia A01/A02 (o BIN1/BIN2 para el B, etc.) o invierte la lógica en software.

• LIDAR:

• Conecta la base/adaptador USB oficial del RPLIDAR A1 a un puerto USB 3.0 (azul). No es necesario cablear nada al GPIO para el LIDAR.

Código completo (Python 3.11)

A continuación se presenta un único script Python que:
– Controla los motores a través del TB6612FNG usando pigpio a 20 kHz de PWM.
– Lee el RPLIDAR A1 por /dev/ttyUSB0.
– Ejecuta SLAM 2D (breezyslam) construyendo un mapa en tiempo real.
– Implementa dos modos: exploración (explore) y navegación a objetivo (goto).
– Guarda periódicamente la imagen del mapa en PNG como validación.

Guarda este archivo como slam_nav.py en tu proyecto.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# Proyecto: lidar-2d-slam-navigation
# Hardware: Raspberry Pi 5 + Slamtec RPLIDAR A1 + TB6612FNG motor driver
# Toolchain:
#   - Raspberry Pi OS Bookworm 64-bit (Debian 12), kernel 6.6+
#   - Python 3.11.2, pip 24.2, setuptools 70.0.0, wheel 0.43.0
#   - pigpio daemon v79 (apt), pigpio Python 1.78
#   - rplidar 0.9.3, breezyslam 0.0.8, numpy 1.26.4, pillow 10.4.0
#
import argparse
import math
import os
import signal
import sys
import time

import numpy as np
from PIL import Image

import pigpio
from rplidar import RPLidar
from breezyslam.slam import RMHC_SLAM
from breezyslam.sensors import Laser

# ----------------------------
# Parámetros de hardware
# ----------------------------
# Raspberry Pi 5 -> TB6612FNG
GPIO_PWMA = 12   # Motor A PWM (PWM0) pin físico 32
GPIO_AIN1 = 23
GPIO_AIN2 = 24
GPIO_PWMB = 13   # Motor B PWM (PWM1) pin físico 33
GPIO_BIN1 = 27
GPIO_BIN2 = 22
GPIO_STBY = 5

PWM_FREQ = 20000  # 20 kHz para evitar zumbidos audibles
PWM_RANGE = 255   # Rango estándar pigpio para duty

# Geometría del robot (ajústala a tu base)
WHEEL_BASE_M = 0.16   # distancia entre ruedas en metros
MAX_ABS_SPEED = 1.0   # escala normalizada [-1..1] para motores

# ----------------------------
# Parámetros de LIDAR y SLAM
# ----------------------------
LIDAR_PORT = '/dev/ttyUSB0'   # Común en RPLIDAR A1 con adaptador USB
LIDAR_SCAN_SIZE = 360         # 1 grado por muestra
LIDAR_MAX_RANGE_MM = 6000     # 6 m
LIDAR_MIN_RANGE_MM = 150      # ignorar ecos muy cercanos no fiables

# Mapa SLAM
MAP_SIZE_PIXELS = 800
MAP_SIZE_METERS = 16.0  # 2 cm/pixel aprox.

# ----------------------------
# Clases auxiliares
# ----------------------------
class RPLidarA1(Laser):
    def __init__(self):
        super().__init__(scan_size=LIDAR_SCAN_SIZE,
                         scan_rate_hz=10,               # A1 ~5-10 Hz
                         detection_margin=15,           # grados ignorados al límite
                         distance_noise=50.0,           # mm
                         min_range=LIDAR_MIN_RANGE_MM,
                         max_range=LIDAR_MAX_RANGE_MM)

class Motor:
    def __init__(self, pi, pwm_pin, in1_pin, in2_pin, name="M"):
        self.pi = pi
        self.pwm_pin = pwm_pin
        self.in1_pin = in1_pin
        self.in2_pin = in2_pin
        self.name = name

        for p in (self.pwm_pin, self.in1_pin, self.in2_pin):
            self.pi.set_mode(p, pigpio.OUTPUT)
        self.pi.set_PWM_frequency(self.pwm_pin, PWM_FREQ)
        self.pi.set_PWM_range(self.pwm_pin, PWM_RANGE)
        self.set_speed(0.0)

    def set_speed(self, speed):
        # speed en [-1.0, 1.0]
        sp = max(-1.0, min(1.0, speed))
        if sp > 0:
            self.pi.write(self.in1_pin, 1)
            self.pi.write(self.in2_pin, 0)
        elif sp < 0:
            self.pi.write(self.in1_pin, 0)
            self.pi.write(self.in2_pin, 1)
        else:
            # freno suave: entradas a 0 (coast) o high-high (brake)
            self.pi.write(self.in1_pin, 0)
            self.pi.write(self.in2_pin, 0)

        duty = int(abs(sp) * PWM_RANGE)
        self.pi.set_PWM_dutycycle(self.pwm_pin, duty)

    def brake(self):
        self.pi.write(self.in1_pin, 1)
        self.pi.write(self.in2_pin, 1)
        self.pi.set_PWM_dutycycle(self.pwm_pin, 0)

class TB6612Robot:
    def __init__(self):
        self.pi = pigpio.pi()
        if not self.pi.connected:
            raise RuntimeError("No se pudo conectar a pigpio. ¿Está el daemon pigpiod en ejecución?")

        # STBY alto para habilitar
        self.pi.set_mode(GPIO_STBY, pigpio.OUTPUT)
        self.pi.write(GPIO_STBY, 1)

        self.motor_left = Motor(self.pi, GPIO_PWMA, GPIO_AIN1, GPIO_AIN2, name="L")
        self.motor_right = Motor(self.pi, GPIO_PWMB, GPIO_BIN1, GPIO_BIN2, name="R")

    def set_twist(self, v_lin, v_ang):
        # v_lin en [-1..1], v_ang en rad/s escalado
        v = max(-1.0, min(1.0, v_lin))
        w = v_ang
        # Mezcla diferencial (normalizada)
        # Asumimos w ya escalado a [-1..1] aprox. tras controlador
        left = v - 0.5 * w
        right = v + 0.5 * w

        # Normalización si se excede
        m = max(1.0, abs(left), abs(right))
        left /= m
        right /= m

        self.motor_left.set_speed(left * MAX_ABS_SPEED)
        self.motor_right.set_speed(right * MAX_ABS_SPEED)

    def stop(self):
        self.motor_left.set_speed(0.0)
        self.motor_right.set_speed(0.0)

    def shutdown(self):
        try:
            self.stop()
            self.motor_left.brake()
            self.motor_right.brake()
            self.pi.write(GPIO_STBY, 0)
        except Exception:
            pass
        self.pi.stop()

class SLAMNavigator:
    def __init__(self, map_pixels=MAP_SIZE_PIXELS, map_meters=MAP_SIZE_METERS):
        self.sensor = RPLidarA1()
        self.slam = RMHC_SLAM(self.sensor, map_pixels, map_meters)
        self.map_bytes = bytearray(map_pixels * map_pixels)
        self.distances = [0] * LIDAR_SCAN_SIZE
        self.pose = (0, 0, 0)  # x_mm, y_mm, theta_deg

    def update_with_scan(self, scan):
        # scan: lista de tuplas (quality, angle_deg, distance_mm)
        # Convertimos a vector de distancias indexado por ángulo entero
        for i in range(LIDAR_SCAN_SIZE):
            self.distances[i] = 0

        for (_, angle, distance) in scan:
            a = int(angle) % 360
            d = int(distance)
            if LIDAR_MIN_RANGE_MM <= d <= LIDAR_MAX_RANGE_MM:
                self.distances[a] = d

        # SLAM
        self.slam.update(self.distances)
        self.pose = self.slam.getpos()
        self.slam.getmap(self.map_bytes)
        return self.pose

    def save_map_png(self, path="map_latest.png"):
        side = int(math.sqrt(len(self.map_bytes)))
        img = np.frombuffer(self.map_bytes, dtype=np.uint8).copy().reshape((side, side))
        # rotar para que "x derecha, y hacia arriba" sea intuitivo al visualizar
        img = np.flipud(img)
        im = Image.fromarray(img, mode='L')
        im.save(path)

def angle_wrap_pi(a):
    while a > math.pi:
        a -= 2*math.pi
    while a < -math.pi:
        a += 2*math.pi
    return a

def polar_to_cartesian(d_mm, angle_deg):
    a = math.radians(angle_deg)
    return d_mm * math.cos(a), d_mm * math.sin(a)

def compute_front_clearance(distances, span_deg=20):
    # Busca la distancia mínima en el sector frontal [-span, +span]
    vals = []
    for i in range(360 - span_deg, 360):
        if distances[i] > 0:
            vals.append(distances[i])
    for i in range(0, span_deg + 1):
        if distances[i] > 0:
            vals.append(distances[i])
    if not vals:
        return LIDAR_MAX_RANGE_MM
    return min(vals)

def compute_free_direction(distances, sector=60):
    # Escoge dirección (signo) con mayor espacio libre: izquierda vs derecha
    left = []
    right = []
    for i in range(1, sector + 1):
        # derecha: 0..+sector
        if distances[i] > 0:
            right.append(distances[i])
        # izquierda: -sector..-1 => 360-sector..359
        j = (360 - i) % 360
        if distances[j] > 0:
            left.append(distances[j])
    left_min = min(left) if left else LIDAR_MAX_RANGE_MM
    right_min = min(right) if right else LIDAR_MAX_RANGE_MM
    return -1 if left_min > right_min else +1  # -1 girar izquierda, +1 derecha

def main():
    parser = argparse.ArgumentParser(description="Lidar 2D SLAM Navigation (RPi5 + RPLIDAR A1 + TB6612FNG)")
    parser.add_argument("--mode", choices=["explore", "goto", "test-motors"], default="explore",
                        help="explore (mapeo reactivo), goto (ir a objetivo), test-motors (prueba PWM)")
    parser.add_argument("--goal-x", type=float, default=2.0, help="Objetivo X en metros (solo modo goto)")
    parser.add_argument("--goal-y", type=float, default=1.0, help="Objetivo Y en metros (solo modo goto)")
    parser.add_argument("--map-save-interval", type=float, default=5.0, help="Intervalo para guardar PNG del mapa (s)")
    parser.add_argument("--forward-speed", type=float, default=0.5, help="Velocidad lineal normalizada base")
    parser.add_argument("--turn-gain", type=float, default=1.0, help="Ganancia de giro (rad -> comando)")
    parser.add_argument("--avoid-dist", type=float, default=0.5, help="Distancia mínima frontal (m)")
    args = parser.parse_args()

    robot = TB6612Robot()
    lidar = RPLidar(port=LIDAR_PORT)
    nav = SLAMNavigator()

    run = True

    def sigint_handler(sig, frame):
        nonlocal run
        run = False

    signal.signal(signal.SIGINT, sigint_handler)
    signal.signal(signal.SIGTERM, sigint_handler)

    last_save = 0.0

    try:
        if args.mode == "test-motors":
            print("Probando motores: adelante 1s, atrás 1s, giro en sitio 1s...")
            robot.set_twist(0.6, 0.0)
            time.sleep(1.0)
            robot.set_twist(-0.6, 0.0)
            time.sleep(1.0)
            robot.set_twist(0.0, 1.0)
            time.sleep(1.0)
            robot.stop()
            print("Test completado.")
            return

        print("Iniciando LIDAR...")
        lidar.start_motor()
        time.sleep(0.5)

        print("Leyendo escaneos y ejecutando SLAM...")
        for scan in lidar.iter_scans(max_buf_meas=5000):
            pose_mm = nav.update_with_scan(scan)
            x_mm, y_mm, theta_deg = pose_mm
            x_m, y_m = x_mm / 1000.0, y_mm / 1000.0
            theta_rad = math.radians(theta_deg)

            # Distancia frontal para evitar colisiones
            front_min_mm = compute_front_clearance(nav.distances, span_deg=20)
            front_min_m = front_min_mm / 1000.0

            v_cmd = 0.0
            w_cmd = 0.0

            if args.mode == "explore":
                # Navegación reactiva para mapear: avanza si despejado, si no esquiva
                if front_min_m < args.avoid_dist:
                    sgn = compute_free_direction(nav.distances, sector=60)
                    v_cmd = 0.0
                    w_cmd = 0.9 * sgn
                else:
                    v_cmd = args.forward_speed
                    # Pequeño giro para explorar suavemente
                    w_cmd = 0.1 * math.sin(time.time())

            elif args.mode == "goto":
                # Control hacia el objetivo con evasión de obstáculos
                dx = args.goal_x - x_m
                dy = args.goal_y - y_m
                goal_dist = math.hypot(dx, dy)
                goal_heading = math.atan2(dy, dx)
                heading_error = angle_wrap_pi(goal_heading - theta_rad)

                # Evasión simple
                if front_min_m < args.avoid_dist:
                    sgn = compute_free_direction(nav.distances, sector=60)
                    v_cmd = 0.0
                    w_cmd = 0.9 * sgn
                else:
                    # Proporcional a distancia y error angular
                    v_cmd = max(0.0, min(args.forward_speed, 0.4 + 0.2 * goal_dist))
                    w_cmd = max(-1.0, min(1.0, args.turn_gain * heading_error))

                # Llega al objetivo (umbral 0.15 m)
                if goal_dist < 0.15:
                    v_cmd = 0.0
                    w_cmd = 0.0
                    robot.stop()
                    print("Objetivo alcanzado.")
                    run = False

            robot.set_twist(v_cmd, w_cmd)

            # Guardar mapa periódicamente
            now = time.time()
            if now - last_save > args.map_save_interval:
                nav.save_map_png("map_latest.png")
                last_save = now
                print(f"Pose: ({x_m:.2f} m, {y_m:.2f} m, {theta_deg:.1f}°), front={front_min_m:.2f} m; mapa actualizado.")

            if not run:
                break

    except KeyboardInterrupt:
        pass
    finally:
        try:
            robot.stop()
        except Exception:
            pass
        try:
            lidar.stop()
            lidar.stop_motor()
            lidar.disconnect()
        except Exception:
            pass
        robot.shutdown()
        print("Apagado seguro completado.")

if __name__ == "__main__":
    main()

Breve explicación de partes clave:
– TB6612Robot/Motor: administración de pines de dirección (IN1/IN2) y PWM por pigpio a 20 kHz. Standby (STBY) se pone en alto para habilitar el driver.
– RPLidarA1 y SLAMNavigator: configuración para 360 muestras por revolución, 6 m de alcance; uso de RMHC_SLAM de BreezySLAM; conversión de escaneo a vector de distancias por ángulo.
– Control reactive y de objetivo: en modo “explore” avanza si no hay obstáculos y explora con leve oscilación; en “goto” combina control al objetivo con evasión frontal simple.
– Map saving: genera map_latest.png periódicamente para validar el SLAM.

Compilación/flash/ejecución

Sigue estos pasos exactamente en la Raspberry Pi 5 con Raspberry Pi OS Bookworm 64-bit:

1) Preparación del entorno y dependencias del sistema:
– sudo apt update
– sudo apt install -y python3.11-venv python3-dev build-essential pkg-config libatlas-base-dev libopenblas-dev pigpio
– sudo systemctl enable –now pigpiod

2) Crear y activar un entorno virtual para aislar dependencias de Python:
– python3 –version
– Debe ser 3.11.x
– python3 -m venv ~/venvs/lidar-slam
– source ~/venvs/lidar-slam/bin/activate
– pip install –upgrade pip==24.2 setuptools==70.0.0 wheel==0.43.0

3) Instalar las librerías Python exactas:
– pip install rplidar==0.9.3 breezyslam==0.0.8 numpy==1.26.4 pigpio==1.78 pillow==10.4.0 gpiozero==2.0

4) Crear el proyecto y copiar el código:
– mkdir -p ~/projects/lidar-2d-slam-navigation
– cd ~/projects/lidar-2d-slam-navigation
– nano slam_nav.py
– Pega el código completo anterior y guarda.

5) Comprobar el LIDAR y permisos:
– ls -l /dev/ttyUSB*
– Debe aparecer /dev/ttyUSB0 (si hay más de un adaptador USB, puede ser ttyUSB1).
– Si no eres parte de dialout (necesario para acceder a /dev/ttyUSB0):
– sudo usermod -aG dialout $USER
– Cierra sesión y vuelve a entrar o ejecuta: newgrp dialout

6) Prueba rápida de motores (sin LIDAR):
– Con la batería de motores conectada (VM), y verificando que las ruedas estén en el aire:
– python3 slam_nav.py –mode test-motors
– Debes ver un avance, retroceso y giro en sitio.

7) Ejecución en modo exploración (mapear y evitar obstáculos):
– python3 slam_nav.py –mode explore –forward-speed 0.5 –avoid-dist 0.5 –map-save-interval 5
– El mapa se guardará cada ~5 s en map_latest.png.

8) Ejecución en modo navegación hacia objetivo (coordenadas en metros desde el origen inicial):
– python3 slam_nav.py –mode goto –goal-x 2.0 –goal-y 1.0 –turn-gain 1.0 –forward-speed 0.5
– El robot intentará alcanzar (2.0 m, 1.0 m) evitando obstáculos.

Opcional: para crear una regla udev que cree un nombre estable /dev/rplidar (en lugar de /dev/ttyUSB0):

# /etc/udev/rules.d/99-rplidar.rules
SUBSYSTEM=="tty", ATTRS{idVendor}=="10c4", ATTRS{idProduct}=="ea60", SYMLINK+="rplidar", GROUP="dialout", MODE="0660"

• sudo udevadm control –reload-rules
• Desconecta y reconecta el RPLIDAR
• Usa LIDAR_PORT = «/dev/rplidar» en el script si aplicas esta regla.

Validación paso a paso

1) Verificar pigpio y PWM:
– pigs -v
– Debe mostrar versión v79. Si “connection failed”, el daemon no está corriendo.
– Ejecuta test de motores:
– python3 slam_nav.py –mode test-motors
– Observa:
– Ruedas giran hacia adelante 1 s, luego hacia atrás 1 s, luego giro en sitio 1 s.
– Si un motor gira invertido, invierte las conexiones A01/A02 o BIN1/BIN2 (o ajusta el signo en software).

2) Verificar LIDAR por USB:
– ls -l /dev/ttyUSB*
– dmesg | grep ttyUSB
– Si aparece /dev/ttyUSB0, procede.

3) Verificar lectura del LIDAR y SLAM en modo exploración:
– Coloca el robot en el suelo con espacio alrededor.
– python3 slam_nav.py –mode explore
– Observa:
– En consola, cada ~5 s, se imprime la pose (x, y, theta) y se guarda map_latest.png.
– Si hay obstáculos, el robot debería evitar chocar y seguir avanzando.

4) Verificar el mapa:
– Abre el archivo map_latest.png desde la propia Pi:
– xdg-open map_latest.png
– O transfiérelo a tu PC por SCP:
– scp pi@:~/projects/lidar-2d-slam-navigation/map_latest.png .
– Compara el entorno real con el mapa generado (paredes y pasillos deben verse coherentes).

5) Validar navegación hacia objetivo:
– Determina un punto objetivo seguro dentro del rango de tu espacio de pruebas (p. ej., +2 m en x, +1 m en y desde la posición inicial).
– python3 slam_nav.py –mode goto –goal-x 2.0 –goal-y 1.0
– Observa:
– El robot debe orientarse hacia el objetivo, esquivar obstáculos simples y detenerse cerca de la meta (< 0.15 m).
– El mapa debe seguir actualizándose y mejorando la consistencia local.

6) Control de estabilidad:
– Si notas oscilaciones de giro, reduce –turn-gain a 0.7 o 0.5.
– Si el robot se detiene por falsas detecciones, reduce –avoid-dist (p. ej., 0.4).

Troubleshooting

1) El LIDAR no aparece como /dev/ttyUSB0:
– Causa: controlador USB a serie distinto, múltiples dispositivos USB, o cable defectuoso.
– Solución:
– Revisa dmesg | tail justo al conectar.
– Prueba con otro puerto USB 3.0 o un cable diferente.
– Confirma permisos (grupo dialout).
– Usa la regla udev para symlink estable /dev/rplidar.

2) Error “No se pudo conectar a pigpio”:
– Causa: el daemon pigpiod no está en ejecución.
– Solución:
– sudo systemctl enable –now pigpiod
– Verifica con pigs -v.
– Si usas un firewall/selinux (raro en Pi OS), desactívalo para pruebas.

3) Motores no giran o zumban sin moverse:
– Causa: STBY en bajo, VM sin batería, PWM a 0, o cableado IN1/IN2 invertido.
– Solución:
– Comprueba STBY en alto (GPIO 5).
– Verifica que la batería VM esté conectada y con tensión suficiente.
– Ejecuta modo test-motors para aislar problema de SLAM/LIDAR.
– Revisa IN1/IN2: con velocidad positiva IN1=1, IN2=0; con negativa, al revés.

4) El mapa sale borroso o inconsistente:
– Causa: vibraciones, superficies con poca textura o velocidad demasiado alta.
– Solución:
– Reduce la velocidad (–forward-speed 0.3–0.4).
– Asegura físicamente el RPLIDAR A1 para minimizar vibraciones.
– Evita moverse cerca de objetos muy brillantes o translúcidos.

5) “Permission denied” al abrir /dev/ttyUSB0:
– Causa: usuario no pertenece a dialout o el archivo cambió de grupo/permisos.
– Solución:
– sudo usermod -aG dialout $USER; cierra sesión y regresa.
– Verifica ls -l /dev/ttyUSB0 (debe ser grupo dialout).
– Aplica regla udev si es necesario.

6) SLAM no converge o la posición deriva:
– Causa: trayectorias con pocos rasgos o escaneo bloqueado.
– Solución:
– Asegura un movimiento gradual (no gires demasiado rápido).
– Evita pasillos demasiado largos sin referencias (añade giros).
– Aumenta el tiempo de estacionario inicial para “anclar” el mapa.

7) PWM ruidoso o motores calientes:
– Causa: frecuencia inadecuada o duty excesivo.
– Solución:
– Mantén 20 kHz para salir del rango audible.
– Reduce MAX_ABS_SPEED si tus motores no están refrigerados.

8) El robot gira en sentido opuesto (invertido):
– Causa: polaridad de motores o cableado IN1/IN2/BIN1/BIN2 invertidos.
– Solución:
– Intercambia A01/A02 (motor A) o B01/B02 (motor B) o invierte las líneas INx.
– Alternativamente, multiplica por -1 el comando de ese motor en software.

Mejoras/variantes

• Encoders en ruedas: Añadir odometría para mejorar la estabilidad del SLAM y la navegación hacia objetivos, fusionando odometría con el scan matching del LIDAR.
• Planeación de rutas: Implementar A o D Lite sobre la cuadrícula de ocupación del mapa, generando trayectorias más suaves que el control reactivo.
• Control avanzado: PID para velocidad de ruedas y control de orientación con límites de aceleración/jerk para proteger la mecánica.
• Fusión sensorial: Integrar IMU (giroscopio) para mejorar la estimación de orientación y reducir deriva en giros.
• Persistencia de mapas: Guardar/cargar mapas con timestamps y permitir localización en mapas previamente construidos.
• ROS 2 (opcional): Migrar a ROS 2 Iron/Humble en RPi 5, con nodos para RPLIDAR, SLAM (Hector SLAM o slam_toolbox) y navegación (nav2). Este caso mantiene la consistencia con Python puro por simplicidad de despliegue.
• Seguridad y energía: Añadir supervisión de voltaje de batería y relé de corte de emergencia por GPIO.

Checklist de verificación

• [ ] Sistema: Raspberry Pi OS Bookworm 64-bit instalado y actualizado (kernel 6.6+).
• [ ] Usuario en grupo dialout; /dev/ttyUSB0 visible al conectar el RPLIDAR A1.
• [ ] pigpio daemon habilitado y en ejecución (pigs -v OK).
• [ ] Cableado TB6612FNG coincidente con la tabla (PWMA=GPIO12, PWMB=GPIO13, AIN1=GPIO23, AIN2=GPIO24, BIN1=GPIO27, BIN2=GPIO22, STBY=GPIO5, VCC=3.3V, GND común).
• [ ] Batería de motores conectada a VM; GND común con la Raspberry Pi 5.
• [ ] Entorno virtual creado y activado; pip 24.2; librerías instaladas con versiones exactas (rplidar 0.9.3, breezyslam 0.0.8, numpy 1.26.4, pigpio 1.78, pillow 10.4.0, gpiozero 2.0).
• [ ] Modo test-motors completado con éxito (adelante, atrás, giro).
• [ ] Modo explore: robot avanza y evita obstáculos; map_latest.png se actualiza periódicamente.
• [ ] Modo goto: robot se orienta y navega hacia las coordenadas objetivo evitando obstáculos; se detiene a ~0.15 m del objetivo.
• [ ] Mapa generado refleja el entorno (paredes y pasillos coherentes).
• [ ] Sin errores persistentes en permisos, pigpio o conexión del LIDAR.

Este caso práctico, completamente centrado en el modelo Raspberry Pi 5 + Slamtec RPLIDAR A1 + TB6612FNG motor driver, proporciona una base sólida para Lidar-2D-SLAM-Navigation con Python 3.11 en Raspberry Pi OS Bookworm 64-bit, incluyendo instrucciones precisas de configuración, cableado, código y validación reproducible.

Objetivo y caso de uso

Qué construirás: Un detector de anomalías de vibración en tiempo real utilizando una Raspberry Pi 4 y un módulo IMU Pimoroni ICM-20948.

Para qué sirve

• Monitoreo de vibraciones en maquinaria industrial para detectar fallos prematuros.
• Detección de anomalías en estructuras civiles mediante análisis de vibraciones.
• Aplicaciones en robótica para estabilización y control de movimiento basado en vibraciones.
• Seguimiento de condiciones de salud en dispositivos médicos a través de vibraciones.

Resultado esperado

• Detección de anomalías con una tasa de precisión del 95% en condiciones de prueba.
• Latencias de respuesta menores a 200 ms en la identificación de anomalías.
• Generación de alertas en tiempo real a través de MQTT con un tiempo de entrega inferior a 1 segundo.
• Capacidad de procesar 1000 paquetes de datos por segundo desde el ICM-20948.

Público objetivo: Ingenieros y estudiantes con experiencia en Linux/embebidos; Nivel: Avanzado

Arquitectura/flujo: Raspberry Pi 4 (I2C) -> ICM-20948 -> Procesamiento de datos -> Detección de anomalías -> Notificación en tiempo real.

Nivel: Avanzado

Este caso práctico guía la implementación, desde cero y de forma reproducible, de un pipeline de detección de anomalías de vibraciones (imu-vibration-anomaly-detection) usando una Raspberry Pi 4 Model B y el módulo IMU Pimoroni ICM-20948 (acelerómetro y giróscopo). Incluye preparación del sistema, conexión por I2C, adquisición cronometrada, extracción de características de vibración en ventanas, entrenamiento de un modelo no supervisado (IsolationForest) y validación en tiempo real.

La propuesta está orientada a ingenieros y alumnos con experiencia en Linux/embebidos, señales y Python, y se ajusta a la toolchain y al modelo de dispositivo solicitados. Se ponen especial cuidado en los comandos y versiones para garantizar reproducibilidad.

Prerrequisitos

• Sistema operativo:
• Raspberry Pi OS (Bookworm) 64-bit
• Python 3.11 (el que viene por defecto en Bookworm)
• Toolchain y versiones exactas (se instalarán/forzarán estas versiones):
• pip 24.2
• setuptools 75.3.0
• wheel 0.44.0
• numpy 2.1.2
• scipy 1.14.1
• scikit-learn 1.5.2
• pandas 2.2.3
• matplotlib 3.9.2
• smbus2 0.5.1
• gpiozero 1.6.2
• joblib 1.4.2
• Accesos/Permisos:
• Usuario en el grupo i2c para acceder al bus sin sudo.
• Acceso SSH o consola local.

Notas importantes para este caso:
– Se usará I2C (bus 1) a 3.3 V.
– Se mostrará cómo habilitar I2C vía raspi-config o editando /boot/firmware/config.txt (propio de Bookworm).
– Se trabajará en un entorno virtual (venv) de Python 3.11.

Materiales

• Raspberry Pi 4 Model B + Pimoroni ICM-20948 9DoF IMU
• Tarjeta microSD (≥16 GB) con Raspberry Pi OS Bookworm 64-bit
• Fuente de alimentación oficial para Raspberry Pi 4 (5 V / 3 A)
• Cables dupont hembra-hembra para I2C (mínimo 4: 3V3, GND, SDA, SCL)
• Opcional (para feedback visual de anomalías, no estrictamente necesario):
• LED rojo y resistencia 330 Ω
• Protoboard y 2 cables adicionales

Consejo de montaje: fija el ICM-20948 rígidamente sobre la máquina o superficie cuyo estado de vibraciones deseas monitorizar; la detección mejora si las señales están bien acopladas mecánicamente.

Preparación y conexión

Habilitar I2C y preparar el sistema

• Actualiza el sistema (recomendado):
  sudo apt update
sudo apt full-upgrade -y
sudo reboot

• Habilita I2C con raspi-config (interactivo):
  sudo raspi-config

• Interface Options → I2C → Enable → Finish → Reboot.

• Alternativa (edición de configuración en Bookworm):
  sudo sed -i 's/^#\?dtparam=i2c_arm=.*/dtparam=i2c_arm=on/' /boot/firmware/config.txt
sudo reboot

• Instala utilidades y cabeceras necesarias:
  sudo apt install -y python3-venv python3-dev python3-pip i2c-tools git build-essential

• Añade tu usuario al grupo i2c y aplica sin reiniciar:
  sudo usermod -aG i2c $USER
newgrp i2c

• Verifica Python 3.11:
  python3 --version
  Debe mostrar algo como: Python 3.11.x.

Conexión eléctrica (I2C)

Conecta el Pimoroni ICM-20948 a la Raspberry Pi 4 Model B por I2C (3.3 V). No uses 5 V para la alimentación del IMU.

Tabla de conexión (cabecera 40 pines de la Raspberry Pi 4):

Notas:
– El breakout de Pimoroni suele venir configurado con dirección I2C 0x69. Si mueves el puente AD0, la dirección cambiará a 0x68.
– No es necesario conectar el pin INT para este caso práctico.
– Si usas LED opcional en GPIO17 (pin 11): serie con 330 Ω a GND, cátodo a GND, ánodo a resistencia → GPIO17.

Verificación del bus y del IMU

• Detecta el dispositivo:
  i2cdetect -y 1
  Debes ver 0x69 (o 0x68 si cambiaste AD0). Si no aparece, revisa la conexión y que I2C esté habilitado.

Código completo

A continuación se proporcionan dos archivos Python:

1) Módulo del sensor (icm-20948) vía I2C con smbus2, lectura de acelerómetro+giroscopio y WHO_AM_I.
2) Script principal con pipeline de adquisición, extracción de características de vibración en ventana y entrenamiento/uso de IsolationForest.

Crea un directorio de trabajo y un entorno virtual:

mkdir -p ~/imu-vibration-anomaly-detection/{data,models,src}
python3 -m venv ~/imu-vibration-anomaly-detection/.venv
source ~/imu-vibration-anomaly-detection/.venv/bin/activate
python -m pip install --upgrade pip==24.2 setuptools==75.3.0 wheel==0.44.0
pip install numpy==2.1.2 scipy==1.14.1 scikit-learn==1.5.2 pandas==2.2.3 matplotlib==3.9.2 smbus2==0.5.1 gpiozero==1.6.2 joblib==1.4.2

Archivo 1: Driver mínimo ICM-20948 por I2C (src/imu_icm20948.py)

Este driver configura lo mínimo para “despertar” el ICM-20948 tras reset, leer WHO_AM_I y muestrear acelerómetro y giroscopio en crudo, convirtiendo a unidades físicas con las sensibilidades por defecto (±2 g, ±250 dps después de reset).

# ~/imu-vibration-anomaly-detection/src/imu_icm20948.py
from __future__ import annotations

import time
from typing import Tuple
from smbus2 import SMBus, i2c_msg

# Direcciones/constantes ICM-20948 (Bank 0 por defecto)
REG_WHO_AM_I = 0x00  # Debería devolver 0xEA
REG_PWR_MGMT_1 = 0x06
REG_PWR_MGMT_2 = 0x07
REG_BANK_SEL = 0x7F

# Registros de datos (Bank 0): acelerómetro y giroscopio
REG_ACCEL_XOUT_H = 0x2D  # XH, XL, YH, YL, ZH, ZL (6 bytes)
REG_GYRO_XOUT_H  = 0x33  # XH, XL, YH, YL, ZH, ZL (6 bytes)

WHO_AM_I_EXPECTED = 0xEA

# Sensibilidades por defecto tras reset (±2g, ±250 dps)
ACC_SENS_LSB_PER_G = 16384.0
GYR_SENS_LSB_PER_DPS = 131.0

class ICM20948:
    """
    Driver mínimo para el ICM-20948 (Pimoroni 9DoF IMU) en I2C.
    Permite:
      - Reset + Wake.
      - Comprobar WHO_AM_I.
      - Leer acelerómetro y giroscopio en SI (m/s^2 y dps).
    No habilita DMP ni magnetómetro.
    """
    def __init__(self, bus: int = 1, address: int = 0x69) -> None:
        self.address = address
        self.bus_num = bus
        self.bus = SMBus(self.bus_num)

    def _write_reg(self, reg: int, val: int) -> None:
        self.bus.write_byte_data(self.address, reg, val & 0xFF)

    def _read_reg(self, reg: int) -> int:
        return self.bus.read_byte_data(self.address, reg)

    def _read_block(self, reg: int, length: int) -> bytes:
        read = i2c_msg.read(self.address, length)
        self.bus.write_byte(self.address, reg)
        self.bus.i2c_rdwr(read)
        return bytes(list(read))

    def set_bank(self, bank: int) -> None:
        # Bank = 0..3; bits [5:4] en REG_BANK_SEL
        self._write_reg(REG_BANK_SEL, (bank & 0x3) << 4)

    def reset(self) -> None:
        # Reset del dispositivo; esperar a que se estabilice
        self._write_reg(REG_PWR_MGMT_1, 0x80)  # DEVICE_RESET
        time.sleep(0.1)

    def wake(self) -> None:
        # Selecciona reloj automático y quita SLEEP
        self._write_reg(REG_PWR_MGMT_1, 0x01)  # CLKSEL=1 (auto), SLEEP=0
        time.sleep(0.01)
        # Habilita acelerómetro y giroscopio (no deshabilitar ejes)
        self._write_reg(REG_PWR_MGMT_2, 0x00)
        time.sleep(0.01)
        # Asegurar Bank 0 para lectura de datos
        self.set_bank(0)

    def who_am_i(self) -> int:
        return self._read_reg(REG_WHO_AM_I)

    def read_accel_gyro_raw(self) -> Tuple[Tuple[int, int, int], Tuple[int, int, int]]:
        # Lee 6 bytes acel + 6 bytes giro
        accel_buf = self._read_block(REG_ACCEL_XOUT_H, 6)
        gyro_buf = self._read_block(REG_GYRO_XOUT_H, 6)

        def to_int16(msb: int, lsb: int) -> int:
            val = (msb << 8) | lsb
            if val & 0x8000:
                val = -((val ^ 0xFFFF) + 1)
            return val

        ax = to_int16(accel_buf[0], accel_buf[1])
        ay = to_int16(accel_buf[2], accel_buf[3])
        az = to_int16(accel_buf[4], accel_buf[5])

        gx = to_int16(gyro_buf[0], gyro_buf[1])
        gy = to_int16(gyro_buf[2], gyro_buf[3])
        gz = to_int16(gyro_buf[4], gyro_buf[5])

        return (ax, ay, az), (gx, gy, gz)

    def read_accel_gyro(self) -> Tuple[Tuple[float, float, float], Tuple[float, float, float]]:
        """
        Devuelve:
          - Aceleración (ax, ay, az) en m/s^2
          - Velocidad angular (gx, gy, gz) en deg/s (dps)
        """
        (ax_r, ay_r, az_r), (gx_r, gy_r, gz_r) = self.read_accel_gyro_raw()
        # Convertir a g y luego a m/s^2
        g = 9.80665
        ax = (ax_r / ACC_SENS_LSB_PER_G) * g
        ay = (ay_r / ACC_SENS_LSB_PER_G) * g
        az = (az_r / ACC_SENS_LSB_PER_G) * g

        # Convertir a deg/s
        gx = gx_r / GYR_SENS_LSB_PER_DPS
        gy = gy_r / GYR_SENS_LSB_PER_DPS
        gz = gz_r / GYR_SENS_LSB_PER_DPS

        return (ax, ay, az), (gx, gy, gz)

    def initialize(self) -> None:
        # Secuencia típica: reset -> wake
        self.reset()
        self.wake()
        who = self.who_am_i()
        if who != WHO_AM_I_EXPECTED:
            raise RuntimeError(f"ICM-20948 WHO_AM_I inesperado: 0x{who:02X} (esperado 0x{WHO_AM_I_EXPECTED:02X})")

    def close(self) -> None:
        try:
            self.bus.close()
        except Exception:
            pass


if __name__ == "__main__":
    # Pequeña prueba interactiva: WHO_AM_I + 10 muestras
    imu = ICM20948(bus=1, address=0x69)
    imu.initialize()
    print(f"WHO_AM_I = 0x{imu.who_am_i():02X} (OK)")
    for i in range(10):
        (ax, ay, az), (gx, gy, gz) = imu.read_accel_gyro()
        print(f"{i:02d}: acc[m/s^2]=({ax:+.3f},{ay:+.3f},{az:+.3f})  gyr[dps]=({gx:+.3f},{gy:+.3f},{gz:+.3f})")
        time.sleep(0.02)
    imu.close()

Puntos clave:
– El driver asume dirección 0x69 (por defecto en el breakout de Pimoroni). Si cambias AD0, usa address=0x68.
– WHO_AM_I debe ser 0xEA. Si no coincide, hay un problema de conexión o de dirección.
– Las sensibilidades usadas son las de reset; para vibraciones en maquinaria, son adecuadas para un primer prototipo. Ajustes avanzados (DLPF/ODR) se pueden explorar como mejora.

Archivo 2: Pipeline de detección (src/imu_vibration_anomaly.py)

Este script ofrece tres modos:
– probe: valida I2C, imprime WHO_AM_I y estima el sample rate efectivo.
– baseline: recolecta ventanas de “estado normal”, extrae características y entrena IsolationForest; guarda modelo y scaler.
– detect: carga el modelo y detecta anomalías en streaming; opcionalmente escribe CSV y activa un LED GPIO al detectar anomalía.

# ~/imu-vibration-anomaly-detection/src/imu_vibration_anomaly.py
from __future__ import annotations

import argparse
import math
import os
import sys
import time
from dataclasses import dataclass
from typing import List, Tuple, Optional

import numpy as np
import pandas as pd
from joblib import dump, load
from gpiozero import LED  # opcional si conectaste LED
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler

# Importar driver IMU
from imu_icm20948 import ICM20948

@dataclass
class FeaturesConfig:
    window_s: float = 2.0
    sample_rate: float = 250.0  # Hz
    # Frecuencia máxima de interés: la mitad por Nyquist
    fmax_hz: float = 100.0

def dominant_freq(x: np.ndarray, fs: float, fmin: float = 1.0, fmax: Optional[float] = None) -> float:
    """
    Calcula la frecuencia dominante (pico) de la magnitud 'x' usando FFT.
    Ignora el DC (frecuencia 0) y busca en [fmin, fmax].
    """
    n = len(x)
    x = x - np.mean(x)
    win = np.hanning(n)
    X = np.fft.rfft(x * win)
    freqs = np.fft.rfftfreq(n, d=1.0/fs)
    mag = np.abs(X)

    # Enmascara rangos
    mask = freqs >= fmin
    if fmax is not None:
        mask &= freqs <= fmax
    if not np.any(mask):
        return 0.0
    idx = np.argmax(mag[mask])
    return float(freqs[mask][idx])

def spectral_centroid(x: np.ndarray, fs: float) -> float:
    """
    Centroides espectrales de la magnitud 'x'.
    """
    n = len(x)
    x = x - np.mean(x)
    X = np.fft.rfft(x * np.hanning(n))
    mag = np.abs(X)
    freqs = np.fft.rfftfreq(n, d=1.0/fs)
    denom = np.sum(mag) + 1e-12
    return float(np.sum(freqs * mag) / denom)

def extract_features(window: np.ndarray, fs: float, fmax: float) -> np.ndarray:
    """
    Extrae características de vibración de una ventana.
    window: array (N, 6) con columnas [ax, ay, az, gx, gy, gz].
    """
    ax, ay, az, gx, gy, gz = window.T
    acc_mag = np.sqrt(ax**2 + ay**2 + az**2)
    gyr_mag = np.sqrt(gx**2 + gy**2 + gz**2)

    feats: List[float] = []

    def stats(v: np.ndarray) -> List[float]:
        # Estadísticos robustos (evitar dependencia exclusiva de var/mean)
        return [
            float(np.mean(v)),
            float(np.std(v, ddof=1)),
            float(np.median(v)),
            float(np.percentile(v, 90) - np.percentile(v, 10)),
            float(np.max(np.abs(v))),
            float(np.sqrt(np.mean(v**2))),  # RMS
        ]

    # Características por eje y por magnitud
    for series in [ax, ay, az, acc_mag, gx, gy, gz, gyr_mag]:
        feats += stats(series)

    # Frecuencia dominante y centroide espectral en magnitudes
    feats += [
        dominant_freq(acc_mag, fs, fmin=1.0, fmax=fmax),
        spectral_centroid(acc_mag, fs),
        dominant_freq(gyr_mag, fs, fmin=1.0, fmax=fmax),
        spectral_centroid(gyr_mag, fs),
    ]
    return np.array(feats, dtype=np.float64)

def acquire_samples(imu: ICM20948, duration_s: float, fs: float) -> np.ndarray:
    """
    Adquiere muestras durante duration_s a fs Hz.
    Devuelve array (N, 6) con [ax, ay, az, gx, gy, gz].
    """
    dt = 1.0 / fs
    n_expected = int(duration_s * fs)
    out = np.zeros((n_expected, 6), dtype=np.float64)

    t_next = time.perf_counter()
    for i in range(n_expected):
        (ax, ay, az), (gx, gy, gz) = imu.read_accel_gyro()
        out[i, :] = [ax, ay, az, gx, gy, gz]
        t_next += dt
        # Espera activa (busy-wait suave) para mayor precisión temporal
        while True:
            now = time.perf_counter()
            if now >= t_next:
                break
            # Sleep corto para liberar CPU
            time.sleep(max(0.0, min(0.0005, t_next - now)))
    return out

def windows(arr: np.ndarray, win_size: int, step: int) -> np.ndarray:
    """
    Crea ventanas deslizantes con paso 'step' y tamaño 'win_size'.
    """
    n = arr.shape[0]
    if n < win_size:
        return np.empty((0, win_size, arr.shape[1]))
    out = []
    for start in range(0, n - win_size + 1, step):
        out.append(arr[start:start + win_size, :])
    return np.stack(out, axis=0)

def build_feature_matrix(raw: np.ndarray, fs: float, cfg: FeaturesConfig) -> np.ndarray:
    """
    Convierte la señal cruda en una matriz de características por ventana.
    """
    win_size = int(cfg.window_s * fs)
    step = win_size  # ventanas sin solape para baseline; ajustar si se desea
    w = windows(raw, win_size, step)
    feats = []
    for wi in w:
        feats.append(extract_features(wi, fs, cfg.fmax_hz))
    if len(feats) == 0:
        return np.empty((0, 0))
    return np.vstack(feats)

def save_csv(path: str, ts0: float, fs: float, raw: np.ndarray) -> None:
    """
    Guarda CSV con timestamp y los 6 canales.
    """
    n = raw.shape[0]
    t = ts0 + np.arange(n) / fs
    cols = ["timestamp", "ax_ms2", "ay_ms2", "az_ms2", "gx_dps", "gy_dps", "gz_dps"]
    df = pd.DataFrame(np.column_stack([t, raw]), columns=cols)
    df.to_csv(path, index=False)

def run_probe(bus: int, addr: int, fs: float, seconds: float) -> None:
    imu = ICM20948(bus=bus, address=addr)
    imu.initialize()
    who = imu.who_am_i()
    print(f"WHO_AM_I=0x{who:02X} (esperado 0xEA)")
    print("Midiendo tasa efectiva...")
    n = max(1, int(seconds * fs))
    t0 = time.perf_counter()
    _ = acquire_samples(imu, seconds, fs)
    t1 = time.perf_counter()
    eff = n / (t1 - t0)
    print(f"Configurado fs={fs:.1f} Hz; efectivo ~{eff:.1f} Hz en {seconds:.2f} s.")
    imu.close()

def run_baseline(bus: int, addr: int, fs: float, duration: float, window_s: float,
                 model_out: str, csv_out: Optional[str]) -> None:
    cfg = FeaturesConfig(window_s=window_s, sample_rate=fs, fmax_hz=min(0.5 * fs, 200.0))
    imu = ICM20948(bus=bus, address=addr)
    imu.initialize()

    print(f"Adquiriendo baseline durante {duration:.1f} s a {fs:.1f} Hz...")
    ts0 = time.time()
    raw = acquire_samples(imu, duration, fs)
    imu.close()
    if csv_out:
        os.makedirs(os.path.dirname(csv_out), exist_ok=True)
        save_csv(csv_out, ts0, fs, raw)
        print(f"Baseline crudo guardado en: {csv_out}")

    X = build_feature_matrix(raw, fs, cfg)
    if X.size == 0:
        raise RuntimeError("Insuficientes muestras para construir ventanas de baseline")

    print(f"Ventanas de baseline: {X.shape[0]}  (dim características: {X.shape[1]})")
    scaler = StandardScaler()
    Xn = scaler.fit_transform(X)

    # IsolationForest: no supervisado; 'contamination' indica fracción estimada de anomalías
    model = IsolationForest(
        n_estimators=200,
        contamination=0.02,
        max_samples='auto',
        random_state=42,
        n_jobs=-1,
        verbose=0
    ).fit(Xn)

    os.makedirs(os.path.dirname(model_out), exist_ok=True)
    dump({
        "scaler": scaler,
        "model": model,
        "fs": fs,
        "window_s": window_s,
        "feature_version": 1,
    }, model_out)
    print(f"Modelo guardado en: {model_out}")

def run_detect(bus: int, addr: int, fs: float, window_s: float, model_path: str,
               csv_stream: Optional[str], led_pin: Optional[int]) -> None:
    pack = load(model_path)
    scaler: StandardScaler = pack["scaler"]
    model: IsolationForest = pack["model"]
    fs_model = float(pack["fs"])
    window_model = float(pack["window_s"])
    if abs(fs_model - fs) > 1e-3 or abs(window_model - window_s) > 1e-3:
        print(f"Advertencia: parámetros de ejecución (fs={fs}Hz, window={window_s}s) difieren del modelo (fs={fs_model}Hz, window={window_model}s).")

    cfg = FeaturesConfig(window_s=window_s, sample_rate=fs, fmax_hz=min(0.5 * fs, 200.0))
    imu = ICM20948(bus=bus, address=addr)
    imu.initialize()

    led = None
    if led_pin is not None:
        try:
            led = LED(led_pin)
        except Exception as e:
            print(f"No se pudo inicializar LED en GPIO{led_pin}: {e}")

    win_size = int(window_s * fs)
    buf = np.zeros((win_size, 6), dtype=np.float64)
    idx = 0

    ts0 = time.time()
    if csv_stream:
        os.makedirs(os.path.dirname(csv_stream), exist_ok=True)

    print("Entrando en modo detección (Ctrl+C para salir).")
    try:
        while True:
            (ax, ay, az), (gx, gy, gz) = imu.read_accel_gyro()
            buf[idx, :] = [ax, ay, az, gx, gy, gz]
            idx += 1

            if idx >= win_size:
                # Extraer y evaluar
                X = extract_features(buf, fs, cfg.fmax_hz).reshape(1, -1)
                Xn = scaler.transform(X)
                pred = model.predict(Xn)[0]  # 1 normal, -1 anomalía
                score = float(model.decision_function(Xn)[0])  # >0 inlier, <0 outlier
                stamp = time.time()
                msg = f"{stamp:.3f} decision={score:+.4f} pred={'NORMAL' if pred==1 else 'ANOMALIA'}"
                print(msg)

                # CSV opcional
                if csv_stream:
                    # Guardar la ventana cruda con timestamp inicial
                    save_csv(csv_stream, ts0, fs, buf)

                # LED opcional
                if led:
                    if pred == -1:
                        led.on()
                    else:
                        led.off()

                # Reiniciar la ventana (no solapada para detección simple)
                idx = 0

            # Cronometría aproximada para lazo ~fs Hz
            time.sleep(max(0.0, (1.0 / fs) * 0.4))  # ligera pausa para no saturar CPU
    except KeyboardInterrupt:
        print("Cancelado por usuario.")
    finally:
        if led:
            led.off()
        imu.close()

def parse_args() -> argparse.Namespace:
    p = argparse.ArgumentParser(description="imu-vibration-anomaly-detection con Raspberry Pi 4 Model B + Pimoroni ICM-20948 9DoF IMU")
    p.add_argument("--mode", choices=["probe", "baseline", "detect"], required=True)
    p.add_argument("--bus", type=int, default=1, help="I2C bus (default 1)")
    p.add_argument("--addr", type=lambda x: int(x, 0), default=0x69, help="Dirección I2C (0x69 por defecto)")
    p.add_argument("--fs", type=float, default=250.0, help="Frecuencia de muestreo objetivo [Hz]")
    p.add_argument("--duration", type=float, default=60.0, help="Duración baseline/probe [s]")
    p.add_argument("--window", type=float, default=2.0, help="Ventana de análisis [s]")
    p.add_argument("--model", type=str, default="models/imu_iforest.joblib", help="Ruta del modelo a guardar/cargar")
    p.add_argument("--csv", type=str, default="", help="Ruta CSV para guardar (baseline) o stream (detect)")
    p.add_argument("--led-pin", type=int, default=None, help="GPIO para LED (opcional) en modo detect")
    return p.parse_args()

def main() -> None:
    args = parse_args()
    csv_path = args.csv if args.csv else None

    if args.mode == "probe":
        run_probe(bus=args.bus, addr=args.addr, fs=args.fs, seconds=args.duration)
    elif args.mode == "baseline":
        run_baseline(
            bus=args.bus, addr=args.addr, fs=args.fs, duration=args.duration,
            window_s=args.window, model_out=args.model, csv_out=csv_path
        )
    elif args.mode == "detect":
        run_detect(
            bus=args.bus, addr=args.addr, fs=args.fs, window_s=args.window,
            model_path=args.model, csv_stream=csv_path, led_pin=args.led_pin
        )
    else:
        raise ValueError("Modo no soportado")

if __name__ == "__main__":
    main()

Puntos clave del código:
– Adquisición cronometrada a fs, con espera activa ligera para mejorar precisión temporal en Linux.
– Extracción de características equilibrando dominio temporal (RMS, std, R-90/10) y frecuencia (pico, centroide) sobre magnitud vectorial (reduce sensibilidad a orientación).
– Modelo no supervisado (IsolationForest) con “contamination” baja (2%) pensado para entornos donde se espera que la mayor parte del tiempo sea “normal”.
– Guardado de baseline crudo y ventanas para auditoría; guardado de scaler+modelo con joblib.
– Modo “detect” en streaming, con feedback por consola y LED opcional.

Compilación/flash/ejecución

No hay “flash” como tal; se ejecuta sobre Raspberry Pi OS. Repite los pasos exactamente.

1) Crear árbol, venv y paquetes con versiones fijadas:

mkdir -p ~/imu-vibration-anomaly-detection/{data,models,src}
python3 -m venv ~/imu-vibration-anomaly-detection/.venv
source ~/imu-vibration-anomaly-detection/.venv/bin/activate
python -m pip install --upgrade pip==24.2 setuptools==75.3.0 wheel==0.44.0
pip install numpy==2.1.2 scipy==1.14.1 scikit-learn==1.5.2 pandas==2.2.3 matplotlib==3.9.2 smbus2==0.5.1 gpiozero==1.6.2 joblib==1.4.2

2) Crear archivos fuente:

cat > ~/imu-vibration-anomaly-detection/src/imu_icm20948.py << 'PY'
[PEGA AQUÍ EL CONTENIDO DEL ARCHIVO 1 COMPLETO]
PY

cat > ~/imu-vibration-anomaly-detection/src/imu_vibration_anomaly.py << 'PY'
[PEGA AQUÍ EL CONTENIDO DEL ARCHIVO 2 COMPLETO]
PY

3) Validar I2C y WHO_AM_I (modo probe, 10–20 s de prueba):

cd ~/imu-vibration-anomaly-detection/src
python imu_vibration_anomaly.py --mode probe --addr 0x69 --fs 250 --duration 10

Deberías ver WHO_AM_I=0xEA y una tasa efectiva cercana a la configurada.

4) Entrenar baseline (ej. 120 s, 2 s por ventana):

python imu_vibration_anomaly.py --mode baseline --addr 0x69 --fs 250 --duration 120 --window 2.0 --model ../models/imu_iforest.joblib --csv ../data/baseline_raw.csv

Se guardará el modelo en models/imu_iforest.joblib y los datos crudos en data/baseline_raw.csv.

5) Detección en tiempo real (mismas condiciones de fs y window):

python imu_vibration_anomaly.py --mode detect --addr 0x69 --fs 250 --window 2.0 --model ../models/imu_iforest.joblib --csv ../data/stream_window.csv

Opcional con LED en GPIO17:

python imu_vibration_anomaly.py --mode detect --addr 0x69 --fs 250 --window 2.0 --model ../models/imu_iforest.joblib --led-pin 17

Notas:
– Si moviste el puente AD0 y ves 0x68 en i2cdetect, usa –addr 0x68.
– Mantén activado el entorno virtual en cada consola: source ~/.venv.../bin/activate.

Validación paso a paso

1) Verificación de hardware/I2C:
– Ejecuta i2cdetect -y 1. Debe aparecer 0x69 (o 0x68).
– Si no aparece:
– Revisa conexiones (3V3, GND, SDA, SCL).
– Asegura que I2C está habilitado.
– Verifica que tu usuario esté en el grupo i2c.

2) WHO_AM_I con el script:
– python src/imu_vibration_anomaly.py --mode probe --addr 0x69 --fs 250 --duration 5
– Debe imprimir WHO_AM_I=0xEA (esperado 0xEA) seguido de una tasa efectiva ~250 Hz ±10–20% (Linux no es RT).

3) Señales crudas plausibles:
– En reposo sobre mesa, aceleración cerca de (0, 0, ±9.8 m/s²) con pequeñas variaciones; giroscopio cerca de 0 dps.
– Mover la IMU debe cambiar valores apreciablemente.

4) Entrenamiento baseline:
– Coloca el IMU en la máquina o superficie en “estado normal” y evita golpes.
– Ejecuta el comando de baseline por 120 s (o la duración deseada).
– Salida esperada:
– “Ventanas de baseline: … (dim características: …)”
– Archivo modelo: models/imu_iforest.joblib creado (tamaño > 100 KB).
– CSV baseline: data/baseline_raw.csv creado (tamaño en función de fs y duración).
– Revisa data/baseline_raw.csv: columnas timestamp y 6 señales; sin NaN.

5) Detección:
– Ejecuta el modo detect y observa la consola.
– En condiciones normales, la mayoría de ventanas deben ser “NORMAL” con decision_function > 0.
– Provoca una anomalía: añade carga excéntrica, desequilibrio, toca o golpea ligeramente la estructura.
– Se deben imprimir eventos “ANOMALIA” con decision_function < 0 y, si LED conectado, encenderse.

6) Coherencia de parámetros:
– Confirma que fs y window en detect coinciden con los usados en baseline.
– Si cambias fs/window, vuelve a entrenar baseline.

7) Rendimiento:
– A 250 Hz, la CPU de la Pi 4 debe sostener la adquisición y cálculo en tiempo real sin saturarse.
– Si ves retardos, reduce fs (p. ej., 200 Hz) o aumenta window (2–3 s) para amortiguar jitter.

Troubleshooting

1) i2cdetect no muestra 0x69/0x68:
– Causa: I2C deshabilitado o conexiones incorrectas.
– Solución: Habilita I2C (raspi-config), reconecta SDA/SCL/3V3/GND, comprueba continuidad. Verifica pull-ups (el breakout Pimoroni suele integrarlos).

2) WHO_AM_I distinto de 0xEA:
– Causa: Dirección equivocada (AD0), bus incorrecto, IMU en mal estado.
– Solución: Prueba –addr 0x68 si AD0 está a GND; confirma con i2cdetect; revisa alimentación a 3.3V.

3) PermissionError al acceder a I2C:
– Causa: Usuario no pertenece al grupo i2c.
– Solución: sudo usermod -aG i2c $USER && newgrp i2c (o cierra y abre sesión).

4) ImportError de paquetes Python:
– Causa: No activaste el venv o versiones distintas a las esperadas.
– Solución: source ~/imu-vibration-anomaly-detection/.venv/bin/activate y reinstala con las versiones fijadas (pip==24.2, etc.).

5) Tasa efectiva (probe) muy inferior a la configurada:
– Causa: Sobrecarga CPU, otros procesos, throttling térmico/energético.
– Solución: Cerrar programas, bajar fs a 200–250 Hz, usar disipador/ventilador, alimentar con PSU oficial, opcional fijar governor “performance”.

6) Ventanas insuficientes o error “Insuficientes muestras…”:
– Causa: duration demasiado corto respecto a window.
– Solución: Aumentar duration o disminuir window; p. ej., window=2.0 s y duration≥20 s.

7) Modelo no detecta anomalías obvias:
– Causas: Baseline poco representativo, “contamination” demasiado alto/bajo, features no separan tu caso.
– Soluciones:
– Recolecta baseline más largo y estable.
– Ajusta contamination (0.01–0.05).
– Amplía features (más métricas espectrales; solape entre ventanas).

8) Señales saturadas o valores extraños:
– Causa: Vibraciones fuera de rango para ±2 g o ±250 dps.
– Solución: Ajustar escalas (FS) y DLPF del ICM-20948 (requeriría extender el driver para configurar Bank 2). Alternativamente, desacoplar ligeramente el sensor o reducir excitación.

Mejoras/variantes

• Ajuste fino del IMU:
• Configurar DLPF y ODR en Bank 2 para mejorar SNR en bandas específicas (p. ej., 20–300 Hz).

• Cambiar full-scale a ±4 g/±500 dps para evitar saturación si vibraciones son intensas.

• Ventaneo y extracción:

• Añadir solape (p. ej., 50%) para mayor resolución temporal.

• Incorporar más features: energía en bandas, crest factor, kurtosis, skewness, entropías espectrales.

• Modelos:

• Probar One-Class SVM o LOF.

• Entrenar autoencoders 1D o CNN ligeras y ejecutar con TensorFlow Lite en la Pi.

• Señalización:

• Publicar eventos por MQTT a un broker (ej. Mosquitto).
• Guardar timeseries en InfluxDB y visualizar con Grafana.

• Añadir buzzer/LEDs multicolor y relé para interlock.

• Robustez:

• Convertir este script en un servicio systemd que se levante al arranque.
• Registrar métricas y logs rotativos en /var/log/imu-vibe/.

• Dockerizar el entorno para despliegues homogéneos.

• Hardware:

• Si necesitas más inmunidad al ruido, usar cable apantallado para I2C y asegurar masa de referencia.
• Explorar SPI si el breakout lo soporta y el entorno EMI es fuerte (I2C es más sensible en cables largos).

Checklist de verificación

• [ ] Raspberry Pi OS Bookworm 64-bit instalado y actualizado.
• [ ] I2C habilitado (raspi-config o /boot/firmware/config.txt).
• [ ] Usuario en grupo i2c; i2cdetect muestra 0x69 (o 0x68).
• [ ] Entorno virtual creado y activado (Python 3.11).
• [ ] Paquetes instalados con versiones fijadas (pip==24.2, numpy==2.1.2, smbus2==0.5.1, scikit-learn==1.5.2, etc.).
• [ ] Código copiado en ~/imu-vibration-anomaly-detection/src/ (dos archivos).
• [ ] Modo probe exitoso: WHO_AM_I=0xEA y tasa efectiva razonable.
• [ ] Baseline recolectado y modelo guardado en models/imu_iforest.joblib.
• [ ] Detección en tiempo real funcionando; eventos “ANOMALIA” al provocar una perturbación.
• [ ] (Opcional) LED en GPIO17 enciende ante anomalía.
• [ ] Datos CSV guardados en data/ para auditoría.

---

Notas finales para el docente:
– Este caso práctico está alineado con el dispositivo especificado “Raspberry Pi 4 Model B + Pimoroni ICM-20948 9DoF IMU”, usando Raspberry Pi OS Bookworm 64-bit y Python 3.11. La toolchain se define con versiones exactas de pip y librerías para reproducibilidad.
– La metodología es extensible y prepara el terreno para Mantenimiento Predictivo (PdM): del muestreo con IMU a la detección de eventos y su integración con sistemas mayores.

Objetivo y caso de uso

Qué construirás: Un servidor web MJPEG que transmite video en tiempo real desde una Raspberry Pi Zero W con Camera Module v2.

Para qué sirve

• Transmisión de video en tiempo real para monitoreo remoto de espacios.
• Integración en sistemas de domótica para visualización de cámaras de seguridad.
• Proyectos de robótica donde se requiere streaming de video en vivo.
• Aplicaciones de telemedicina que permiten la supervisión visual de pacientes.

Resultado esperado

• Latencia de transmisión menor a 100 ms.
• Capacidad de manejar al menos 5 conexiones simultáneas sin pérdida de calidad.
• FPS (fotogramas por segundo) estables en 30 FPS durante la transmisión.
• Consumo de ancho de banda de aproximadamente 1.5 Mbps por flujo de video.

Público objetivo: Desarrolladores y entusiastas de la tecnología; Nivel: Avanzado

Arquitectura/flujo: Raspberry Pi Zero W con Camera Module v2 -> Servidor Flask -> Transmisión MJPEG a través de HTTP.

Nivel: Avanzado

Prerrequisitos

Este caso práctico está verificado y documentado específicamente para el modelo “Raspberry Pi Zero W + Camera Module v2 (IMX219)”, orientado a montar un servidor web que publique un stream MJPEG (multipart/x-mixed-replace) vía CSI. Se asume que trabajas con la pila actual basada en libcamera (sin el stack “legacy”).

• Sistema operativo
• Raspberry Pi OS Bookworm Lite (32-bit, armhf). El Pi Zero W no soporta 64-bit; Bookworm 32-bit incluye Python 3.11 por defecto.
• Imagen recomendada (ejemplo): 2024-10-22 o posterior.
• Kernel Linux: 6.6.x (cualquiera de la serie estable incluida con la imagen anterior).
• Toolchain y versiones probadas (recomendadas para reproducibilidad)
• Python: 3.11.2 (Bookworm)
• Pip: 23.2.1
• Venv: 3.11 (módulo estándar)
• GCC: 12.2.0 (Debian 12)
• CMake: 3.25.1
• libcamera: 0.0.5+ (paquetes de Raspberry Pi OS Bookworm; ver sección de validación para comprobar versión local)
• libcamera-apps: 0.0.5+ (rpi)
• python3-picamera2: 0.3.16–0.3.17 (del repo Raspberry Pi OS)
• Flask: 2.3.3 (pip)
• Pillow: 10.3.0 (pip, vía piwheels)
• Waitress: 2.1.2 (pip, servidor WSGI puro Python, apto para stream chunked)
• Red y acceso
• Wi‑Fi configurado (2.4 GHz), IP accesible desde tu PC.
• Acceso SSH habilitado o terminal local con teclado/monitor.
• Conocimientos previos
• Linux y shell en Raspberry Pi.
• Python 3.11, virtualenvs, pip.
• Conceptos básicos de HTTP y streaming MJPEG.

Comando para verificar versiones (ejecútalos y anota tu entorno real):

python3 --version
pip3 --version
gcc --version | head -n1
cmake --version | head -n1
apt-cache policy libcamera0 libcamera-apps python3-picamera2 | sed -n '1,6p'

Materiales

• Raspberry Pi Zero W (modelo exacto).
• Camera Module v2 (sensor Sony IMX219, con cable CSI de 15 pines).
• Tarjeta microSD (≥ 16 GB, clase A1/A2 preferible).
• Fuente de alimentación: 5 V / 2 A micro-USB.
• Cable plano CSI adecuado para Pi Zero W (el conector es de tamaño “corto”, suele venir con la cámara; si no, compra el cable específico para Zero).
• Adaptador micro-USB OTG (opcional, para teclado/USB si no usas SSH).
• PC para flashear la imagen de Raspberry Pi OS Bookworm Lite (32-bit).

Nota: Este proyecto no requiere hardware adicional de E/S GPIO; todo el flujo es CSI → ISP/libcamera → servidor HTTP.

Preparación y conexión

Preparación del sistema

1. Flashea Raspberry Pi OS Bookworm Lite (32-bit) en la microSD con Raspberry Pi Imager.
2. Selecciona: Raspberry Pi OS Lite (32-bit) — Bookworm.
3. Configura en Imager (opciones avanzadas): hostname, usuario, contraseña, Wi‑Fi y SSH.
4. Inserta la microSD en la Raspberry Pi Zero W.
5. Conecta la alimentación y espera el primer arranque.

Conexión de la cámara CSI

• Apaga completamente la Raspberry Pi antes de manipular la cámara.
• Localiza el conector CSI de la Raspberry Pi Zero W (interfaz de cámara).
• Inserta el cable plano:
• Contactos del cable hacia los contactos del conector CSI.
• Bloquea la pestaña con cuidado para que no se suelte.
• Conecta el otro extremo al Camera Module v2 respetando la orientación (contactos a contactos).
• Enciende la Raspberry Pi.

Tabla de puertos y orientación (Pi Zero W + Camera Module v2)

Habilitar interfaces y cámara (Bookworm)

En Bookworm con libcamera, normalmente no es necesario habilitar “Legacy Camera”. Aun así, verifica:

• Opción A: raspi-config
• sudo raspi-config
• Interface Options:
  • I2C: Enable (útil para periféricos; no estrictamente necesario para libcamera, pero conveniente).
  • Legacy Camera: Disabled (para mantener libcamera).
• Localisation Options: ajusta zona horaria y teclado si lo necesitas.

• Finish y reboot.

• Opción B: edición de /boot/firmware/config.txt

• sudo nano /boot/firmware/config.txt
• Asegúrate de tener (o añade) las líneas:
  camera_auto_detect=1
# Forzar overlay del sensor si fuese necesario:
# dtoverlay=imx219
• Guarda y reinicia: sudo reboot

Comprobación rápida de la cámara con libcamera:

libcamera-hello -t 2000

Deberías ver en consola que se abre la cámara (en Pi Zero W sin pantalla HDMI, la aplicación igualmente informa si detecta el sensor). Si falla, revisa Troubleshooting.

Código completo

Implementaremos un servidor web Python que:
– Inicializa Picamera2 con una configuración “video-like” a baja resolución para la CPU del Pi Zero W.
– Captura frames periódicamente como arrays RGB.
– Codifica cada frame a JPEG con Pillow (calidad ajustable).
– Expone:
– /stream.mjpg → MJPEG multipart (para navegadores y VLC).
– /snapshot.jpg → fotograma actual (single shot).
– /health → estado simple.

Notas de diseño:
– Elegimos 640×480 a ~8–10 fps para un equilibrio entre CPU/latencia/calidad en el Zero W.
– Creamos un hilo productor que inserta frames JPEG en una cola bounded (drop-old) para evitar backlog.
– Usamos Flask por sencillez y Waitress como servidor WSGI robusto en producción ligera (ambos Python puro).

Estructura del proyecto

• ~/mjpeg-csi/
• app.py
• .env (opcional, variables de configuración)
• venv/ (virtualenv con paquetes pip)
• run.sh (opcional, arranque)
• servicio systemd (descrito más adelante)

app.py

#!/usr/bin/env python3
import io
import os
import signal
import threading
import time
from queue import Queue, Full, Empty
from typing import Generator, Optional

from flask import Flask, Response, jsonify, stream_with_context
from PIL import Image
from picamera2 import Picamera2

# Configuración desde entorno (con valores por defecto seguros para Pi Zero W)
WIDTH = int(os.getenv("MJPEG_WIDTH", "640"))
HEIGHT = int(os.getenv("MJPEG_HEIGHT", "480"))
FPS = float(os.getenv("MJPEG_FPS", "8"))  # 8 fps razonable en Zero W
JPEG_QUALITY = int(os.getenv("MJPEG_QUALITY", "70"))  # 50-80 recomendado
BOUNDARY = os.getenv("MJPEG_BOUNDARY", "frameboundary")
PORT = int(os.getenv("MJPEG_PORT", "8080"))
HOST = os.getenv("MJPEG_HOST", "0.0.0.0")
QUEUE_SIZE = int(os.getenv("MJPEG_QUEUE_SIZE", "2"))
CAPTURE_SLEEP = float(os.getenv("MJPEG_CAPTURE_SLEEP", str(1.0 / FPS)))

app = Flask(__name__)

# Cola de frames JPEG
frame_queue: "Queue[bytes]" = Queue(maxsize=QUEUE_SIZE)
shutdown_event = threading.Event()


def camera_thread():
    """
    Hilo productor:
    - Inicializa Picamera2
    - Captura arrays RGB a la resolución dada
    - Codifica a JPEG con Pillow
    - Inserta frames a la cola con política drop-old
    """
    picam2 = Picamera2()

    # Configuración "video" con salida RGB888 para codificar fácilmente a JPEG
    video_config = picam2.create_video_configuration(
        main={"size": (WIDTH, HEIGHT), "format": "RGB888"},
        transform=None  # sin rotación
    )
    picam2.configure(video_config)

    # Ajustes de control (opcionales)
    # Nota: En Zero W, mantener autoexposición/awb suele ser suficiente
    # picam2.set_controls({"AwbEnable": True, "AeEnable": True})

    picam2.start()
    try:
        while not shutdown_event.is_set():
            # Captura frame como array RGB
            frame = picam2.capture_array("main")

            # Codifica a JPEG
            with io.BytesIO() as buf:
                # Pillow espera array en RGB ordenado (ya lo tenemos)
                Image.fromarray(frame).save(
                    buf, format="JPEG", quality=JPEG_QUALITY, optimize=True
                )
                jpg = buf.getvalue()

            # Inserta en la cola, descartando el más antiguo si está llena
            try:
                frame_queue.put(jpg, timeout=0.01)
            except Full:
                try:
                    frame_queue.get_nowait()
                except Empty:
                    pass
                # Reintenta tras descartar
                try:
                    frame_queue.put_nowait(jpg)
                except Full:
                    pass

            # Ritmo de captura
            if CAPTURE_SLEEP > 0:
                time.sleep(CAPTURE_SLEEP)
    finally:
        picam2.stop()


def mjpeg_generator() -> Generator[bytes, None, None]:
    """
    Generador de stream MJPEG en formato multipart/x-mixed-replace.
    Cada iteración produce:
    --BOUNDARY
    Content-Type: image/jpeg
    Content-Length: <n>

    <bytes JPEG>
    """
    boundary_bytes = BOUNDARY.encode("ascii")
    while not shutdown_event.is_set():
        try:
            jpg = frame_queue.get(timeout=1.0)
        except Empty:
            continue

        header = (
            b"--" + boundary_bytes + b"\r\n"
            b"Content-Type: image/jpeg\r\n"
            b"Content-Length: " + str(len(jpg)).encode("ascii") + b"\r\n\r\n"
        )
        yield header + jpg + b"\r\n"


@app.route("/stream.mjpg")
def stream():
    return Response(
        stream_with_context(mjpeg_generator()),
        mimetype=f"multipart/x-mixed-replace; boundary={BOUNDARY}",
    )


@app.route("/snapshot.jpg")
def snapshot():
    # Toma el último frame disponible (o espera un poco)
    try:
        jpg = frame_queue.get(timeout=2.0)
    except Empty:
        return Response(status=503)
    return Response(jpg, mimetype="image/jpeg")


@app.route("/health")
def health():
    return jsonify(
        status="ok",
        width=WIDTH,
        height=HEIGHT,
        fps=FPS,
        quality=JPEG_QUALITY,
        queue_size=QUEUE_SIZE,
        boundary=BOUNDARY,
    )


def handle_sigterm(signum, frame):
    shutdown_event.set()


def main():
    # Inicia el hilo de cámara
    t = threading.Thread(target=camera_thread, name="camera-thread", daemon=True)
    t.start()

    # Señales para apagado limpio
    signal.signal(signal.SIGTERM, handle_sigterm)
    signal.signal(signal.SIGINT, handle_sigterm)

    # Servidor WSGI
    use_waitress = os.getenv("USE_WAITRESS", "1") == "1"
    if use_waitress:
        try:
            from waitress import serve
            # threads controla peticiones concurrentes (streams simultáneos)
            serve(app, host=HOST, port=PORT, threads=4)
        except Exception as e:
            print(f"[waitress] Error: {e}; usando servidor Flask de desarrollo.")
            app.run(host=HOST, port=PORT, threaded=True)
    else:
        app.run(host=HOST, port=PORT, threaded=True)

    shutdown_event.set()
    t.join(timeout=2.0)


if __name__ == "__main__":
    main()

Puntos clave del código:
– create_video_configuration con formato RGB888 simplifica la compresión JPEG con Pillow. Para el Zero W, bajar resolución o FPS si ves CPU alta.
– Cola bounded con política drop-old para minimizar latencia en el stream.
– Endpoints claros y reusables en validación.
– Waitress para servir responses “chunked” eficientes y estables; en caso de fallo, usa el servidor de desarrollo de Flask.

Compilación/flash/ejecución

A continuación, pasos reproducibles, exactos y ordenados.

1) Actualización de sistema y paquetes base

sudo apt update
sudo apt full-upgrade -y
sudo reboot

2) Instalar dependencias del sistema

• libcamera y herramientas, Picamera2, Python 3.11, venv y utilidades:

sudo apt install -y \
  libcamera-apps \
  python3-picamera2 \
  python3-pip \
  python3-venv \
  python3-numpy \
  git \
  gcc \
  cmake

Comprobación rápida de cámara:

libcamera-hello -t 2000

Si ves que abre la cámara sin error, continúa.

3) Preparar directorio de proyecto y virtualenv

mkdir -p ~/mjpeg-csi
cd ~/mjpeg-csi
python3 -m venv --system-site-packages venv
source venv/bin/activate

Usamos –system-site-packages para que el venv vea python3-picamera2 instalado por apt.

4) Configurar piwheels y pip

Piwheels acelera instalación en Raspberry Pi y ofrece ruedas para ARMv6 (Zero W).

pip install --upgrade pip==23.2.1
pip config set global.index-url https://www.piwheels.org/simple

5) Instalar dependencias Python (pinned)

pip install \
  Flask==2.3.3 \
  Pillow==10.3.0 \
  waitress==2.1.2

Comprueba versiones:

python -c "import flask, PIL; import waitress; \
print('Flask', flask.__version__); \
print('Pillow', PIL.__version__); \
import pkgutil; import waitress; print('Waitress', waitress.__version__)"

6) Crear el archivo app.py

Copia el contenido de la sección anterior en:

nano app.py
# pega el código, guarda con Ctrl+O y sal con Ctrl+X

Dale permisos de ejecución:

chmod +x app.py

7) Ejecutar en modo desarrollo (prueba rápida)

source venv/bin/activate
export USE_WAITRESS=0
python app.py

Abre desde tu PC:
– Stream MJPEG: http://:8080/stream.mjpg
– Snapshot: http://:8080/snapshot.jpg
– Salud: http://:8080/health

Para detener: Ctrl+C.

8) Ejecutar con Waitress (recomendado para uso continuado)

source venv/bin/activate
export USE_WAITRESS=1
python app.py

Si todo funciona, automatiza con systemd.

9) Systemd service para autoarranque

Crea el servicio:

sudo tee /etc/systemd/system/mjpeg-csi.service >/dev/null <<'EOF'
[Unit]
Description=Servidor MJPEG (Picamera2) en Raspberry Pi Zero W
After=network-online.target
Wants=network-online.target

[Service]
Type=simple
User=pi
WorkingDirectory=/home/pi/mjpeg-csi
Environment="USE_WAITRESS=1"
Environment="MJPEG_WIDTH=640"
Environment="MJPEG_HEIGHT=480"
Environment="MJPEG_FPS=8"
Environment="MJPEG_QUALITY=70"
Environment="MJPEG_PORT=8080"
Environment="MJPEG_HOST=0.0.0.0"
Environment="MJPEG_QUEUE_SIZE=2"
Environment="MJPEG_BOUNDARY=frameboundary"
Environment="MJPEG_CAPTURE_SLEEP=0.125"
ExecStart=/home/pi/mjpeg-csi/venv/bin/python /home/pi/mjpeg-csi/app.py
Restart=on-failure
RestartSec=5

[Install]
WantedBy=multi-user.target
EOF

Recarga, habilita e inicia:

sudo systemctl daemon-reload
sudo systemctl enable --now mjpeg-csi.service
systemctl status mjpeg-csi.service --no-pager

Valida en el navegador: http://:8080/stream.mjpg

Validación paso a paso

1. Comprobación de hardware CSI:
2. Cámara correctamente conectada (cable firme, pestañas ZIF cerradas).

3. Reinicio tras conexión.

4. Comprobación libcamera:

5. libcamera-hello -t 2000 no debe mostrar “No cameras available”.

6. libcamera-hello --version muestra versión de libcamera-apps. Ejemplo:

   • libcamera-apps version 1.1.x (en Bookworm, varía según build de RPi OS).

7. Comprobación de Picamera2 en Python:

8. Inicia Python: python -c "from picamera2 import Picamera2; print('OK picamera2')".

9. Arranque de servidor:

10. systemctl status mjpeg-csi.service → “active (running)”.

11. Logs iniciales en journalctl -u mjpeg-csi.service -n 50 --no-pager.

12. Conectividad:

13. curl -sI http://<IP>:8080/health → HTTP/1.1 200 OK y JSON con parámetros.
14. curl -sI http://<IP>:8080/stream.mjpg → Content-Type: multipart/x-mixed-replace; boundary=frameboundary.

15. curl -o /dev/null http://<IP>:8080/snapshot.jpg -v → Content-Type: image/jpeg.

16. Visualización:

17. Navegador (Chrome/Firefox): abrir http://:8080/stream.mjpg; deberías ver vídeo fluido (~8 fps).
18. VLC: Medio → Abrir ubicación de red → http://:8080/stream.mjpg.

19. ffplay (opcional en PC): ffplay -fflags nobuffer -flags low_delay -framedrop http://<IP>:8080/stream.mjpg.

20. Rendimiento en el Zero W:

21. top o htop: CPU del proceso Python entre 60–95% según calidad/fps/resolución.

22. Ajusta variables (anchura, altura, FPS, calidad) si observas throttling o cortes.

23. Persistencia:

24. Reinicia el Pi: sudo reboot.
25. Valida que el servicio levanta automáticamente y el stream responde.

Troubleshooting

1) Error: “No cameras available” con libcamera-hello
– Causas:
– Cable CSI mal orientado o suelto.
– Cámara defectuosa.
– dt/config no detecta el sensor.
– Soluciones:
– Apaga, reconecta cable CSI en ambos extremos; verifica orientación de contactos.
– En /boot/firmware/config.txt añade: camera_auto_detect=1 y, si persiste, fuerza dtoverlay=imx219.
– Actualiza firmware: sudo apt update && sudo apt full-upgrade -y && sudo reboot.

2) ImportError: No module named ‘picamera2’
– Causas: Falta el paquete apt o no se ve desde el venv.
– Solución:
– sudo apt install -y python3-picamera2
– Crea el venv con --system-site-packages y actívalo de nuevo.
– Comprueba en Python: from picamera2 import Picamera2.

3) “Address already in use” al arrancar el servidor
– Causa: Puerto 8080 ocupado.
– Solución:
– Cambia MJPEG_PORT (p. ej., 8081) en el servicio systemd y systemctl daemon-reload && systemctl restart mjpeg-csi.service.
– Verifica puertos: ss -ltnp | grep :8080.

4) Latencia alta o stream entrecortado
– Causas: CPU saturada (Zero W es limitado).
– Ajustes:
– Reduce resolución: MJPEG_WIDTH=426, MJPEG_HEIGHT=240.
– Baja FPS: MJPEG_FPS=6 (y MJPEG_CAPTURE_SLEEP acorde ~0.166).
– Baja calidad JPEG: MJPEG_QUALITY=60.
– Verifica Wi‑Fi (RSSI) y evita congestión 2.4 GHz.

5) Imagen muy oscura/borrosa
– Causas: Iluminación insuficiente, exposiciones largas.
– Soluciones:
– Mejora iluminación.
– Baja FPS (para permitir mayor exposición) o considera fijar controles en Picamera2 (AeEnable=True, ISO).
– Limpia el protector/óptica.

6) Waitress no responde al stream tras varios clientes
– Causas: Límite de threads insuficiente en Waitress vs. clientes simultáneos.
– Solución:
– Incrementa threads en el servicio (p. ej., 6–8).
– Recuerda que más clientes aumentan la carga en CPU.

7) El stream no abre en Safari/iOS
– Causas: Implementación MJPEG y/o caché.
– Soluciones:
– Asegura cabecera correcta multipart con boundary constante (ya lo hace el código).
– Prueba otra app (VLC) o un navegador alternativo; Safari a veces interrumpe MJPEG en ciertas condiciones de red.

8) “Broken pipe” o desconexiones frecuentes
– Causas: Cortes Wi‑Fi, clientes que cierran y servidor aún enviando frames.
– Soluciones:
– Estas excepciones suelen ser benignas; observa logs y mantén la red estable.
– Reduce FPS/Quality para ahorrar ancho de banda.

Mejoras/variantes

• Encoder alternativo con MJPEGEncoder de Picamera2:
• Implementar un Output personalizado que reciba frames JPEG directos del encoder, reduciendo CPU. Requiere profundizar en la API de encoders/outputs de Picamera2.
• Doble endpoint con distintas calidades:
• /stream_small.mjpg (426×240 @ 8 fps, Q=60)
• /stream_large.mjpg (640×480 @ 8 fps, Q=75) si la CPU lo permite.
• Autenticación básica:
• Añade un decorador simple o usa un reverse proxy Nginx con auth.
• HTTPS:
• Coloca Nginx delante (TLS) y haz proxy_pass a 127.0.0.1:8080.
• Control de cámara:
• Endpoints para cambiar parámetros en vivo (FPS, calidad), guardándolos en un archivo .env.
• Grabación puntual:
• Endpoint /record?n=10 para capturar N segundos a JPEGs o MKV con ffmpeg (ojo con carga del Zero W).
• Métricas:
• Exponer /metrics con tiempos de frame, colas, uso de CPU (psutil).

Checklist de verificación

• [ ] Hardware:
• [ ] Raspberry Pi Zero W con fuente 5 V/2 A estable.
• [ ] Camera Module v2 con cable CSI bien orientado y pestillos cerrados.
• [ ] microSD con Raspberry Pi OS Bookworm Lite (32-bit).
• [ ] Sistema:
• [ ] sudo apt update && sudo apt full-upgrade -y ejecutado sin errores.
• [ ] libcamera-hello -t 2000 detecta la cámara.
• [ ] Toolchain / Paquetes:
• [ ] python3-picamera2 instalado por apt.
• [ ] venv creado con –system-site-packages.
• [ ] pip configurado con piwheels.
• [ ] Flask==2.3.3, Pillow==10.3.0, Waitress==2.1.2 instalados.
• [ ] Código:
• [ ] app.py creado y ejecutable.
• [ ] Variables (MJPEG_WIDTH/HEIGHT/FPS/QUALITY) ajustadas a tu entorno.
• [ ] Ejecución:
• [ ] python app.py entrega /health con status ok.
• [ ] /stream.mjpg visible en navegador o VLC.
• [ ] /snapshot.jpg devuelve un JPEG válido.
• [ ] Servicio:
• [ ] systemd mjpeg-csi.service en estado “active (running)”.
• [ ] Arranca al boot y responde tras reinicio.
• [ ] Rendimiento:
• [ ] CPU en rango aceptable; si no, reduce resolución/FPS/calidad.
• [ ] Sin cortes frecuentes; Wi‑Fi estable.

Con este caso práctico, dispones de un servidor web MJPEG eficiente y reproducible con el conjunto “Raspberry Pi Zero W + Camera Module v2” usando la pila libcamera y Python 3.11 en Raspberry Pi OS Bookworm (32-bit). Si en el futuro migras a placas más potentes (por ejemplo, Pi 3/4/5), puedes incrementar resolución, FPS y calidad, o sustituir la compresión por encoders más exigentes sin alterar la arquitectura básica del servidor.