Flask mehreren Kamera -Wiederholungen mit OpenCV ⇐ Python

[Anonymous]

Ich möchte feststellen, ob es möglich ist, Flask und OpenCV zu verwenden, um eine Anwendung zu erstellen, um gleichzeitig Feeds von 4 Kamera zu nehmen, und auf der Demand spielen die letzten 30 Sekunden zurück. Eine Art einfache VAR -Wiederholung. (HINWEIS: Derzeit fällt Kameras 2,3,4 mehr Kameras, wenn ich beweise, dass dies funktioniert.) Ich habe es geschafft, eine Verbindung zu Kameras herzustellen und Inhalte zu sehen, wenn die grundlegende Python -App ohne Flasche verwendet wird. Aber wenn ich in Flask laufe, ist der Feed schwarz mit gelegentlichen farbigen Zeilen.

Code: Select all

import cv2
import threading
import time
import numpy as np
from collections import deque
from flask import Flask, Response, render_template, request, jsonify

app = Flask(__name__)

# Parameters
FPS = 30  # Frames per second
BUFFER_SIZE = 30 * FPS  # Store last 30 sec of frames
NUM_CAMERAS = 4  # Number of cameras
buffers = [deque(maxlen=BUFFER_SIZE) for _ in range(NUM_CAMERAS)]
#caps = [cv2.VideoCapture(i) for i in range(NUM_CAMERAS)]  # Open all cameras

caps = [
cv2.VideoCapture(0, cv2.CAP_DSHOW),  # Camera 0
cv2.VideoCapture(0, cv2.CAP_DSHOW),  # Fake Camera 2 (copy of 0)
cv2.VideoCapture(0, cv2.CAP_DSHOW),  # Fake Camera 2 (copy of 0)
cv2.VideoCapture(0, cv2.CAP_DSHOW)   # Fake Camera 3 (copy of 0)
]

# Capture video from ca
# meras
def capture_video(cam_index):
while True:
ret, frame = caps[cam_index].read()
if ret:
buffers[cam_index].append(frame)
time.sleep(1 / FPS)

# Start threads for each camera
for i in range(NUM_CAMERAS):
threading.Thread(target=capture_video, args=(i,), daemon=True).start()

@app.route('/video_feed/0')
def video_feed_0():
return Response(generate_feed(0), mimetype='multipart/x-mixed-replace; boundary=frame')

@app.route('/video_feed/1')
def video_feed_1():
return Response(generate_feed(1), mimetype='multipart/x-mixed-replace; boundary=frame')

@app.route('/video_feed/')
def video_feed(cam_index):
return Response(generate_feed(cam_index),
mimetype='multipart/x-mixed-replace; boundary=frame')

# Stream live video
def generate_feed(cam_index):
while True:
success, frame = caps[cam_index].read()

if not success:
print(f"❌ Failed to capture frame from Camera {cam_index}")
continue

cv2.imshow(f"Camera {cam_index}", frame)  # Display the frame in OpenCV window

ret, buffer = cv2.imencode(".jpg", frame)
if not ret:
print(f"❌ Failed to encode frame from Camera {cam_index}")
continue

yield (b"--frame\r\n"
b"Content-Type: image/jpeg\r\n\r\n" + buffer.tobytes() + b"\r\n")

# Replay last 30 sec
@app.route('/replay/')
def replay(speed):
replay_buffers = [list(buffers[i]) for i in range(NUM_CAMERAS)]
min_frames = min(len(frames) for frames in replay_buffers)

for i in range(min_frames):
frames = [cv2.resize(replay_buffers[c][i], (320, 240)) for c in range(NUM_CAMERAS)]
combined_frame = np.vstack([np.hstack(frames[:2]), np.hstack(frames[2:])])

_, buffer = cv2.imencode(".jpg", combined_frame)
yield (b"--frame\r\n"
b"Content-Type: image/jpeg\r\n\r\n" + buffer.tobytes() + b"\r\n")
time.sleep((1 / FPS) / speed)

@app.route('/replay_feed/')
def replay_feed(speed):
return Response(replay(speed), mimetype="multipart/x-mixed-replace; boundary=frame")

@app.route('/')
def index():
return render_template("index.html")

if __name__ == "__main__":
app.run(host="0.0.0.0", port=5000, debug=True)
< /code>
HTML: < /p>





Multi-Camera Replay

body { font-family: Arial, sans-serif; text-align: center; }
.grid { display: grid; grid-template-columns: repeat(2, 1fr); gap: 10px; }
img { width: 100%; height: auto; }
button { margin: 10px; padding: 10px; font-size: 16px; cursor: pointer; }




Live Camera Feeds

[img]/video_feed/0[/img]
[img]/video_feed/1[/img]
[img]/video_feed/2[/img]
[img]/video_feed/3[/img]

Replay Last 30 Sec
Replay Normal
Replay 2x Speed
Replay Slow Motion

Replay View



function startReplay(speed) {
document.getElementById("replayFeed").src = "/replay_feed/" + speed;
}