Practical case: level monitor with Raspberry Pi 5

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