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()