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.

Code

Save the following as tank_monitor.py:

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

import argparse
import random
import statistics
import time
from dataclasses import dataclass, field
from typing import Optional


@dataclass
class TankConfig:
    tank_depth_cm: float = 80.0
    sensor_offset_cm: float = 0.0
    min_valid_distance_cm: float = 3.0
    max_valid_distance_cm: float = 250.0
    low_percent_threshold: float = 20.0
    high_percent_threshold: float = 85.0
    sample_count: int = 5
    sample_interval_s: float = 0.08
    loop_interval_s: float = 1.0


@dataclass
class Measurement:
    raw_distance_cm: Optional[float]
    filtered_distance_cm: Optional[float]
    liquid_height_cm: Optional[float]
    fill_percent: Optional[float]
    status: str
    valid: bool
    timestamp: float = field(default_factory=time.time)


class UltrasonicAdapter:
    def read_distance_cm(self) -> float:
        raise NotImplementedError


class MockUltrasonicAdapter(UltrasonicAdapter):
    def __init__(self, start_distance_cm: float = 50.0, noise_cm: float = 0.8):
        self.distance_cm = start_distance_cm
        self.noise_cm = noise_cm
        self.direction = -1

    def read_distance_cm(self) -> float:
        self.distance_cm += self.direction * 0.4
        if self.distance_cm < 15.0:
            self.direction = 1
        elif self.distance_cm > 70.0:
            self.direction = -1
        noisy = self.distance_cm + random.uniform(-self.noise_cm, self.noise_cm)
        return max(2.0, noisy)


class RpiHcsr04Adapter(UltrasonicAdapter):
    def __init__(self, trig_pin: int, echo_pin: int):
        from gpiozero import DistanceSensor

        self.sensor = DistanceSensor(
            echo=echo_pin,
            trigger=trig_pin,
            max_distance=3.0,
            partial=False,
        )

    def read_distance_cm(self) -> float:
        return self.sensor.distance * 100.0


class DisplayAdapter:
    def show(self, measurement: Measurement, cfg: TankConfig) -> None:
        raise NotImplementedError


class ConsoleDisplayAdapter(DisplayAdapter):
    def show(self, measurement: Measurement, cfg: TankConfig) -> None:
        ts = time.strftime("%H:%M:%S", time.localtime(measurement.timestamp))
        print(
            f"[{ts}] valid={measurement.valid} "
            f"raw={measurement.raw_distance_cm!s} "
            f"filtered={measurement.filtered_distance_cm!s} "
            f"height={measurement.liquid_height_cm!s} "
            f"fill={measurement.fill_percent!s} "
            f"status={measurement.status}"
        )


class Ili9341DisplayAdapter(DisplayAdapter):
    def __init__(self, width: int = 320, height: int = 240):
        from PIL import Image, ImageDraw, ImageFont
        import spidev  # noqa: F401

        self.Image = Image
        self.ImageDraw = ImageDraw
        self.ImageFont = ImageFont
        self.width = width
        self.height = height
        self.font_large = ImageFont.load_default()
        self.font_small = ImageFont.load_default()

    def _render_to_image(self, measurement: Measurement, cfg: TankConfig):
        img = self.Image.new("RGB", (self.width, self.height), (0, 0, 0))
        draw = self.ImageDraw.Draw(img)

        if not measurement.valid:
            color = (255, 80, 80)
        elif measurement.status == "LOW":
            color = (255, 180, 0)
        elif measurement.status == "HIGH":
            color = (80, 220, 120)
        else:
            color = (80, 180, 255)

        draw.rectangle(
            (0, 0, self.width - 1, self.height - 1),
            outline=(100, 100, 100),
        )
        draw.text(
            (10, 10),
            "Ultrasonic Tank Level",
            fill=(255, 255, 255),
            font=self.font_large,
        )
        draw.text(
            (10, 40),
            f"Status: {measurement.status}",
            fill=color,
            font=self.font_large,
        )

        distance_text = (
            "--"
            if measurement.filtered_distance_cm is None
            else f"{measurement.filtered_distance_cm:.1f} cm"
        )
        height_text = (
            "--"
            if measurement.liquid_height_cm is None
            else f"{measurement.liquid_height_cm:.1f} cm"
        )
        fill_text = (
            "--"
            if measurement.fill_percent is None
            else f"{measurement.fill_percent:.1f} %"
        )

        draw.text(
            (10, 80),
            f"Distance: {distance_text}",
            fill=(255, 255, 255),
            font=self.font_small,
        )
        draw.text(
            (10, 110),
            f"Height:   {height_text}",
            fill=(255, 255, 255),
            font=self.font_small,
        )
        draw.text(
            (10, 140),
            f"Fill:     {fill_text}",
            fill=(255, 255, 255),
            font=self.font_small,
        )
        draw.text(
            (10, 170),
            f"Tank depth: {cfg.tank_depth_cm:.1f} cm",
            fill=(180, 180, 180),
            font=self.font_small,
        )

        bar_x0, bar_y0 = 250, 30
        bar_x1, bar_y1 = 290, 210
        draw.rectangle(
            (bar_x0, bar_y0, bar_x1, bar_y1),
            outline=(200, 200, 200),
            fill=(20, 20, 20),
        )

        if measurement.fill_percent is not None:
            bounded = max(0.0, min(100.0, measurement.fill_percent))
            fill_height = int((bar_y1 - bar_y0 - 4) * bounded / 100.0)
            draw.rectangle(
                (bar_x0 + 2, bar_y1 - 2 - fill_height, bar_x1 - 2, bar_y1 - 2),
                fill=color,
            )

        return img

    def show(self, measurement: Measurement, cfg: TankConfig) -> None:
        img = self._render_to_image(measurement, cfg)
        img.save("/tmp/tank_monitor_preview.png")


