Application code

air_quality_alarm.py

#!/usr/bin/env python3
"""
Touchscreen air-quality alarm for:
- Raspberry Pi 4 Model B
- BME680 over I2C
- HyperPixel 4.0 Touch

The IAQ, eCO2, and TVOC values here are educational heuristic estimates.
They are useful for trend display and UI behavior validation, not certified measurement.
"""

from __future__ import annotations

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

import pygame


@dataclass
class SensorReading:
    temperature_c: float
    humidity_pct: float
    pressure_hpa: float
    gas_ohms: float
    iaq_score: float
    eco2_ppm: int
    tvoc_ppb: int
    timestamp: float


def clamp(value: float, low: float, high: float) -> float:
    return max(low, min(high, value))


def estimate_iaq_score(humidity_pct: float, gas_ohms: float) -> float:
    """
    Educational heuristic only.
    Lower score is better, with an approximate 0..500 scale.
    """
    humidity_target = 40.0
    humidity_offset = abs(humidity_pct - humidity_target)
    humidity_score = clamp(humidity_offset / 40.0 * 25.0, 0.0, 25.0)

    gas_reference = 50000.0
    gas_ratio = clamp((gas_reference - gas_ohms) / gas_reference, 0.0, 1.0)
    gas_score = gas_ratio * 475.0

    return round(humidity_score + gas_score, 1)


def estimate_eco2_ppm(iaq_score: float, gas_ohms: float) -> int:
    """
    Educational heuristic only.
    Maps worsening gas behavior to an estimated eCO2 range.
    """
    base = 420
    score_component = int(iaq_score * 4.2)
    gas_component = int(clamp((50000.0 - gas_ohms) / 80.0, 0.0, 2000.0))
    return int(clamp(base + score_component + gas_component, 400, 2500))


def estimate_tvoc_ppb(iaq_score: float, gas_ohms: float) -> int:
    """
    Educational heuristic only.
    """
    base = 20
    score_component = int(iaq_score * 1.5)
    gas_component = int(clamp((50000.0 - gas_ohms) / 120.0, 0.0, 1000.0))
    return int(clamp(base + score_component + gas_component, 0, 1200))


def air_quality_state(eco2_ppm: int) -> Tuple[str, Tuple[int, int, int]]:
    if eco2_ppm < 800:
        return ("GOOD", (40, 140, 70))
    if eco2_ppm < 1200:
        return ("ELEVATED", (180, 150, 40))
    if eco2_ppm < 1600:
        return ("POOR", (190, 100, 30))
    return ("ALARM", (170, 40, 40))


class BME680Adapter:
    """Real sensor adapter using the bme680 package."""

    def __init__(self, i2c_addr: int = 0x76):
        self.i2c_addr = i2c_addr
        self.sensor = None

    def open(self) -> None:
        import bme680

        self.sensor = bme680.BME680(i2c_addr=self.i2c_addr)
        self.sensor.set_humidity_oversample(bme680.OS_2X)
        self.sensor.set_pressure_oversample(bme680.OS_4X)
        self.sensor.set_temperature_oversample(bme680.OS_8X)
        self.sensor.set_filter(bme680.FILTER_SIZE_3)
        self.sensor.set_gas_status(bme680.ENABLE_GAS_MEAS)
        self.sensor.set_gas_heater_temperature(320)
        self.sensor.set_gas_heater_duration(150)
        self.sensor.select_gas_heater_profile(0)

    def read(self) -> SensorReading:
        if self.sensor is None:
            raise RuntimeError("Sensor not opened")

        if not self.sensor.get_sensor_data():
            raise RuntimeError("No sensor data available")

        data = self.sensor.data
        gas_ohms = float(getattr(data, "gas_resistance", 0.0) or 0.0)

        iaq_score = estimate_iaq_score(float(data.humidity), gas_ohms)
        eco2_ppm = estimate_eco2_ppm(iaq_score, gas_ohms)
        tvoc_ppb = estimate_tvoc_ppb(iaq_score, gas_ohms)

        return SensorReading(
            temperature_c=float(data.temperature),
            humidity_pct=float(data.humidity),
            pressure_hpa=float(data.pressure),
            gas_ohms=gas_ohms,
            iaq_score=iaq_score,
            eco2_ppm=eco2_ppm,
            tvoc_ppb=tvoc_ppb,
            timestamp=time.time(),
        )


