tweaks

- Read `ttl` from the global notifications config and hide it on the camera form
- Show it under Advanced Settings on the Notifications page
- Remove extra padding that misaligned the Notifications cards with the rest of the form
- Shorten the ttl warning

.devcontainer/devcontainer-lock.json

@@ -0,0 +1,9 @@
+{
+  "features": {
+    "ghcr.io/devcontainers/features/common-utils:2": {
+      "version": "2.7.1",
+      "resolved": "ghcr.io/devcontainers/features/common-utils@sha256:e76a6a8a35d1ad824daad1540725e0ec9f8ed2dc00d2850e990a9830013c7d08",
+      "integrity": "sha256:e76a6a8a35d1ad824daad1540725e0ec9f8ed2dc00d2850e990a9830013c7d08"
+    }
+  }
+}

.github/workflows/pull_request.yml

@@ -125,4 +125,4 @@ jobs:
       - name: Check API spec is up to date
         run: devcontainer exec --workspace-folder . bash -lc "python3 generate_api_auth_spec.py --check"
       - name: Run unit tests in devcontainer
-        run: devcontainer exec --workspace-folder . bash -lc "python3 -u -m unittest"
+        run: devcontainer exec --workspace-folder . bash -lc "python3 -u -m frigate.test"

AGENTS.md

@@ -36,7 +36,7 @@ When reviewing code, do NOT comment on:
 - **Formatting**: Ruff (configured in `pyproject.toml`)
 - **Linting**: Ruff with rules defined in project config
 - **Type Checking**: Use type hints consistently
-- **Testing**: unittest framework - use `python3 -u -m unittest` to run tests
+- **Testing**: unittest framework - use `python3 -u -m frigate.test` to run the suite in parallel groups
 - **Language**: American English for all code, comments, and documentation
 - **Punctuation**: Do not use em dashes in documentation, comments, or strings; reword with standard punctuation (commas, colons, parentheses, or separate sentences)
 
@@ -196,7 +196,7 @@ Key rules enforced:
 ### Backend Testing
 
 - **Framework**: Python unittest
-- **Run Command**: `python3 -u -m unittest`
+- **Run Command**: `python3 -u -m frigate.test` (parallel groups; plain `python3 -u -m unittest` runs serially)
 - **Location**: `frigate/test/`
 - **Coverage**: Aim for comprehensive test coverage of core functionality
 - **Pattern**: Use `TestCase` classes with descriptive test method names
@@ -219,8 +219,8 @@ Key rules enforced:
 ### Python Backend
 
 ```bash
-# Run all tests
-python3 -u -m unittest
+# Run all tests in parallel groups (-j to set the number of jobs)
+python3 -u -m frigate.test
 
 # Run specific test file
 python3 -u -m unittest frigate.test.test_ffmpeg_presets

CONTRIBUTING.md

@@ -64,7 +64,7 @@ Pull requests that appear to be unreviewed AI output will be closed without revi
 - **Python** — use modern language features (type hints, pattern matching, f-strings, dataclasses)
 - **Formatting**: Ruff (configured in `pyproject.toml`)
 - **Linting**: Ruff
-- **Testing**: `python3 -u -m unittest`
+- **Testing**: `python3 -u -m frigate.test` (runs the suite in parallel groups)
 - **Logging**: Use module-level `logger = logging.getLogger(__name__)` with lazy formatting
 - **Async**: All external I/O must be async. No blocking calls in async functions.
 - **Error handling**: Use specific exception types. Keep try blocks minimal.
@@ -82,7 +82,7 @@ Pull requests that appear to be unreviewed AI output will be closed without revi
 
 ```bash
 # Python
-python3 -u -m unittest                              # Run all tests
+python3 -u -m frigate.test                           # Run all tests in parallel groups
 python3 -u -m unittest frigate.test.test_ffmpeg_presets  # Run specific test
 ruff format frigate                                  # Format
 ruff check frigate                                   # Lint

Makefile

@@ -54,7 +54,7 @@ run: local
 
 run_tests: local
 	docker run --rm --workdir=/opt/frigate --entrypoint= frigate:latest \
-		python3 -u -m unittest
+		python3 -u -m frigate.test
 	docker run --rm --workdir=/opt/frigate --entrypoint= frigate:latest \
 		python3 -u -m mypy --config-file frigate/mypy.ini frigate
 

docs/docs/configuration/advanced/reference.md

@@ -578,6 +578,9 @@ notifications:
   email: "admin@example.com"
   # Optional: Cooldown time for notifications in seconds (default: shown below)
   cooldown: 0
+  # Optional: Seconds the push service keeps retrying a notification sent while an alert is active (default: shown below)
+  # NOTE: A non-zero value can deliver late notifications that still link to the live view
+  ttl: 0
 
 # Optional: Record configuration
 # NOTE: Can be overridden at the camera level

docs/docs/configuration/genai/config.md

@@ -75,11 +75,12 @@ Running Generative AI models on CPU is not recommended, as high inference times
 
 You must use a vision-capable model with Frigate. The following models are recommended for local deployment of the `descriptions` and `chat` roles:
 
