import base64 from datetime import datetime, timedelta, timezone import copy import glob import logging import json import os import subprocess as sp import pytz import time import traceback from functools import reduce from pathlib import Path from tzlocal import get_localzone_name from urllib.parse import unquote import cv2 import numpy as np from flask import ( Blueprint, Flask, Response, current_app, jsonify, make_response, request, ) from peewee import SqliteDatabase, operator, fn, DoesNotExist from playhouse.shortcuts import model_to_dict from frigate.config import FrigateConfig from frigate.const import CLIPS_DIR, MAX_SEGMENT_DURATION, RECORD_DIR from frigate.models import Event, Recordings from frigate.object_processing import TrackedObject from frigate.stats import stats_snapshot from frigate.util import ( clean_camera_user_pass, ffprobe_stream, restart_frigate, vainfo_hwaccel, get_tz_modifiers, ) from frigate.storage import StorageMaintainer from frigate.version import VERSION logger = logging.getLogger(__name__) bp = Blueprint("frigate", __name__) def create_app( frigate_config, database: SqliteDatabase, stats_tracking, detected_frames_processor, storage_maintainer: StorageMaintainer, plus_api, ): app = Flask(__name__) @app.before_request def _db_connect(): if database.is_closed(): database.connect() @app.teardown_request def _db_close(exc): if not database.is_closed(): database.close() app.frigate_config = frigate_config app.stats_tracking = stats_tracking app.detected_frames_processor = detected_frames_processor app.storage_maintainer = storage_maintainer app.plus_api = plus_api app.camera_error_image = None app.hwaccel_errors = [] app.register_blueprint(bp) return app @bp.route("/") def is_healthy(): return "Frigate is running. Alive and healthy!" @bp.route("/events/summary") def events_summary(): tz_name = request.args.get("timezone", default="utc", type=str) hour_modifier, minute_modifier = get_tz_modifiers(tz_name) has_clip = request.args.get("has_clip", type=int) has_snapshot = request.args.get("has_snapshot", type=int) clauses = [] if not has_clip is None: clauses.append((Event.has_clip == has_clip)) if not has_snapshot is None: clauses.append((Event.has_snapshot == has_snapshot)) if len(clauses) == 0: clauses.append((True)) groups = ( Event.select( Event.camera, Event.label, fn.strftime( "%Y-%m-%d", fn.datetime( Event.start_time, "unixepoch", hour_modifier, minute_modifier ), ).alias("day"), Event.zones, fn.COUNT(Event.id).alias("count"), ) .where(reduce(operator.and_, clauses)) .group_by( Event.camera, Event.label, fn.strftime( "%Y-%m-%d", fn.datetime( Event.start_time, "unixepoch", hour_modifier, minute_modifier ), ), Event.zones, ) ) return jsonify([e for e in groups.dicts()]) @bp.route("/events/", methods=("GET",)) def event(id): try: return model_to_dict(Event.get(Event.id == id)) except DoesNotExist: return "Event not found", 404 @bp.route("/events//retain", methods=("POST",)) def set_retain(id): try: event = Event.get(Event.id == id) except DoesNotExist: return make_response( jsonify({"success": False, "message": "Event " + id + " not found"}), 404 ) event.retain_indefinitely = True event.save() return make_response( jsonify({"success": True, "message": "Event " + id + " retained"}), 200 ) @bp.route("/events//plus", methods=("POST",)) def send_to_plus(id): if not current_app.plus_api.is_active(): message = "PLUS_API_KEY environment variable is not set" logger.error(message) return make_response( jsonify( { "success": False, "message": message, } ), 400, ) try: event = Event.get(Event.id == id) except DoesNotExist: message = f"Event {id} not found" logger.error(message) return make_response(jsonify({"success": False, "message": message}), 404) if event.end_time is None: logger.error(f"Unable to load clean png for in-progress event: {event.id}") return make_response( jsonify( { "success": False, "message": "Unable to load clean png for in-progress event", } ), 400, ) if event.plus_id: message = "Already submitted to plus" logger.error(message) return make_response(jsonify({"success": False, "message": message}), 400) # load clean.png try: filename = f"{event.camera}-{event.id}-clean.png" image = cv2.imread(os.path.join(CLIPS_DIR, filename)) except Exception: logger.error(f"Unable to load clean png for event: {event.id}") return make_response( jsonify( {"success": False, "message": "Unable to load clean png for event"} ), 400, ) if image is None or image.size == 0: logger.error(f"Unable to load clean png for event: {event.id}") return make_response( jsonify( {"success": False, "message": "Unable to load clean png for event"} ), 400, ) try: plus_id = current_app.plus_api.upload_image(image, event.camera) except Exception as ex: logger.exception(ex) return make_response( jsonify({"success": False, "message": str(ex)}), 400, ) # store image id in the database event.plus_id = plus_id event.save() return make_response(jsonify({"success": True, "plus_id": plus_id}), 200) @bp.route("/events//retain", methods=("DELETE",)) def delete_retain(id): try: event = Event.get(Event.id == id) except DoesNotExist: return make_response( jsonify({"success": False, "message": "Event " + id + " not found"}), 404 ) event.retain_indefinitely = False event.save() return make_response( jsonify({"success": True, "message": "Event " + id + " un-retained"}), 200 ) @bp.route("/events//sub_label", methods=("POST",)) def set_sub_label(id): try: event: Event = Event.get(Event.id == id) except DoesNotExist: return make_response( jsonify({"success": False, "message": "Event " + id + " not found"}), 404 ) if request.json: new_sub_label = request.json.get("subLabel") else: new_sub_label = None if new_sub_label and len(new_sub_label) > 20: return make_response( jsonify( { "success": False, "message": new_sub_label + " exceeds the 20 character limit for sub_label", } ), 400, ) if not event.end_time: tracked_obj: TrackedObject = ( current_app.detected_frames_processor.camera_states[ event.camera ].tracked_objects.get(event.id) ) if tracked_obj: tracked_obj.obj_data["sub_label"] = new_sub_label event.sub_label = new_sub_label event.save() return make_response( jsonify( { "success": True, "message": "Event " + id + " sub label set to " + new_sub_label, } ), 200, ) @bp.route("/sub_labels") def get_sub_labels(): split_joined = request.args.get("split_joined", type=int) try: events = Event.select(Event.sub_label).distinct() except Exception as e: return jsonify( {"success": False, "message": f"Failed to get sub_labels: {e}"}, "404" ) sub_labels = [e.sub_label for e in events] if None in sub_labels: sub_labels.remove(None) if split_joined: original_labels = sub_labels.copy() for label in original_labels: if "," in label: sub_labels.remove(label) parts = label.split(",") for part in parts: if not (part.strip()) in sub_labels: sub_labels.append(part.strip()) sub_labels.sort() return jsonify(sub_labels) @bp.route("/events/", methods=("DELETE",)) def delete_event(id): try: event = Event.get(Event.id == id) except DoesNotExist: return make_response( jsonify({"success": False, "message": "Event " + id + " not found"}), 404 ) media_name = f"{event.camera}-{event.id}" if event.has_snapshot: media = Path(f"{os.path.join(CLIPS_DIR, media_name)}.jpg") media.unlink(missing_ok=True) media = Path(f"{os.path.join(CLIPS_DIR, media_name)}-clean.png") media.unlink(missing_ok=True) if event.has_clip: media = Path(f"{os.path.join(CLIPS_DIR, media_name)}.mp4") media.unlink(missing_ok=True) event.delete_instance() return make_response( jsonify({"success": True, "message": "Event " + id + " deleted"}), 200 ) @bp.route("/events//thumbnail.jpg") def event_thumbnail(id, max_cache_age=2592000): format = request.args.get("format", "ios") thumbnail_bytes = None event_complete = False try: event = Event.get(Event.id == id) if not event.end_time is None: event_complete = True thumbnail_bytes = base64.b64decode(event.thumbnail) except DoesNotExist: # see if the object is currently being tracked try: camera_states = current_app.detected_frames_processor.camera_states.values() for camera_state in camera_states: if id in camera_state.tracked_objects: tracked_obj = camera_state.tracked_objects.get(id) if not tracked_obj is None: thumbnail_bytes = tracked_obj.get_thumbnail() except: return "Event not found", 404 if thumbnail_bytes is None: return "Event not found", 404 # android notifications prefer a 2:1 ratio if format == "android": jpg_as_np = np.frombuffer(thumbnail_bytes, dtype=np.uint8) img = cv2.imdecode(jpg_as_np, flags=1) thumbnail = cv2.copyMakeBorder( img, 0, 0, int(img.shape[1] * 0.5), int(img.shape[1] * 0.5), cv2.BORDER_CONSTANT, (0, 0, 0), ) ret, jpg = cv2.imencode(".jpg", thumbnail, [int(cv2.IMWRITE_JPEG_QUALITY), 70]) thumbnail_bytes = jpg.tobytes() response = make_response(thumbnail_bytes) response.headers["Content-Type"] = "image/jpeg" if event_complete: response.headers["Cache-Control"] = f"private, max-age={max_cache_age}" else: response.headers["Cache-Control"] = "no-store" return response @bp.route("//