class MockBME680Adapter:
    """Mock adapter for validation without hardware."""

    def __init__(self) -> None:
        self.t0 = time.time()

    def open(self) -> None:
        return

    def read(self) -> SensorReading:
        elapsed = time.time() - self.t0

        temperature_c = 22.0 + 1.5 * math.sin(elapsed / 40.0)
        humidity_pct = 45.0 + 8.0 * math.sin(elapsed / 55.0)
        pressure_hpa = 1012.0 + 1.2 * math.sin(elapsed / 120.0)

        baseline = 45000.0
        drop = min(32000.0, elapsed * 350.0)
        noise = random.uniform(-1200.0, 1200.0)
        gas_ohms = max(5000.0, baseline - drop + noise)

        iaq_score = estimate_iaq_score(humidity_pct, gas_ohms)
        eco2_ppm = estimate_eco2_ppm(iaq_score, gas_ohms)
        tvoc_ppb = estimate_tvoc_ppb(iaq_score, gas_ohms)

        return SensorReading(
            temperature_c=temperature_c,
            humidity_pct=humidity_pct,
            pressure_hpa=pressure_hpa,
            gas_ohms=gas_ohms,
            iaq_score=iaq_score,
            eco2_ppm=eco2_ppm,
            tvoc_ppb=tvoc_ppb,
            timestamp=time.time(),
        )


class HyperPixelUI:
    def __init__(self, width: int = 800, height: int = 480):
        pygame.init()
        pygame.font.init()
        self.screen = pygame.display.set_mode((width, height))
        pygame.display.set_caption("Air Quality Alarm")
        self.width = width
        self.height = height
        self.font_big = pygame.font.SysFont("Arial", 52)
        self.font_med = pygame.font.SysFont("Arial", 32)
        self.font_small = pygame.font.SysFont("Arial", 24)
        self.show_detail = False
        self.alarm_ack_until = 0.0

    def draw(self, reading: SensorReading, alarm_threshold: int) -> None:
        state, bg = air_quality_state(reading.eco2_ppm)
        self.screen.fill(bg)

        now = time.time()
        alarm_active = (
            reading.eco2_ppm >= alarm_threshold and now >= self.alarm_ack_until
        )

        title = self.font_big.render(state, True, (255, 255, 255))
        self.screen.blit(title, (30, 20))

        eco2_text = self.font_big.render(
            f"eCO2 {reading.eco2_ppm} ppm", True, (255, 255, 255)
        )
        self.screen.blit(eco2_text, (30, 95))

        iaq_text = self.font_med.render(
            f"IAQ score: {reading.iaq_score:.1f}", True, (255, 255, 255)
        )
        self.screen.blit(iaq_text, (30, 170))

        temp_text = self.font_small.render(
            f"Temp: {reading.temperature_c:.1f} C", True, (255, 255, 255)
        )
        hum_text = self.font_small.render(
            f"Humidity: {reading.humidity_pct:.1f} %", True, (255, 255, 255)
        )
        pres_text = self.font_small.render(
            f"Pressure: {reading.pressure_hpa:.1f} hPa", True, (255, 255, 255)
        )
        gas_text = self.font_small.render(
            f"Gas: {reading.gas_ohms:.0f} ohms", True, (255, 255, 255)
        )
        tvoc_text = self.font_small.render(
            f"TVOC est: {reading.tvoc_ppb} ppb", True, (255, 255, 255)
        )

        self.screen.blit(temp_text, (30, 230))
        self.screen.blit(hum_text, (30, 265))
        self.screen.blit(pres_text, (30, 300))

        if self.show_detail:
            self.screen.blit(gas_text, (30, 335))
            self.screen.blit(tvoc_text, (30, 370))

        info = self.font_small.render(
            "Tap left: details  |  Tap right: acknowledge 60 s",
            True,
            (255, 255, 255),
        )
        self.screen.blit(info, (20, 440))

        if alarm_active:
            banner = pygame.Rect(500, 20, 260, 90)
            pygame.draw.rect(self.screen, (255, 230, 230), banner, border_radius=12)
            msg1 = self.font_med.render("VENTILATE ROOM", True, (120, 0, 0))
            msg2 = self.font_small.render(
                f"Threshold {alarm_threshold} ppm exceeded", True, (120, 0, 0)
            )
            self.screen.blit(msg1, (520, 35))
            self.screen.blit(msg2, (520, 75))

        pygame.display.flip()

    def handle_event(self, event: pygame.event.Event) -> None:
        if event.type == pygame.MOUSEBUTTONDOWN:
            x, _y = event.pos
            if x < self.width // 2:
                self.show_detail = not self.show_detail
            else:
                self.alarm_ack_until = time.time() + 60.0


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Touchscreen CO2 air-quality alarm")
    parser.add_argument("--mock", action="store_true", help="Run without hardware")
    parser.add_argument(
        "--i2c-addr",
        default="0x76",
        help="BME680 I2C address such as 0x76 or 0x77",
    )
    parser.add_argument(
        "--alarm-ppm",
        type=int,
        default=1200,
        help="Estimated eCO2 alarm threshold",
    )
    parser.add_argument(
        "--interval",
        type=float,
        default=2.0,
        help="Seconds between sensor updates",
    )
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    i2c_addr = int(args.i2c_addr, 16)

    sensor = MockBME680Adapter() if args.mock else BME680Adapter(i2c_addr=i2c_addr)
    sensor.open()

    ui = HyperPixelUI()
    last_read = 0.0
    reading: Optional[SensorReading] = None
    running = True

    while running:
        for event in pygame.event.get():
            if event.type == pygame.QUIT:
                running = False
            ui.handle_event(event)

        now = time.time()
        if reading is None or (now - last_read) >= args.interval:
            reading = sensor.read()
            last_read = now

        if reading is not None:
            ui.draw(reading, args.alarm_ppm)

        time.sleep(0.05)

    pygame.quit()
    return 0


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