-| Model               | Review [frame mode](/configuration/genai/genai_review#frame-mode) | Notes                                                                                                                                                                |
-| ------------------- | --------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
-| `qwen3-vl`          | `frames`                                                              | Strong visual and situational understanding, enhanced ability to identify smaller objects and interactions with object. Follows a sequence of frames on its own.     |
-| `qwen3.6`/`qwen3.8` | `frames`                                                              | Strong situational understanding, but missing DeepStack from qwen3-vl leading to worse performance for identifying objects in people's hand and other small details. |
-| `gemma4`            | `annotated_frames`                                                    | Strong situational understanding, sometimes resorts to more vague terms like 'interacts' instead of assigning a specific action. Loses track of activity that repeats or reverses, so it benefits from annotated frames. |
+| Model               | Review [frame mode](/configuration/genai/genai_review#frame-mode) | Review [response style](/configuration/genai/genai_review#response-style) | Notes                                                                                                                                                                                                                    |
+| ------------------- | ----------------------------------------------------------------- | ------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
+| `qwen3-vl`          | `frames`                                                          | `natural`                                                                 | Strong visual and situational understanding, enhanced ability to identify smaller objects and interactions with object. Follows a sequence of frames on its own.                                                         |
+| `qwen3.6`/`qwen3.8` | `frames`                                                          | `natural`                                                                 | Strong situational understanding, but missing DeepStack from qwen3-vl leading to worse performance for identifying objects in people's hand and other small details.                                                     |
+| `gemma4`            | `annotated_frames`                                                | `default`                                                                 | Strong situational understanding, sometimes resorts to more vague terms like 'interacts' instead of assigning a specific action. Loses track of activity that repeats or reverses, so it benefits from annotated frames. |
+| `lfm3-vl`           | `annotated_frames`                                                | `concise`                                                                 | Very efficient with fast inference, making it a good choice for less powerful hardware or high volumes of review items.                                                                                                  |
 
 #### Embedding models
 
@@ -95,7 +96,7 @@ The `embeddings` role needs a different kind of model. Text queries are matched
 The `transcribe` role needs a model that accepts audio input. A text-only or vision-only model cannot serve this role. The following are recommended for local deployment of the `transcribe` role:
 
 | Model       | Notes                                                                                                                                                                                                      |
-| ----------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
+| ----------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
 | `qwen3-asr` | Dedicated speech recognition model covering 30 languages, and the better choice for transcription quality. It only transcribes, so it cannot be shared with the `descriptions` or `chat` roles.            |
 | `gemma4`    | General multimodal model that accepts audio as well as images, so one served model can cover `transcribe` alongside the other roles. Transcript quality is below `qwen3-asr`, particularly on noisy audio. |
 

docs/docs/configuration/notifications.md

@@ -43,7 +43,7 @@ Notifications will be prevented if either:
 <ConfigTabs>
 <TabItem value="ui">
 
-1. Navigate to <NavPath path="Settings > Notifications > Notifications" />.
+1. Navigate to <NavPath path="Settings > Notifications" />.
    - Set **Email** to your email address
    - Enable notifications for the desired cameras
 
@@ -108,6 +108,29 @@ Currently, only Chrome supports images in notifications. Safari and Firefox will
 
 Different platforms handle notifications differently, some settings changes may be required to get optimal notification delivery.
 
+### Notification TTL
+
+By default, Frigate sends notifications for an active alert with a TTL of `0`, so the push service either delivers them immediately or drops them. If your device is briefly unreachable (asleep, switching from Wi-Fi to cellular, or on a weak signal), those notifications are lost and you only receive the notification sent when the alert ends, which is always kept for up to an hour.
+
+Setting `ttl` tells the push service to keep retrying notifications sent while an alert is active for that many seconds. This makes delivery more reliable, but a notification delivered late will still say an object is detected, even if the object has already left. Keep the value small.
+
+<ConfigTabs>
+<TabItem value="ui">
+
+1. Navigate to <NavPath path="Settings > Notifications" />.
+   - Expand **Advanced Settings** and set **Notification TTL** to the number of seconds to keep retrying (e.g. `10`)
+
+</TabItem>
+<TabItem value="yaml">
+
+```yaml
+notifications:
+  ttl: 10 # retry an undelivered notification for up to 10 seconds
+```
+
+</TabItem>
+</ConfigTabs>
+
 ### Android
 
 Most Android phones have battery optimization settings. To get reliable Notification delivery the browser (Chrome, Firefox) should have battery optimizations disabled. If Frigate is running as a PWA then the Frigate app should have battery optimizations disabled as well.

docs/docs/development/contributing.md

@@ -94,7 +94,7 @@ After closing VS Code, you may still have containers running. To close everythin
 GitHub will execute unit tests on new PRs. You must ensure that all tests pass.
 
 ```shell
-python3 -u -m unittest
+python3 -u -m frigate.test
 ```
 
 #### FFMPEG Hardware Acceleration
@@ -207,7 +207,7 @@ npm run lint
 - Ensure the backend [unit tests](#unit-tests) pass. Your PR cannot be merged unless tests pass.
 
 ```shell
-python3 -u -m unittest
+python3 -u -m frigate.test
 ```
 
 - Ensure the end-to-end tests pass. They run in Playwright against a production build with mocked API data, so they don't need a running Frigate instance. Add or update tests in `web/e2e/specs/` when you change UI behavior.

frigate/api/auth.py

@@ -934,7 +934,8 @@ def profile(request: Request):
 )
 def logout(request: Request):
     auth_config: AuthConfig = request.app.frigate_config.auth
-    response = RedirectResponse("/login", status_code=303)
+    base_path = request.headers.get("x-ingress-path", "").rstrip("/")
+    response = RedirectResponse(f"{base_path}/login", status_code=303)
     response.delete_cookie(auth_config.cookie_name)
     return response
 

frigate/comms/webpush.py

@@ -623,7 +623,7 @@ class WebPushClient(Communicator):
 
         # if event is ongoing open to live view otherwise open to recordings view
         direct_url = f"/review?id={reviewId}" if ended else f"/#{camera}"
-        ttl = 3600 if ended else 0
+        ttl = 3600 if ended else self.config.notifications.ttl
 
         logger.debug(f"Sending push notification for {camera}, review ID {reviewId}")
 

frigate/config/camera/notification.py

@@ -24,6 +24,12 @@ class NotificationConfig(FrigateBaseModel):
         title="Cooldown period",
         description="Cooldown (seconds) between notifications to avoid spamming recipients.",
     )
+    ttl: int = Field(
+        default=0,
+        ge=0,
+        title="Notification TTL",
+        description="Seconds the push service keeps retrying a notification sent while an alert is active if your device can't be reached. 0 delivers immediately or drops it. The notification sent when an alert ends is always kept for up to an hour.",
+    )
     enabled_in_config: bool | None = Field(
         default=None,
         title="Original notifications state",

frigate/detectors/detector_utils.py

@@ -1,8 +1,13 @@
 import logging
+import math
 import os
 
+import cv2
 import numpy as np
 
+from frigate.detectors.detector_config import ModelConfig, ModelTypeEnum
+from frigate.util.model import xyxy_to_xywh_for_nms
+
 try:
     from tflite_runtime.interpreter import Interpreter, load_delegate
 except ModuleNotFoundError:
@@ -12,39 +17,324 @@ except ModuleNotFoundError:
 logger = logging.getLogger(__name__)
 
 
-def tflite_init(self, interpreter):
-    self.interpreter = interpreter
+class TFLiteDetectionModel:
+    """Object detection pre and post processing for a TFLite interpreter.
 
-    self.interpreter.allocate_tensors()
+    Every TFLite detector shares this, differing only in how it builds the
+    interpreter (on the CPU, or with the EdgeTPU or Teflon delegate).
+    """
 
-    self.tensor_input_details = self.interpreter.get_input_details()
-    self.tensor_output_details = self.interpreter.get_output_details()
+    supported_models = [
+        ModelTypeEnum.ssd,
+        ModelTypeEnum.yologeneric,
+    ]
 
+    def __init__(
+        self, interpreter: Interpreter, model_config: ModelConfig, detector_key: str
+    ):
+        """Prepare the interpreter and work out the model's output layout.
 
-def tflite_detect_raw(self, tensor_input):
-    self.interpreter.set_tensor(self.tensor_input_details[0]["index"], tensor_input)
-    self.interpreter.invoke()
+        Args:
+            interpreter: The TFLite interpreter, with any delegate loaded
+            model_config: The model being run
+            detector_key: The detector's type, for error messages
+        """
+        self.model_type = model_config.model_type
 
-    boxes = self.interpreter.tensor(self.tensor_output_details[0]["index"])()[0]
-    class_ids = self.interpreter.tensor(self.tensor_output_details[1]["index"])()[0]
-    scores = self.interpreter.tensor(self.tensor_output_details[2]["index"])()[0]
-    count = int(self.interpreter.tensor(self.tensor_output_details[3]["index"])()[0])
+        if self.model_type not in self.supported_models:
+            raise ValueError(
+                f"{self.model_type.value} is currently not supported for {detector_key}. See the docs for more info on supported models."
+            )
 
-    detections = np.zeros((20, 6), np.float32)
+        self.interpreter = interpreter
 
-    for i in range(count):
-        if scores[i] < 0.4 or i == 20:
-            break
-        detections[i] = [
-            class_ids[i],
-            float(scores[i]),
-            boxes[i][0],
-            boxes[i][1],
-            boxes[i][2],
-            boxes[i][3],
-        ]
+        try:
+            self.interpreter.allocate_tensors()
+        except RuntimeError as e:
+            # an EdgeTPU compiled model wraps its graph in a custom op that only
+            # the EdgeTPU delegate can run
+            if "edgetpu-custom-op" in str(e):
+                logger.error(
+                    "%s is compiled for the EdgeTPU and can only be run by an edgetpu detector",
+                    model_config.path,
+                )
+            raise
 
-    return detections
+        self.tensor_input_details = self.interpreter.get_input_details()
+        self.tensor_output_details = self.interpreter.get_output_details()
+        self.model_width = model_config.width
+        self.model_height = model_config.height
+
+        self.min_score = 0.4
+        self.max_detections = 20
+
+        self.model_requires_int8 = self.tensor_input_details[0]["dtype"] == np.int8
+
+        if self.model_type == ModelTypeEnum.yologeneric:
+            logger.debug("Using YOLO preprocessing/postprocessing")
+
+            if len(self.tensor_output_details) not in [2, 3]:
+                raise ValueError(
+                    f"Invalid count of output tensors in YOLO model. Found {len(self.tensor_output_details)}, expecting 2 or 3."
+                )
+
+            self.reg_max = 16  # = 64 dfl_channels // 4 # YOLO standard
+            self.min_logit_value = np.log(
+                self.min_score / (1 - self.min_score)
+            )  # for filtering
+            self._generate_anchors_and_strides()  # decode bounding box DFL
+            self.project = np.arange(
+                self.reg_max, dtype=np.float32
+            )  # for decoding bounding box DFL information
+
+            # Determine YOLO tensor indices and quantization scales for
+            # boxes and class_scores the tensor ordering and names are
+            # not reliable, so use tensor shape to detect which tensor
+            # holds boxes or class scores.
+            # The tensors have shapes (B, N, C)
+            # where N is the number of candidates (=2100 for 320x320)
+            # this may guess wrong if the number of classes is exactly 64
+            output_boxes_index = None
+            output_classes_index = None
+            for i, x in enumerate(self.tensor_output_details):
+                # the nominal index seems to start at 1 instead of 0
+                if len(x["shape"]) == 3 and x["shape"][2] == 64:
+                    output_boxes_index = i
+                elif len(x["shape"]) == 3 and x["shape"][2] > 1:
+                    # require the number of classes to be more than 1
+                    # to differentiate from (not used) max score tensor
+                    output_classes_index = i
+            if output_boxes_index is None or output_classes_index is None:
+                logger.warning("Unrecognized model output, unexpected tensor shapes.")
+                output_classes_index = (
+                    0
+                    if (output_boxes_index is None or output_classes_index == 1)
+                    else 1
+                )  # 0 is default guess
+                output_boxes_index = 1 if (output_boxes_index == 0) else 0
+
+            scores_details = self.tensor_output_details[output_classes_index]
+            boxes_details = self.tensor_output_details[output_boxes_index]
+
+            # post processing works on the raw quantized values, which a float
+            # model doesn't have
+            if np.issubdtype(scores_details["dtype"], np.floating) or np.issubdtype(
+                boxes_details["dtype"], np.floating
+            ):
+                raise ValueError(
+                    f"{model_config.path} has float outputs. YOLO models for TFLite must be int8 quantized."
+                )
+
+            self.scores_tensor_index = scores_details["index"]
+            self.scores_scale, self.scores_zero_point = scores_details["quantization"]
+            # calculate the quantized version of the min_score
+            self.min_score_quantized = int(
+                (self.min_logit_value / self.scores_scale) + self.scores_zero_point
+            )
+            self.logit_shift_to_positive_values = (
+                max(0, math.ceil((128 + self.scores_zero_point) * self.scores_scale))
+                + 1
+            )  # round up
+
+            self.boxes_tensor_index = boxes_details["index"]
+            self.boxes_scale, self.boxes_zero_point = boxes_details["quantization"]
+
+        elif self.model_type == ModelTypeEnum.ssd:
+            logger.debug("Using SSD preprocessing/postprocessing")
+
+            # SSD models end in TFLite_Detection_PostProcess, which outputs
+            # boxes, class ids, scores and count
+            if len(self.tensor_output_details) < 4:
+                raise ValueError(
+                    f"Invalid count of output tensors in SSD model. Found {len(self.tensor_output_details)}, expecting 4. YOLO models need a model_type of yolo-generic."
+                )
+
+    def _generate_anchors_and_strides(self):
+        # for decoding the bounding box DFL information into xy coordinates
+        all_anchors = []
+        all_strides = []
+        strides = (8, 16, 32)  # YOLO's small, medium, large detection heads
+
+        for stride in strides:
+            feat_h, feat_w = self.model_height // stride, self.model_width // stride
+
+            grid_y, grid_x = np.meshgrid(
+                np.arange(feat_h, dtype=np.float32),
+                np.arange(feat_w, dtype=np.float32),
+                indexing="ij",
+            )
+
+            grid_coords = np.stack((grid_x.flatten(), grid_y.flatten()), axis=1)
+            anchor_points = grid_coords + 0.5
+
+            all_anchors.append(anchor_points)
+            all_strides.append(np.full((feat_h * feat_w, 1), stride, dtype=np.float32))
+
+        self.anchors = np.concatenate(all_anchors, axis=0)
+        self.anchor_strides = np.concatenate(all_strides, axis=0)
+
+    def pre_process(self, tensor_input):
+        if self.model_requires_int8:
+            tensor_input = np.bitwise_xor(tensor_input, 128).view(
+                np.int8
+            )  # shift by -128
+        return tensor_input
+
+    def detect_raw(self, tensor_input):
+        """Run the model on a frame.
+
+        Args:
+            tensor_input: The frame, already resized to the model's input
+
+        Returns:
+            Up to 20 detections as [class_id, score, ymin, xmin, ymax, xmax],
+            padded with zeros
+        """
+        tensor_input = self.pre_process(tensor_input)
+
+        self.interpreter.set_tensor(self.tensor_input_details[0]["index"], tensor_input)
+        self.interpreter.invoke()
+
+        if self.model_type == ModelTypeEnum.yologeneric:
+            return self._post_process_yolo()
+
+        return self._post_process_ssd()
+
+    def _post_process_yolo(self):
+        # Multi-tensor YOLO model with (non-standard B(H*W)C output format).
+        # (the comments indicate the shape of tensors,
+        # using "2100" as the anchor count (for image size of 320x320),
+        # "NC" as number of classes,
+        # "N" as the count that survive after min-score filtering)
+        # TENSOR A) class scores (1, 2100, NC) with logit values
+        # TENSOR B) box coordinates (1, 2100, 64) encoded as dfl scores
+        # Recommend that the model clamp the logit values in tensor (A)
+        # to the range [-4,+4] to preserve precision from [2%,98%]
+        # and because NMS requires the min_score parameter to be >= 0
+
+        # don't dequantize scores data yet, wait until the low-confidence
+        # candidates are filtered out from the overall result set.
+        # This reduces the work and makes post-processing faster.
+        # this method works with raw quantized numbers when possible,
+        # which relies on the value of the scale factor to be >0.
+        # This speeds up max and argmax operations.
+        # Get max confidence for each detection and create the mask
+        detections = np.zeros(
+            (self.max_detections, 6), np.float32
+        )  # initialize zero results
+        scores_output_quantized = self.interpreter.get_tensor(self.scores_tensor_index)[
+            0
+        ]  # (2100, NC)
+        max_scores_quantized = np.max(scores_output_quantized, axis=1)  # (2100,)
+        mask = max_scores_quantized >= self.min_score_quantized  # (2100,)
+
+        if not np.any(mask):
+            return detections  # empty results
+
+        max_scores_filtered_shiftedpositive = (
+            (max_scores_quantized[mask] - self.scores_zero_point) * self.scores_scale
+        ) + self.logit_shift_to_positive_values  # (N,1) shifted logit values
+        scores_output_quantized_filtered = scores_output_quantized[mask]
+
+        # dequantize boxes. NMS needs them to be in float format
+        # remove candidates with probabilities < threshold
+        boxes_output_quantized_filtered = (
+            self.interpreter.get_tensor(self.boxes_tensor_index)[0]
+        )[mask]  # (N, 64)
+        boxes_output_filtered = (
+            boxes_output_quantized_filtered.astype(np.float32) - self.boxes_zero_point
+        ) * self.boxes_scale
+
+        # 2. Decode DFL to distances (ltrb)
+        dfl_distributions = boxes_output_filtered.reshape(
+            -1, 4, self.reg_max
+        )  # (N, 4, 16)
+
+        # Softmax over the 16 bins
+        dfl_max = np.max(dfl_distributions, axis=2, keepdims=True)
+        dfl_exp = np.exp(dfl_distributions - dfl_max)
+        dfl_probs = dfl_exp / np.sum(dfl_exp, axis=2, keepdims=True)  # (N, 4, 16)
+
+        # Weighted sum: (N, 4, 16) * (16,) -> (N, 4)
+        distances = np.einsum("pcr,r->pc", dfl_probs, self.project)
+
+        # Calculate box corners in pixel coordinates
+        anchors_filtered = self.anchors[mask]
+        anchor_strides_filtered = self.anchor_strides[mask]
+        x1y1 = (
+            anchors_filtered - distances[:, [0, 1]]
+        ) * anchor_strides_filtered  # (N, 2)
+        x2y2 = (
+            anchors_filtered + distances[:, [2, 3]]
+        ) * anchor_strides_filtered  # (N, 2)
+        boxes_filtered_decoded = np.concatenate((x1y1, x2y2), axis=-1)  # (N, 4)
+
+        # 9. Apply NMS. Use logit scores here to defer sigmoid()
+        # until after filtering out redundant boxes
+        # Shift the logit scores to be non-negative (required by cv2)
+        indices = cv2.dnn.NMSBoxes(
+            bboxes=xyxy_to_xywh_for_nms(boxes_filtered_decoded),
+            scores=max_scores_filtered_shiftedpositive,
+            score_threshold=(
+                self.min_logit_value + self.logit_shift_to_positive_values
+            ),
+            nms_threshold=0.4,  # should this be a model config setting?
+        )
+        num_detections = len(indices)
+        if num_detections == 0:
+            return detections  # empty results
+
+        nms_indices = np.array(indices, dtype=np.int32).ravel()  # or .flatten()
+        if num_detections > self.max_detections:
+            nms_indices = nms_indices[: self.max_detections]
+            num_detections = self.max_detections
+        kept_logits_quantized = scores_output_quantized_filtered[nms_indices]
+        class_ids_post_nms = np.argmax(kept_logits_quantized, axis=1)
+
+        # Extract the final boxes and scores using fancy indexing
+        final_boxes = boxes_filtered_decoded[nms_indices]
+        final_scores_logits = (
+            max_scores_filtered_shiftedpositive[nms_indices]
+            - self.logit_shift_to_positive_values
+        )  # Unshifted logits
+
+        # Detections array format: [class_id, score, ymin, xmin, ymax, xmax]
+        detections[:num_detections, 0] = class_ids_post_nms
+        detections[:num_detections, 1] = 1.0 / (
+            1.0 + np.exp(-final_scores_logits)
+        )  # sigmoid
+        detections[:num_detections, 2] = final_boxes[:, 1] / self.model_height
+        detections[:num_detections, 3] = final_boxes[:, 0] / self.model_width
+        detections[:num_detections, 4] = final_boxes[:, 3] / self.model_height
+        detections[:num_detections, 5] = final_boxes[:, 2] / self.model_width
+        return detections
+
+    def _post_process_ssd(self):
+        boxes = self.interpreter.tensor(self.tensor_output_details[0]["index"])()[0]
+        class_ids = self.interpreter.tensor(self.tensor_output_details[1]["index"])()[0]
+        scores = self.interpreter.tensor(self.tensor_output_details[2]["index"])()[0]
+        count = int(
+            self.interpreter.tensor(self.tensor_output_details[3]["index"])()[0]
+        )
+
+        detections = np.zeros((self.max_detections, 6), np.float32)
+
+        for i in range(count):
+            if scores[i] < self.min_score:
+                break
+            if i == self.max_detections:
+                logger.debug("Too many detections (%s)", count)
+                break
+            detections[i] = [
+                class_ids[i],
+                float(scores[i]),
+                boxes[i][0],
+                boxes[i][1],
+                boxes[i][2],
+                boxes[i][3],
+            ]
+
+        return detections
 
 
 def tflite_load_delegate_interpreter(

frigate/detectors/plugins/cpu_tfl.py

@@ -7,7 +7,7 @@ from frigate.detectors.detection_api import DetectionApi
 from frigate.detectors.detector_config import BaseDetectorConfig
 from frigate.log import suppress_stderr_during
 
-from ..detector_utils import tflite_detect_raw, tflite_init
+from ..detector_utils import TFLiteDetectionModel
 
 try:
     from tflite_runtime.interpreter import Interpreter
@@ -40,6 +40,7 @@ class CpuDetectorConfig(BaseDetectorConfig):
 
 class CpuTfl(DetectionApi):
     type_key = DETECTOR_KEY
+    supported_models = TFLiteDetectionModel.supported_models
 
     def __init__(self, detector_config: CpuDetectorConfig):
         # Suppress TFLite delegate creation messages that bypass Python logging
@@ -49,7 +50,9 @@ class CpuTfl(DetectionApi):
                 num_threads=detector_config.num_threads or 3,
             )
 
-        tflite_init(self, interpreter)
+        self.tflite_model = TFLiteDetectionModel(
+            interpreter, detector_config.model, DETECTOR_KEY
+        )
 
     def detect_raw(self, tensor_input):
-        return tflite_detect_raw(self, tensor_input)
+        return self.tflite_model.detect_raw(tensor_input)

frigate/detectors/plugins/edgetpu_tfl.py

@@ -1,15 +1,13 @@
 import logging
-import math
 import os
 from typing import ClassVar, Literal
 
-import cv2
-import numpy as np
 from pydantic import ConfigDict, Field
 
 from frigate.detectors.detection_api import DetectionApi
-from frigate.detectors.detector_config import BaseDetectorConfig, ModelTypeEnum
-from frigate.util.model import xyxy_to_xywh_for_nms
+from frigate.detectors.detector_config import BaseDetectorConfig
+
+from ..detector_utils import TFLiteDetectionModel
 
 try:
     from tflite_runtime.interpreter import Interpreter, load_delegate
@@ -41,10 +39,7 @@ class EdgeTpuDetectorConfig(BaseDetectorConfig):
 
 class EdgeTpuTfl(DetectionApi):
     type_key = DETECTOR_KEY
-    supported_models = [
-        ModelTypeEnum.ssd,
-        ModelTypeEnum.yologeneric,
-    ]
+    supported_models = TFLiteDetectionModel.supported_models
 
     def __init__(self, detector_config: EdgeTpuDetectorConfig):
         device_config = {}
@@ -60,7 +55,7 @@ class EdgeTpuTfl(DetectionApi):
             logger.info(f"Attempting to load TPU as {device_type}")
             edge_tpu_delegate = load_delegate("libedgetpu.so.1.0", device_config)
             logger.info("TPU found")
-            self.interpreter = Interpreter(
+            interpreter = Interpreter(
                 model_path=detector_config.model.path,
                 experimental_delegates=[edge_tpu_delegate],
             )
@@ -78,298 +73,9 @@ class EdgeTpuTfl(DetectionApi):
 
             raise
 
-        self.interpreter.allocate_tensors()
-
-        self.tensor_input_details = self.interpreter.get_input_details()
-        self.tensor_output_details = self.interpreter.get_output_details()
-        self.model_width = detector_config.model.width
-        self.model_height = detector_config.model.height
-
-        self.min_score = 0.4
-        self.max_detections = 20
-
-        self.model_type = detector_config.model.model_type
-        self.model_requires_int8 = self.tensor_input_details[0]["dtype"] == np.int8
-
-        if self.model_type == ModelTypeEnum.yologeneric:
-            logger.debug("Using YOLO preprocessing/postprocessing")
-
-            if len(self.tensor_output_details) not in [2, 3]:
-                logger.error(
-                    f"Invalid count of output tensors in YOLO model. Found {len(self.tensor_output_details)}, expecting 2 or 3."
-                )
-                raise
-
-            self.reg_max = 16  # = 64 dfl_channels // 4 # YOLO standard
-            self.min_logit_value = np.log(
-                self.min_score / (1 - self.min_score)
-            )  # for filtering
-            self._generate_anchors_and_strides()  # decode bounding box DFL
-            self.project = np.arange(
-                self.reg_max, dtype=np.float32
-            )  # for decoding bounding box DFL information
-
-            # Determine YOLO tensor indices and quantization scales for
-            # boxes and class_scores the tensor ordering and names are
-            # not reliable, so use tensor shape to detect which tensor
-            # holds boxes or class scores.
-            # The tensors have shapes (B, N, C)
-            # where N is the number of candidates (=2100 for 320x320)
-            # this may guess wrong if the number of classes is exactly 64
-            output_boxes_index = None
-            output_classes_index = None
-            for i, x in enumerate(self.tensor_output_details):
-                # the nominal index seems to start at 1 instead of 0
-                if len(x["shape"]) == 3 and x["shape"][2] == 64:
-                    output_boxes_index = i
-                elif len(x["shape"]) == 3 and x["shape"][2] > 1:
-                    # require the number of classes to be more than 1
-                    # to differentiate from (not used) max score tensor
-                    output_classes_index = i
-            if output_boxes_index is None or output_classes_index is None:
-                logger.warning("Unrecognized model output, unexpected tensor shapes.")
-                output_classes_index = (
-                    0
-                    if (output_boxes_index is None or output_classes_index == 1)
-                    else 1
-                )  # 0 is default guess
-                output_boxes_index = 1 if (output_boxes_index == 0) else 0
-
-            scores_details = self.tensor_output_details[output_classes_index]
-            self.scores_tensor_index = scores_details["index"]
-            self.scores_scale, self.scores_zero_point = scores_details["quantization"]
-            # calculate the quantized version of the min_score
-            self.min_score_quantized = int(
-                (self.min_logit_value / self.scores_scale) + self.scores_zero_point
-            )
-            self.logit_shift_to_positive_values = (
-                max(0, math.ceil((128 + self.scores_zero_point) * self.scores_scale))
-                + 1
-            )  # round up
-
-            boxes_details = self.tensor_output_details[output_boxes_index]
-            self.boxes_tensor_index = boxes_details["index"]
-            self.boxes_scale, self.boxes_zero_point = boxes_details["quantization"]
-
-        elif self.model_type == ModelTypeEnum.ssd:
-            logger.debug("Using SSD preprocessing/postprocessing")
-
-            # SSD model indices (4 outputs: boxes, class_ids, scores, count)
-            for x in self.tensor_output_details:
-                if len(x["shape"]) == 3:
-                    self.output_boxes_index = x["index"]
-                elif len(x["shape"]) == 1:
-                    self.output_count_index = x["index"]
-
-            self.output_class_ids_index = None
-            self.output_class_scores_index = None
-
-        else:
-            raise Exception(
-                f"{self.model_type} is currently not supported for edgetpu. See the docs for more info on supported models."
-            )
-
-    def _generate_anchors_and_strides(self):
-        # for decoding the bounding box DFL information into xy coordinates
-        all_anchors = []
-        all_strides = []
-        strides = (8, 16, 32)  # YOLO's small, medium, large detection heads
-
-        for stride in strides:
-            feat_h, feat_w = self.model_height // stride, self.model_width // stride
-
-            grid_y, grid_x = np.meshgrid(
-                np.arange(feat_h, dtype=np.float32),
-                np.arange(feat_w, dtype=np.float32),
-                indexing="ij",
-            )
-
-            grid_coords = np.stack((grid_x.flatten(), grid_y.flatten()), axis=1)
-            anchor_points = grid_coords + 0.5
-
-            all_anchors.append(anchor_points)
-            all_strides.append(np.full((feat_h * feat_w, 1), stride, dtype=np.float32))
-
-        self.anchors = np.concatenate(all_anchors, axis=0)
-        self.anchor_strides = np.concatenate(all_strides, axis=0)
-
-    def determine_indexes_for_non_yolo_models(self):
-        """Legacy method for SSD models."""
-        if (
-            self.output_class_ids_index is None
-            or self.output_class_scores_index is None
-        ):
-            for i in range(4):
-                index = self.tensor_output_details[i]["index"]
-                if (
-                    index != self.output_boxes_index
-                    and index != self.output_count_index
-                ):
-                    if (
-                        np.mod(np.float32(self.interpreter.tensor(index)()[0][0]), 1)
-                        == 0.0
-                    ):
-                        self.output_class_ids_index = index
-                    else:
-                        self.output_scores_index = index
-
-    def pre_process(self, tensor_input):
-        if self.model_requires_int8:
-            tensor_input = np.bitwise_xor(tensor_input, 128).view(
-                np.int8
-            )  # shift by -128
-        return tensor_input
+        self.tflite_model = TFLiteDetectionModel(
+            interpreter, detector_config.model, DETECTOR_KEY
+        )
 
     def detect_raw(self, tensor_input):
-        tensor_input = self.pre_process(tensor_input)
-
-        self.interpreter.set_tensor(self.tensor_input_details[0]["index"], tensor_input)
-        self.interpreter.invoke()
-
-        if self.model_type == ModelTypeEnum.yologeneric:
-            # Multi-tensor YOLO model with (non-standard B(H*W)C output format).
-            # (the comments indicate the shape of tensors,
-            # using "2100" as the anchor count (for image size of 320x320),
-            # "NC" as number of classes,
-            # "N" as the count that survive after min-score filtering)
-            # TENSOR A) class scores (1, 2100, NC) with logit values
-            # TENSOR B) box coordinates (1, 2100, 64) encoded as dfl scores
-            # Recommend that the model clamp the logit values in tensor (A)
-            # to the range [-4,+4] to preserve precision from [2%,98%]
-            # and because NMS requires the min_score parameter to be >= 0
-
-            # don't dequantize scores data yet, wait until the low-confidence
-            # candidates are filtered out from the overall result set.
-            # This reduces the work and makes post-processing faster.
-            # this method works with raw quantized numbers when possible,
-            # which relies on the value of the scale factor to be >0.
-            # This speeds up max and argmax operations.
-            # Get max confidence for each detection and create the mask
-            detections = np.zeros(
-                (self.max_detections, 6), np.float32
-            )  # initialize zero results
-            scores_output_quantized = self.interpreter.get_tensor(
-                self.scores_tensor_index
-            )[0]  # (2100, NC)
-            max_scores_quantized = np.max(scores_output_quantized, axis=1)  # (2100,)
-            mask = max_scores_quantized >= self.min_score_quantized  # (2100,)
-
-            if not np.any(mask):
-                return detections  # empty results
-
-            max_scores_filtered_shiftedpositive = (
-                (max_scores_quantized[mask] - self.scores_zero_point)
-                * self.scores_scale
-            ) + self.logit_shift_to_positive_values  # (N,1) shifted logit values
-            scores_output_quantized_filtered = scores_output_quantized[mask]
-
-            # dequantize boxes. NMS needs them to be in float format
-            # remove candidates with probabilities < threshold
-            boxes_output_quantized_filtered = (
-                self.interpreter.get_tensor(self.boxes_tensor_index)[0]
-            )[mask]  # (N, 64)
-            boxes_output_filtered = (
-                boxes_output_quantized_filtered.astype(np.float32)
-                - self.boxes_zero_point
-            ) * self.boxes_scale
-
-            # 2. Decode DFL to distances (ltrb)
-            dfl_distributions = boxes_output_filtered.reshape(
-                -1, 4, self.reg_max
-            )  # (N, 4, 16)
-
-            # Softmax over the 16 bins
-            dfl_max = np.max(dfl_distributions, axis=2, keepdims=True)
-            dfl_exp = np.exp(dfl_distributions - dfl_max)
-            dfl_probs = dfl_exp / np.sum(dfl_exp, axis=2, keepdims=True)  # (N, 4, 16)
-
-            # Weighted sum: (N, 4, 16) * (16,) -> (N, 4)
-            distances = np.einsum("pcr,r->pc", dfl_probs, self.project)
-
-            # Calculate box corners in pixel coordinates
-            anchors_filtered = self.anchors[mask]
-            anchor_strides_filtered = self.anchor_strides[mask]
-            x1y1 = (
-                anchors_filtered - distances[:, [0, 1]]
-            ) * anchor_strides_filtered  # (N, 2)
-            x2y2 = (
-                anchors_filtered + distances[:, [2, 3]]
-            ) * anchor_strides_filtered  # (N, 2)
-            boxes_filtered_decoded = np.concatenate((x1y1, x2y2), axis=-1)  # (N, 4)
-
-            # 9. Apply NMS. Use logit scores here to defer sigmoid()
-            # until after filtering out redundant boxes
-            # Shift the logit scores to be non-negative (required by cv2)
-            indices = cv2.dnn.NMSBoxes(
-                bboxes=xyxy_to_xywh_for_nms(boxes_filtered_decoded),
-                scores=max_scores_filtered_shiftedpositive,
-                score_threshold=(
-                    self.min_logit_value + self.logit_shift_to_positive_values
-                ),
-                nms_threshold=0.4,  # should this be a model config setting?
-            )
-            num_detections = len(indices)
-            if num_detections == 0:
-                return detections  # empty results
-
-            nms_indices = np.array(indices, dtype=np.int32).ravel()  # or .flatten()
-            if num_detections > self.max_detections:
-                nms_indices = nms_indices[: self.max_detections]
-                num_detections = self.max_detections
-            kept_logits_quantized = scores_output_quantized_filtered[nms_indices]
-            class_ids_post_nms = np.argmax(kept_logits_quantized, axis=1)
-
-            # Extract the final boxes and scores using fancy indexing
-            final_boxes = boxes_filtered_decoded[nms_indices]
-            final_scores_logits = (
-                max_scores_filtered_shiftedpositive[nms_indices]
-                - self.logit_shift_to_positive_values
-            )  # Unshifted logits
-
-            # Detections array format: [class_id, score, ymin, xmin, ymax, xmax]
-            detections[:num_detections, 0] = class_ids_post_nms
-            detections[:num_detections, 1] = 1.0 / (
-                1.0 + np.exp(-final_scores_logits)
-            )  # sigmoid
-            detections[:num_detections, 2] = final_boxes[:, 1] / self.model_height
-            detections[:num_detections, 3] = final_boxes[:, 0] / self.model_width
-            detections[:num_detections, 4] = final_boxes[:, 3] / self.model_height
-            detections[:num_detections, 5] = final_boxes[:, 2] / self.model_width
-            return detections
-
-        elif self.model_type == ModelTypeEnum.ssd:
-            self.determine_indexes_for_non_yolo_models()
-            boxes = self.interpreter.tensor(self.tensor_output_details[0]["index"])()[0]
-            class_ids = self.interpreter.tensor(
-                self.tensor_output_details[1]["index"]
-            )()[0]
-            scores = self.interpreter.tensor(self.tensor_output_details[2]["index"])()[
-                0
-            ]
-            count = int(
-                self.interpreter.tensor(self.tensor_output_details[3]["index"])()[0]
-            )
-
-            detections = np.zeros((self.max_detections, 6), np.float32)
-
-            for i in range(count):
-                if scores[i] < self.min_score:
-                    break
-                if i == self.max_detections:
-                    logger.debug(f"Too many detections ({count})!")
-                    break
-                detections[i] = [
-                    class_ids[i],
-                    float(scores[i]),
-                    boxes[i][0],
-                    boxes[i][1],
-                    boxes[i][2],
-                    boxes[i][3],
-                ]
-
-            return detections
-
-        else:
-            raise Exception(
-                f"{self.model_type} is currently not supported for edgetpu. See the docs for more info on supported models."
-            )
+        return self.tflite_model.detect_raw(tensor_input)

frigate/detectors/plugins/teflon_tfl.py

@@ -7,8 +7,7 @@ from frigate.detectors.detection_api import DetectionApi
 from frigate.detectors.detector_config import BaseDetectorConfig
 
 from ..detector_utils import (
-    tflite_detect_raw,
-    tflite_init,
+    TFLiteDetectionModel,
     tflite_load_delegate_interpreter,
 )
 
@@ -30,6 +29,7 @@ class TeflonDetectorConfig(BaseDetectorConfig):
 
 class TeflonTfl(DetectionApi):
     type_key = DETECTOR_KEY
+    supported_models = TFLiteDetectionModel.supported_models
 
     def __init__(self, detector_config: TeflonDetectorConfig):
         # Location in Debian's mesa-teflon-delegate
@@ -39,7 +39,9 @@ class TeflonTfl(DetectionApi):
         interpreter = tflite_load_delegate_interpreter(
             delegate_library, detector_config, device_config
         )
-        tflite_init(self, interpreter)
+        self.tflite_model = TFLiteDetectionModel(
+            interpreter, detector_config.model, DETECTOR_KEY
+        )
 
     def detect_raw(self, tensor_input):
-        return tflite_detect_raw(self, tensor_input)
+        return self.tflite_model.detect_raw(tensor_input)

frigate/events/audio.py

@@ -178,6 +178,7 @@ class AudioProcessor(FrigateProcess):
             [
                 CameraConfigUpdateEnum.add,
                 CameraConfigUpdateEnum.audio,
+                CameraConfigUpdateEnum.enabled,
                 CameraConfigUpdateEnum.ffmpeg,
                 CameraConfigUpdateEnum.remove,
             ],

frigate/events/cleanup.py

@@ -93,6 +93,7 @@ class EventCleanup(threading.Thread):
                     Event.start_time < expire_after,
                     Event.label == event.label,
                     Event.retain_indefinitely == False,
+                    Event.has_snapshot == True,
                 )
                 .namedtuples()
                 .iterator()
@@ -114,6 +115,7 @@ class EventCleanup(threading.Thread):
                 Event.start_time < expire_after,
                 Event.label == event.label,
                 Event.retain_indefinitely == False,
+                Event.has_snapshot == True,
             )
 
             events_to_update: list[str] = []
@@ -169,6 +171,7 @@ class EventCleanup(threading.Thread):
                         Event.start_time < expire_after,
                         Event.label == event.label,
                         Event.retain_indefinitely == False,
+                        Event.has_snapshot == True,
                     )
                     .namedtuples()
                     .iterator()

