Source:0x0 Canvas:0x0
NPU Inference: 0.0 FPS
Active Detections: 0
"""
return render_template_string(html_template, active_src=src_active)
@app.route('/api/stream_info')
def stream_info():
with lock:
return jsonify({
"native_w": STATE_REPO["native_w"],
"native_h": STATE_REPO["native_h"],
"width": STATE_REPO["stream_w"],
"height": STATE_REPO["stream_h"],
"fps": STATE_REPO["inference_fps"],
"detections": len(STATE_REPO["detections"])
})
@app.route('/api/swap_config', methods=['POST'])
def swap_config():
payload = request.get_json()
src_selected = payload.get("source")
mdl_selected = payload.get("model")
with lock:
trigger_restart = False
if src_selected in STATE_REPO["source_registry"] and STATE_REPO["active_source"] != src_selected:
STATE_REPO["active_source"] = src_selected
trigger_restart = True
if mdl_selected in STATE_REPO["model_registry"] and STATE_REPO["active_model_name"] != mdl_selected:
base_dir = app.config["MODEL_DIR"]
STATE_REPO["active_model_name"] = mdl_selected
STATE_REPO["active_model_path"] = os.path.join(base_dir, mdl_selected, "model.dvm")
trigger_restart = True
if trigger_restart:
STATE_REPO["restart_flag"] = True
return jsonify({"status": "success"})
def generate_mjpeg_stream_generator():
MODEL_W = float(STATE_REPO["MODEL_W"])
MODEL_H = float(STATE_REPO["MODEL_H"])
while True:
time.sleep(0.04)
frame_copy = STATE_REPO["frame"]
local_dets = list(STATE_REPO["detections"])
if frame_copy is not None:
frame = frame_copy.copy()
h_native, w_native, _ = frame_copy.shape
for class_id, confidence, rx1, ry1, rx2, ry2 in local_dets:
cx1 = int(rx1 * (float(w_native) / MODEL_W))
cx2 = int(rx2 * (float(w_native) / MODEL_W))
cy1 = int(ry1 * (float(h_native) / MODEL_H))
cy2 = int(ry2 * (float(h_native) / MODEL_H))
label = f"{COCO_LABELS.get(class_id, f'Class {class_id}')} ({confidence*100:.1f}%)"
cv2.rectangle(frame, (cx1, cy1), (cx2, cy2), (0, 255, 97), 2)
cv2.putText(frame, label, (cx1, max(15, cy1 - 5)), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 255, 97), 2)
_, encoded_img = cv2.imencode(".jpg", frame)
yield (b'--frame\r\n'
b'Content-Type: image/jpeg\r\n\r\n' + encoded_img.tobytes() + b'\r\n')
else:
waiting_canvas = np.zeros((480, 640, 3), dtype=np.uint8)
cv2.putText(waiting_canvas, "AWAITING MEDIA INPUT SELECTION...", (140, 240), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 255, 97), 1)
_, encoded_img = cv2.imencode(".jpg", waiting_canvas)
yield (b'--frame\r\n'
b'Content-Type: image/jpeg\r\n\r\n' + encoded_img.tobytes() + b'\r\n')
@app.route('/stream.mjpg')
def video_feed_stream_route():
return Response(generate_mjpeg_stream_generator(), mimetype='multipart/x-mixed-replace; boundary=frame')
def main():
parser = argparse.ArgumentParser(description="Wiki Template: Ara Flask Video Engine")
parser.add_argument("--camera", default=None, help="Camera context device node path")
parser.add_argument("--mp4", default=None, help="Directory containing target mp4 sample videos")
parser.add_argument("--port", type=int, default=8080, help="Target port mapping")
parser.add_argument("--model-dir", default="/usr/share/cnn/detection", help="Directory containing target models")
parser.add_argument("--model", default="yolov8n", help="Initial model selection")
args = parser.parse_args()
app.config["MODEL_DIR"] = args.model_dir
STATE_REPO["source_registry"] = []
if args.camera and os.path.exists(args.camera):
STATE_REPO["source_registry"].append(args.camera)
if args.mp4 and os.path.exists(args.mp4):
local_videos = glob.glob(os.path.join(args.mp4, "*.mp4"))
for vid in sorted(local_videos):
STATE_REPO["source_registry"].append(vid)
if os.path.exists(args.model_dir):
discovered_models = []
for entry in sorted(os.listdir(args.model_dir)):
full_subdir = os.path.join(args.model_dir, entry)
if os.path.isdir(full_subdir) and os.path.exists(os.path.join(full_subdir, "model.dvm")):
discovered_models.append(entry)
if discovered_models:
STATE_REPO["model_registry"] = discovered_models
STATE_REPO["active_model_name"] = args.model if args.model in discovered_models else discovered_models[0]
STATE_REPO["active_model_path"] = os.path.join(args.model_dir, STATE_REPO["active_model_name"], "model.dvm")
threading.Thread(target=gstreamer_orchestration_loop, daemon=True).start()
print(f"Server serving on: http://localhost:{args.port}/")
app.run(host='0.0.0.0', port=args.port, threaded=True, use_reloader=False, debug=False)
if __name__ == '__main__':
main()