image alarm

audio_server.py

@@ -19,6 +19,7 @@ from pathlib import Path
 from random import randint
 
 import websockets
+from PIL import Image
 
 logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
 log = logging.getLogger("audio_server")
@@ -26,6 +27,8 @@ log = logging.getLogger("audio_server")
 PORT = 8766
 CHUNK_SIZE = 4096
 AUDIO_DIR = Path(__file__).parent / "assets" / "alarm"
+IMG_DIR = Path(__file__).parent / "assets" / "img"
+STATUS_IMG_SIZE = 120
 
 
 def find_wav() -> Path:
@@ -49,6 +52,34 @@ def read_wav(path: Path) -> tuple[bytes, int, int, int]:
     return pcm, sr, ch, bits
 
 
+def load_status_image(path: Path) -> bytes:
+    """Load a PNG, convert to 1-bit 120x120 monochrome bitmap (MSB-first, black=1)."""
+    img = Image.open(path).convert("L")
+
+    # Resize to fit within 120x120, preserving aspect ratio
+    img.thumbnail((STATUS_IMG_SIZE, STATUS_IMG_SIZE), Image.LANCZOS)
+
+    # Paste centered onto white canvas
+    canvas = Image.new("L", (STATUS_IMG_SIZE, STATUS_IMG_SIZE), 255)
+    x_off = (STATUS_IMG_SIZE - img.width) // 2
+    y_off = (STATUS_IMG_SIZE - img.height) // 2
+    canvas.paste(img, (x_off, y_off))
+
+    # Threshold to 1-bit: black (< 128) → 1, white → 0
+    bw = canvas.point(lambda p: 1 if p < 128 else 0, "1")
+    raw = bw.tobytes()
+    log.info("Status image loaded: %s → %d bytes", path.name, len(raw))
+    return raw
+
+
+async def send_status_image(ws, img_bytes: bytes):
+    """Send a status image over the WebSocket (text header + binary payload)."""
+    header = json.dumps({"type": "status_image", "width": STATUS_IMG_SIZE, "height": STATUS_IMG_SIZE})
+    await ws.send(header)
+    await ws.send(img_bytes)
+    log.info("Sent status image (%d bytes)", len(img_bytes))
+
+
 def chunk_bytes(data: bytes, size: int):
     """Yield data in fixed-size chunks."""
     for i in range(0, len(data), size):
@@ -92,12 +123,24 @@ async def handler(ws):
     wav_path = find_wav()
     pcm, sr, ch, bits = read_wav(wav_path)
 
+    # Load status images
+    img_idle = load_status_image(IMG_DIR / "idle.png")
+    img_alarm = load_status_image(IMG_DIR / "on_alarm.png")
+
     try:
+        # Send idle image on connect
+        await send_status_image(ws, img_idle)
+
         while True:
             delay = randint(30, 60)
             log.info("Next alarm in %ds", delay)
             await asyncio.sleep(delay)
+
+            # Switch to alarm image before audio
+            await send_status_image(ws, img_alarm)
             await stream_alarm(ws, pcm, sr, ch, bits)
+            # Switch back to idle after alarm
+            await send_status_image(ws, img_idle)
     except websockets.exceptions.ConnectionClosed:
         log.info("Client disconnected: %s:%d", remote[0], remote[1])
 

requirements.txt

@@ -1,2 +1,3 @@
 websockets>=12.0
 psutil>=5.9.0
+Pillow>=10.0