"""MJPEG streaming handler for HTTP multipart/x-mixed-replace.
Each client gets its own handler thread. Frames are pulled from the shared
camera capture singleton — no queuing, just the latest frame.
"""
import io
import threading
import time
import socketserver
from http import server
BOUNDARY = b"--myboundary"
# HTML template for the web UI
INDEX_HTML = (
""
""
""
""
""
"Camera Feed"
""
""
""
""
"
"
""
""
)
class MJPEGHandler(server.BaseHTTPRequestHandler):
"""Serve a single MJPEG stream to one HTTP client."""
server_version = "CameraWebUI/0.1"
def log_message(self, format, *args): # noqa: A002
"""Suppress per-request logging to reduce noise."""
pass
def do_GET(self):
if self.path == "/" or self.path == "":
self._serve_index()
elif self.path == "/stream":
self._serve_stream()
else:
self.send_error(404)
def _serve_index(self):
self.send_response(200)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.end_headers()
self.wfile.write(INDEX_HTML.encode("utf-8"))
def _serve_stream(self):
self.send_response(200)
self.send_header(
"Content-Type", "multipart/x-mixed-replace; boundary=%s" % BOUNDARY.decode()
)
self.send_header("Cache-Control", "no-cache")
self.send_header("Connection", "close")
self.end_headers()
from camera.capture import get_camera
cam = get_camera()
try:
last_frame = None
while True:
frame = cam.frame()
if frame is not None and frame is not last_frame:
self._write_frame(frame)
last_frame = frame
else:
time.sleep(0.05)
except (BrokenPipeError, ConnectionResetError, OSError):
pass
def _write_frame(self, frame: bytes) -> None:
"""Write one MJPEG frame with boundary and headers."""
w = self.wfile
w.write(BOUNDARY)
w.write(b"\r\n")
w.write(b"Content-Type: image/jpeg\r\n")
w.write(b"Content-Length: %d\r\n\r\n" % len(frame))
w.write(frame)
w.write(b"\r\n")
w.flush()
class ThreadedServer(socketserver.ThreadingMixIn, server.HTTPServer):
"""HTTP server that handles each request in a new thread."""
daemon_threads = True
allow_reuse_address = True
def run(host: str = "0.0.0.0", port: int = 8080):
"""Start the HTTP server and block."""
from camera.capture import get_camera
cam = get_camera()
try:
cam.start()
except FileNotFoundError:
print(f"Error: {cam._device} not found — is the camera connected?")
raise SystemExit(1)
server = ThreadedServer((host, port), MJPEGHandler)
print(f"Server running on http://{host}:{port}")
try:
server.serve_forever()
except KeyboardInterrupt:
pass
finally:
cam.stop()
server.server_close()