import logging
import threading
import time
import cv2
from flask import Flask, Response, render_template_string
app = Flask(__name__)
logging.basicConfig(
level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
)
class UsbCamera:
def __init__(self, device_index=0, resolution=(640, 480), framerate=30):
self.device_index = device_index
self.resolution = resolution
self.framerate = framerate
self.cap = None
self.frame = None
self.lock = threading.Lock()
self.running = False
self.thread = None
def start(self):
self.cap = cv2.VideoCapture(self.device_index)
if not self.cap.isOpened():
raise RuntimeError(f"Cannot open camera device {self.device_index}")
self.cap.set(cv2.CAP_PROP_FRAME_WIDTH, self.resolution[0])
self.cap.set(cv2.CAP_PROP_FRAME_HEIGHT, self.resolution[1])
self.cap.set(cv2.CAP_PROP_FPS, self.framerate)
# Give the driver a moment to apply settings
time.sleep(0.5)
self.running = True
self.thread = threading.Thread(target=self._capture, daemon=True)
self.thread.start()
logging.info("USB camera started.")
def _capture(self):
while self.running:
ret, img = self.cap.read()
if not ret:
logging.warning("Failed to read frame from camera.")
time.sleep(0.05)
continue
ret2, buf = cv2.imencode(".jpg", img)
if ret2:
with self.lock:
self.frame = buf.tobytes()
# Throttle slightly to avoid maxing CPU
time.sleep(1.0 / max(self.framerate, 1))
def get_frame(self):
with self.lock:
return self.frame
def stop(self):
self.running = False
if self.thread:
self.thread.join(timeout=2.0)
if self.cap:
self.cap.release()
logging.info("USB camera stopped.")
camera = None
HTML_PAGE = """
PiCam Stream
PiCam Live Stream
{{ resolution }} @ {{ framerate }} fps
 }})
"""
@app.route("/")
def index():
return render_template_string(
HTML_PAGE, resolution=camera.resolution, framerate=camera.framerate
)
@app.route("/video_feed")
def video_feed():
def generate():
while True:
frame = camera.get_frame()
if frame is None:
time.sleep(0.01)
continue
yield (b"--frame\r\nContent-Type: image/jpeg\r\n\r\n" + frame + b"\r\n")
return Response(generate(), mimetype="multipart/x-mixed-replace; boundary=frame")
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser(
description="PiCam MJPEG Streamer using OpenCV (USB camera)"
)
parser.add_argument(
"--host", default="0.0.0.0", help="Interface to bind (default: 0.0.0.0)"
)
parser.add_argument(
"--port", type=int, default=5000, help="Port to bind (default: 5000)"
)
parser.add_argument(
"--device", type=int, default=0, help="Video device index (default: 0)"
)
parser.add_argument("--width", type=int, default=640, help="Capture width")
parser.add_argument("--height", type=int, default=480, help="Capture height")
parser.add_argument("--fps", type=int, default=30, help="Target framerate")
args = parser.parse_args()
camera = UsbCamera(
device_index=args.device,
resolution=(args.width, args.height),
framerate=args.fps,
)
camera.start()
try:
logging.info(f"Starting server on http://{args.host}:{args.port}")
app.run(host=args.host, port=args.port, threaded=True, debug=False)
except KeyboardInterrupt:
pass
finally:
camera.stop()