Caso práctico: registrador MQTT con Raspberry Pi 4

#!/usr/bin/env python3
"""
MQTT environment data logger for:
Raspberry Pi 4 Model B + BME680 + DS3231 RTC

Features:
- Dry-run mode for validation on normal computers
- Hardware mode via smbus2 on Raspberry Pi
- MQTT publishing with paho-mqtt
- JSON payloads
- DS3231 RTC time read
- BME680 basic sensor read placeholder via adapter logic

This file is py_compile-valid and runnable without hardware in --dry-run mode.
"""

from __future__ import annotations

import argparse
import datetime as dt
import json
import math
import os
import random
import socket
import sys
import time
from dataclasses import dataclass, asdict
from typing import Optional


def bcd_to_int(value: int) -> int:
    return ((value >> 4) * 10) + (value & 0x0F)


@dataclass
class EnvReading:
    iso_time: str
    unix_time: int
    temperature_c: float
    humidity_pct: float
    pressure_hpa: float
    gas_ohm: float
    source_time: str
    hostname: str
    sequence: int


class I2CBusBase:
    def read_i2c_block_data(self, addr: int, reg: int, length: int) -> list[int]:
        raise NotImplementedError

    def write_byte_data(self, addr: int, reg: int, value: int) -> None:
        raise NotImplementedError


