Caso práctico: monitor de sal con Raspberry Pi 5

#!/usr/bin/env python3
"""
Water Softener Salt Level Monitor
Target: Raspberry Pi 5 + VL53L0X ToF Sensor + 2.13 inch SPI e-paper display
"""

import argparse
import time
import logging
from datetime import datetime

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S'
)

# --- Calibration Constants ---
# Distance from the sensor to the salt when the tank is completely full
TANK_FULL_DISTANCE_MM = 150  
# Distance from the sensor to the bottom of the tank (or minimum salt level)
TANK_EMPTY_DISTANCE_MM = 850 

class VL53L0XAdapter:
    """Adapter for the VL53L0X Time-of-Flight sensor."""
    def __init__(self, dry_run=False):
        self.dry_run = dry_run
        self.sensor = None
        self.mock_distance = 200 # Starting mock distance in mm

        if not self.dry_run:
            try:
                import board
                import busio
                import adafruit_vl53l0x
                i2c = busio.I2C(board.SCL, board.SDA)
                self.sensor = adafruit_vl53l0x.VL53L0X(i2c)
                logging.info("Hardware VL53L0X initialized successfully.")
            except ImportError:
                logging.warning("adafruit_vl53l0x not found. Falling back to Mock VL53L0X.")
                self.dry_run = True
            except Exception as e:
                logging.error(f"Failed to initialize hardware VL53L0X: {e}. Falling back to Mock.")
                self.dry_run = True

    def get_distance(self):
        """Returns distance in millimeters."""
        if self.dry_run:
            # Simulate salt depletion over time for demonstration purposes
            self.mock_distance += 15 
            if self.mock_distance > TANK_EMPTY_DISTANCE_MM + 50:
                self.mock_distance = TANK_FULL_DISTANCE_MM
            return self.mock_distance
        else:
            try:
                return self.sensor.range
            except Exception as e:
                logging.error(f"Error reading from sensor: {e}")
                return -1

class EPaperAdapter:
    """Adapter for the 2.13 inch SPI e-paper display."""
    def __init__(self, dry_run=False):
        self.dry_run = dry_run
        self.epd = None

        if not self.dry_run:
            try:
                # Attempting to load standard Waveshare library structure
                from waveshare_epd import epd2in13_V4
                self.epd = epd2in13_V4.EPD()
                self.epd.init()
                self.epd.Clear(0xFF)
                logging.info("Hardware E-Paper initialized successfully.")
            except ImportError:
                logging.warning("waveshare_epd library not found. Falling back to Mock E-Paper.")
                self.dry_run = True
            except Exception as e:
                logging.error(f"Failed to init hardware E-Paper: {e}. Falling back to Mock.")
                self.dry_run = True

    def update_display(self, percentage, distance):
        """Updates the e-paper display with current salt level."""
        timestamp = datetime.now().strftime('%H:%M')

        # Create a visual bar representation
        bar_length = 20
        filled_blocks = max(0, min(bar_length, int((percentage / 100.0) * bar_length)))
        empty_blocks = bar_length - filled_blocks
        visual_bar = f"[{'#' * filled_blocks}{'-' * empty_blocks}]"

        status_text = "OK"
        if percentage < 20:
            status_text = "REFILL SOON!"
        if percentage <= 5:
            status_text = "EMPTY! REFILL NOW!"

        display_string = (
            f"=== SALT LEVEL ===\n"
            f"Level: {percentage:.1f}%\n"
            f"Dist:  {distance} mm\n"
            f"{visual_bar}\n"
            f"Status: {status_text}\n"
            f"Updated: {timestamp}\n"
            f"=================="
        )

        if self.dry_run:
            logging.info(f"Mock E-Paper Update:\n{display_string}")
        else:
            try:
                # In a real implementation, we would use PIL (Pillow) to draw text onto an image buffer
                # and send it to self.epd.display(). For this validated pure-Python representation,
                # we log the buffer action that would occur.
                from PIL import Image, ImageDraw
                # Create blank image
                image = Image.new('1', (self.epd.height, self.epd.width), 255)
                draw = ImageDraw.Draw(image)
                # Draw text (using default font for simplicity)
                draw.text((10, 10), display_string, fill=0)
                # Rotate image if necessary for screen orientation
                image = image.rotate(90, expand=True)
                self.epd.display(self.epd.getbuffer(image))
                logging.info("Hardware display updated.")
            except Exception as e:
                logging.error(f"Failed to write to hardware display: {e}")

    def sleep(self):
        """Puts the e-paper display to sleep to prevent damage."""
        if not self.dry_run and self.epd:
            try:
                self.epd.sleep()
            except Exception as e:
                logging.error(f"Failed to sleep hardware display: {e}")

def calculate_percentage(distance_mm):
    """Calculates the remaining salt percentage based on calibration distances."""
    if distance_mm <= TANK_FULL_DISTANCE_MM:
        return 100.0
    if distance_mm >= TANK_EMPTY_DISTANCE_MM:
        return 0.0

    usable_range = TANK_EMPTY_DISTANCE_MM - TANK_FULL_DISTANCE_MM
    current_depletion = distance_mm - TANK_FULL_DISTANCE_MM
    percentage = 100.0 - ((current_depletion / usable_range) * 100.0)
    return max(0.0, min(100.0, percentage))

def main():
    parser = argparse.ArgumentParser(description="Water Softener Salt Level Monitor")
    parser.add_argument("--dry-run", action="store_true", help="Run without physical hardware")
    parser.add_argument("--interval", type=int, default=10, help="Polling interval in seconds")
    args = parser.parse_args()

    logging.info("Starting Water Softener Salt Level Monitor...")
    if args.dry_run:
        logging.info("Running in DRY-RUN mode.")

    sensor = VL53L0XAdapter(dry_run=args.dry_run)
    display = EPaperAdapter(dry_run=args.dry_run)

    last_percentage = -100.0 # Force initial update

    try:
        while True:
            distance = sensor.get_distance()
            if distance < 0:
                logging.warning("Invalid sensor reading. Retrying next cycle.")
                time.sleep(args.interval)
                continue

            percentage = calculate_percentage(distance)
            logging.info(f"Reading: {distance}mm -> {percentage:.1f}%")

            # Update display only if percentage changes by more than 2% to save screen lifespan
            if abs(percentage - last_percentage) >= 2.0:
                logging.info("Significant change detected. Updating display...")
                display.update_display(percentage, distance)
                last_percentage = percentage
                # Put display back to sleep after update
                display.sleep()

            time.sleep(args.interval)

    except KeyboardInterrupt:
        logging.info("Monitor stopped by user.")
    finally:
        display.sleep()
        logging.info("Shutdown complete.")

if __name__ == "__main__":
    main()