blakeblackshear.frigate/frigate/http.py

import base64
from collections import OrderedDict
from datetime import datetime, timedelta
import json
import glob
import logging
import os
import re
import time
from functools import reduce
from pathlib import Path

import cv2
import gevent
import numpy as np
from flask import (
    Blueprint,
    Flask,
    Response,
    current_app,
    jsonify,
    make_response,
    request,
)
from flask_sockets import Sockets
from peewee import SqliteDatabase, operator, fn, DoesNotExist, Value
from playhouse.shortcuts import model_to_dict

from frigate.const import CLIPS_DIR, RECORD_DIR
from frigate.models import Event, Recordings
from frigate.stats import stats_snapshot
from frigate.util import calculate_region
from frigate.version import VERSION

logger = logging.getLogger(__name__)

bp = Blueprint("frigate", __name__)
ws = Blueprint("ws", __name__)


class MqttBackend:
    """Interface for registering and updating WebSocket clients."""

    def __init__(self, mqtt_client, topic_prefix):
        self.clients = list()
        self.mqtt_client = mqtt_client
        self.topic_prefix = topic_prefix

    def register(self, client):
        """Register a WebSocket connection for Mqtt updates."""
        self.clients.append(client)

    def publish(self, message):
        try:
            json_message = json.loads(message)
            json_message = {
                "topic": f"{self.topic_prefix}/{json_message['topic']}",
                "payload": json_message["payload"],
                "retain": json_message.get("retain", False),
            }
        except:
            logger.warning("Unable to parse websocket message as valid json.")
            return

        logger.debug(
            f"Publishing mqtt message from websockets at {json_message['topic']}."
        )
        self.mqtt_client.publish(
            json_message["topic"],
            json_message["payload"],
            retain=json_message["retain"],
        )

    def run(self):
        def send(client, userdata, message):
            """Sends mqtt messages to clients."""
            try:
                logger.debug(f"Received mqtt message on {message.topic}.")
                ws_message = json.dumps(
                    {
                        "topic": message.topic.replace(f"{self.topic_prefix}/", ""),
                        "payload": message.payload.decode(),
                    }
                )
            except:
                # if the payload can't be decoded don't relay to clients
                logger.debug(
                    f"MQTT payload for {message.topic} wasn't text. Skipping..."
                )
                return

            for client in self.clients:
                try:
                    client.send(ws_message)
                except:
                    logger.debug(
                        "Removing websocket client due to a closed connection."
                    )
                    self.clients.remove(client)

        self.mqtt_client.message_callback_add(f"{self.topic_prefix}/#", send)

    def start(self):
        """Maintains mqtt subscription in the background."""
        gevent.spawn(self.run)


def create_app(
    frigate_config,
    database: SqliteDatabase,
    stats_tracking,
    detected_frames_processor,
    mqtt_client,
):
    app = Flask(__name__)
    sockets = Sockets(app)

    @app.before_request
    def _db_connect():
        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.register_blueprint(bp)
    sockets.register_blueprint(ws)

    app.mqtt_backend = MqttBackend(mqtt_client, frigate_config.mqtt.topic_prefix)
    app.mqtt_backend.start()

    return app


@bp.route("/")
def is_healthy():
    return "Frigate is running. Alive and healthy!"


@bp.route("/events/summary")
def events_summary():
    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((1 == 1))

    groups = (
        Event.select(
            Event.camera,
            Event.label,
            fn.strftime(
                "%Y-%m-%d", fn.datetime(Event.start_time, "unixepoch", "localtime")
            ).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", "localtime")
            ),
            Event.zones,
        )
    )

    return jsonify([e for e in groups.dicts()])


