Registrador en Python

Crea meter_logger.py:

#!/usr/bin/env python3
import csv
import os
import signal
import sys
import time
from datetime import datetime, timezone

import RPi.GPIO as GPIO

GPIO_PIN = 17
CSV_PATH = os.path.expanduser("~/utility-meter-logger/meter_log.csv")
STATE_PATH = os.path.expanduser("~/utility-meter-logger/state.txt")

# Example: 1000 pulses per kWh for an electricity meter.
# Change this value to match your utility meter.
PULSES_PER_UNIT = 1000.0
UNIT_NAME = "kWh"

# Debounce time in milliseconds. Increase if your contact bounces.
BOUNCE_MS = 50

running = True
pulse_count = 0


def load_state(path: str) -> int:
    if not os.path.exists(path):
        return 0
    with open(path, "r", encoding="utf-8") as f:
        text = f.read().strip()
    if text == "":
        return 0
    return int(text)


def save_state(path: str, count: int) -> None:
    tmp_path = path + ".tmp"
    with open(tmp_path, "w", encoding="utf-8") as f:
        f.write(f"{count}\n")
    os.replace(tmp_path, path)


def ensure_csv_header(path: str) -> None:
    if os.path.exists(path) and os.path.getsize(path) > 0:
        return
    os.makedirs(os.path.dirname(path), exist_ok=True)
    with open(path, "a", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        writer.writerow(
            ["timestamp_utc", "pulse_count_total", "delta_pulses", "consumption_total", "unit"]
        )


def append_csv(path: str, count: int, delta: int) -> None:
    consumption = count / PULSES_PER_UNIT
    timestamp = datetime.now(timezone.utc).isoformat()
    with open(path, "a", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        writer.writerow([timestamp, count, delta, f"{consumption:.6f}", UNIT_NAME])


def handle_pulse(channel: int) -> None:
    global pulse_count
    pulse_count += 1
    save_state(STATE_PATH, pulse_count)
    append_csv(CSV_PATH, pulse_count, 1)
    print(
        f"{datetime.now(timezone.utc).isoformat()} pulse={pulse_count} "
        f"consumption={pulse_count / PULSES_PER_UNIT:.6f} {UNIT_NAME}",
        flush=True,
    )


def handle_signal(signum, frame) -> None:
    global running
    running = False


def main() -> int:
    global pulse_count

    pulse_count = load_state(STATE_PATH)
    ensure_csv_header(CSV_PATH)

    GPIO.setmode(GPIO.BCM)
    GPIO.setup(GPIO_PIN, GPIO.IN, pull_up_down=GPIO.PUD_UP)

    signal.signal(signal.SIGINT, handle_signal)
    signal.signal(signal.SIGTERM, handle_signal)

    GPIO.add_event_detect(
        GPIO_PIN,
        GPIO.FALLING,
        callback=handle_pulse,
        bouncetime=BOUNCE_MS,
    )

    print(
        f"Utility meter pulse logger started on GPIO{GPIO_PIN}. "
        f"Current total pulses: {pulse_count}",
        flush=True,
    )

    try:
        while running:
            time.sleep(1.0)
    finally:
        GPIO.cleanup()

    return 0


if __name__ == "__main__":
    sys.exit(main())

Hazlo ejecutable:

chmod +x ~/utility-meter-logger/meter_logger.py

Método de validación

Para validar que este registrador de pulsos de contador de suministros funciona correctamente, usa una fuente de pulsos conocida.

Método 1: Prueba manual de contacto seco

1. Detén el servicio si está en ejecución:

   sudo systemctl stop utility-meter-logger.service
   ```

2. Ejecuta el registrador en una terminal:

```bash
   cd ~/utility-meter-logger
   python3 meter_logger.py
   ```

3. Cortocircuita brevemente `GPIO17` con `GND` exactamente 10 veces usando un pulsador o un cable jumper.
4. Comprueba que se imprimen 10 líneas nuevas y que el conteo total de pulsos aumenta en 10.

Evidencia esperada:

- la salida de consola incrementa una vez por cada cortocircuito
- `tail -n 10 ~/utility-meter-logger/meter_log.csv` muestra 10 filas nuevas
- `cat ~/utility-meter-logger/state.txt` coincide con el conteo total esperado

### Método 2: Comprobación de la constante del contador

Si tu contador eléctrico está etiquetado como `1000 imp/kWh` o `1000 pulses/kWh`, entonces:

- 1000 pulsos registrados deben equivaler a `1.000000 kWh`
- 100 pulsos deben equivaler a `0.100000 kWh`

Evidencia esperada:

- el campo `consumption_total` en el CSV coincide con `pulse_count_total / 1000`

## Solución de problemas

### No se cuentan pulsos

Comprueba:

- que la salida del sensor realmente lleva la línea a GND
- que el número GPIO es correcto: `GPIO17`, no el número de pin físico `17`
- que la salida del contador es segura y de bajo voltaje
- que el tiempo antirrebote no es demasiado corto para un contacto reed

### Conteo doble

Si ves conteos extra desde un interruptor reed, aumenta:


```python
BOUNCE_MS = 100

Luego reinicia el script.

Unidades de consumo incorrectas

Configura correctamente estos valores para tu contador de suministros:

PULSES_PER_UNIT = 1000.0
UNIT_NAME = "kWh"

Ejemplos:

• contador de agua con 1 pulse/litre:
• PULSES_PER_UNIT = 1.0
• UNIT_NAME = "litre"
• contador eléctrico con 1000 pulses/kWh:
• PULSES_PER_UNIT = 1000.0
• UNIT_NAME = "kWh"

Aplicación en Python

Guarda este archivo como /home/pi/mailbox-alert/mailbox_alert.py.

#!/usr/bin/env python3
from gpiozero import Button
from signal import pause
from datetime import datetime
import os
import socket
import requests

PIN = 17
LOG_FILE = "/home/pi/mailbox-alert/mailbox_events.log"
HOSTNAME = socket.gethostname()
TOPIC = f"{HOSTNAME}-smart-mailbox-open-alert"
NTFY_URL = f"https://ntfy.sh/{TOPIC}"

sensor = Button(PIN, pull_up=False, bounce_time=0.2)

def write_log(message: str) -> None:
    timestamp = datetime.now().isoformat(timespec="seconds")
    line = f"{timestamp} {message}\n"
    with open(LOG_FILE, "a", encoding="utf-8") as log_file:
        log_file.write(line)

def send_alert() -> None:
    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    title = "Mailbox opened"
    body = f"Smart mailbox open alert from {HOSTNAME} at {now}"
    headers = {
        "Title": title,
        "Priority": "default",
        "Tags": "mailbox,postbox,package"
    }

    try:
        response = requests.post(
            NTFY_URL,
            data=body.encode("utf-8"),
            headers=headers,
            timeout=10
        )
        response.raise_for_status()
        write_log(f"ALERT_SENT topic={TOPIC} status={response.status_code}")
    except requests.RequestException as exc:
        write_log(f"ALERT_FAILED error={exc}")

def on_open() -> None:
    write_log("MAILBOX_OPEN")
    send_alert()

def main() -> None:
    write_log(f"START topic={TOPIC} pin={PIN}")
    sensor.when_pressed = on_open
    pause()

if __name__ == "__main__":
    os.makedirs("/home/pi/mailbox-alert", exist_ok=True)
    main()

Hazlo ejecutable:

chmod +x /home/pi/mailbox-alert/mailbox_alert.py

Monitor de corriente en Python

Guarda esto como current_monitor.py:

#!/usr/bin/env python3
import csv
import sys
import time
from datetime import datetime

import board
import busio
from adafruit_ina219 import ADCResolution, BusVoltageRange, Gain, INA219


def create_sensor():
    i2c = busio.I2C(board.SCL, board.SDA)
    sensor = INA219(i2c)

    # Conservative configuration suitable for many low-voltage bench setups.
    sensor.bus_adc_resolution = ADCResolution.ADCRES_12BIT_32S
    sensor.shunt_adc_resolution = ADCResolution.ADCRES_12BIT_32S
    sensor.bus_voltage_range = BusVoltageRange.RANGE_16V
    sensor.gain = Gain.DIV_8_320MV
    return sensor


def open_csv(path):
    file_obj = open(path, "a", newline="", encoding="utf-8")
    writer = csv.writer(file_obj)
    if file_obj.tell() == 0:
        writer.writerow(
            ["timestamp", "bus_voltage_V", "shunt_voltage_mV", "current_A", "power_W"]
        )
        file_obj.flush()
    return file_obj, writer


def main():
    csv_path = sys.argv[1] if len(sys.argv) > 1 else "current_log.csv"
    interval_s = float(sys.argv[2]) if len(sys.argv) > 2 else 1.0

    sensor = create_sensor()
    csv_file, writer = open_csv(csv_path)

    print("Raspberry Pi lab power supply current monitor")
    print(f"Logging to: {csv_path}")
    print("Press Ctrl+C to stop.")
    print("timestamp,bus_voltage_V,shunt_voltage_mV,current_A,power_W")

    try:
        while True:
            timestamp = datetime.now().isoformat(timespec="seconds")
            bus_voltage = sensor.bus_voltage
            shunt_voltage_mv = sensor.shunt_voltage / 1000.0
            current_a = sensor.current / 1000.0
            power_w = sensor.power / 1000.0

            row = [
                timestamp,
                f"{bus_voltage:.3f}",
                f"{shunt_voltage_mv:.3f}",
                f"{current_a:.6f}",
                f"{power_w:.6f}",
            ]
            print(",".join(row))
            writer.writerow(row)
            csv_file.flush()
            time.sleep(interval_s)
    except KeyboardInterrupt:
        print("\nStopped.")
    finally:
        csv_file.close()


if __name__ == "__main__":
    main()

Hazlo ejecutable y ejecútalo:

chmod +x current_monitor.py
python3 current_monitor.py current_log.csv 1.0

Programa del controlador

Guarda este archivo como co2_ventilation_controller.py.

#!/usr/bin/env python3
import time
import serial
import RPi.GPIO as GPIO

UART_DEVICE = "/dev/serial0"
UART_BAUD = 9600

RELAY_GPIO = 17
RELAY_ACTIVE_LOW = True

CO2_ON_THRESHOLD = 1000   # ppm: turn ventilation ON at or above this level
CO2_OFF_THRESHOLD = 800   # ppm: turn ventilation OFF at or below this level

POLL_INTERVAL_SECONDS = 5.0


def relay_write(enabled: bool) -> None:
    if RELAY_ACTIVE_LOW:
        GPIO.output(RELAY_GPIO, GPIO.LOW if enabled else GPIO.HIGH)
    else:
        GPIO.output(RELAY_GPIO, GPIO.HIGH if enabled else GPIO.LOW)


def mhz19_checksum(packet: bytes) -> int:
    checksum = 0xFF - (sum(packet[1:8]) % 256) + 1
    return checksum & 0xFF


def read_co2_ppm(ser: serial.Serial) -> int:
    command = bytes([0xFF, 0x01, 0x86, 0x00, 0x00, 0x00, 0x00, 0x00, 0x79])
    ser.reset_input_buffer()
    ser.write(command)
    response = ser.read(9)

    if len(response) != 9:
        raise RuntimeError(f"Incomplete response: got {len(response)} bytes")

    if response[0] != 0xFF or response[1] != 0x86:
        raise RuntimeError(f"Unexpected response header: {response.hex()}")

    if response[8] != mhz19_checksum(response):
        raise RuntimeError(f"Checksum error: {response.hex()}")

    co2_ppm = response[2] * 256 + response[3]
    return co2_ppm


def main() -> None:
    GPIO.setmode(GPIO.BCM)
    GPIO.setup(RELAY_GPIO, GPIO.OUT)
    relay_enabled = False
    relay_write(relay_enabled)

    print("Starting CO2 ventilation relay controller")
    print(f"ON threshold : {CO2_ON_THRESHOLD} ppm")
    print(f"OFF threshold: {CO2_OFF_THRESHOLD} ppm")

    try:
        with serial.Serial(UART_DEVICE, UART_BAUD, timeout=1.0) as ser:
            while True:
                try:
                    co2_ppm = read_co2_ppm(ser)

                    if not relay_enabled and co2_ppm >= CO2_ON_THRESHOLD:
                        relay_enabled = True
                        relay_write(relay_enabled)
                    elif relay_enabled and co2_ppm <= CO2_OFF_THRESHOLD:
                        relay_enabled = False
                        relay_write(relay_enabled)

                    state = "ON" if relay_enabled else "OFF"
                    print(f"CO2={co2_ppm} ppm, ventilation relay={state}")
                except RuntimeError as exc:
                    print(f"Read error: {exc}")

                time.sleep(POLL_INTERVAL_SECONDS)
    finally:
        relay_write(False)
        GPIO.cleanup()


if __name__ == "__main__":
    main()

Código 2: Controlador de bomba de riego para plantas

Guarda como irrigation_controller.py:

#!/usr/bin/env python3
from gpiozero import DigitalInputDevice, OutputDevice
import time
from datetime import datetime

SENSOR_PIN = 17
RELAY_PIN = 27

CHECK_INTERVAL_SECONDS = 30
PUMP_ON_SECONDS = 5
POST_WATERING_DELAY_SECONDS = 60
MAX_CYCLES_PER_HOUR = 6

# Assumption for this tutorial:
# sensor.value == 0 means dry soil
# sensor.value == 1 means wet soil
DRY_SENSOR_VALUE = 0

sensor = DigitalInputDevice(SENSOR_PIN, pull_up=False)
relay = OutputDevice(RELAY_PIN, active_high=False, initial_value=False)

cycle_timestamps = []

def now_text() -> str:
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

def is_dry() -> bool:
    return sensor.value == DRY_SENSOR_VALUE

def prune_old_cycles() -> None:
    global cycle_timestamps
    cutoff = time.time() - 3600
    cycle_timestamps = [t for t in cycle_timestamps if t >= cutoff]

def can_water() -> bool:
    prune_old_cycles()
    return len(cycle_timestamps) < MAX_CYCLES_PER_HOUR

def water_once() -> None:
    print(f"{now_text()} Pump ON for {PUMP_ON_SECONDS} s")
    relay.on()
    time.sleep(PUMP_ON_SECONDS)
    relay.off()
    cycle_timestamps.append(time.time())
    print(f"{now_text()} Pump OFF")

def main() -> None:
    print(f"{now_text()} Irrigation controller started")
    while True:
        dry = is_dry()
        print(f"{now_text()} Sensor value={sensor.value} dry={dry}")

        if dry:
            if can_water():
                water_once()
                print(f"{now_text()} Waiting {POST_WATERING_DELAY_SECONDS} s after watering")
                time.sleep(POST_WATERING_DELAY_SECONDS)
            else:
                print(f"{now_text()} Watering skipped: hourly cycle limit reached")
                time.sleep(CHECK_INTERVAL_SECONDS)
        else:
            time.sleep(CHECK_INTERVAL_SECONDS)

if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        relay.off()
        print(f"{now_text()} Stopped by user")

Ejecútalo:

python3 irrigation_controller.py

Aplicación

Crea freezer_alarm.py:

#!/usr/bin/env python3
from __future__ import annotations

import csv
import glob
import os
import time
from datetime import datetime, timezone
from pathlib import Path

from flask import Flask, jsonify, render_template_string

# Alarm threshold in degrees Celsius.
# Example: alarm if freezer temperature rises above -12.0 C.
ALARM_THRESHOLD_C = -12.0

# Logging interval in seconds.
LOG_INTERVAL_S = 60

# CSV log file
LOG_FILE = Path("/home/pi/freezer_temperature_log.csv")

# DS18B20 sysfs base path
W1_BASE = "/sys/bus/w1/devices"

app = Flask(__name__)

HTML_TEMPLATE = """<!doctype html>
<html lang="en">
<head>
  <meta charset="utf-8">
  <meta http-equiv="refresh" content="15">
  <meta name="viewport" content="width=device-width, initial-scale=1">
  <title>Freezer Temperature Web Alarm</title>
  <style>
    body { font-family: Arial, sans-serif; margin: 2rem; background: #f5f7fa; color: #222; }
    .card { max-width: 36rem; padding: 1.5rem; border-radius: 12px; background: white; box-shadow: 0 2px 8px rgba(0,0,0,0.08); }
    .ok { color: #0a7d21; font-weight: bold; }
    .alarm { color: #b00020; font-weight: bold; }
    .value { font-size: 2rem; margin: 0.5rem 0; }
    code { background: #eef; padding: 0.1rem 0.3rem; border-radius: 4px; }
  </style>
</head>
<body>
  <div class="card">
    <h1>Freezer Temperature Web Alarm</h1>
    <p class="value">{{ temp_text }}</p>
    <p>Status:
      {% if alarm %}
        <span class="alarm">ALARM</span>
      {% else %}
        <span class="ok">OK</span>
      {% endif %}
    </p>
    <p>Alarm threshold: <code>{{ threshold }} C</code></p>
    <p>Last update: <code>{{ timestamp }}</code></p>
    <p>JSON endpoint: <code>/api/status</code></p>
  </div>
</body>
</html>
"""


def find_sensor_file() -> str:
    matches = glob.glob(os.path.join(W1_BASE, "28-*", "w1_slave"))
    if not matches:
        raise FileNotFoundError(
            "No DS18B20 sensor found under /sys/bus/w1/devices/. "
            "Check wiring and 1-Wire configuration."
        )
    return matches[0]


def read_temperature_c() -> float:
    sensor_file = find_sensor_file()
    with open(sensor_file, "r", encoding="ascii") as f:
        lines = f.read().strip().splitlines()

    if len(lines) != 2 or not lines[0].endswith("YES"):
        raise RuntimeError("Sensor CRC check failed")

    marker = "t="
    pos = lines[1].find(marker)
    if pos == -1:
        raise RuntimeError("Temperature value not found in sensor output")

    milli_c = int(lines[1][pos + len(marker):])
    return milli_c / 1000.0


def alarm_state(temp_c: float) -> bool:
    return temp_c > ALARM_THRESHOLD_C


def ensure_log_header() -> None:
    if not LOG_FILE.exists():
        LOG_FILE.parent.mkdir(parents=True, exist_ok=True)
        with open(LOG_FILE, "w", newline="", encoding="utf-8") as f:
            writer = csv.writer(f)
            writer.writerow(["timestamp_utc", "temperature_c", "alarm"])


def append_log_once() -> None:
    ensure_log_header()
    temp_c = read_temperature_c()
    state = alarm_state(temp_c)
    now = datetime.now(timezone.utc).isoformat()

    with open(LOG_FILE, "a", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        writer.writerow([now, f"{temp_c:.3f}", int(state)])


@app.route("/")
def index():
    temp_c = read_temperature_c()
    state = alarm_state(temp_c)
    timestamp = datetime.now(timezone.utc).isoformat()

    return render_template_string(
        HTML_TEMPLATE,
        temp_text=f"{temp_c:.2f} C",
        alarm=state,
        threshold=f"{ALARM_THRESHOLD_C:.1f}",
        timestamp=timestamp,
    )


@app.route("/api/status")
def api_status():
    temp_c = read_temperature_c()
    state = alarm_state(temp_c)
    timestamp = datetime.now(timezone.utc).isoformat()
    return jsonify(
        {
            "device": "Raspberry Pi",
            "application": "freezer-temperature-web-alarm",
            "temperature_c": round(temp_c, 3),
            "alarm": state,
            "alarm_threshold_c": ALARM_THRESHOLD_C,
            "timestamp_utc": timestamp,
        }
    )


def main() -> None:
    append_log_once()
    app.run(host="0.0.0.0", port=8000, debug=False)


if __name__ == "__main__":
    main()

Hazlo ejecutable:

chmod +x freezer_alarm.py

Servicio de registro opcional

La aplicación web registra una vez al iniciarse. Para un registro continuo en segundo plano cada minuto, crea freezer_logger.py:

#!/usr/bin/env python3
from __future__ import annotations

import csv
import glob
import os
import time
from datetime import datetime, timezone
from pathlib import Path

ALARM_THRESHOLD_C = -12.0
LOG_INTERVAL_S = 60
LOG_FILE = Path("/home/pi/freezer_temperature_log.csv")
W1_BASE = "/sys/bus/w1/devices"


def find_sensor_file() -> str:
    matches = glob.glob(os.path.join(W1_BASE, "28-*", "w1_slave"))
    if not matches:
        raise FileNotFoundError("No DS18B20 sensor found")
    return matches[0]


def read_temperature_c() -> float:
    sensor_file = find_sensor_file()
    with open(sensor_file, "r", encoding="ascii") as f:
        lines = f.read().strip().splitlines()

    if len(lines) != 2 or not lines[0].endswith("YES"):
        raise RuntimeError("Sensor CRC check failed")

    pos = lines[1].find("t=")
    if pos == -1:
        raise RuntimeError("Temperature value not found")

    return int(lines[1][pos + 2:]) / 1000.0


def ensure_log_header() -> None:
    if not LOG_FILE.exists():
        LOG_FILE.parent.mkdir(parents=True, exist_ok=True)
        with open(LOG_FILE, "w", newline="", encoding="utf-8") as f:
            csv.writer(f).writerow(["timestamp_utc", "temperature_c", "alarm"])


def main() -> None:
    ensure_log_header()
    while True:
        now = datetime.now(timezone.utc).isoformat()
        temp_c = read_temperature_c()
        alarm = int(temp_c > ALARM_THRESHOLD_C)
        with open(LOG_FILE, "a", newline="", encoding="utf-8") as f:
            csv.writer(f).writerow([now, f"{temp_c:.3f}", alarm])
        time.sleep(LOG_INTERVAL_S)


if __name__ == "__main__":
    main()

Ejecuta la aplicación web:

python3 freezer_alarm.py

En otra terminal, ejecuta el registrador:

python3 freezer_logger.py

Abre la página de estado desde otro dispositivo en la misma red:

hostname -I

Luego navega a:

http://<raspberry-pi-ip>:8000/