POC: Frigate HTTP API using FastAPI

docker/main/rootfs/usr/local/nginx/conf/nginx.conf

@@ -222,12 +222,16 @@ http {
  include proxy.conf;
  }
 
- location ~* /api/.*\.(jpg|jpeg|png|webp|gif)$ {
- include auth_request.conf;
- rewrite ^/api/(.*)$ $1 break;
- proxy_pass http://frigate_api;
- include proxy.conf;
- }
+# FIXME: Needed to disabled this rule, otherwise it fails for endpoints that end with one of those file extensions
+# 1. with httptools it passes the auth.conf but then throws a 400 error "WARN "Invalid HTTP request received." -> https://github.com/encode/uvicorn/blob/47304d9ae76321f0f5f649ff4f73e09b17085933/uvicorn/protocols/http/httptools_impl.py#L165
+# 2. With h11 it goes through the auth.conf but returns a 404 error
+# We might need to add extra rules that will allow endpoint that end with an extension OR find a fix without creating other rules
+# location ~* /api/.*\.(jpg|jpeg|png|webp|gif)$ {
+# include auth_request.conf;
+# rewrite ^/api/(.*)$ $1 break;
+# proxy_pass http://frigate_api;
+# include proxy.conf;
+# }
 
  location /api/ {
  include auth_request.conf;

frigate/api/app.py

@@ -10,6 +10,9 @@
 from typing import Optional
 
 import requests
+from fastapi import APIRouter
+from fastapi.encoders import jsonable_encoder
+from fastapi.responses import JSONResponse
 from flask import Blueprint, Flask, current_app, jsonify, make_response, request
 from markupsafe import escape
 from peewee import operator
@@ -21,7 +24,6 @@
 from frigate.api.export import ExportBp
 from frigate.api.media import MediaBp
 from frigate.api.notification import NotificationBp
-from frigate.api.preview import PreviewBp
 from frigate.api.review import ReviewBp
 from frigate.config import FrigateConfig
 from frigate.const import CONFIG_DIR
@@ -47,11 +49,12 @@
 bp.register_blueprint(EventBp)
 bp.register_blueprint(ExportBp)
 bp.register_blueprint(MediaBp)
-bp.register_blueprint(PreviewBp)
 bp.register_blueprint(ReviewBp)
 bp.register_blueprint(AuthBp)
 bp.register_blueprint(NotificationBp)
 
+router = APIRouter()
+
 
 def create_app(
  frigate_config,
@@ -454,19 +457,26 @@ def vainfo():
  )
 
 
-@bp.route("/logs/<service>", methods=["GET"])
-def logs(service: str):
+@router.get("/logs/{service}", tags=["Logs"])
+def logs(
+ service: str,
+ download: Optional[str] = None,
+ start: Optional[int] = 0,
+ end: Optional[int] = None,
+):
+ """Get logs for the requested service (frigate/nginx/go2rtc/chroma)"""
+
  def download_logs(service_location: str):
  try:
  file = open(service_location, "r")
  contents = file.read()
  file.close()
- return jsonify(contents)
+ return JSONResponse(jsonable_encoder(contents))
  except FileNotFoundError as e:
  logger.error(e)
- return make_response(
- jsonify({"success": False, "message": "Could not find log file"}),
- 500,
+ return JSONResponse(
+ content={"success": False, "message": "Could not find log file"},
+ status_code=500,
  )
 
  log_locations = {
@@ -478,17 +488,14 @@ def download_logs(service_location: str):
  service_location = log_locations.get(service)
 
  if not service_location:
- return make_response(
- jsonify({"success": False, "message": "Not a valid service"}),
- 404,
+ return JSONResponse(
+ content={"success": False, "message": "Not a valid service"},
+ status_code=404,
  )
 
- if request.args.get("download", type=bool, default=False):
+ if download:
  return download_logs(service_location)
 
- start = request.args.get("start", type=int, default=0)
- end = request.args.get("end", type=int)
-
  try:
  file = open(service_location, "r")
  contents = file.read()
@@ -529,15 +536,15 @@ def download_logs(service_location: str):
 
  logLines.append(currentLine)
 
- return make_response(
- jsonify({"totalLines": len(logLines), "lines": logLines[start:end]}),
- 200,
+ return JSONResponse(
+ content={"totalLines": len(logLines), "lines": logLines[start:end]},
+ status_code=200,
  )
  except FileNotFoundError as e:
  logger.error(e)
- return make_response(
- jsonify({"success": False, "message": "Could not find log file"}),
- 500,
+ return JSONResponse(
+ content={"success": False, "message": "Could not find log file"},
+ status_code=500,
  )
 
 

frigate/api/media.py

@@ -2,16 +2,20 @@
 
 import base64
 import glob
+import io
 import logging
 import os
 import subprocess as sp
 import time
 from datetime import datetime, timedelta, timezone
+from typing import Optional
 from urllib.parse import unquote
 
 import cv2
 import numpy as np
 import pytz
+from fastapi import APIRouter, Request
+from fastapi.responses import JSONResponse, StreamingResponse
 from flask import Blueprint, Response, current_app, jsonify, make_response, request
 from peewee import DoesNotExist, fn
 from tzlocal import get_localzone_name
@@ -32,6 +36,8 @@
 
 MediaBp = Blueprint("media", __name__)
 
+router = APIRouter(tags=["Media"])
+
 
 @MediaBp.route("/<camera_name>")
 def mjpeg_feed(camera_name):
@@ -92,90 +98,102 @@ def camera_ptz_info(camera_name):
  404,
  )
 
-
-@MediaBp.route("/<camera_name>/latest.jpg")
-@MediaBp.route("/<camera_name>/latest.webp")
-def latest_frame(camera_name):
+@router.get("/{camera_name}/latest.{extension}")
+def latest_frame(
+ request: Request,
+ camera_name: str,
+ extension: str, # jpg/jpeg/png/webp
+ bbox: Optional[int] = None,
+ timestamp: Optional[int] = None,
+ zones: Optional[int] = None,
+ mask: Optional[int] = None,
+ motion: Optional[int] = None,
+ regions: Optional[int] = None,
+ quality: Optional[int] = 70,
+ h: Optional[int] = None,
+):
  draw_options = {
- "bounding_boxes": request.args.get("bbox", type=int),
- "timestamp": request.args.get("timestamp", type=int),
- "zones": request.args.get("zones", type=int),
- "mask": request.args.get("mask", type=int),
- "motion_boxes": request.args.get("motion", type=int),
- "regions": request.args.get("regions", type=int),
+ "bounding_boxes": bbox,
+ "timestamp": timestamp,
+ "zones": zones,
+ "mask": mask,
+ "motion_boxes": motion,
+ "regions": regions,
  }
- resize_quality = request.args.get("quality", default=70, type=int)
- extension = os.path.splitext(request.path)[1][1:]
 
- if camera_name in current_app.frigate_config.cameras:
- frame = current_app.detected_frames_processor.get_current_frame(
+ if camera_name in request.app.frigate_config.cameras:
+ frame = request.app.detected_frames_processor.get_current_frame(
  camera_name, draw_options
  )
  retry_interval = float(
- current_app.frigate_config.cameras.get(camera_name).ffmpeg.retry_interval
+ request.app.frigate_config.cameras.get(camera_name).ffmpeg.retry_interval
  or 10
  )
 
  if frame is None or datetime.now().timestamp() > (
- current_app.detected_frames_processor.get_current_frame_time(camera_name)
+ request.app.detected_frames_processor.get_current_frame_time(camera_name)
  + retry_interval
  ):
- if current_app.camera_error_image is None:
+ if request.app.camera_error_image is None:
  error_image = glob.glob("/opt/frigate/frigate/images/camera-error.jpg")
 
  if len(error_image) > 0:
- current_app.camera_error_image = cv2.imread(
+ request.app.camera_error_image = cv2.imread(
  error_image[0], cv2.IMREAD_UNCHANGED
  )
 
- frame = current_app.camera_error_image
+ frame = request.app.camera_error_image
 
- height = int(request.args.get("h", str(frame.shape[0])))
+ height = int(h or str(frame.shape[0]))
  width = int(height * frame.shape[1] / frame.shape[0])
 
  if frame is None:
- return make_response(
- jsonify({"success": False, "message": "Unable to get valid frame"}),
- 500,
+ return JSONResponse(
+ content={"success": False, "message": "Unable to get valid frame"},
+ status_code=500,
  )
 
  if height < 1 or width < 1:
- return (
- "Invalid height / width requested :: {} / {}".format(height, width),
- 400,
+ return JSONResponse(
+ content="Invalid height / width requested :: {} / {}".format(
+ height, width
+ ),
+ status_code=400,
  )
 
  frame = cv2.resize(frame, dsize=(width, height), interpolation=cv2.INTER_AREA)
 
  ret, img = cv2.imencode(
- f".{extension}", frame, [int(cv2.IMWRITE_WEBP_QUALITY), resize_quality]
+ f".{extension}", frame, [int(cv2.IMWRITE_WEBP_QUALITY), quality]
  )
- response = make_response(img.tobytes())
- response.headers["Content-Type"] = f"image/{extension}"
- response.headers["Cache-Control"] = "no-store"
- return response
- elif camera_name == "birdseye" and current_app.frigate_config.birdseye.restream:
+ return StreamingResponse(
+ io.BytesIO(img.tobytes()),
+ media_type=f"image/{extension}",
+ headers={"Content-Type": f"image/{extension}", "Cache-Control": "no-store"},
+ )
+ elif camera_name == "birdseye" and request.app.frigate_config.birdseye.restream:
  frame = cv2.cvtColor(
- current_app.detected_frames_processor.get_current_frame(camera_name),
+ request.app.detected_frames_processor.get_current_frame(camera_name),
  cv2.COLOR_YUV2BGR_I420,
  )
 
- height = int(request.args.get("h", str(frame.shape[0])))
+ height = int(h or str(frame.shape[0]))
  width = int(height * frame.shape[1] / frame.shape[0])
 
  frame = cv2.resize(frame, dsize=(width, height), interpolation=cv2.INTER_AREA)
 
  ret, img = cv2.imencode(
- f".{extension}", frame, [int(cv2.IMWRITE_WEBP_QUALITY), resize_quality]
+ f".{extension}", frame, [int(cv2.IMWRITE_WEBP_QUALITY), quality]
+ )
+ return StreamingResponse(
+ io.BytesIO(img.tobytes()),
+ media_type=f"image/{extension}",
+ headers={"Content-Type": f"image/{extension}", "Cache-Control": "no-store"},
  )
- response = make_response(img.tobytes())
- response.headers["Content-Type"] = f"image/{extension}"
- response.headers["Cache-Control"] = "no-store"
- return response
  else:
- return make_response(
- jsonify({"success": False, "message": "Camera not found"}),
- 404,
+ return JSONResponse(
+ content={"success": False, "message": "Camera not found"},
+ status_code=404,
  )
 
 

frigate/api/new_app.py

@@ -0,0 +1,37 @@
+import logging
+
+from fastapi import FastAPI
+
+from frigate.api import app as main_app
+from frigate.api import media, preview
+
+logger = logging.getLogger(__name__)
+
+# https://fastapi.tiangolo.com/tutorial/metadata/#use-your-tags
+tags_metadata = [
+ {
+ "name": "Preview",
+ "description": "Preview routes",
+ },
+ {
+ "name": "Logs",
+ "description": "Logs routes",
+ },
+ {
+ "name": "Media",
+ "description": "Media routes",
+ },
+]
+
+
+def create_fastapi_app(frigate_config, detected_frames_processor):
+ logger.info("Starting FastAPI app")
+ app = FastAPI(debug=False, tags_metadata=tags_metadata)
+ app.include_router(preview.router)
+ app.include_router(media.router)
+ app.include_router(main_app.router)
+ app.frigate_config = frigate_config
+ app.detected_frames_processor = detected_frames_processor
+ app.camera_error_image = None
+
+ return app