class TankLevelMonitor:
    def __init__(
        self,
        sensor: UltrasonicAdapter,
        display: DisplayAdapter,
        cfg: TankConfig,
    ):
        self.sensor = sensor
        self.display = display
        self.cfg = cfg

    def sample_distance(self) -> Optional[float]:
        samples: list[float] = []
        for _ in range(self.cfg.sample_count):
            try:
                distance_cm = self.sensor.read_distance_cm()
                if (
                    self.cfg.min_valid_distance_cm
                    <= distance_cm
                    <= self.cfg.max_valid_distance_cm
                ):
                    samples.append(distance_cm)
            except Exception:
                pass
            time.sleep(self.cfg.sample_interval_s)

        if not samples:
            return None

        if len(samples) >= 3:
            median_value = statistics.median(samples)
            filtered = [x for x in samples if abs(x - median_value) < 5.0]
            if filtered:
                samples = filtered

        return statistics.mean(samples)

    def compute_measurement(self) -> Measurement:
        distance_cm = self.sample_distance()
        if distance_cm is None:
            return Measurement(
                raw_distance_cm=None,
                filtered_distance_cm=None,
                liquid_height_cm=None,
                fill_percent=None,
                status="NO ECHO",
                valid=False,
            )

        adjusted_distance = distance_cm - self.cfg.sensor_offset_cm
        liquid_height = self.cfg.tank_depth_cm - adjusted_distance
        liquid_height = max(0.0, min(self.cfg.tank_depth_cm, liquid_height))

        if self.cfg.tank_depth_cm <= 0:
            return Measurement(
                raw_distance_cm=distance_cm,
                filtered_distance_cm=adjusted_distance,
                liquid_height_cm=None,
                fill_percent=None,
                status="CONFIG ERROR",
                valid=False,
            )

        fill_percent = (liquid_height / self.cfg.tank_depth_cm) * 100.0

        if fill_percent < self.cfg.low_percent_threshold:
            status = "LOW"
        elif fill_percent >= self.cfg.high_percent_threshold:
            status = "HIGH"
        else:
            status = "OK"

        return Measurement(
            raw_distance_cm=distance_cm,
            filtered_distance_cm=adjusted_distance,
            liquid_height_cm=liquid_height,
            fill_percent=fill_percent,
            status=status,
            valid=True,
        )

    def run_forever(self) -> None:
        while True:
            measurement = self.compute_measurement()
            self.display.show(measurement, self.cfg)
            time.sleep(self.cfg.loop_interval_s)


