import threading
import time
from collections import deque
import cv2
from flask import Flask, Response, jsonify, render_template_string, request
import FFT_RTSP
class WebSink(FFT_RTSP.BaseSink):
def __init__(self, max_points=2000):
self.lock = threading.Lock()
self.latest_roi_jpg = None
self.latest_frame_jpg = None
self.latest_t = 0.0
self.latest_intensity = 0.0
self.series_t = deque(maxlen=max_points)
self.series_y = deque(maxlen=max_points)
self._stop = False
self.running = False
def _encode(self, img, quality=80):
ok, jpg = cv2.imencode(".jpg", img, [int(cv2.IMWRITE_JPEG_QUALITY), int(quality)])
return jpg.tobytes() if ok else None
def on_roi(self, roi_view, t_s: float, intensity: float):
with self.lock:
self.latest_roi_jpg = self._encode(roi_view)
def on_frame(self, frame, t_s: float, intensity: float):
with self.lock:
self.latest_frame_jpg = self._encode(frame)
self.latest_t = float(t_s)
self.latest_intensity = float(intensity)
self.series_t.append(float(t_s))
self.series_y.append(float(intensity))
def on_series(self, t_list, y_list):
# optional: if using the original arrays rather than deque appends
pass
def should_stop(self) -> bool:
with self.lock:
return self._stop
def stop(self):
with self.lock:
self._stop = True
def snapshot_state(self):
with self.lock:
return {
"running": self.running,
"time": self.latest_t,
"intensity": self.latest_intensity,
"series_t": list(self.series_t),
"series_y": list(self.series_y),
}
def snapshot_jpg(self, which: str):
with self.lock:
return self.latest_roi_jpg if which == "roi" else self.latest_frame_jpg
app = Flask(__name__)
sink = WebSink()
worker_thread = None
# default ROI points
roi_points = [(677, 1288), (1325, 1418), (1425, 1171), (893, 1051)]
def run_worker(source: str):
sink.running = True
try:
FFT_RTSP.process_rtsp_stream(source, roi_points, sink=sink)
finally:
sink.running = False
@app.get("/")
def index():
return render_template_string("""
CV Web UI
Full Frame
""")
@app.get("/api/state")
def api_state():
return jsonify(sink.snapshot_state())
@app.post("/api/start")
def api_start():
global worker_thread
data = request.get_json(force=True) or {}
source = data.get("source") or FFT_RTSP.RTSP_URL
if sink.running:
return jsonify({"ok": True, "already_running": True})
sink._stop = False
worker_thread = threading.Thread(target=run_worker, args=(source,), daemon=True)
worker_thread.start()
return jsonify({"ok": True, "source": source})
@app.post("/api/stop")
def api_stop():
sink.stop()
return jsonify({"ok": True})
def mjpeg_stream(which: str):
boundary = b"frame"
while True:
jpg = sink.snapshot_jpg(which)
if jpg is None:
time.sleep(0.05)
continue
yield b"--" + boundary + b"\r\nContent-Type: image/jpeg\r\n\r\n" + jpg + b"\r\n"
time.sleep(0.03)
@app.get("/mjpeg/roi")
def mjpeg_roi():
return Response(mjpeg_stream("roi"), mimetype="multipart/x-mixed-replace; boundary=frame")
@app.get("/mjpeg/frame")
def mjpeg_frame():
return Response(mjpeg_stream("frame"), mimetype="multipart/x-mixed-replace; boundary=frame")
if __name__ == "__main__":
app.run(host="0.0.0.0", port=8000, threaded=True)