diff --git a/frigate/api/event.py b/frigate/api/event.py index fc7c58c375..605073a696 100644 --- a/frigate/api/event.py +++ b/frigate/api/event.py @@ -389,82 +389,106 @@ def events_explore( limit: int = 10, allowed_cameras: List[str] = Depends(get_allowed_cameras_for_filter), ): - # get distinct labels for all events - distinct_labels = ( - Event.select(Event.label) - .where(Event.camera << allowed_cameras) - .distinct() - .order_by(Event.label) + if not allowed_cameras: + return JSONResponse(content=[]) + + explore_columns = ( + Event.id, + Event.camera, + Event.label, + Event.sub_label, + Event.zones, + Event.start_time, + Event.end_time, + Event.has_clip, + Event.has_snapshot, + Event.plus_id, + Event.retain_indefinitely, + Event.top_score, + Event.false_positive, + Event.box, + Event.data, ) - label_counts = {} - - def event_generator(): - for label_obj in distinct_labels.iterator(): - label = label_obj.label - - # get most recent events for this label - label_events = ( - Event.select() - .where((Event.label == label) & (Event.camera << allowed_cameras)) - .order_by(Event.start_time.desc()) - .limit(limit) - .iterator() - ) - - # count total events for this label - label_counts[label] = ( - Event.select() - .where((Event.label == label) & (Event.camera << allowed_cameras)) - .count() - ) - - yield from label_events - - def process_events(): - for event in event_generator(): - processed_event = { - "id": event.id, - "camera": event.camera, - "label": event.label, - "zones": event.zones, - "start_time": event.start_time, - "end_time": event.end_time, - "has_clip": event.has_clip, - "has_snapshot": event.has_snapshot, - "plus_id": event.plus_id, - "retain_indefinitely": event.retain_indefinitely, - "sub_label": event.sub_label, - "top_score": event.top_score, - "false_positive": event.false_positive, - "box": event.box, - "data": { - k: v - for k, v in event.data.items() - if k - in [ - "type", - "score", - "top_score", - "description", - "sub_label_score", - "average_estimated_speed", - "velocity_angle", - "path_data", - "recognized_license_plate", - "recognized_license_plate_score", - ] - }, - "event_count": label_counts[event.label], - } - yield processed_event - - # convert iterator to list and sort - processed_events = sorted( - process_events(), - key=lambda x: (x["event_count"], x["start_time"]), - reverse=True, + # Single query: per-label COUNT and top-N ranking by start_time computed + # via window functions in a CTE, then filtered to rn <= limit + event_count = ( + fn.COUNT(Event.id).over(partition_by=[Event.label]).alias("event_count") ) + rn = ( + fn.ROW_NUMBER() + .over(partition_by=[Event.label], order_by=[Event.start_time.desc()]) + .alias("rn") + ) + + base_query = Event.select( + *explore_columns, + event_count, + rn, + ).where(Event.camera << allowed_cameras) + ranked = base_query.cte("ranked") + query = ( + Event.select( + ranked.c.id, + ranked.c.camera, + ranked.c.label, + ranked.c.sub_label, + ranked.c.zones, + ranked.c.start_time, + ranked.c.end_time, + ranked.c.has_clip, + ranked.c.has_snapshot, + ranked.c.plus_id, + ranked.c.retain_indefinitely, + ranked.c.top_score, + ranked.c.false_positive, + ranked.c.box, + ranked.c.data, + ranked.c.event_count, + ) + .from_(ranked) + .with_cte(ranked) + .where(ranked.c.rn <= limit) + .order_by(ranked.c.event_count.desc(), ranked.c.start_time.desc()) + .objects() + ) + + allowed_data_keys = { + "type", + "score", + "top_score", + "description", + "sub_label_score", + "average_estimated_speed", + "velocity_angle", + "path_data", + "recognized_license_plate", + "recognized_license_plate_score", + } + + processed_events = [ + { + "id": event.id, + "camera": event.camera, + "label": event.label, + "zones": event.zones, + "start_time": event.start_time, + "end_time": event.end_time, + "has_clip": event.has_clip, + "has_snapshot": event.has_snapshot, + "plus_id": event.plus_id, + "retain_indefinitely": event.retain_indefinitely, + "sub_label": event.sub_label, + "top_score": event.top_score, + "false_positive": event.false_positive, + "box": event.box, + "data": { + k: v for k, v in (event.data or {}).items() if k in allowed_data_keys + }, + "event_count": event.event_count, + } + for event in query + ] return JSONResponse(content=processed_events) @@ -487,22 +511,18 @@ async def event_ids(ids: str, request: Request): status_code=400, ) - for event_id in ids: - try: - event = Event.get(Event.id == event_id) - await require_camera_access(event.camera, request=request) - except DoesNotExist: - # we should not fail the entire request if an event is not found - continue - try: - events = Event.select().where(Event.id << ids).dicts().iterator() - return JSONResponse(list(events)) + events = list(Event.select().where(Event.id << ids).dicts().iterator()) except Exception: return JSONResponse( content=({"success": False, "message": "Events not found"}), status_code=400 ) + for event in events: + await require_camera_access(event["camera"], request=request) + + return JSONResponse(events) + @router.get( "/events/search", diff --git a/frigate/api/review.py b/frigate/api/review.py index ccf0be9d12..dbc89b41ca 100644 --- a/frigate/api/review.py +++ b/frigate/api/review.py @@ -10,7 +10,7 @@ import pandas as pd from fastapi import APIRouter, Request from fastapi.params import Depends from fastapi.responses import JSONResponse -from peewee import Case, DoesNotExist, IntegrityError, fn, operator +from peewee import Case, DoesNotExist, fn, operator from playhouse.shortcuts import model_to_dict from frigate.api.auth import ( @@ -172,11 +172,19 @@ async def review_ids(request: Request, ids: str): status_code=400, ) + try: + reviews = list( + ReviewSegment.select().where(ReviewSegment.id << ids).dicts().iterator() + ) + except Exception: + return JSONResponse( + content=({"success": False, "message": "Review segments not found"}), + status_code=400, + ) + + found_ids = {r["id"] for r in reviews} for review_id in ids: - try: - review = ReviewSegment.get(ReviewSegment.id == review_id) - await require_camera_access(review.camera, request=request) - except DoesNotExist: + if review_id not in found_ids: return JSONResponse( content=( {"success": False, "message": f"Review {review_id} not found"} @@ -184,16 +192,10 @@ async def review_ids(request: Request, ids: str): status_code=404, ) - try: - reviews = ( - ReviewSegment.select().where(ReviewSegment.id << ids).dicts().iterator() - ) - return JSONResponse(list(reviews)) - except Exception: - return JSONResponse( - content=({"success": False, "message": "Review segments not found"}), - status_code=400, - ) + for review in reviews: + await require_camera_access(review["camera"], request=request) + + return JSONResponse(reviews) @router.get( @@ -490,27 +492,52 @@ async def set_multiple_reviewed( user_id = current_user["username"] - for review_id in body.ids: - try: - review = ReviewSegment.get(ReviewSegment.id == review_id) - await require_camera_access(review.camera, request=request) - review_status = UserReviewStatus.get( - UserReviewStatus.user_id == user_id, - UserReviewStatus.review_segment == review_id, + reviews = list( + ReviewSegment.select(ReviewSegment.id, ReviewSegment.camera).where( + ReviewSegment.id << body.ids + ) + ) + + for review in reviews: + await require_camera_access(review.camera, request=request) + + found_ids = [r.id for r in reviews] + + if found_ids: + existing_statuses = list( + UserReviewStatus.select().where( + (UserReviewStatus.user_id == user_id) + & (UserReviewStatus.review_segment << found_ids) ) - # Update based on the reviewed parameter - if review_status.has_been_reviewed != body.reviewed: - review_status.has_been_reviewed = body.reviewed - review_status.save() - except DoesNotExist: - try: - UserReviewStatus.create( - user_id=user_id, - review_segment=ReviewSegment.get(id=review_id), - has_been_reviewed=body.reviewed, + ) + + status_by_review = {s.review_segment_id: s for s in existing_statuses} + + to_update = [] + to_create = [] + + for review_id in found_ids: + if review_id in status_by_review: + status = status_by_review[review_id] + if status.has_been_reviewed != body.reviewed: + status.has_been_reviewed = body.reviewed + to_update.append(status) + else: + to_create.append( + { + "user_id": user_id, + "review_segment_id": review_id, + "has_been_reviewed": body.reviewed, + } ) - except (DoesNotExist, IntegrityError): - pass + + if to_update: + UserReviewStatus.bulk_update( + to_update, fields=[UserReviewStatus.has_been_reviewed], batch_size=100 + ) + + if to_create: + UserReviewStatus.insert_many(to_create).execute() return JSONResponse( content=( diff --git a/migrations/036_add_perf_indexes.py b/migrations/036_add_perf_indexes.py new file mode 100644 index 0000000000..5354e5c09f --- /dev/null +++ b/migrations/036_add_perf_indexes.py @@ -0,0 +1,27 @@ +"""Peewee migrations -- 036_add_perf_indexes.py. + +Adds composite/single-column indexes to speed up the most common queries +issued by the web UI on initial page load: + +- event(camera, start_time DESC): /events list filtered by camera + time range +- reviewsegment(camera, start_time DESC): /api/review filtered by camera + time range +- reviewsegment(end_time): supports the end_time > after half of /api/review's range + +The existing event(label, start_time DESC) index from migration 027 already +covers /events/explore, so it is intentionally not duplicated here. +""" + +import peewee as pw + +SQL = pw.SQL + + +def migrate(migrator, database, fake=False, **kwargs): + migrator.sql( + 'CREATE INDEX IF NOT EXISTS "event_camera_start_time" ' + 'ON "event" ("camera", "start_time" DESC)' + ) + + +def rollback(migrator, database, fake=False, **kwargs): + migrator.sql('DROP INDEX IF EXISTS "event_camera_start_time"')