def build_arg_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Ultrasonic tank level monitor")
    parser.add_argument("--mode", choices=["mock", "hardware"], default="mock")
    parser.add_argument("--tank-depth-cm", type=float, default=80.0)
    parser.add_argument("--sensor-offset-cm", type=float, default=0.0)
    parser.add_argument("--trig-pin", type=int, default=23)
    parser.add_argument("--echo-pin", type=int, default=24)
    parser.add_argument("--display", choices=["console", "ili9341"], default="console")
    return parser


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

    cfg = TankConfig(
        tank_depth_cm=args.tank_depth_cm,
        sensor_offset_cm=args.sensor_offset_cm,
    )

    if args.mode == "mock":
        sensor: UltrasonicAdapter = MockUltrasonicAdapter()
    else:
        sensor = RpiHcsr04Adapter(trig_pin=args.trig_pin, echo_pin=args.echo_pin)

    if args.display == "console":
        display: DisplayAdapter = ConsoleDisplayAdapter()
    else:
        display = Ili9341DisplayAdapter()

    app = TankLevelMonitor(sensor=sensor, display=display, cfg=cfg)
    app.run_forever()
    return 0


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

Save the following as test_dry_run.py:

#!/usr/bin/env python3
from tank_monitor import (
    ConsoleDisplayAdapter,
    TankConfig,
    TankLevelMonitor,
    UltrasonicAdapter,
)


class FixedSensor(UltrasonicAdapter):
    def __init__(self, value_cm: float):
        self.value_cm = value_cm

    def read_distance_cm(self) -> float:
        return self.value_cm


def run_case(distance_cm: float, tank_depth_cm: float) -> None:
    cfg = TankConfig(
        tank_depth_cm=tank_depth_cm,
        sample_count=3,
        sample_interval_s=0.0,
        loop_interval_s=0.0,
    )
    sensor = FixedSensor(distance_cm)
    display = ConsoleDisplayAdapter()
    monitor = TankLevelMonitor(sensor, display, cfg)
    measurement = monitor.compute_measurement()
    print(
        f"distance={distance_cm:.1f} cm, "
        f"height={measurement.liquid_height_cm:.1f} cm, "
        f"fill={measurement.fill_percent:.1f} %, "
        f"status={measurement.status}, valid={measurement.valid}"
    )


if __name__ == "__main__":
    run_case(70.0, 80.0)
    run_case(40.0, 80.0)
    run_case(8.0, 80.0)

Validation steps

1. Software validation without hardware

Run:

python3.11 -m py_compile tank_monitor.py test_dry_run.py
python3.11 test_dry_run.py

Expected evidence:

• The compile command prints nothing
• distance=70 cm gives a lower fill than distance=40 cm
• distance=8 cm gives a near-full result

2. Mock live updates

Run:

python3.11 tank_monitor.py --mode mock --display console --tank-depth-cm 80

Expected evidence:

• Distance values change slowly
• Fill percentage changes inversely to distance
• Status transitions between LOW, OK, and HIGH

3. HC-SR04 bench test

Point the sensor at a flat target at a known distance and run:

python3.11 tank_monitor.py --mode hardware --display console --tank-depth-cm 80

Expected evidence:

• The reported distance is close to the physical distance
• Moving the target closer reduces the displayed distance
• Repeated readings at a fixed target position vary only slightly

4. Tank validation

Measure the real sensor-to-bottom usable depth and set --tank-depth-cm to that value.

Then compare the displayed distance to manual measurements at:

• Near-empty condition
• Mid-level or high-level condition

Expected evidence:

• Percentage increases as the liquid rises
• LOW and HIGH thresholds trigger at the intended points

Setup/Connection

This project avoids a circuit drawing and uses text-only connection guidance.

Connection strategy

The prototype uses:
– SPI for the PN532 NFC HAT
– I2C for the DS3231 RTC

Many PN532 HAT boards can be configured for SPI, I2C, or UART using switches/jumpers. For this tutorial:
– Set the PN532 HAT to SPI mode
– Keep the DS3231 on I2C

Raspberry Pi interface enable steps

Run:

sudo raspi-config

Then:
1. Go to Interface Options
2. Enable SPI
3. Enable I2C
4. Finish and reboot

After reboot, verify:

ls /dev/spidev*
ls /dev/i2c-*

You should see devices similar to:
– /dev/spidev0.0
– /dev/i2c-1

