Initiales Halloween-Projekt

halloween/camera.py

@@ -0,0 +1,118 @@
+"""Ein Kameraproduzent für mehrere Livebild-Betrachter."""
+
+from io import BytesIO
+import threading
+import time
+from datetime import datetime
+
+from PIL import Image, ImageDraw
+
+
+class CameraStream:
+    def __init__(self, demo: bool = False):
+        self.demo = demo
+        self._condition = threading.Condition()
+        self._thread = None
+        self._stop = threading.Event()
+        self._frame = None
+        self._number = 0
+        self._time = None
+        self.error = None
+
+    def start(self):
+        if self._thread is not None:
+            return
+        self._thread = threading.Thread(target=self._run, daemon=True, name="camera")
+        self._thread.start()
+
+    def stop(self):
+        self._stop.set()
+        with self._condition:
+            self._condition.notify_all()
+        if self._thread is not None:
+            self._thread.join(timeout=3)
+
+    @staticmethod
+    def _jpeg(image: Image.Image) -> bytes:
+        buffer = BytesIO()
+        image.save(buffer, format="JPEG", quality=75)
+        return buffer.getvalue()
+
+    def _publish(self, frame: bytes):
+        with self._condition:
+            self._frame = frame
+            self._number += 1
+            self._time = time.monotonic()
+            self._condition.notify_all()
+
+    def _run(self):
+        try:
+            if self.demo:
+                self._run_demo()
+            else:
+                self._run_pi()
+        except Exception as exc:
+            with self._condition:
+                self.error = f"Kamera nicht verfügbar: {type(exc).__name__}: {exc}"
+                self._condition.notify_all()
+
+    def _run_demo(self):
+        while not self._stop.is_set():
+            image = Image.new("RGB", (640, 480), (21, 28, 32))
+            draw = ImageDraw.Draw(image)
+            draw.rectangle((40, 105, 600, 400), outline=(144, 111, 65), width=8)
+            draw.ellipse((165, 205, 475, 340), fill=(45, 52, 36), outline=(110, 91, 59), width=4)
+            draw.text((42, 30), "DEMO - KEINE ECHTE KAMERA", fill=(255, 172, 72))
+            draw.text((42, 440), datetime.now().strftime("%Y-%m-%d %H:%M:%S"), fill="white")
+            self._publish(self._jpeg(image))
+            self._stop.wait(0.2)
+
+    def _run_pi(self):
+        from picamera2 import Picamera2
+
+        camera = Picamera2()
+        started = False
+        try:
+            # BGR888 liefert laut Picamera2-Dokumentation RGB-Arrays für PIL.
+            config = camera.create_video_configuration(
+                main={"size": (640, 480), "format": "BGR888"}
+            )
+            camera.configure(config)
+            camera.start()
+            started = True
+            while not self._stop.is_set():
+                pixels = camera.capture_array("main")
+                self._publish(self._jpeg(Image.fromarray(pixels, "RGB")))
+                self._stop.wait(0.2)
+        finally:
+            try:
+                if started:
+                    camera.stop()
+            finally:
+                camera.close()
+
+    def latest(self):
+        with self._condition:
+            return self._frame
+
+    def latest_numbered(self):
+        with self._condition:
+            return self._number, self._frame
+
+    def wait_next(self, number: int, timeout: float = 10):
+        with self._condition:
+            self._condition.wait_for(
+                lambda: self._number > number or self.error or self._stop.is_set(),
+                timeout=timeout,
+            )
+            return self._number, self._frame
+
+    def status(self):
+        with self._condition:
+            age = None if self._time is None else round(time.monotonic() - self._time, 1)
+            return {
+                "source": "Demo" if self.demo else "Pi-Kamera",
+                "ready": self._frame is not None and age is not None and age < 3,
+                "frame_age_seconds": age,
+                "error": self.error,
+            }