nicoboy
d416f82889
- Pipeline 640x480@30fps au lieu de 1280x720@15fps - appsink max-buffers=1 sync=false pour éliminer le buffering - Suppression du time.sleep(0.066) dans generate_mjpeg Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
148 lines
4.3 KiB
Python
148 lines
4.3 KiB
Python
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")
|
|
while True:
|
|
ok, frame = cam.read()
|
|
if ok:
|
|
frame = apply_settings(frame)
|
|
with frame_lock:
|
|
current_frame = frame.copy()
|
|
time.sleep(0.066) # ~15fps
|
|
|
|
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 """
|
|
<html>
|
|
<head><title>pi-test-cam</title></head>
|
|
<body style="background:#111;color:#eee;font-family:monospace;padding:20px">
|
|
<h2>pi-test-cam</h2>
|
|
<img src="/stream" width="640" height="360"
|
|
style="border:1px solid #444;border-radius:4px"><br><br>
|
|
|
|
<div style="display:flex;gap:20px;margin-top:10px">
|
|
<label>Brightness
|
|
<input type="range" min="-100" max="100" value="0"
|
|
oninput="update('brightness', this.value)">
|
|
</label>
|
|
<label>Contrast
|
|
<input type="range" min="50" max="300" value="100"
|
|
oninput="update('contrast', this.value/100)">
|
|
</label>
|
|
<label>Saturation
|
|
<input type="range" min="0" max="300" value="100"
|
|
oninput="update('saturation', this.value/100)">
|
|
</label>
|
|
</div>
|
|
<br>
|
|
<a href="/snapshot" style="color:#4af">📸 Snapshot</a>
|
|
|
|
<script>
|
|
function update(key, value) {
|
|
fetch('/settings', {
|
|
method: 'POST',
|
|
headers: {'Content-Type': 'application/json'},
|
|
body: JSON.stringify({[key]: parseFloat(value)})
|
|
})
|
|
}
|
|
</script>
|
|
</body>
|
|
</html>
|
|
"""
|
|
|
|
@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) |