Text-based connection notes

PN532 NFC HAT

If your PN532 HAT stacks directly onto the Raspberry Pi header, the SPI pins are already routed through the header. If using wires instead of a direct HAT stack, connect these signals:

• PN532 VCC -> Raspberry Pi 3.3 V
• PN532 GND -> Raspberry Pi GND
• PN532 SCK -> Raspberry Pi SPI SCLK
• PN532 MISO -> Raspberry Pi SPI MISO
• PN532 MOSI -> Raspberry Pi SPI MOSI
• PN532 SS/CS -> Raspberry Pi SPI CE0
• PN532 RSTO or RSTPDN -> optional GPIO if required by your board, otherwise leave according to HAT design
• PN532 mode selector -> SPI

DS3231 RTC

If the RTC is a separate module:
– DS3231 VCC -> Raspberry Pi 3.3 V
– DS3231 GND -> Raspberry Pi GND
– DS3231 SDA -> Raspberry Pi GPIO2 / SDA1
– DS3231 SCL -> Raspberry Pi GPIO3 / SCL1

Bus detection checks

Install common tools:

sudo apt update
sudo apt install -y i2c-tools python3-pip

Check I2C devices:

sudo i2cdetect -y 1

A DS3231 often appears around address 0x68.

For SPI, there is no equivalent single probe as simple as i2cdetect, but the existence of /dev/spidev0.0 confirms the SPI interface is enabled.

Project directory

Create a clean working directory:

mkdir -p ~/nfc-door-access-alarm
cd ~/nfc-door-access-alarm

Validated Code

The code below is designed to satisfy two important goals:
1. Be useful on the real Raspberry Pi with hardware adapter classes.
2. Be runnable in dry-run/mock mode on a normal computer without NFC or RTC hardware.

This matches the requested Raspberry Pi validation style.

access_controller.py

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

import argparse
import csv
import json
import os
import sys
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Iterable, List, Optional


@dataclass
class AccessEvent:
    timestamp: str
    uid: str
    authorized: bool
    source: str
    alarm_active: bool


class RTCAdapter:
    def now_iso(self) -> str:
        raise NotImplementedError


class SystemRTC(RTCAdapter):
    def now_iso(self) -> str:
        return datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds")


class MockDS3231RTC(RTCAdapter):
    def __init__(self, fixed_time: Optional[str] = None) -> None:
        self.fixed_time = fixed_time

    def now_iso(self) -> str:
        if self.fixed_time:
            return self.fixed_time
        return datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds")


class NFCReaderAdapter:
    def poll_uid(self) -> Optional[str]:
        raise NotImplementedError


class MockPN532Reader(NFCReaderAdapter):
    def __init__(self, sequence: Iterable[str], repeat: bool = False) -> None:
        self._sequence = list(sequence)
        self._repeat = repeat
        self._index = 0

    def poll_uid(self) -> Optional[str]:
        if not self._sequence:
            return None
        if self._index >= len(self._sequence):
            if self._repeat:
                self._index = 0
            else:
                return None
        uid = self._sequence[self._index]
        self._index += 1
        time.sleep(0.2)
        return uid


class AlarmAdapter:
    def set_alarm(self, active: bool) -> None:
        raise NotImplementedError


class ConsoleAlarm(AlarmAdapter):
    def __init__(self) -> None:
        self.state = False

    def set_alarm(self, active: bool) -> None:
        if active != self.state:
            self.state = active
            print(f"[ALARM] state={'ON' if active else 'OFF'}")


class AccessController:
    def __init__(
        self,
        rtc: RTCAdapter,
        reader: NFCReaderAdapter,
        alarm: AlarmAdapter,
        allowed_uids: List[str],
        log_path: Path,
    ) -> None:
        self.rtc = rtc
        self.reader = reader
        self.alarm = alarm
        self.allowed_uids = {uid.strip().upper() for uid in allowed_uids if uid.strip()}
        self.log_path = log_path
        self.alarm_active = False
