from flask import Flask, Response, render_template_string
from picamera2 import Picamera2
import io
import logging
import threading
import time
app = Flask(__name__)
# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
class Camera:
def __init__(self, resolution=(640, 480), framerate=30):
self.resolution = resolution
self.framerate = framerate
self.picam2 = None
self.frame = None
self.lock = threading.Lock()
self.running = False
self.thread = None
def start(self):
self.picam2 = Picamera2()
config = self.picam2.create_video_configuration(
main={"size": self.resolution, "format": "RGB888"},
controls={"FrameRate": self.framerate}
)
self.picam2.configure(config)
self.picam2.start()
# Allow camera to warm up
time.sleep(2)
self.running = True
self.thread = threading.Thread(target=self._capture, daemon=True)
self.thread.start()
logging.info("Camera started.")
def _capture(self):
while self.running:
try:
arr = self.picam2.capture_array()
# Encode as JPEG
import cv2
ret, buf = cv2.imencode('.jpg', arr)
if ret:
with self.lock:
self.frame = buf.tobytes()
time.sleep(0.001) # Brief sleep to yield
except Exception as e:
logging.error(f"Capture error: {e}")
time.sleep(0.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.picam2:
self.picam2.stop()
self.picam2.close()
logging.info("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\n'
b'Content-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 picamera2')
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('--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 = Camera(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()