from flask import Flask, Response, request, jsonify
import cv2
import threading
import time
import numpy as np
app = Flask(__name__)
# Paramètres image modifiables à chaud
settings = {
"brightness": 0, # -100 à +100
"contrast": 1.0, # 0.5 à 3.0
"saturation": 1.0, # 0.0 à 3.0
}
frame_lock = threading.Lock()
current_frame = None
def apply_settings(frame):
# Contrast + Brightness
frame = cv2.convertScaleAbs(
frame,
alpha=settings["contrast"],
beta=settings["brightness"]
)
# Saturation via HSV
hsv = cv2.cvtColor(frame, cv2.COLOR_BGR2HSV).astype("float32")
hsv[:, :, 1] *= settings["saturation"]
hsv[:, :, 1] = np.clip(hsv[:, :, 1], 0, 255)
frame = cv2.cvtColor(hsv.astype("uint8"), cv2.COLOR_HSV2BGR)
return frame
def capture_loop():
global current_frame
# Pipeline GStreamer pour CSI sur Pi Bookworm
pipeline = (
"libcamerasrc ! "
"video/x-raw,width=640,height=480,framerate=30/1 ! "
"videoconvert ! "
"video/x-raw,format=BGR ! "
"appsink drop=1 max-buffers=1 sync=false"
)
cam = cv2.VideoCapture(pipeline, cv2.CAP_GSTREAMER)
if not cam.isOpened():
print("ERREUR : impossible d'ouvrir la caméra")
return
print("Caméra initialisée")
fps_counter = 0
fps_display = 0.0
fps_timer = time.time()
while True:
ok, frame = cam.read()
if ok:
fps_counter += 1
now = time.time()
if now - fps_timer >= 1.0:
fps_display = fps_counter / (now - fps_timer)
fps_counter = 0
fps_timer = now
frame = apply_settings(frame)
h, w = frame.shape[:2]
label = f"{w}x{h} | {fps_display:.1f} fps"
cv2.putText(frame, label, (8, h - 8),
cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 0, 0), 2)
cv2.putText(frame, label, (8, h - 8),
cv2.FONT_HERSHEY_SIMPLEX, 0.5, (255, 255, 255), 1)
with frame_lock:
current_frame = frame.copy()
def generate_mjpeg():
while True:
with frame_lock:
if current_frame is None:
time.sleep(0.1)
continue
_, jpeg = cv2.imencode(
".jpg", current_frame,
[cv2.IMWRITE_JPEG_QUALITY, 70]
)
yield (
b"--frame\r\n"
b"Content-Type: image/jpeg\r\n\r\n"
+ jpeg.tobytes()
+ b"\r\n"
)
@app.route("/")
def index():
return """
pi-test-cam
pi-test-cam
📸 Snapshot
"""
@app.route("/stream")
def stream():
return Response(
generate_mjpeg(),
mimetype="multipart/x-mixed-replace; boundary=frame"
)
@app.route("/snapshot")
def snapshot():
with frame_lock:
if current_frame is None:
return "Pas de frame disponible", 503
_, jpeg = cv2.imencode(
".jpg", current_frame,
[cv2.IMWRITE_JPEG_QUALITY, 95]
)
return Response(jpeg.tobytes(), mimetype="image/jpeg")
@app.route("/settings", methods=["GET", "POST"])
def handle_settings():
if request.method == "POST":
data = request.json
for k in ["brightness", "contrast", "saturation"]:
if k in data:
settings[k] = float(data[k])
return jsonify({"ok": True, "settings": settings})
return jsonify(settings)
if __name__ == "__main__":
t = threading.Thread(target=capture_loop, daemon=True)
t.start()
time.sleep(2)
print("Serveur démarré sur http://0.0.0.0:5000")
app.run(host="0.0.0.0", port=5000, threaded=True)