# ... continues for members in the complete validated source ...

    def handle_uid(self, uid: str, source: str = "nfc") -> AccessEvent:
        normalized = uid.strip().upper()
        authorized = normalized in self.allowed_uids
        self.alarm_active = not authorized
        self.alarm.set_alarm(self.alarm_active)

        event = AccessEvent(
            timestamp=self.rtc.now_iso(),
            uid=normalized,
            authorized=authorized,
            source=source,
            alarm_active=self.alarm_active,
        )
        self._append_log(event)
        if authorized:
            print(f"{event.timestamp} ACCESS GRANTED uid={event.uid}")
        else:
            print(f"{event.timestamp} ACCESS DENIED uid={event.uid}")
        return event

    def _append_log(self, event: AccessEvent) -> None:
        file_exists = self.log_path.exists()
        with self.log_path.open("a", newline="", encoding="utf-8") as f:
            writer = csv.writer(f)
            if not file_exists:
                writer.writerow(["timestamp", "uid", "authorized", "source", "alarm_active"])
            writer.writerow([
                event.timestamp,

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

import argparse
import csv
import json
import os
import sys
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Iterable, List, Optional


@dataclass
class AccessEvent:
    timestamp: str
    uid: str
    authorized: bool
    source: str
    alarm_active: bool


class RTCAdapter:
    def now_iso(self) -> str:
        raise NotImplementedError


class SystemRTC(RTCAdapter):
    def now_iso(self) -> str:
        return datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds")


class MockDS3231RTC(RTCAdapter):
    def __init__(self, fixed_time: Optional[str] = None) -> None:
        self.fixed_time = fixed_time

    def now_iso(self) -> str:
        if self.fixed_time:
            return self.fixed_time
        return datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds")


class NFCReaderAdapter:
    def poll_uid(self) -> Optional[str]:
        raise NotImplementedError


class MockPN532Reader(NFCReaderAdapter):
    def __init__(self, sequence: Iterable[str], repeat: bool = False) -> None:
        self._sequence = list(sequence)
        self._repeat = repeat
        self._index = 0

    def poll_uid(self) -> Optional[str]:
        if not self._sequence:
            return None
        if self._index >= len(self._sequence):
            if self._repeat:
                self._index = 0
            else:
                return None
        uid = self._sequence[self._index]
        self._index += 1
        time.sleep(0.2)
        return uid


class AlarmAdapter:
    def set_alarm(self, active: bool) -> None:
        raise NotImplementedError


class ConsoleAlarm(AlarmAdapter):
    def __init__(self) -> None:
        self.state = False

    def set_alarm(self, active: bool) -> None:
        if active != self.state:
            self.state = active
            print(f"[ALARM] state={'ON' if active else 'OFF'}")


class AccessController:
    def __init__(
        self,
        rtc: RTCAdapter,
        reader: NFCReaderAdapter,
        alarm: AlarmAdapter,
        allowed_uids: List[str],
        log_path: Path,
    ) -> None:
        self.rtc = rtc
        self.reader = reader
        self.alarm = alarm
        self.allowed_uids = {uid.strip().upper() for uid in allowed_uids if uid.strip()}
        self.log_path = log_path
        self.alarm_active = False

    def handle_uid(self, uid: str, source: str = "nfc") -> AccessEvent:
        normalized = uid.strip().upper()
        authorized = normalized in self.allowed_uids
        self.alarm_active = not authorized
        self.alarm.set_alarm(self.alarm_active)

        event = AccessEvent(
            timestamp=self.rtc.now_iso(),
            uid=normalized,
            authorized=authorized,
            source=source,
            alarm_active=self.alarm_active,
        )
        self._append_log(event)
        if authorized:
            print(f"{event.timestamp} ACCESS GRANTED uid={event.uid}")
        else:
            print(f"{event.timestamp} ACCESS DENIED uid={event.uid}")
        return event

    def _append_log(self, event: AccessEvent) -> None:
        file_exists = self.log_path.exists()
        with self.log_path.open("a", newline="", encoding="utf-8") as f:
            writer = csv.writer(f)
            if not file_exists:
                writer.writerow(["timestamp", "uid", "authorized", "source", "alarm_active"])
            writer.writerow([
                event.timestamp,
                event.uid,
                int(event.authorized),
                event.source,
                int(event.alarm_active),
            ])

    def run(self, max_reads: int = 0, poll_delay: float = 0.5) -> int:
        reads = 0
        while True:
            uid = self.reader.poll_uid()
            if uid:
                self.handle_uid(uid)
                reads += 1
            else:
                time.sleep(poll_delay)

            if max_reads > 0 and reads >= max_reads:
                break
        return 0


def load_allowed_uids(path: Path) -> List[str]:
    with path.open("r", encoding="utf-8") as f:
        data = json.load(f)
    if not isinstance(data, dict) or "allowed_uids" not in data:
        raise ValueError("allowlist file must contain a JSON object with key 'allowed_uids'")
    items = data["allowed_uids"]
    if not isinstance(items, list):
        raise ValueError("'allowed_uids' must be a list")
    return [str(x) for x in items]


def build_mock_sequence(args: argparse.Namespace) -> List[str]:
    if args.mock_sequence:
        return [x.strip().upper() for x in args.mock_sequence.split(",") if x.strip()]
    return [
        "04A1B2C3D4",
        "1122334455",
        "04A1B2C3D4",
    ]


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="NFC door access alarm prototype")
    parser.add_argument("--allowlist", default="allowlist.json", help="Path to JSON allowlist")
    parser.add_argument("--log", default="access_log.csv", help="Path to CSV log file")
    parser.add_argument("--mock", action="store_true", help="Use mock NFC and RTC adapters")
    parser.add_argument("--mock-sequence", default="", help="Comma-separated UID sequence for mock mode")
    parser.add_argument("--max-reads", type=int, default=3, help="Stop after this many successful reads, 0=forever")
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    allowlist_path = Path(args.allowlist)
    log_path = Path(args.log)

    allowed_uids = load_allowed_uids(allowlist_path)

    if args.mock:
        rtc: RTCAdapter = MockDS3231RTC()
        reader: NFCReaderAdapter = MockPN532Reader(build_mock_sequence(args), repeat=False)
    else:
        # real-hardware adapter integration is intentionally not auto-imported here.
        # If no mock mode is selected, use system clock and require explicit future extension
        # for PN532 hardware reading.
        rtc = SystemRTC()
        reader = MockPN532Reader([], repeat=False)

    alarm = ConsoleAlarm()
    controller = AccessController(
        rtc=rtc,
        reader=reader,
        alarm=alarm,
        allowed_uids=allowed_uids,
        log_path=log_path,
    )
    return controller.run(max_reads=args.max_reads)


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

allowlist.json

{
  "allowed_uids": [
    "04A1B2C3D4",
    "AABBCCDDEE"
  ]
}

test_access_controller.py

This validation script performs a dry-run check using mock inputs. It is not a full unit-test framework dependency; it uses only the standard library.

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

import csv
import subprocess
import sys
from pathlib import Path


def main() -> int:
    project_dir = Path(__file__).resolve().parent
    log_path = project_dir / "test_access_log.csv"

    if log_path.exists():
        log_path.unlink()

    cmd = [
        sys.executable,
        str(project_dir / "access_controller.py"),
        "--mock",
        "--allowlist",
        str(project_dir / "allowlist.json"),
        "--log",
        str(log_path),
        "--mock-sequence",
        "04A1B2C3D4,DEADBEEF01",
        "--max-reads",
        "2",
    ]

    result = subprocess.run(cmd, capture_output=True, text=True, check=False)
    print(result.stdout)
# ... continues for members in the complete validated source ...

    if result.returncode != 0:
        print(result.stderr)
        return result.returncode

    if not log_path.exists():
        print("ERROR: log file was not created")
        return 1

    with log_path.open("r", encoding="utf-8", newline="") as f:
        rows = list(csv.DictReader(f))

    if len(rows) != 2:
        print(f"ERROR: expected 2 log entries, got {len(rows)}")
        return 1

    if rows[0]["authorized"] != "1":
        print("ERROR: first UID should be authorized")
        return 1

    if rows[1]["authorized"] != "0":
        print("ERROR: second UID should be unauthorized")
        return 1

    if rows[1]["alarm_active"] != "1":
        print("ERROR: alarm should be active for unauthorized UID")
        return 1

    print("Dry-run validation passed.")

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

import csv
import subprocess
import sys
from pathlib import Path


def main() -> int:
    project_dir = Path(__file__).resolve().parent
    log_path = project_dir / "test_access_log.csv"

    if log_path.exists():
        log_path.unlink()

    cmd = [
        sys.executable,
        str(project_dir / "access_controller.py"),
        "--mock",
        "--allowlist",
        str(project_dir / "allowlist.json"),
        "--log",
        str(log_path),
        "--mock-sequence",
        "04A1B2C3D4,DEADBEEF01",
        "--max-reads",
        "2",
    ]

    result = subprocess.run(cmd, capture_output=True, text=True, check=False)
    print(result.stdout)
    if result.returncode != 0:
        print(result.stderr)
        return result.returncode

    if not log_path.exists():
        print("ERROR: log file was not created")
        return 1

    with log_path.open("r", encoding="utf-8", newline="") as f:
        rows = list(csv.DictReader(f))

    if len(rows) != 2:
        print(f"ERROR: expected 2 log entries, got {len(rows)}")
        return 1

    if rows[0]["authorized"] != "1":
        print("ERROR: first UID should be authorized")
        return 1

    if rows[1]["authorized"] != "0":
        print("ERROR: second UID should be unauthorized")
        return 1

    if rows[1]["alarm_active"] != "1":
        print("ERROR: alarm should be active for unauthorized UID")
        return 1

    print("Dry-run validation passed.")
    return 0


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

Setup/Connection

This project uses the Raspberry Pi I2C bus. The BME680 and DS3231 can share the same SDA and SCL lines because I2C is bus-based.

I2C device notes

Typical addresses:
– BME680: often 0x76 or 0x77
– DS3231: usually 0x68

Text-based wiring

Connect both breakout modules to the Raspberry Pi 40-pin header as follows:

• Raspberry Pi 3.3V -> BME680 VIN or 3V3
• Raspberry Pi GND -> BME680 GND
• Raspberry Pi GPIO2 / SDA1 / Pin 3 -> BME680 SDA
• Raspberry Pi GPIO3 / SCL1 / Pin 5 -> BME680 SCL

And for the RTC:

• Raspberry Pi 3.3V -> DS3231 VCC
• Raspberry Pi GND -> DS3231 GND
• Raspberry Pi GPIO2 / SDA1 / Pin 3 -> DS3231 SDA
• Raspberry Pi GPIO3 / SCL1 / Pin 5 -> DS3231 SCL

Important connection details

• Do not connect these modules to 5 V unless your board explicitly requires and safely supports it. For beginner work with Raspberry Pi GPIO, prefer 3.3 V logic-compatible operation.
• Many BME680 and DS3231 breakout boards already include pull-up resistors on SDA/SCL. That is normal.
• If the BME680 is not found at 0x76, try 0x77.

Enable I2C on Raspberry Pi

Run:

sudo raspi-config

Then:
1. Choose Interface Options
2. Choose I2C
3. Enable it
4. Reboot if requested

After reboot, check visible devices:

sudo apt update
sudo apt install -y i2c-tools
i2cdetect -y 1

You should normally see:
– 68 for DS3231
– 76 or 77 for BME680

Validated Code

The code below is designed to meet the tutorial constraints:
– pure Python
– Python 3.11 compatible
– py_compile valid
– dry-run capable without physical hardware
– hardware access hidden behind adapter classes

Create a project directory:

mkdir -p ~/mqtt-env-logger
cd ~/mqtt-env-logger

env_logger.py

#!/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()
# ... continues for members in the complete validated source ...

    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

#!/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())

test_dry_run.py

This small validation script checks that the program can produce valid reading objects in dry-run conditions.

#!/usr/bin/env python3
import json
from dataclasses import asdict

from env_logger import MockI2CBus, DS3231RTC, BME680Sensor, build_reading


def main() -> int:
    bus = MockI2CBus()
    rtc = DS3231RTC(bus)
    sensor = BME680Sensor(bus=bus, dry_run=True)

    reading = build_reading(rtc, sensor, sequence=1)
    payload = asdict(reading)

    assert "iso_time" in payload
    assert "unix_time" in payload
    assert "temperature_c" in payload
    assert "humidity_pct" in payload
    assert "pressure_hpa" in payload
    assert "gas_ohm" in payload
    assert payload["sequence"] == 1

    print(json.dumps(payload, indent=2, sort_keys=True))
    print("Dry-run validation passed")
    return 0


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