frigate/review/maintainer.py

@@ -100,6 +100,14 @@ class PendingReviewSegment:
             if zone not in self.zones:
                 self.zones.append(zone)
 
+    def add_manual_detection(self, info: dict[str, Any]) -> None:
+        """Add a manual event's label and sub label to the segment."""
+        if info.get("sub_label"):
+            self.detections[info["event_id"]] = f"{info['label']}-verified"
+            self.sub_labels[info["event_id"]] = info["sub_label"]
+        else:
+            self.detections[info["event_id"]] = info["label"]
+
     def update_frame(
         self,
         camera_config: CameraConfig,
@@ -483,8 +491,6 @@ class ReviewSegmentMaintainer(threading.Thread):
         alerts so manual events keep their historical severity.
         """
         review_config = self.config.cameras[camera].review
-        # label contains 'label: sub_label', only the label is categorized
-        label = label.split(": ")[0]
 
         if review_config.alerts.enabled and label in review_config.alerts.labels:
             return SeverityEnum.alert
@@ -894,9 +900,7 @@ class ReviewSegmentMaintainer(threading.Thread):
                             current_segment.last_detection_time = frame_time
                 elif topic == DetectionTypeEnum.api or topic == DetectionTypeEnum.lpr:
                     if manual_info["state"] == ManualEventState.complete:
-                        current_segment.detections[manual_info["event_id"]] = (
-                            manual_info["label"]
-                        )
+                        current_segment.add_manual_detection(manual_info)
                         if topic == DetectionTypeEnum.api:
                             severity = self.get_manual_event_severity(
                                 camera, manual_info["label"]
@@ -921,9 +925,7 @@ class ReviewSegmentMaintainer(threading.Thread):
                         self.indefinite_events[camera][manual_info["event_id"]] = (
                             manual_info["label"]
                         )
-                        current_segment.detections[manual_info["event_id"]] = (
-                            manual_info["label"]
-                        )
+                        current_segment.add_manual_detection(manual_info)
                         if topic == DetectionTypeEnum.api:
                             if (
                                 self.get_manual_event_severity(
@@ -1014,11 +1016,12 @@ class ReviewSegmentMaintainer(threading.Thread):
                             camera,
                             frame_time,
                             severity,
-                            {manual_info["event_id"]: manual_info["label"]},
+                            {},
                             {},
                             [],
                             set(),
                         )
+                        api_segment.add_manual_detection(manual_info)
                         self._activate_segment(api_segment)
 
                         if manual_info["state"] == ManualEventState.start:
@@ -1041,11 +1044,12 @@ class ReviewSegmentMaintainer(threading.Thread):
                             camera,
                             frame_time,
                             SeverityEnum.detection,
-                            {manual_info["event_id"]: manual_info["label"]},
+                            {},
                             {},
                             [],
                             set(),
                         )
+                        lpr_segment.add_manual_detection(manual_info)
                         self._activate_segment(lpr_segment)
 
                         if manual_info["state"] == ManualEventState.start:

frigate/test/__init__.py

@@ -0,0 +1,11 @@
+from frigate.config.config import stream_info_retriever
+
+
+def _unreachable_stream_info(ffmpeg, path: str) -> dict:
+    return {"width": 0, "height": 0, "fourcc": None}
+
+
+# Validating a camera without detect dimensions probes its stream. Test streams
+# are unreachable, so return the failed probe result instead of waiting ~12s per
+# stream for ffprobe to time out
+stream_info_retriever.get_stream_info = _unreachable_stream_info

frigate/test/__main__.py

@@ -0,0 +1,278 @@
+"""Run the unit test suite in parallel groups.
+
+Test modules are split into groups balanced by test count, similar to Home
+Assistant's split_tests.py, and each group runs in its own unittest process.
+
+Usage (from the repo root):
+    python3 -u -m frigate.test [-j JOBS] [--groups GROUPS] [--timeout SECONDS]
+"""
+
+import argparse
+import math
+import os
+import re
+import signal
+import subprocess
+import sys
+import tempfile
+import threading
+import time
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from dataclasses import dataclass, field
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+TEST_METHOD = re.compile(r"^\s+(?:async )?def test", re.MULTILINE)
+RAN_TESTS = re.compile(r"^Ran (\d+) tests? in", re.MULTILINE)
+
+# Each group pays for its own interpreter and imports, so smaller groups only
+# help when there are more cores than groups
+TESTS_PER_GROUP = 250
+
+# Each group runs in its own process group, which ctrl+c in the terminal does
+# not reach, so the runner passes the interrupt on to these
+interrupted = threading.Event()
+running_groups: set[int] = set()
+
+
+@dataclass
+class TestGroup:
+    index: int
+    modules: list[str] = field(default_factory=list)
+    weight: int = 0
+
+
+@dataclass
+class GroupResult:
+    group: TestGroup
+    returncode: int
+    output: str
+    duration: float
+
+    @property
+    def tests_run(self) -> int:
+        """Number of tests unittest reported running, 0 if it never finished."""
+        match = RAN_TESTS.search(self.output)
+        return int(match.group(1)) if match else 0
+
+
+def discover_modules() -> dict[str, int]:
+    """Find test modules the same way unittest discovery does.
+
+    Returns:
+        Module name mapped to the number of test methods it defines
+    """
+    modules: dict[str, int] = {}
+
+    for path in sorted((REPO_ROOT / "frigate").rglob("test*.py")):
+        relative = path.relative_to(REPO_ROOT)
+
+        # discovery only descends into packages
+        if not all(
+            (REPO_ROOT / parent / "__init__.py").exists()
+            for parent in relative.parents
+            if parent != Path(".")
+        ):
+            continue
+
+        name = ".".join(relative.with_suffix("").parts)
+        modules[name] = len(TEST_METHOD.findall(path.read_text()))
+
+    return modules
+
+
+def split_groups(modules: dict[str, int], count: int) -> list[TestGroup]:
+    """Split modules into groups with a similar number of tests.
+
+    Largest modules are placed first, each into the currently lightest group.
+    """
+    groups = [TestGroup(index) for index in range(1, count + 1)]
+
+    for name, weight in sorted(modules.items(), key=lambda m: m[1], reverse=True):
+        group = min(groups, key=lambda g: g.weight)
+        group.modules.append(name)
+        group.weight += weight
+
+    return [group for group in groups if group.modules]
+
+
+def signal_group(pgid: int, sig: signal.Signals) -> bool:
+    """Send a signal to every process in a process group.
+
+    Returns:
+        False if no process is left in the group
+    """
+    try:
+        os.killpg(pgid, sig)
+    except ProcessLookupError:
+        return False
+
+    return True
+
+
+def stop_group(pgid: int) -> None:
+    """Stop the processes a group's tests left running.
+
+    They get a moment to exit before being killed, so the multiprocessing
+    resource tracker, which ignores SIGTERM, can remove leaked shared memory
+    once the processes holding it have exited.
+    """
+    if signal_group(pgid, signal.SIGTERM):
+        time.sleep(2)
+        signal_group(pgid, signal.SIGKILL)
+
+
+def run_group(group: TestGroup, timeout: float) -> GroupResult:
+    """Run a group's modules in a unittest process of its own.
+
+    Args:
+        group: The modules to run
+        timeout: Seconds before the group is stopped
+
+    Returns:
+        The exit code and combined output of the unittest process
+    """
+    start = time.monotonic()
+
+    # a file instead of a pipe so a process left behind by a test can't keep
+    # the group from finishing by holding the pipe open
+    with tempfile.TemporaryFile("w+") as output:
+        proc = subprocess.Popen(
+            [sys.executable, "-u", "-X", "faulthandler", "-m", "unittest"]
+            + group.modules,
+            cwd=REPO_ROOT,
+            # a group in the background that read the terminal would be
+            # stopped, and a stopped process doesn't act on SIGABRT
+            stdin=subprocess.DEVNULL,
+            stdout=output,
+            stderr=subprocess.STDOUT,
+            # a process group of its own so the processes its tests start can
+            # be stopped with it
+            process_group=0,
+        )
+        running_groups.add(proc.pid)
+
+        # the runner may have been interrupted while this group was starting
+        if interrupted.is_set():
+            signal_group(proc.pid, signal.SIGINT)
+
+        timed_out = False
+
+        try:
+            proc.wait(timeout)
+        except subprocess.TimeoutExpired:
+            # faulthandler prints the stack of every thread on SIGABRT, which
+            # shows where the group is stuck
+            timed_out = True
+            proc.send_signal(signal.SIGABRT)
+            proc.wait()
+
+            # aborting skips the cleanup of the running test, which would
+            # have stopped the processes it started
+            stop_group(proc.pid)
+
+        running_groups.discard(proc.pid)
+        output.seek(0)
+        text = output.read()
+
+    if timed_out:
+        text += f"\nGroup timed out after {timeout:.0f}s\n"
+
+    return GroupResult(group, proc.returncode, text, time.monotonic() - start)
+
+
+def main() -> int:
+    """Split the tests into groups, run them in parallel, and report the results.
+
+    Returns:
+        The exit code, 1 if any group failed
+    """
+    cpus = os.cpu_count() or 1
+    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    parser.add_argument(
+        "-j",
+        "--jobs",
+        type=int,
+        default=cpus,
+        help=f"number of groups to run at once (default: {cpus})",
+    )
+    parser.add_argument(
+        "--groups",
+        type=int,
+        help=(
+            "number of groups to split the tests into (default: one per job, "
+            f"or more to keep groups under about {TESTS_PER_GROUP} tests)"
+        ),
+    )
+    parser.add_argument(
+        "--timeout",
+        type=float,
+        default=600,
+        help="seconds before a group is stopped and reported as failed",
+    )
+    parser.add_argument(
+        "-v",
+        "--verbose",
+        action="store_true",
+        help="print the output of passing groups as well as failing ones",
+    )
+    args = parser.parse_args()
+
+    modules = discover_modules()
+    group_count = args.groups or max(
+        args.jobs, math.ceil(sum(modules.values()) / TESTS_PER_GROUP)
+    )
+    groups = split_groups(modules, group_count)
+    print(f"Running {len(groups)} test groups with {args.jobs} jobs", flush=True)
+
+    start = time.monotonic()
+    failed: list[GroupResult] = []
+    total_tests = 0
+
+    with ThreadPoolExecutor(max_workers=args.jobs) as executor:
+        futures = [executor.submit(run_group, group, args.timeout) for group in groups]
+
+        try:
+            for future in as_completed(futures):
+                result = future.result()
+                total_tests += result.tests_run
+                status = "OK" if result.returncode == 0 else "FAILED"
+                print(
+                    f"Group {result.group.index}/{len(groups)}: {status}, "
+                    f"{result.tests_run} tests in {result.duration:.1f}s",
+                    flush=True,
+                )
+
+                if result.returncode != 0:
+                    failed.append(result)
+
+                if result.returncode != 0 or args.verbose:
+                    print(result.output, flush=True)
+        except KeyboardInterrupt:
+            interrupted.set()
+            executor.shutdown(wait=False, cancel_futures=True)
+
+            for pgid in list(running_groups):
+                signal_group(pgid, signal.SIGINT)
+
+            raise
+
+    print(f"\nRan {total_tests} tests in {time.monotonic() - start:.1f}s")
+
+    if failed:
+        print(f"{len(failed)} of {len(groups)} groups FAILED, rerun one with:")
+
+        for result in sorted(failed, key=lambda r: r.group.index):
+            print(
+                f"  Group {result.group.index}: python3 -u -m unittest "
+                f"{' '.join(result.group.modules)}"
+            )
+
+        return 1
+
+    print("OK")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

frigate/test/const.py

@@ -1,4 +1,12 @@
 """Constants for testing."""
 
-TEST_DB = "test.db"
-TEST_DB_CLEANUPS = ["test.db", "test.db-shm", "test.db-wal"]
+import os
+import tempfile
+
+# Unique per process so test groups can run in parallel without sharing a database
+TEST_DB = os.path.join(tempfile.gettempdir(), f"frigate_test_{os.getpid()}.db")
+TEST_DB_CLEANUPS = [TEST_DB, f"{TEST_DB}-shm", f"{TEST_DB}-wal"]
+
+# The database is thrown away after each test, so skip syncing every commit to
+# disk, which is slow when several test processes are writing at once
+TEST_DB_PRAGMAS = {"synchronous": "off"}

frigate/test/http_api/base_http_test.py

@@ -18,7 +18,7 @@ from frigate.debug_replay import DebugReplayManager
 from frigate.jobs.export import JobStatePublisher
 from frigate.models import Event, Recordings, ReviewSegment
 from frigate.review.types import SeverityEnum
-from frigate.test.const import TEST_DB, TEST_DB_CLEANUPS
+from frigate.test.const import TEST_DB, TEST_DB_CLEANUPS, TEST_DB_PRAGMAS
 
 
 class AuthTestClient(TestClient):
@@ -38,12 +38,12 @@ class AuthTestClient(TestClient):
 class BaseTestHttp(unittest.TestCase):
     def setUp(self, models):
         # setup clean database for each test run
-        migrate_db = SqliteDatabase("test.db")
+        migrate_db = SqliteDatabase(TEST_DB, pragmas=TEST_DB_PRAGMAS)
         del logging.getLogger("peewee_migrate").handlers[:]
         router = Router(migrate_db)
         router.run()
         migrate_db.close()
-        self.db = SqliteQueueDatabase(TEST_DB)
+        self.db = SqliteQueueDatabase(TEST_DB, pragmas=TEST_DB_PRAGMAS)
         self.db.bind(models)
 
         # The export job manager broadcasts via JobStatePublisher on

frigate/test/http_api/test_http_latest_frame.py

@@ -1,11 +1,12 @@
 import os
 import shutil
-from unittest.mock import MagicMock
+import tempfile
+from unittest.mock import MagicMock, patch
 
 import cv2
 import numpy as np
 
-from frigate.output.preview import PREVIEW_CACHE_DIR, PREVIEW_FRAME_TYPE
+from frigate.output.preview import PREVIEW_FRAME_TYPE
 from frigate.test.http_api.base_http_test import AuthTestClient, BaseTestHttp
 
 
@@ -15,14 +16,15 @@ class TestHttpLatestFrame(BaseTestHttp):
         self.app = super().create_app()
         self.app.detected_frames_processor = MagicMock()
 
-        if os.path.exists(PREVIEW_CACHE_DIR):
-            shutil.rmtree(PREVIEW_CACHE_DIR)
-        os.makedirs(PREVIEW_CACHE_DIR)
-
-    def tearDown(self):
-        if os.path.exists(PREVIEW_CACHE_DIR):
-            shutil.rmtree(PREVIEW_CACHE_DIR)
-        super().tearDown()
+        # a directory of its own so test groups running in parallel can't
+        # delete each other's preview frames
+        self.preview_dir = tempfile.mkdtemp()
+        self.addCleanup(shutil.rmtree, self.preview_dir, ignore_errors=True)
+        cache_dir_patch = patch(
+            "frigate.output.preview.PREVIEW_CACHE_DIR", self.preview_dir
+        )
+        cache_dir_patch.start()
+        self.addCleanup(cache_dir_patch.stop)
 
     def test_latest_frame_fallback_to_preview(self):
         camera = "front_door"
@@ -45,7 +47,7 @@ class TestHttpLatestFrame(BaseTestHttp):
             2,
         )
         preview_path = os.path.join(
-            PREVIEW_CACHE_DIR, f"preview_{camera}-1234567890.0.{PREVIEW_FRAME_TYPE}"
+            self.preview_dir, f"preview_{camera}-1234567890.0.{PREVIEW_FRAME_TYPE}"
         )
         cv2.imwrite(preview_path, dummy_frame)
 
@@ -80,7 +82,7 @@ class TestHttpLatestFrame(BaseTestHttp):
 
         # 2. Create a dummy preview file
         preview_path = os.path.join(
-            PREVIEW_CACHE_DIR, f"preview_{camera}-999.0.{PREVIEW_FRAME_TYPE}"
+            self.preview_dir, f"preview_{camera}-999.0.{PREVIEW_FRAME_TYPE}"
         )
         cv2.imwrite(preview_path, dummy_frame)
 

frigate/test/test_audio_spawn.py

@@ -1,4 +1,4 @@
-"""Tests for audio maintainer spawning on runtime camera adds."""
+"""Tests for audio maintainer spawning on runtime camera updates."""
 
 import threading
 import unittest
@@ -42,6 +42,42 @@ class TestAudioSpawnIfNeeded(unittest.TestCase):
         processor.logger = MagicMock()
         return processor
 
+    def test_initially_disabled_camera_spawns_after_enabled_update(self):
+        """Consume an enabled update through the manager's real subscriber."""
+        config = _build_config()
+        config.cameras["front_door"].enabled = False
+        stop = MagicMock()
+        stop.wait.side_effect = [False, True]
+        processor = AudioProcessor(
+            config, {"front_door": MagicMock()}, MagicMock(), stop
+        )
+        processor.logger = MagicMock()
+        updates = iter([("config/cameras/front_door/enabled", True), (None, None)])
+        self.addCleanup(
+            setattr, threading.current_thread(), "name", threading.current_thread().name
+        )
+
+        with (
+            patch.object(processor, "pre_run_setup"),
+            patch("frigate.config.camera.updater.ConfigSubscriber") as transport,
+            patch("frigate.events.audio.AudioEventMaintainer") as maintainer,
+        ):
+
+            def receive_update():
+                self.assertEqual(processor.audio_threads, {})
+                maintainer.assert_not_called()
+                return next(updates)
+
+            transport.return_value.check_for_update.side_effect = receive_update
+            maintainer.return_value.is_alive.return_value = False
+            processor.run()
+
+        self.assertEqual(
+            processor.audio_threads, {"front_door": maintainer.return_value}
+        )
+        maintainer.assert_called_once()
+        maintainer.return_value.start.assert_called_once()
+
     def test_skips_camera_whose_metrics_have_not_been_created_yet(self):
         """The camera maintainer creates metrics on its own poll of the add
         update, so the audio processor can see the camera first."""

frigate/test/test_event_cleanup_snapshots.py

@@ -0,0 +1,87 @@
+"""Tests for snapshot expiration in event cleanup."""
+
+import datetime
+import unittest
+from unittest.mock import MagicMock, patch
+
+from peewee import SqliteDatabase
+
+from frigate.config import FrigateConfig
+from frigate.events.cleanup import EventCleanup
+from frigate.models import Event, Timeline
+
+
+class TestEventCleanupSnapshots(unittest.TestCase):
+    def setUp(self):
+        self.db = SqliteDatabase(":memory:")
+        models = [Event, Timeline]
+        self.db.bind(models)
+        self.db.create_tables(models)
+
+        config = FrigateConfig(
+            **{
+                "mqtt": {"host": "mqtt"},
+                "snapshots": {"enabled": True, "retain": {"default": 14}},
+                "cameras": {
+                    "front_door": {
+                        "ffmpeg": {
+                            "inputs": [
+                                {
+                                    "path": "rtsp://10.0.0.1:554/video",
+                                    "roles": ["detect"],
+                                },
+                            ]
+                        },
+                    }
+                },
+            }
+        )
+        self.cleanup = EventCleanup(config, MagicMock(), MagicMock())
+
+    def tearDown(self):
+        self.db.close()
+
+    def _insert_event(self, id: str, camera: str, has_snapshot: bool) -> None:
+        end_time = (datetime.datetime.now() - datetime.timedelta(days=20)).timestamp()
+        Event.create(
+            id=id,
+            label="person",
+            camera=camera,
+            start_time=end_time - 10,
+            end_time=end_time,
+            top_score=0.9,
+            score=0.9,
+            false_positive=False,
+            zones=[],
+            thumbnail="",
+            has_clip=True,
+            has_snapshot=has_snapshot,
+            region=[],
+            box=[],
+            area=0,
+            retain_indefinitely=False,
+            plus_id="",
+            model_hash="",
+            detector_type="cpu",
+            model_type="ssd",
+            data={},
+        )
+
+    @patch("frigate.events.cleanup.delete_event_snapshot", return_value=True)
+    def test_expired_snapshots_are_only_deleted_once(self, delete_snapshot):
+        # events stay in the db while their clip is retained, so events whose
+        # snapshot already expired must not be selected again
+        for camera in ("front_door", "removed_camera"):
+            self._insert_event(f"{camera}-pending", camera, has_snapshot=True)
+            self._insert_event(f"{camera}-expired", camera, has_snapshot=False)
+
+        self.cleanup.expire_snapshots()
+
+        deleted = sorted(c.args[0].id for c in delete_snapshot.call_args_list)
+        assert deleted == ["front_door-pending", "removed_camera-pending"]
+        assert not Event.select().where(Event.has_snapshot == True).exists()
+
+        delete_snapshot.reset_mock()
+        self.cleanup.expire_snapshots()
+
+        delete_snapshot.assert_not_called()

frigate/test/test_export.py

@@ -617,9 +617,13 @@ class TestExportTimelineAlignment(unittest.TestCase):
         ):
             windows = exporter._staged_chapter_windows()
 
