#!/usr/bin/env python3
"""
Serveur Flask - Stream vidéo Raspberry Pi HQ Camera M12
Compatible Raspberry Pi 4 et 5 avec Picamera2
"""
import io
import os
import threading
import time
from datetime import datetime
from flask import Flask, Response, jsonify, render_template_string
from picamera2 import Picamera2
from picamera2.encoders import JpegEncoder
from picamera2.outputs import FileOutput
# ─── Configuration ────────────────────────────────────────────────────────────
RESOLUTION = (1280, 720) # Résolution du stream (largeur, hauteur)
FRAMERATE = 30 # Images par seconde
QUALITY = 85 # Qualité JPEG (1-100)
HOST = "0.0.0.0" # Ecoute sur toutes les interfaces réseau
PORT = 5000
PHOTOS_DIR = os.path.join(os.path.dirname(__file__), "photos")
# ─── Buffer de frames thread-safe ─────────────────────────────────────────────
class StreamOutput(io.BufferedIOBase):
def __init__(self):
self.frame = None
self.condition = threading.Condition()
def write(self, buf):
with self.condition:
self.frame = buf
self.condition.notify_all()
# ─── Initialisation caméra ────────────────────────────────────────────────────
picam2 = Picamera2()
video_config = picam2.create_video_configuration(
main={"size": RESOLUTION, "format": "RGB888"},
controls={"FrameRate": FRAMERATE}
)
picam2.configure(video_config)
output = StreamOutput()
encoder = JpegEncoder(q=QUALITY)
picam2.start_recording(encoder, FileOutput(output))
print(f"[✓] Caméra démarrée ({RESOLUTION[0]}x{RESOLUTION[1]} @ {FRAMERATE}fps)")
os.makedirs(PHOTOS_DIR, exist_ok=True)
# ─── Générateur MJPEG ─────────────────────────────────────────────────────────
def generate_frames():
"""Génère un flux MJPEG frame par frame."""
while True:
with output.condition:
output.condition.wait()
frame = output.frame
yield (
b"--frame\r\n"
b"Content-Type: image/jpeg\r\n\r\n" + frame + b"\r\n"
)
# ─── Interface HTML ───────────────────────────────────────────────────────────
HTML_PAGE = """
Pi Camera Stream
"""
# ─── Application Flask ────────────────────────────────────────────────────────
app = Flask(__name__)
@app.route("/")
def index():
return render_template_string(
HTML_PAGE,
width=RESOLUTION[0],
height=RESOLUTION[1],
quality=QUALITY
)
@app.route("/capture")
def capture():
"""Sauvegarde la frame courante dans le dossier photos/."""
with output.condition:
output.condition.wait()
frame = output.frame
filename = datetime.now().strftime("%Y%m%d_%H%M%S.jpg")
filepath = os.path.join(PHOTOS_DIR, filename)
with open(filepath, "wb") as f:
f.write(frame)
return jsonify({"status": "ok", "file": filename})
@app.route("/video_feed")
def video_feed():
"""Endpoint du flux MJPEG."""
return Response(
generate_frames(),
mimetype="multipart/x-mixed-replace; boundary=frame"
)
# ─── Lancement ────────────────────────────────────────────────────────────────
if __name__ == "__main__":
try:
print(f"[✓] Serveur démarré → http://{HOST}:{PORT}")
print(f" Accès local : http://localhost:{PORT}")
print(f" Accès réseau : http://:{PORT}")
app.run(host=HOST, port=PORT, threaded=True)
except KeyboardInterrupt:
print("\n[!] Arrêt du serveur...")
finally:
picam2.stop_recording()
picam2.close()
print("[✓] Caméra arrêtée proprement.")