class MockI2CBus(I2CBusBase):
    def __init__(self) -> None:
        self._start = time.time()

    def read_i2c_block_data(self, addr: int, reg: int, length: int) -> list[int]:
        if addr == 0x68 and reg == 0x00 and length == 7:
            now = dt.datetime.now()
            return [
                ((now.second // 10) << 4) | (now.second % 10),
                ((now.minute // 10) << 4) | (now.minute % 10),
                ((now.hour // 10) << 4) | (now.hour % 10),
                1,
                ((now.day // 10) << 4) | (now.day % 10),
                ((now.month // 10) << 4) | (now.month % 10),
                (((now.year - 2000) // 10) << 4) | ((now.year - 2000) % 10),
            ]
        return [0] * length

    def write_byte_data(self, addr: int, reg: int, value: int) -> None:
        return


class SMBusAdapter(I2CBusBase):
    def __init__(self, bus_id: int) -> None:
        from smbus2 import SMBus
        self._bus = SMBus(bus_id)

    def read_i2c_block_data(self, addr: int, reg: int, length: int) -> list[int]:
        return self._bus.read_i2c_block_data(addr, reg, length)

    def write_byte_data(self, addr: int, reg: int, value: int) -> None:
        self._bus.write_byte_data(addr, reg, value)


class DS3231RTC:
    def __init__(self, bus: I2CBusBase, address: int = 0x68) -> None:
        self.bus = bus
        self.address = address

    def read_datetime(self) -> dt.datetime:
        raw = self.bus.read_i2c_block_data(self.address, 0x00, 7)
        second = bcd_to_int(raw[0] & 0x7F)
        minute = bcd_to_int(raw[1] & 0x7F)
        hour = bcd_to_int(raw[2] & 0x3F)
        day = bcd_to_int(raw[4] & 0x3F)
        month = bcd_to_int(raw[5] & 0x1F)
        year = 2000 + bcd_to_int(raw[6])
        return dt.datetime(year, month, day, hour, minute, second)


class BME680Sensor:
    """
    Educational adapter.

    In dry-run mode it generates plausible values.
    In hardware mode without a full external driver, it still provides
    a clear adapter boundary for later extension.

    For beginner education, the practical project goal is MQTT logging flow.
    """

    def __init__(self, bus: I2CBusBase, address: int = 0x76, dry_run: bool = False) -> None:
        self.bus = bus
        self.address = address
        self.dry_run = dry_run
        self._t0 = time.time()

    def read(self) -> tuple[float, float, float, float]:
        if self.dry_run:
            elapsed = time.time() - self._t0
            temperature_c = 23.0 + 2.0 * math.sin(elapsed / 90.0) + random.uniform(-0.2, 0.2)
            humidity_pct = 48.0 + 5.0 * math.sin(elapsed / 120.0) + random.uniform(-0.5, 0.5)
            pressure_hpa = 1012.0 + 1.5 * math.sin(elapsed / 200.0) + random.uniform(-0.3, 0.3)
            gas_ohm = 12000.0 + 1500.0 * math.sin(elapsed / 150.0) + random.uniform(-100.0, 100.0)
            return (
                round(temperature_c, 2),
                round(humidity_pct, 2),
                round(pressure_hpa, 2),
                round(gas_ohm, 2),
            )

        raise RuntimeError(
            "Hardware BME680 raw driver not included in this basic tutorial. "
            "Use --dry-run for validation, or extend the BME680 adapter with a tested hardware library."
        )


class MQTTClientAdapter:
    def __init__(self, broker: str, port: int, topic: str, client_id: str, dry_run: bool = False) -> None:
        self.broker = broker
        self.port = port
        self.topic = topic
        self.client_id = client_id
        self.dry_run = dry_run
        self._client = None

    def connect(self) -> None:
        if self.dry_run:
            print(f"[DRY-RUN] MQTT connect to {self.broker}:{self.port} as {self.client_id}")
            return

        import paho.mqtt.client as mqtt
        self._client = mqtt.Client(client_id=self.client_id)
        self._client.connect(self.broker, self.port, 60)
        self._client.loop_start()

    def publish(self, payload: dict) -> None:
        payload_text = json.dumps(payload, separators=(",", ":"), sort_keys=True)
        if self.dry_run:
            print(f"[DRY-RUN] MQTT publish topic={self.topic} payload={payload_text}")
            return

        if self._client is None:
            raise RuntimeError("MQTT client not connected")
        result = self._client.publish(self.topic, payload_text, qos=0, retain=False)
        if result.rc != 0:
            raise RuntimeError(f"MQTT publish failed with rc={result.rc}")

    def close(self) -> None:
        if self._client is not None:
            self._client.loop_stop()
            self._client.disconnect()


def build_reading(
    rtc: DS3231RTC,
    sensor: BME680Sensor,
    sequence: int,
    fallback_to_system_time: bool = True
) -> EnvReading:
    source_time = "rtc"
    try:
        ts = rtc.read_datetime()
    except Exception:
        if not fallback_to_system_time:
            raise
        ts = dt.datetime.now()
        source_time = "system"

    temperature_c, humidity_pct, pressure_hpa, gas_ohm = sensor.read()
    return EnvReading(
        iso_time=ts.isoformat(),
        unix_time=int(ts.timestamp()),
        temperature_c=temperature_c,
        humidity_pct=humidity_pct,
        pressure_hpa=pressure_hpa,
        gas_ohm=gas_ohm,
        source_time=source_time,
        hostname=socket.gethostname(),
        sequence=sequence,
    )


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="MQTT environment data logger")
    parser.add_argument("--broker", default="localhost", help="MQTT broker hostname or IP")
    parser.add_argument("--port", type=int, default=1883, help="MQTT broker port")
    parser.add_argument("--topic", default="lab/env/pi4/logger", help="MQTT topic")
    parser.add_argument("--interval", type=int, default=30, help="Publish interval in seconds")
    parser.add_argument("--count", type=int, default=0, help="Number of messages to publish, 0 means infinite")
    parser.add_argument("--dry-run", action="store_true", help="Run without physical hardware or broker")
    parser.add_argument("--bus", type=int, default=1, help="I2C bus number")
    parser.add_argument("--bme680-addr", type=lambda x: int(x, 0), default=0x76, help="BME680 I2C address")
    parser.add_argument("--rtc-addr", type=lambda x: int(x, 0), default=0x68, help="DS3231 I2C address")
    return parser.parse_args()


def main() -> int:
    args = parse_args()

    if args.dry_run:
        bus = MockI2CBus()
    else:
        bus = SMBusAdapter(args.bus)

    rtc = DS3231RTC(bus=bus, address=args.rtc_addr)
    sensor = BME680Sensor(bus=bus, address=args.bme680_addr, dry_run=args.dry_run)
    mqtt_client = MQTTClientAdapter(
        broker=args.broker,
        port=args.port,
        topic=args.topic,
        client_id=f"pi4-env-{os.getpid()}",
        dry_run=args.dry_run,
    )

    mqtt_client.connect()

    sent = 0
    try:
        while True:
            reading = build_reading(rtc, sensor, sequence=sent + 1)
            payload = asdict(reading)
            print(json.dumps(payload, indent=2, sort_keys=True))
            mqtt_client.publish(payload)

            sent += 1
            if args.count > 0 and sent >= args.count:
                break
            time.sleep(args.interval)
    except KeyboardInterrupt:
        print("Stopped by user")
    finally:
        mqtt_client.close()

    return 0


if __name__ == "__main__":
    raise SystemExit(main())