Improved readability

This reverts commit 9715339c43.

requirements.txt

@@ -5,3 +5,4 @@ uvicorn
 fastapi-cors
 Adafruit_DHT
 requests
+mh_z19

src/actions.py

@@ -1,54 +1,38 @@
 import asyncio
 from datetime import datetime
-import logging
 import os
 from config import climate_log_file, dht22_pin
 
-from handler.matrix.matrix_display import MatrixDisplayAdapter
-from handler.climate.climate import ClimateAdapter
-
-climate_sensor: ClimateAdapter = None
-matrix_display: MatrixDisplayAdapter = None
-
-try:
-    from handler.matrix.luma_matrix_display import LumaMatrixDisplay
-
-    matrix_display = LumaMatrixDisplay()
-except:
-    from handler.matrix.dummy_matrix_display import DummyMatrixDisplay
-
-    logging.info("Using Dummy Matrix Display")
-    matrix_display = DummyMatrixDisplay()
-
-try:
-    from handler.climate.dht22_climate import Dht22Climate
-
-    climate_sensor = Dht22Climate(dht22_pin)
-except:
-    from handler.climate.dummy_climate import DummyClimate
-
-    logging.info("Using Dummy Climate")
-    climate_sensor = DummyClimate()
+from handler.dht22_climate import Dht22Climate
+from handler.matrix_display import MatrixDisplay
+from handler.mhz19_co2 import Mhz19Co2
 
 
-async def log_temperature():
+climate_sensor = Dht22Climate(dht22_pin)
+co2_sensor = Mhz19Co2()
+matrix_display = MatrixDisplay()
+
+
+async def log_climate(delay_sec: int = 60):
     # If file does not exist, create it and write header
     if not os.path.isfile(climate_log_file):
         with open(climate_log_file, "w") as f:
-            f.write("timestamp,temperature,humidity\n")
+            f.write("timestamp,temperature,humidity,co2\n")
 
     while True:
         measurements = climate_sensor.read()