-        path = exporter._build_recording_segment_chapter_metadata_file(windows)
-        self.addCleanup(os.remove, path)
-        content = Path(path).read_text()
+        # the real cache dir only exists once the container has started
+        with tempfile.TemporaryDirectory() as tmpdir:
+            chapter_path = os.path.join(tmpdir, "chapters.txt")
+            exporter._chapter_metadata_path = lambda: chapter_path  # type: ignore[method-assign]
+            path = exporter._build_recording_segment_chapter_metadata_file(windows)
+            self.assertEqual(path, chapter_path)
+            content = Path(chapter_path).read_text()
 
         self.assertIn("START=0\nEND=21500", content)
         self.assertIn("START=21500\nEND=43000", content)

frigate/test/test_media_auth.py

@@ -21,7 +21,7 @@ from frigate.api.media_auth import (
 )
 from frigate.config import FrigateConfig
 from frigate.models import Event, Export, Recordings, ReviewSegment
-from frigate.test.const import TEST_DB, TEST_DB_CLEANUPS
+from frigate.test.const import TEST_DB, TEST_DB_CLEANUPS, TEST_DB_PRAGMAS
 
 _CONFIG = {
     "mqtt": {"host": "mqtt"},
@@ -222,11 +222,11 @@ class TestExportResolution(unittest.TestCase):
     """Export resolution requires a DB lookup."""
 
     def setUp(self):
-        migrate_db = SqliteDatabase("test.db")
+        migrate_db = SqliteDatabase(TEST_DB, pragmas=TEST_DB_PRAGMAS)
         del logging.getLogger("peewee_migrate").handlers[:]
         Router(migrate_db).run()
         migrate_db.close()
-        self.db = SqliteQueueDatabase(TEST_DB)
+        self.db = SqliteQueueDatabase(TEST_DB, pragmas=TEST_DB_PRAGMAS)
         self.db.bind([Event, ReviewSegment, Recordings, Export])
         self.config = FrigateConfig(**_CONFIG)
 

frigate/test/test_notice_registry.py

@@ -12,7 +12,7 @@ from playhouse.sqliteq import SqliteQueueDatabase
 from frigate.models import Notice, NoticeStats
 from frigate.notices.registry import NoticeRegistry
 from frigate.notices.types import NOTICE_KINDS, NoticeKind, NoticeSeverity, notice_id
-from frigate.test.const import TEST_DB, TEST_DB_CLEANUPS
+from frigate.test.const import TEST_DB, TEST_DB_CLEANUPS, TEST_DB_PRAGMAS
 
 # kinds that switch on the lifecycle knobs, so the tests do not depend on the
 # values the real catalog picks
@@ -49,12 +49,12 @@ class TestNoticeKinds(unittest.TestCase):
 
 class RegistryTestCase(unittest.TestCase):
     def setUp(self):
-        migrate_db = SqliteDatabase("test.db")
+        migrate_db = SqliteDatabase(TEST_DB, pragmas=TEST_DB_PRAGMAS)
         del logging.getLogger("peewee_migrate").handlers[:]
         router = Router(migrate_db)
         router.run()
         migrate_db.close()
-        self.db = SqliteQueueDatabase(TEST_DB)
+        self.db = SqliteQueueDatabase(TEST_DB, pragmas=TEST_DB_PRAGMAS)
         self.db.bind([Notice, NoticeStats])
         self.registry = NoticeRegistry()
         self.listener = MagicMock()

frigate/test/test_preview_loader.py

@@ -1,23 +1,23 @@
 import os
 import shutil
+import tempfile
 import unittest
+from unittest.mock import patch
 
-from frigate.output.preview import (
-    PREVIEW_CACHE_DIR,
-    PREVIEW_FRAME_TYPE,
-    get_most_recent_preview_frame,
-)
+from frigate.output.preview import PREVIEW_FRAME_TYPE, get_most_recent_preview_frame
 
 
 class TestPreviewLoader(unittest.TestCase):
     def setUp(self):
-        if os.path.exists(PREVIEW_CACHE_DIR):
-            shutil.rmtree(PREVIEW_CACHE_DIR)
-        os.makedirs(PREVIEW_CACHE_DIR)
-
-    def tearDown(self):
-        if os.path.exists(PREVIEW_CACHE_DIR):
-            shutil.rmtree(PREVIEW_CACHE_DIR)
+        # a directory of its own so test groups running in parallel can't
+        # delete each other's preview frames
+        self.preview_dir = tempfile.mkdtemp()
+        self.addCleanup(shutil.rmtree, self.preview_dir, ignore_errors=True)
+        cache_dir_patch = patch(
+            "frigate.output.preview.PREVIEW_CACHE_DIR", self.preview_dir
+        )
+        cache_dir_patch.start()
+        self.addCleanup(cache_dir_patch.stop)
 
     def test_get_most_recent_preview_frame_missing(self):
         self.assertIsNone(get_most_recent_preview_frame("test_camera"))
@@ -28,14 +28,14 @@ class TestPreviewLoader(unittest.TestCase):
         for ts in ["1000.0", "2000.0", "1500.0"]:
             with open(
                 os.path.join(
-                    PREVIEW_CACHE_DIR, f"preview_{camera}-{ts}.{PREVIEW_FRAME_TYPE}"
+                    self.preview_dir, f"preview_{camera}-{ts}.{PREVIEW_FRAME_TYPE}"
                 ),
                 "w",
             ) as f:
                 f.write(f"test_{ts}")
 
         expected_path = os.path.join(
-            PREVIEW_CACHE_DIR, f"preview_{camera}-2000.0.{PREVIEW_FRAME_TYPE}"
+            self.preview_dir, f"preview_{camera}-2000.0.{PREVIEW_FRAME_TYPE}"
         )
         self.assertEqual(get_most_recent_preview_frame(camera), expected_path)
 
@@ -45,7 +45,7 @@ class TestPreviewLoader(unittest.TestCase):
         for ts in ["1000.0", "2000.0"]:
             with open(
                 os.path.join(
-                    PREVIEW_CACHE_DIR, f"preview_{camera}-{ts}.{PREVIEW_FRAME_TYPE}"
+                    self.preview_dir, f"preview_{camera}-{ts}.{PREVIEW_FRAME_TYPE}"
                 ),
                 "w",
             ) as f:
@@ -53,7 +53,7 @@ class TestPreviewLoader(unittest.TestCase):
 
         # Test finding frame before or at 1500
         expected_path = os.path.join(
-            PREVIEW_CACHE_DIR, f"preview_{camera}-1000.0.{PREVIEW_FRAME_TYPE}"
+            self.preview_dir, f"preview_{camera}-1000.0.{PREVIEW_FRAME_TYPE}"
         )
         self.assertEqual(
             get_most_recent_preview_frame(camera, before=1500.0), expected_path
@@ -67,7 +67,7 @@ class TestPreviewLoader(unittest.TestCase):
         other_camera = "other_camera"
         with open(
             os.path.join(
-                PREVIEW_CACHE_DIR, f"preview_{other_camera}-3000.0.{PREVIEW_FRAME_TYPE}"
+                self.preview_dir, f"preview_{other_camera}-3000.0.{PREVIEW_FRAME_TYPE}"
             ),
             "w",
         ) as f:
@@ -78,12 +78,12 @@ class TestPreviewLoader(unittest.TestCase):
     def test_get_most_recent_preview_frame_hyphenated_camera(self):
         for name in ("preview_front-2000.0", "preview_front-door-3000.0"):
             with open(
-                os.path.join(PREVIEW_CACHE_DIR, f"{name}.{PREVIEW_FRAME_TYPE}"), "w"
+                os.path.join(self.preview_dir, f"{name}.{PREVIEW_FRAME_TYPE}"), "w"
             ) as f:
                 f.write("test")
 
         expected_path = os.path.join(
-            PREVIEW_CACHE_DIR, f"preview_front-2000.0.{PREVIEW_FRAME_TYPE}"
+            self.preview_dir, f"preview_front-2000.0.{PREVIEW_FRAME_TYPE}"
         )
         self.assertEqual(get_most_recent_preview_frame("front"), expected_path)
         self.assertEqual(
@@ -91,5 +91,5 @@ class TestPreviewLoader(unittest.TestCase):
         )
 
     def test_get_most_recent_preview_frame_no_directory(self):
-        shutil.rmtree(PREVIEW_CACHE_DIR)
+        shutil.rmtree(self.preview_dir)
         self.assertIsNone(get_most_recent_preview_frame("test_camera"))

frigate/test/test_review_flow.py

@@ -19,6 +19,7 @@ import numpy as np
 from frigate.comms.detections_updater import DetectionTypeEnum
 from frigate.config import FrigateConfig
 from frigate.review.maintainer import ReviewSegmentMaintainer
+from frigate.track.object_processing import ManualEventState
 
 CAMERA = "front_door"
 
@@ -122,13 +123,19 @@ class ReviewFlowTestCase(unittest.TestCase):
 
     def feed(self, *frames: tuple[float, list[dict[str, Any]]]) -> None:
         """Run the maintainer loop over the given (frame_time, objects) frames."""
-        queue = [
-            (
-                DetectionTypeEnum.video.value,
-                (CAMERA, f"{CAMERA}_{frame_time}", frame_time, objects, [], []),
-            )
-            for frame_time, objects in frames
-        ]
+        self.feed_messages(
+            *[
+                (
+                    DetectionTypeEnum.video.value,
+                    (CAMERA, f"{CAMERA}_{frame_time}", frame_time, objects, [], []),
+                )
+                for frame_time, objects in frames
+            ]
+        )
+
+    def feed_messages(self, *messages: tuple[str, tuple[Any, ...]]) -> None:
+        """Run the maintainer loop over the given (topic, data) messages."""
+        queue = list(messages)
         self.maintainer.stop_event = threading.Event()
 
         def next_update(timeout: float) -> Any:
@@ -853,5 +860,46 @@ class TestSplitDetectionLabels(ReviewFlowTestCase):
         self.assertCountEqual(detection["data"]["zones"], ["yard", "driveway"])
 
 
+class TestManualEvents(ReviewFlowTestCase):
+    def manual_event(
+        self, event_id: str, frame_time: float, sub_label: str | None = None
+    ) -> tuple[str, tuple[Any, ...]]:
+        """Build a completed manual event as published by the object processor."""
+        return (
+            DetectionTypeEnum.api.value,
+            (
+                CAMERA,
+                frame_time,
+                {
+                    "state": ManualEventState.complete,
+                    "label": "person",
+                    "sub_label": sub_label,
+                    "event_id": event_id,
+                    "end_time": frame_time + 5,
+                },
+            ),
+        )
+
+    def test_sub_label_is_verified_on_new_segment(self) -> None:
+        self.feed_messages(self.manual_event("m1", 1, "Bob"))
+
+        data = self.maintainer.active_review_segments[CAMERA].get_data(False)["data"]
+        self.assertEqual(data["objects"], ["person-verified"])
+        self.assertEqual(data["verified_objects"], ["person-verified"])
+        self.assertEqual(data["sub_labels"], ["Bob"])
+
+    def test_sub_label_is_verified_on_active_segment(self) -> None:
+        self.feed(
+            (1, [self.tracked("dog1", "dog", 1)]),
+        )
+        self.feed_messages(
+            self.manual_event("m1", 2, "Bob"), self.manual_event("m2", 2)
+        )
+
+        data = self.maintainer.active_review_segments[CAMERA].get_data(False)["data"]
+        self.assertCountEqual(data["objects"], ["dog", "person-verified", "person"])
+        self.assertEqual(data["sub_labels"], ["Bob"])
+
+
 if __name__ == "__main__":
     unittest.main()

frigate/test/test_review_manual_event_severity.py

@@ -100,24 +100,6 @@ class TestManualEventSeverity(unittest.TestCase):
             SeverityEnum.alert,
         )
 
-    def test_sub_label_is_stripped_before_categorizing(self) -> None:
-        maintainer = self._make_maintainer(
-            """
-    review:
-      alerts:
-        labels:
-          - person
-      detections:
-        labels:
-          - person
-"""
-        )
-
-        self.assertEqual(
-            maintainer.get_manual_event_severity("front_door", "person: Bob"),
-            SeverityEnum.alert,
-        )
-
     def test_alert_label_is_detection_when_alerts_disabled(self) -> None:
         maintainer = self._make_maintainer(
             """

frigate/test/test_storage.py

@@ -13,22 +13,28 @@ from frigate.config import FrigateConfig
 from frigate.const import STREAM_TYPE_MAIN, STREAM_TYPE_SUB
 from frigate.models import Event, Recordings
 from frigate.storage import MAX_CALCULATED_BANDWIDTH, StorageMaintainer
-from frigate.test.const import TEST_DB, TEST_DB_CLEANUPS
+from frigate.test.const import TEST_DB, TEST_DB_CLEANUPS, TEST_DB_PRAGMAS
 
 
 class TestHttp(unittest.TestCase):
     def setUp(self):
         # setup clean database for each test run
-        migrate_db = SqliteDatabase("test.db")
+        migrate_db = SqliteDatabase(TEST_DB, pragmas=TEST_DB_PRAGMAS)
         del logging.getLogger("peewee_migrate").handlers[:]
         router = Router(migrate_db)
         router.run()
         migrate_db.close()
-        self.db = SqliteQueueDatabase(TEST_DB)
+        self.db = SqliteQueueDatabase(TEST_DB, pragmas=TEST_DB_PRAGMAS)
         models = [Event, Recordings]
         self.db.bind(models)
         self.test_dir = tempfile.mkdtemp()
 
+        # The maintainers are never stopped, so a real subscriber leaks a zmq
+        # context whose garbage collection can hang a later test in term()
+        subscriber_patch = patch("frigate.storage.CameraConfigUpdateSubscriber")
+        subscriber_patch.start()
+        self.addCleanup(subscriber_patch.stop)
+
         self.minimal_config = {
             "mqtt": {"host": "mqtt"},
             "cameras": {

frigate/test/test_tflite_detection_model.py

@@ -0,0 +1,229 @@
+"""Tests for the TFLite detection model shared by the TFLite detectors."""
+
+import unittest
+
+import numpy as np
+
+from frigate.detectors.detector_config import ModelConfig, ModelTypeEnum
+from frigate.detectors.detector_utils import TFLiteDetectionModel
+
+
+class FakeInterpreter:
+    """Stands in for a TFLite interpreter with fixed output tensors."""
+
+    def __init__(self, outputs, input_dtype=np.uint8, allocate_error=None):
+        self.outputs = {}
+        self.output_details = []
+        self.allocate_error = allocate_error
+        self.input_details = [
+            {"index": 0, "dtype": input_dtype, "shape": np.array([1, 320, 320, 3])}
+        ]
+        self.input = None
+
+        for i, (tensor, quantization) in enumerate(outputs, start=1):
+            self.outputs[i] = tensor
+            self.output_details.append(
+                {
+                    "index": i,
+                    "dtype": tensor.dtype,
+                    "shape": np.array(tensor.shape),
+                    "quantization": quantization,
+                }
+            )
+
+    def allocate_tensors(self):
+        if self.allocate_error:
+            raise self.allocate_error
+
+    def get_input_details(self):
+        return self.input_details
+
+    def get_output_details(self):
+        return self.output_details
+
+    def set_tensor(self, index, value):
+        self.input = value
+
+    def invoke(self):
+        pass
+
+    def tensor(self, index):
+        return lambda: self.outputs[index]
+
+    def get_tensor(self, index):
+        return self.outputs[index]
+
+
+def model_with_type(model_type) -> ModelConfig:
+    return ModelConfig(
+        path="/config/model.tflite",
+        model_type=model_type,
+        labelmap_path=None,
+        labelmap={0: "person"},
+        width=320,
+        height=320,
+    )
+
+
+def ssd_outputs(scores, count=None):
+    num = len(scores)
+    boxes = np.tile(np.array([0.1, 0.2, 0.3, 0.4], np.float32), (1, num, 1))
+    class_ids = np.arange(num, dtype=np.float32).reshape(1, num)
+    scores = np.array(scores, np.float32).reshape(1, num)
+    count = np.array([num if count is None else count], np.float32)
+    return [
+        (boxes, (0.0, 0)),
+        (class_ids, (0.0, 0)),
+        (scores, (0.0, 0)),
+        (count, (0.0, 0)),
+    ]
+
+
+# 320x320 YOLO heads at strides 8, 16 and 32 have 40*40 + 20*20 + 10*10 anchors
+YOLO_ANCHORS = 2100
+YOLO_CLASSES = 3
+SCORES_SCALE = 1 / 32
+BOXES_SCALE = 0.1
+
+
+def yolo_outputs(candidates, dtype=np.int8):
+    """Build quantized YOLO outputs.
+
+    Args:
+        candidates: (anchor index, class id, logit, ltrb distances in strides)
+        dtype: The output tensors' type
+
+    Returns:
+        The class score and DFL box tensors with their quantization
+    """
+    scores = np.full((1, YOLO_ANCHORS, YOLO_CLASSES), -128, dtype)
+    boxes = np.full((1, YOLO_ANCHORS, 4, 16), -100, dtype)
+
+    for anchor, class_id, logit, distances in candidates:
+        scores[0, anchor, class_id] = round(logit / SCORES_SCALE)
+
+        # a single hot bin per side decodes to that bin's distance
+        for side, distance in enumerate(distances):
+            boxes[0, anchor, side, distance] = 100
+
+    boxes = boxes.reshape(1, YOLO_ANCHORS, 64)
+
+    if np.issubdtype(dtype, np.floating):
+        return [(scores, (0.0, 0)), (boxes, (0.0, 0))]
+
+    return [(scores, (SCORES_SCALE, 0)), (boxes, (BOXES_SCALE, 0))]
+
+
+class TestTFLiteDetectionModel(unittest.TestCase):
+    def test_ssd_returns_detections_above_min_score(self):
+        interpreter = FakeInterpreter(ssd_outputs([0.9, 0.5, 0.3, 0.8]))
+        model = TFLiteDetectionModel(
+            interpreter, model_with_type(ModelTypeEnum.ssd), "cpu"
+        )
+
+        detections = model.detect_raw(np.zeros((1, 320, 320, 3), np.uint8))
+
+        # stops at the first score under the minimum, as SSD scores are sorted
+        np.testing.assert_allclose(
+            detections[:2],
+            [[0, 0.9, 0.1, 0.2, 0.3, 0.4], [1, 0.5, 0.1, 0.2, 0.3, 0.4]],
+            rtol=1e-6,
+        )
+        self.assertFalse(detections[2:].any())
+
+    def test_ssd_caps_detections_at_twenty(self):
+        interpreter = FakeInterpreter(ssd_outputs([0.9] * 25))
+        model = TFLiteDetectionModel(
+            interpreter, model_with_type(ModelTypeEnum.ssd), "cpu"
+        )
+
+        detections = model.detect_raw(np.zeros((1, 320, 320, 3), np.uint8))
+
+        self.assertEqual(detections.shape, (20, 6))
+        self.assertTrue((detections[:, 1] == np.float32(0.9)).all())
+
+    def test_ssd_model_without_four_outputs_is_rejected(self):
+        interpreter = FakeInterpreter(yolo_outputs([]))
+
+        with self.assertRaisesRegex(ValueError, "yolo-generic"):
+            TFLiteDetectionModel(interpreter, model_with_type(ModelTypeEnum.ssd), "cpu")
+
+    def test_int8_input_is_shifted_from_uint8(self):
+        interpreter = FakeInterpreter(ssd_outputs([0.9], count=0), np.int8)
+        model = TFLiteDetectionModel(
+            interpreter, model_with_type(ModelTypeEnum.ssd), "cpu"
+        )
+        frame = np.zeros((1, 320, 320, 3), np.uint8)
+        frame[0, 0, 0] = [0, 128, 255]
+
+        model.detect_raw(frame)
+
+        self.assertEqual(interpreter.input.dtype, np.int8)
+        self.assertEqual(interpreter.input[0, 0, 0].tolist(), [-128, 0, 127])
+
+    def test_yolo_decodes_quantized_outputs(self):
+        # anchor 420 is row 10, column 20 of the stride 8 head, centered on
+        # (20.5, 10.5) strides. Its neighbor overlaps it and should be removed
+        # by NMS as it has a lower score.
+        interpreter = FakeInterpreter(
+            yolo_outputs([(420, 2, 2.0, (2, 3, 4, 5)), (421, 1, 1.0, (2, 3, 4, 5))])
+        )
+        model = TFLiteDetectionModel(
+            interpreter, model_with_type(ModelTypeEnum.yologeneric), "cpu"
+        )
+
+        detections = model.detect_raw(np.zeros((1, 320, 320, 3), np.uint8))
+
+        # box corners are (20.5 - 2, 10.5 - 3) and (20.5 + 4, 10.5 + 5) strides
+        np.testing.assert_allclose(
+            detections[0],
+            [2, 1 / (1 + np.exp(-2.0)), 60 / 320, 148 / 320, 124 / 320, 196 / 320],
+            rtol=1e-4,
+        )
+        self.assertFalse(detections[1:].any())
+
+    def test_yolo_without_candidates_returns_no_detections(self):
+        interpreter = FakeInterpreter(yolo_outputs([]))
+        model = TFLiteDetectionModel(
+            interpreter, model_with_type(ModelTypeEnum.yologeneric), "cpu"
+        )
+
+        detections = model.detect_raw(np.zeros((1, 320, 320, 3), np.uint8))
+
+        self.assertFalse(detections.any())
+
+    def test_yolo_model_with_float_outputs_is_rejected(self):
+        interpreter = FakeInterpreter(yolo_outputs([], np.float32))
+
+        with self.assertRaisesRegex(ValueError, "int8 quantized"):
+            TFLiteDetectionModel(
+                interpreter, model_with_type(ModelTypeEnum.yologeneric), "cpu"
+            )
+
+    def test_unsupported_model_type_is_rejected(self):
+        interpreter = FakeInterpreter(ssd_outputs([0.9]))
+
+        with self.assertRaisesRegex(ValueError, "yolox is currently not supported"):
+            TFLiteDetectionModel(
+                interpreter, model_with_type(ModelTypeEnum.yolox), "cpu"
+            )
+
+    def test_edgetpu_compiled_model_is_explained(self):
+        interpreter = FakeInterpreter(
+            ssd_outputs([0.9]),
+            allocate_error=RuntimeError(
+                "Encountered unresolved custom op: edgetpu-custom-op."
+            ),
+        )
+
+        with (
+            self.assertLogs("frigate.detectors.detector_utils", "ERROR") as logs,
+            self.assertRaises(RuntimeError),
+        ):
+            TFLiteDetectionModel(interpreter, model_with_type(ModelTypeEnum.ssd), "cpu")
+
+        self.assertIn("compiled for the EdgeTPU", logs.output[0])
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)

frigate/track/object_processing.py

@@ -635,7 +635,8 @@ class TrackedObjectProcessor(threading.Thread):
                             if end_time
                             else ManualEventState.start
                         ),
-                        "label": f"{label}: {sub_label}" if sub_label else label,
+                        "label": label,
+                        "sub_label": sub_label,
                         "event_id": event_id,
                         "end_time": end_time,
                     },
@@ -696,7 +697,8 @@ class TrackedObjectProcessor(threading.Thread):
                 frame_time,
                 {
                     "state": ManualEventState.start,
-                    "label": f"{label}: {sub_label}" if sub_label else label,
+                    "label": label,
+                    "sub_label": sub_label,
                     "event_id": event_id,
                     "end_time": None,
                 },

web/public/locales/en/config/cameras.json

@@ -858,6 +858,10 @@
       "label": "Cooldown period",
       "description": "Cooldown (seconds) between notifications to avoid spamming recipients."
     },
+    "ttl": {
+      "label": "Notification TTL",
+      "description": "Seconds the push service keeps retrying a notification sent while an alert is active if your device can't be reached. 0 delivers immediately or drops it. The notification sent when an alert ends is always kept for up to an hour."
+    },
     "enabled_in_config": {
       "label": "Original notifications state",
       "description": "Indicates whether notifications were enabled in the original static configuration."

web/public/locales/en/config/global.json

@@ -154,6 +154,10 @@
       "label": "Cooldown period",
       "description": "Cooldown (seconds) between notifications to avoid spamming recipients."
     },
+    "ttl": {
+      "label": "Notification TTL",
+      "description": "Seconds the push service keeps retrying a notification sent while an alert is active if your device can't be reached. 0 delivers immediately or drops it. The notification sent when an alert ends is always kept for up to an hour."
+    },
     "enabled_in_config": {
       "label": "Original notifications state",
       "description": "Indicates whether notifications were enabled in the original static configuration."

web/public/locales/en/views/settings.json

@@ -2051,7 +2051,8 @@
       "profileBaseDisabled": "Recording is disabled in this camera's base config, so enabling it in a profile has no effect. Enable recording in the base config and use a profile to disable it instead."
     },
     "notifications": {
-      "profileBaseDisabled": "No cameras have notifications enabled in their base config, so enabling them in a profile has no effect. Enable notifications in the base config of at least one camera."
+      "profileBaseDisabled": "No cameras have notifications enabled in their base config, so enabling them in a profile has no effect. Enable notifications in the base config of at least one camera.",
+      "ttlDelay": "Notifications can arrive up to this many seconds late and still link to the live view after the review item has ended."
     },
     "birdseye": {
       "objectTrackingDetectDisabled": "Birdseye includes tracked objects, but object detection is disabled for this camera. The camera will not appear in Birdseye.",

web/src/components/config-form/section-configs/notifications.ts

@@ -7,7 +7,7 @@ const notifications: SectionConfigOverrides = {
     fieldOrder: ["enabled", "email"],
     fieldGroups: {},
     hiddenFields: ["enabled_in_config"],
-    advancedFields: [],
+    advancedFields: ["ttl"],
     fieldMessages: [
       {
         key: "profile-base-notifications-disabled",
@@ -24,6 +24,15 @@ const notifications: SectionConfigOverrides = {
               camera.notifications.enabled_in_config,
           ),
       },
+      {
+        key: "notifications-ttl-delay",
+        field: "ttl",
+        messageKey: "configMessages.notifications.ttlDelay",
+        severity: "warning",
+        position: "after",
+        condition: (ctx) =>
+          typeof ctx.formData?.ttl === "number" && ctx.formData.ttl > 0,
+      },
     ],
   },
   global: {
@@ -36,7 +45,7 @@ const notifications: SectionConfigOverrides = {
     },
   },
   camera: {
-    hiddenFields: ["enabled_in_config", "email"],
+    hiddenFields: ["enabled_in_config", "email", "ttl"],
   },
 };
 

web/src/components/config-form/sectionExtras/NotificationsSettingsExtras.tsx

@@ -340,7 +340,7 @@ export default function NotificationsSettingsExtras({
       set(nextNotifications, "enabled", desiredEnabled);
       const sanitizedNotifications = sanitizeSectionData(
         nextNotifications as ConfigSectionData,
-        ["enabled_in_config", "email"],
+        ["enabled_in_config", "email", "ttl"],
       );
       onPendingDataChange("notifications", camera.name, sanitizedNotifications);
       pendingCameraOverridesRef.current.add(camera.name);
@@ -490,7 +490,7 @@ export default function NotificationsSettingsExtras({
 
   return (
     <div className="flex size-full flex-col md:flex-row">
-      <div className="scrollbar-container order-last mb-2 mt-2 flex h-full w-full flex-col overflow-y-auto px-2 md:order-none">
+      <div className="scrollbar-container order-last mb-2 mt-2 flex h-full w-full flex-col overflow-y-auto md:order-none">
         <div className={cn("w-full max-w-5xl space-y-6")}>
           {isAdmin && (
             <SettingsGroupCard