@bp.route("/events/<id>", 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/<id>", 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)
    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/<id>/thumbnail.jpg")
def event_thumbnail(id):
    format = request.args.get("format", "ios")
    thumbnail_bytes = None
    try:
        event = Event.get(Event.id == id)
        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/jpg"
    return response


@bp.route("/events/<id>/snapshot.jpg")
def event_snapshot(id):
    jpg_bytes = None
    try:
        event = Event.get(Event.id == id)
        if not event.has_snapshot:
            return "Snapshot not available", 404
        # read snapshot from disk
        with open(
            os.path.join(CLIPS_DIR, f"{event.camera}-{id}.jpg"), "rb"
        ) as image_file:
            jpg_bytes = image_file.read()
    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:
                        jpg_bytes = tracked_obj.get_jpg_bytes(
                            timestamp=request.args.get("timestamp", type=int),
                            bounding_box=request.args.get("bbox", type=int),
                            crop=request.args.get("crop", type=int),
                            height=request.args.get("h", type=int),
                        )
        except:
            return "Event not found", 404
    except:
        return "Event not found", 404

    response = make_response(jpg_bytes)
    response.headers["Content-Type"] = "image/jpg"
    return response


@bp.route("/events")
def events():
    limit = request.args.get("limit", 100)
    camera = request.args.get("camera")
    label = request.args.get("label")
    zone = request.args.get("zone")
    after = request.args.get("after", type=float)
    before = request.args.get("before", type=float)
    has_clip = request.args.get("has_clip", type=int)
    has_snapshot = request.args.get("has_snapshot", type=int)
    include_thumbnails = request.args.get("include_thumbnails", default=1, type=int)

    clauses = []
    excluded_fields = []

    if camera:
        clauses.append((Event.camera == camera))

    if label:
        clauses.append((Event.label == label))

    if zone:
        clauses.append((Event.zones.cast("text") % f'*"{zone}"*'))

    if after:
        clauses.append((Event.start_time >= after))

    if before:
        clauses.append((Event.start_time <= before))

    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 not include_thumbnails:
        excluded_fields.append(Event.thumbnail)

    if len(clauses) == 0:
        clauses.append((1 == 1))

    events = (
        Event.select()
        .where(reduce(operator.and_, clauses))
        .order_by(Event.start_time.desc())
        .limit(limit)
    )

    return jsonify([model_to_dict(e, exclude=excluded_fields) for e in events])


@bp.route("/config")
def config():
    return jsonify(current_app.frigate_config.to_dict())


@bp.route("/version")
def version():
    return VERSION


@bp.route("/stats")
def stats():
    stats = stats_snapshot(current_app.stats_tracking)
    return jsonify(stats)


@bp.route("/<camera_name>/<label>/best.jpg")
def best(camera_name, label):
    if camera_name in current_app.frigate_config.cameras:
        best_object = current_app.detected_frames_processor.get_best(camera_name, label)
        best_frame = best_object.get("frame")
        if best_frame is None:
            best_frame = np.zeros((720, 1280, 3), np.uint8)
        else:
            best_frame = cv2.cvtColor(best_frame, cv2.COLOR_YUV2BGR_I420)

        crop = bool(request.args.get("crop", 0, type=int))
        if crop:
            box = best_object.get("box", (0, 0, 300, 300))
            region = calculate_region(
                best_frame.shape, box[0], box[1], box[2], box[3], 1.1
            )
            best_frame = best_frame[region[1] : region[3], region[0] : region[2]]

        height = int(request.args.get("h", str(best_frame.shape[0])))
        width = int(height * best_frame.shape[1] / best_frame.shape[0])

        best_frame = cv2.resize(
            best_frame, dsize=(width, height), interpolation=cv2.INTER_AREA
        )
        ret, jpg = cv2.imencode(".jpg", best_frame, [int(cv2.IMWRITE_JPEG_QUALITY), 70])
        response = make_response(jpg.tobytes())
        response.headers["Content-Type"] = "image/jpg"
        return response
    else:
        return "Camera named {} not found".format(camera_name), 404


@bp.route("/<camera_name>")
def mjpeg_feed(camera_name):
    fps = int(request.args.get("fps", "3"))
    height = int(request.args.get("h", "360"))
    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),
    }
    if camera_name in current_app.frigate_config.cameras:
        # return a multipart response
        return Response(
            imagestream(
                current_app.detected_frames_processor,
                camera_name,
                fps,
                height,
                draw_options,
            ),
            mimetype="multipart/x-mixed-replace; boundary=frame",
        )
    else:
        return "Camera named {} not found".format(camera_name), 404