test_logic.py

#!/usr/bin/env python3

import air_quality_alarm as app


def run() -> None:
    assert app.clamp(5, 0, 10) == 5
    assert app.clamp(-1, 0, 10) == 0
    assert app.clamp(11, 0, 10) == 10

    iaq_good = app.estimate_iaq_score(40.0, 50000.0)
    iaq_bad = app.estimate_iaq_score(70.0, 10000.0)
    assert iaq_good <= iaq_bad

    eco2_good = app.estimate_eco2_ppm(iaq_good, 50000.0)
    eco2_bad = app.estimate_eco2_ppm(iaq_bad, 10000.0)
    assert eco2_good < eco2_bad

    assert app.air_quality_state(700)[0] == "GOOD"
    assert app.air_quality_state(900)[0] == "ELEVATED"
    assert app.air_quality_state(1300)[0] == "POOR"
    assert app.air_quality_state(1800)[0] == "ALARM"

    mock = app.MockBME680Adapter()
    mock.open()
    reading = mock.read()
    assert 0 <= reading.iaq_score <= 500
    assert 400 <= reading.eco2_ppm <= 2500
    assert reading.pressure_hpa > 800

    print("logic tests passed")


if __name__ == "__main__":
    run()

Validation steps

1. Validate syntax

cd ~/projects/touchscreen-co2-air-quality-alarm
. .venv/bin/activate
python3 -m py_compile air_quality_alarm.py test_logic.py

Expected evidence:

• No output from py_compile

2. Validate logic tests

cd ~/projects/touchscreen-co2-air-quality-alarm
. .venv/bin/activate
python3 test_logic.py

Expected evidence:

logic tests passed

3. Validate mock mode

cd ~/projects/touchscreen-co2-air-quality-alarm
. .venv/bin/activate
python3 air_quality_alarm.py --mock --alarm-ppm 1200 --interval 1.5

Expected evidence:

• A window opens at approximately 800 x 480
• Values update every few seconds
• The background color changes as the simulated room worsens
• Clicking the left half toggles details
• Clicking the right half acknowledges the alarm for 60 seconds
• After enough simulated time, the VENTILATE ROOM banner appears

4. Validate real sensor mode

If the sensor is at 0x76:

cd ~/projects/touchscreen-co2-air-quality-alarm
. .venv/bin/activate
python3 air_quality_alarm.py --alarm-ppm 1200 --interval 2.0

If the sensor is at 0x77:

cd ~/projects/touchscreen-co2-air-quality-alarm
. .venv/bin/activate
python3 air_quality_alarm.py --i2c-addr 0x77 --alarm-ppm 1200 --interval 2.0

Expected evidence:

• Temperature is plausible for the room
• Humidity is plausible for the room
• Pressure is in a normal atmospheric range
• Gas resistance is non-zero
• The touchscreen updates continuously

5. Practical room validation

To validate the tutorial’s main objective, test the device in a real room:

1. Put the device in a small enclosed room
2. Leave the door and windows closed for a period
3. Observe whether the estimated eCO2 trend rises over time
4. Open a door or window
5. Observe whether the displayed trend begins to improve over subsequent updates

Expected evidence:

• The status is easy to interpret on screen
• The visible alarm appears when the configured estimated threshold is exceeded
• Ventilation changes produce a visible trend change

This validates the prototype as a trend-based educational room reminder, not as a certified CO2 meter.