+        co2_density = co2_sensor.read()
         if measurements is not None:
             with open(climate_log_file, "a") as f:
                 f.write(
-                    "{},{},{}\n".format(
+                    "{},{},{},{}\n".format(
                         datetime.now().isoformat(),
                         measurements["temperature"],
                         measurements["humidity"],
+                        co2_density,
                     )
                 )
-        await asyncio.sleep(60)
+        await asyncio.sleep(delay_sec)
 
 
 async def display_time():
@@ -57,3 +41,8 @@ async def display_time():
 
         seconds_until_next_minute = 60 - datetime.now().second
         await asyncio.sleep(seconds_until_next_minute)
+
+
+async def display_pattern(*args, **kwargs):
+    while True:
+        await matrix_display.pattern(*args, **kwargs)

src/handler/climate/climate.py

@@ -1,11 +0,0 @@
-class ClimateAdapter:
-    def __init__(self):
-        self.last_read = None
-
-    def read(self):
-        raise NotImplemented()
-
-    def get_last_read(self):
-        if self.last_read is None:
-            return self.read()
-        return self.last_read

src/handler/climate/dummy_climate.py

@@ -1,7 +0,0 @@
-from handler.climate.climate import ClimateAdapter
-
-
-class DummyClimate(ClimateAdapter):
-    def read(self):
-        self.last_read = {"temperature": 22.3, "humidity": 50.1}
-        return self.last_read

src/handler/dht22_climate.py

@@ -1,14 +1,19 @@
 import Adafruit_DHT
-from handler.climate.climate import ClimateAdapter
 
 
-class Dht22Climate(ClimateAdapter):
-    def __init__(self, pin):
-        super().__init__()
+class Dht22Climate:
+
+    def __init__(self, pin: int):
         self.sensor = Adafruit_DHT.AM2302
         self.pin = pin
+        self.last_read = None
 
-    def read(self):
+    def get_last_read(self) -> dict | None:
+        if self.last_read is None:
+            return self.read()
+        return self.last_read
+
+    def read(self) -> dict | None:
         humidity, temperature = Adafruit_DHT.read_retry(self.sensor, self.pin)
         if humidity is not None and temperature is not None:
             self.last_read = {"temperature": temperature, "humidity": humidity}

src/handler/matrix/dummy_matrix_display.py

@@ -1,40 +0,0 @@
-from handler.matrix.matrix_display import MatrixDisplayAdapter
-import logging
-
-
-class DummyMatrixDisplay(MatrixDisplayAdapter):
-    def __init__(self, *, contrast=0, text_speed=0.02) -> None:
-        self.contrast = contrast
-        self.text_speed = text_speed
-        logging.info("DummyMatrixDisplay: Initialized")
-
-    def set_contrast(self, contrast: int):
-        """Set contrast for all actions.
-
-        Args:
-            contrast (int): [0, 255]
-        """
-        self.contrast = contrast
-        logging.info(f"DummyMatrixDisplay: setting contrast to {contrast}")
-
-    def show_text(self, text):
-        logging.info(f"DummyMatrixDisplay: Showing text '{text}'")
-
-    def flash(self, count=1, contrast=None):
-        if contrast:
-            logging.info(
-                f"DummyMatrixDisplay: flashing {count} times with contrast {contrast}"
-            )
-        else:
-            logging.info(
-                f"DummyMatrixDisplay: flashing {count} times with general contrast"
-            )
-
-    def turn_off(self):
-        logging.info(f"DummyMatrixDisplay: turning off")
-
-    def turn_full(self):
-        logging.info(f"DummyMatrixDisplay: turning full")
-
-    def show_current_time(self):
-        logging.info(f"DummyMatrixDisplay: showing time")

src/handler/matrix/matrix_display.py

@@ -1,27 +0,0 @@
-class MatrixDisplayAdapter:
-    def __init__(self, *, contrast=0, text_speed=0.02) -> None:
-        self.contrast = contrast
-        self.text_speed = text_speed
-
-    def set_contrast(self, contrast: int):
-        """Set contrast for all actions.
-
-        Args:
-            contrast (int): [0, 255]
-        """
-        self.contrast = contrast
-
-    def show_text(self, text):
-        raise NotImplementedError()
-
-    def flash(self, count=1, contrast=None):
-        raise NotImplementedError()
-
-    def turn_off(self):
-        raise NotImplementedError()
-
-    def turn_full(self):
-        raise NotImplementedError()
-
-    def show_current_time(self):
-        raise NotImplementedError()

src/handler/matrix_display.py

@@ -1,3 +1,4 @@
+import asyncio
 import time
 from datetime import datetime
 from luma.core.interface.serial import spi, noop
@@ -7,10 +8,8 @@ from luma.core import legacy
 from luma.core.virtual import viewport
 from luma.core.legacy.font import proportional, CP437_FONT
 
-from handler.matrix.matrix_display import MatrixDisplayAdapter
 
-
-class LumaMatrixDisplay(MatrixDisplayAdapter):
+class MatrixDisplay:
     def __init__(self, *, contrast=0, text_speed=0.02) -> None:
         self.contrast = contrast
         self.text_speed = text_speed
@@ -27,7 +26,7 @@ class LumaMatrixDisplay(MatrixDisplayAdapter):
         Args:
             contrast (int): [0, 255]
         """
-        super().set_contrast(contrast)
+        self.contrast = contrast
         self.device.contrast(self.contrast)
 
     def show_text(self, text):
@@ -69,6 +68,17 @@ class LumaMatrixDisplay(MatrixDisplayAdapter):
         with canvas(self.device) as draw:
             draw.rectangle((0, 0, 31, 7), outline="white", fill="white")
 
+    async def pattern(self, pattern: str = "01", step_ms: int = 500):
+        # Parse
+        pattern_steps = [step.strip() == "1" for step in pattern]
+        # Execute
+        for step in pattern_steps:
+            if step:
+                self.turn_full()
+            else:
+                self.turn_off()
+            await asyncio.sleep(step_ms / 1000)
+
     def show_current_time(self):
         self.device.contrast(self.contrast)
         hour = str(datetime.now().hour).rjust(2, "0")

src/handler/mhz19_co2.py

@@ -0,0 +1,48 @@
+import serial
+
+
+class Mhz19Co2:
+    def __init__(self):
+        self.last_read = None
+
+        self.serial_port = "/dev/serial0"
+        self.baud_rate = 9600
+        self.byte_size = 8
+        self.parity = "N"
+        self.stop_bits = 1
+        self.timeout = None
+
+    def get_last_read(self) -> int | None:
+        if self.last_read is None:
+            return self.read()
+        return self.last_read
+
+    def read(self) -> int | None:
+        ser = None
+        try:
+            ser = serial.Serial(
+                port=self.serial_port,
+                baudrate=self.baud_rate,
+                bytesize=self.byte_size,
+                parity=self.parity,
+                stopbits=self.stop_bits,
+                timeout=self.timeout,
+            )
+
+            # send "Read CO2" command
+            command_data = bytes([0xFF, 0x01, 0x86, 0x00, 0x00, 0x00, 0x00, 0x00, 0x79])
+            ser.write(command_data)
+
+            # read "Return Value (CO2 concentration)"
+            data = ser.read(9)
+            concentration = data[2] * 256 + data[3]
+
+        except Exception as e:
+            print(f"Error reading data: {e}")
+        finally:
+            if ser:
+                ser.close()
+                ser = None
+
+        self.last_read = concentration
+        return concentration

src/main.py

@@ -5,16 +5,25 @@ from typing import Optional
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 
-from actions import climate_sensor, display_time, log_temperature, matrix_display
+from actions import (
+    climate_sensor,
+    display_time,
+    matrix_display,
+    display_pattern,
+    co2_sensor,
+)
 from config import climate_log_file
 from handler.action_queue import ActionQueue
 from handler.history import get_recent_entries
 
 logging.getLogger().setLevel(logging.INFO)
 
+
+# Start services
 queue = ActionQueue(matrix_display.show_current_time)
 app = FastAPI()
 
+
 origins = [
     "http://localhost",
     "http://localhost:8000",
@@ -52,7 +61,7 @@ async def turn_off():
 
 @app.post("/temperature")
 async def temperature():
-    measurements = climate_sensor.get_last_read()
+    measurements = climate_sensor.read()
     if measurements is None:
         return {"message": "Failed to read temperature"}
 
@@ -65,7 +74,7 @@ async def temperature():
 
 @app.post("/humidity")
 async def humidity():
-    measurements = climate_sensor.get_last_read()
+    measurements = climate_sensor.read()
     if measurements is None:
         return {"message": "Failed to read humidity"}
 
@@ -76,6 +85,30 @@ async def humidity():
     return measurements
 
 
+@app.post("/co2")
+async def co2():
+    co2 = co2_sensor.read()
+    if co2 is None:
+        return {"message": "Failed to read co2"}
+
+    await queue.add_action_to_queue(matrix_display.show_text, f"{co2} ppm")
+
+    return {"co2": co2}
+
+
+@app.post("/climate")
+async def climate():
+    measurements = climate_sensor.read()
+    if measurements is None:
+        return {"message": "Failed to read humidy and temperature"}
+
+    co2 = co2_sensor.read()
+    if co2 is None:
+        return {"message": "Failed to read co2"}
+
+    return {"co2": co2, **measurements}
+
+
 @app.post("/history")
 async def history():
     day_entry_count = 24 * 60
@@ -90,6 +123,12 @@ async def flash(count: int = 1, contrast: Optional[int] = None):
     return {"message": "Display flashed"}
 
 
+@app.post("/pattern")
+async def flash(pattern: str = "01", step_ms: int = 500):
+    await queue.set_idle_action(display_pattern, pattern=pattern, step_ms=step_ms)
+    return {"message": "Activated pattern."}
+
+
 @app.post("/contrast")
 async def contrast(contrast: Optional[int] = None):
     if contrast:
@@ -103,7 +142,3 @@ async def display_message(body: dict):
     message_text = body.get("message")
     await queue.add_action_to_queue(matrix_display.show_text, message_text)
     return {"message": "Message displayed"}
-
-
-if __name__ == "__main__":
-    asyncio.create_task(log_temperature())

start.sh

@@ -1,3 +0,0 @@
-cd /home/pi/matrix-clock/src
-
-uvicorn main:app --reload --host 0.0.0.0