@bp.route("/<camera_name>/latest.jpg")
def latest_frame(camera_name):
    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),
    }
    if camera_name in current_app.frigate_config.cameras:
        # max out at specified FPS
        frame = current_app.detected_frames_processor.get_current_frame(
            camera_name, draw_options
        )
        if frame is None:
            frame = np.zeros((720, 1280, 3), np.uint8)

        height = int(request.args.get("h", 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, jpg = cv2.imencode(".jpg", frame, [int(cv2.IMWRITE_JPEG_QUALITY), 70])
        response = make_response(jpg.tobytes())
        response.headers["Content-Type"] = "image/jpg"
        return response
    else:
        return "Camera named {} not found".format(camera_name), 404


@bp.route("/<camera_name>/recordings")
def recordings(camera_name):
    dates = OrderedDict()

    # Retrieve all recordings for this camera
    recordings = (
        Recordings.select()
        .where(Recordings.camera == camera_name)
        .order_by(Recordings.start_time.asc())
    )

    last_end = 0
    recording: Recordings
    for recording in recordings:
        date = datetime.fromtimestamp(recording.start_time)
        key = date.strftime("%Y-%m-%d")
        hour = date.strftime("%H")

        # Create Day Record
        if key not in dates:
            dates[key] = OrderedDict()

        # Create Hour Record
        if hour not in dates[key]:
            dates[key][hour] = {"delay": {}, "events": []}

        # Check for delay
        the_hour = datetime.strptime(f"{key} {hour}", "%Y-%m-%d %H").timestamp()
        # diff current recording start time and the greater of the previous end time or top of the hour
        diff = recording.start_time - max(last_end, the_hour)
        # Determine seconds into recording
        seconds = 0
        if datetime.fromtimestamp(last_end).strftime("%H") == hour:
            seconds = int(last_end - the_hour)
        # Determine the delay
        delay = min(int(diff), 3600 - seconds)
        if delay > 1:
            # Add an offset for any delay greater than a second
            dates[key][hour]["delay"][seconds] = delay

        last_end = recording.end_time

    # Packing intervals to return all events with same label and overlapping times as one row.
    # See: https://blogs.solidq.com/en/sqlserver/packing-intervals/
    events = Event.raw(
        """WITH C1 AS
        (
        SELECT id, label, camera, top_score, start_time AS ts, +1 AS type, 1 AS sub
        FROM event
        WHERE camera = ?
        UNION ALL
        SELECT id, label, camera, top_score, end_time + 15 AS ts, -1 AS type, 0 AS sub
        FROM event
        WHERE camera = ?
        ),
        C2 AS
        (
        SELECT C1.*,
        SUM(type) OVER(PARTITION BY label ORDER BY ts, type DESC
        ROWS BETWEEN UNBOUNDED PRECEDING
        AND CURRENT ROW) - sub AS cnt
        FROM C1
        ),
        C3 AS
        (
        SELECT id, label, camera, top_score, ts,
        (ROW_NUMBER() OVER(PARTITION BY label ORDER BY ts) - 1) / 2 + 1
        AS grpnum
        FROM C2
        WHERE cnt = 0
        )
        SELECT MIN(id) as id, label, camera, MAX(top_score) as top_score, MIN(ts) AS start_time, max(ts) AS end_time
        FROM C3
        GROUP BY label, grpnum
        ORDER BY start_time;""",
        camera_name,
        camera_name,
    )

    event: Event
    for event in events:
        date = datetime.fromtimestamp(event.start_time)
        key = date.strftime("%Y-%m-%d")
        hour = date.strftime("%H")
        if key in dates and hour in dates[key]:
            dates[key][hour]["events"].append(
                model_to_dict(
                    event,
                    exclude=[
                        Event.false_positive,
                        Event.zones,
                        Event.thumbnail,
                        Event.has_clip,
                        Event.has_snapshot,
                    ],
                )
            )

    return jsonify(
        [
            {
                "date": date,
                "events": sum([len(value["events"]) for value in hours.values()]),
                "recordings": [
                    {"hour": hour, "delay": value["delay"], "events": value["events"]}
                    for hour, value in hours.items()
                ],
            }
            for date, hours in dates.items()
        ]
    )


@bp.route("/vod/<year_month>/<day>/<hour>/<camera>")
def vod(year_month, day, hour, camera):
    start_date = datetime.strptime(f"{year_month}-{day} {hour}", "%Y-%m-%d %H")
    end_date = start_date + timedelta(hours=1)
    start_ts = start_date.timestamp()
    end_ts = end_date.timestamp()

    # Select all recordings where either the start or end dates fall in the requested hour
    recordings = (
        Recordings.select()
        .where(
            (Recordings.start_time.between(start_ts, end_ts))
            | (Recordings.end_time.between(start_ts, end_ts))
        )
        .where(Recordings.camera == camera)
        .order_by(Recordings.start_time.asc())
    )

    clips = []
    durations = []

    recording: Recordings
    for recording in recordings:
        clip = {"type": "source", "path": recording.path}
        duration = int(recording.duration * 1000)
        # Determine if offset is needed for first clip
        if recording.start_time < start_ts:
            offset = int((start_ts - recording.start_time) * 1000)
            clip["clipFrom"] = offset
            duration -= offset
        # Determine if we need to end the last clip early
        if recording.end_time > end_ts:
            duration -= int((recording.end_time - end_ts) * 1000)
        clips.append(clip)
        durations.append(duration)

    return jsonify(
        {
            "cache": datetime.now() - timedelta(hours=1) > start_date,
            "discontinuity": False,
            "durations": durations,
            "sequences": [{"clips": clips}],
        }
    )


def imagestream(detected_frames_processor, camera_name, fps, height, draw_options):
    while True:
        # max out at specified FPS
        gevent.sleep(1 / fps)
        frame = detected_frames_processor.get_current_frame(camera_name, draw_options)
        if frame is None:
            frame = np.zeros((height, int(height * 16 / 9), 3), np.uint8)

        width = int(height * frame.shape[1] / frame.shape[0])
        frame = cv2.resize(frame, dsize=(width, height), interpolation=cv2.INTER_LINEAR)

        ret, jpg = cv2.imencode(".jpg", frame, [int(cv2.IMWRITE_JPEG_QUALITY), 70])
        yield (
            b"--frame\r\n"
            b"Content-Type: image/jpeg\r\n\r\n" + jpg.tobytes() + b"\r\n\r\n"
        )


@ws.route("/ws")
def echo_socket(socket):
    current_app.mqtt_backend.register(socket)

    while not socket.closed:
        # Sleep to prevent *constant* context-switches.
        gevent.sleep(0.1)

        message = socket.receive()
        if message:
            current_app.mqtt_backend.publish(message)