QUELLCODE / PY
device.py
src/hardware_box/capture/device.py
1# OpenAss: nur privater, experimenteller Gebrauch; auf eigene Verantwortung.2# Niemals Unternehmensrechner, Produktivsysteme, Industrieanlagen oder3# sicherheitskritische Systeme anschließen oder bedienen.4# Haftungshinweis und gesetzliche Grenzen: ../../../DISCLAIMER.txt.5import asyncio6import threading7import time89from PIL import Image, ImageDraw10from hardware_box.vision.frame import prepare111213class CaptureError(RuntimeError):14 pass151617class DemoCapture:18 def __init__(self, settings):19 self.settings, self.sequence = settings, 02021 async def start(self):22 pass2324 async def close(self):25 pass2627 def health(self):28 return {"ready": True, "error": None, "age": 0}2930 async def snapshot(self, after=0):31 self.sequence += 132 image = Image.new("RGB", (1920, 1080), "#172334")33 draw = ImageDraw.Draw(image)34 draw.text((90, 80), "HARDWARE BOX / DEMO - no real PC attached", fill="white", font_size=36)35 draw.rectangle((150, 200, 1770, 860), outline="#46ddd0", width=4)36 draw.text((190, 240), "Monatsbericht.xlsx | B7: [250]", fill="white", font_size=32)37 draw.text((190, 340), "This image is simulated. HID events are logged only.", fill="#a8bfd7", font_size=28)38 return await asyncio.to_thread(prepare, image, self.settings, self.sequence)394041class UVCCapture:42 def __init__(self, settings):43 self.settings = settings44 self._stop = threading.Event()45 self._lock = threading.Lock()46 self._latest = None47 self._error = None48 self._thread = None49 self._cap = None50 self._negotiated = {}5152 async def start(self):53 import cv254 self._cv2 = cv255 self._cap = await asyncio.to_thread(cv2.VideoCapture, self.settings.capture_device, cv2.CAP_V4L2)56 if not self._cap.isOpened():57 raise CaptureError("UVC-Gerät konnte nicht geöffnet werden")58 self._cap.set(cv2.CAP_PROP_FOURCC, cv2.VideoWriter_fourcc(*self.settings.capture_format))59 self._cap.set(cv2.CAP_PROP_FRAME_WIDTH, self.settings.capture_width)60 self._cap.set(cv2.CAP_PROP_FRAME_HEIGHT, self.settings.capture_height)61 self._cap.set(cv2.CAP_PROP_FPS, self.settings.capture_fps)62 negotiated_fps = self._cap.get(cv2.CAP_PROP_FPS)63 fourcc = int(self._cap.get(cv2.CAP_PROP_FOURCC))64 reported_format = "".join(chr((fourcc >> (8 * i)) & 255) for i in range(4))65 accepted_formats = {self.settings.capture_format}66 if self.settings.capture_format == "YUYV":67 accepted_formats.add("YUY2")68 self._negotiated = {"fps": negotiated_fps, "format": reported_format,69 "width": self._cap.get(cv2.CAP_PROP_FRAME_WIDTH),70 "height": self._cap.get(cv2.CAP_PROP_FRAME_HEIGHT)}71 if negotiated_fps < self.settings.capture_fps - 0.5 or reported_format not in accepted_formats:72 self._cap.release()73 raise CaptureError("Treiber meldet nicht den gewählten Capture-Modus; UVC-Format und 60 fps mit v4l2-ctl prüfen")74 self._cap.set(cv2.CAP_PROP_BUFFERSIZE, 1) # advisory; continuously drain regardless75 self._thread = threading.Thread(target=self._reader, daemon=True, name="uvc-latest-frame")76 self._thread.start()77 await self.snapshot()7879 def _reader(self):80 seq = 081 try:82 while not self._stop.is_set():83 ok, frame = self._cap.read()84 if not ok:85 raise CaptureError("UVC-Aufnahme unterbrochen")86 if frame.shape[1::-1] != (self.settings.capture_width, self.settings.capture_height):87 raise CaptureError("UVC-Auflösung stimmt nicht mit der Konfiguration überein")88 seq += 189 with self._lock:90 self._latest = (frame, time.monotonic(), seq)91 except Exception as exc:92 with self._lock:93 self._error = str(exc)9495 async def snapshot(self, after=0):96 deadline = time.monotonic() + 297 while time.monotonic() < deadline:98 with self._lock:99 item, error = self._latest, self._error100 if error:101 raise CaptureError(error)102 if item and item[1] > after and time.monotonic() - item[1] < 0.5:103 raw, timestamp, seq = item104 rgb = await asyncio.to_thread(self._cv2.cvtColor, raw, self._cv2.COLOR_BGR2RGB)105 return await asyncio.to_thread(prepare, Image.fromarray(rgb), self.settings, seq, timestamp)106 await asyncio.sleep(0.02)107 raise CaptureError("Kein aktuelles HDMI-Bild innerhalb von zwei Sekunden")108109 def health(self):110 with self._lock:111 item, error = self._latest, self._error112 return {"ready": bool(item), "error": error,113 "age": time.monotonic() - item[1] if item else 9999,114 "negotiated": self._negotiated}115116 async def close(self):117 self._stop.set()118 if self._thread:119 await asyncio.to_thread(self._thread.join, 0.5)120 if self._cap and (not self._thread or not self._thread.is_alive()):121 self._cap.release()