Add ffprobe endpoint

frigate/http.py

@@ -26,11 +26,12 @@ from flask import (
 from peewee import SqliteDatabase, operator, fn, DoesNotExist
 from playhouse.shortcuts import model_to_dict
 
+from frigate.config import CameraConfig
 from frigate.const import CLIPS_DIR
 from frigate.models import Event, Recordings
 from frigate.object_processing import TrackedObject, TrackedObjectProcessor
 from frigate.stats import stats_snapshot
-from frigate.util import clean_camera_user_pass
+from frigate.util import clean_camera_user_pass, ffprobe_stream
 from frigate.version import VERSION
 
 logger = logging.getLogger(__name__)
@@ -607,6 +608,40 @@ def stats():
     return jsonify(stats)
 
 
+@bp.route("/<camera_name>/ffprobe")
+def ffprobe(camera_name):
+    if camera_name not in current_app.frigate_config.cameras:
+        return jsonify(
+            {"success": False, "message": f"Camera name {camera_name} not found"}, "404"
+        )
+
+    config: CameraConfig = current_app.frigate_config.cameras[camera_name]
+
+    if len(config.ffmpeg.inputs) > 1:
+        # user has multiple streams
+        ffprobe_cmd = [
+            "ffprobe",
+            "-rtsp_transport",
+            "tcp",
+            config.ffmpeg.inputs[0].path,
+        ]
+    else:
+        # user has single stream
+        ffprobe = ffprobe_stream(config.ffmpeg.inputs[0].path)
+        if not ffprobe:
+            return jsonify(
+                {
+                    "success": False,
+                    "message": f"ffprobe unable to get info for {camera_name}",
+                },
+                "500",
+            )
+        else:
+            return jsonify(
+                {"success": True, "message": ffprobe}, "200"
+            )
+
+
 @bp.route("/<camera_name>")
 def mjpeg_feed(camera_name):
     fps = int(request.args.get("fps", "3"))

frigate/util.py

@@ -712,6 +712,27 @@ def get_cpu_stats() -> dict[str, dict]:
         return usages
 
 
+def ffprobe_stream(path: str) -> str:
+    """Run ffprobe on stream."""
+    ffprobe_cmd = [
+        "ffprobe",
+        "-rtsp_transport",
+        "tcp",
+        "-show_entries",
+        "stream=codec_long_name,width,height,bit_rate,duration,display_aspect_ratio,avg_frame_rate",
+        "-loglevel",
+        "quiet",
+        path,
+    ]
+    p = sp.run(ffprobe_cmd, capture_output=True)
+
+    if p.returncode != 0:
+        logger.error(f"ffprobe unable to get result for stream: {p.stderr}")
+        return None
+    else:
+        return p.stdout.decode().strip()
+
+
 class FrameManager(ABC):
     @abstractmethod
     def create(self, name, size) -> AnyStr: