administrator/frigate
1
0
Fork 0
mirror of https://github.com/blakeblackshear/frigate.git synced 2026-04-16 12:02:09 +03:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

Include function correctly

Browse Source
This commit is contained in:
Nicolas Mowen 2025-10-02 12:36:50 -06:00
parent 2115d85cc1
commit aba9cfba33
2 changed files with 42 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

44
frigate/data_processing/post/object_descriptions.py
View File

@ -7,6 +7,7 @@ import threading
from pathlib import Path
from typing import TYPE_CHECKING, Any
import cv2
import numpy as np
from peewee import DoesNotExist
@ -158,7 +159,7 @@ class ObjectDescriptionProcessor(PostProcessorApi):
)
if event.has_snapshot and source == "snapshot":
snapshot_image = self._read_and_crop_snapshot(event, camera_config)
snapshot_image = self._read_and_crop_snapshot(event)
if not snapshot_image:
return
@ -203,11 +204,50 @@ class ObjectDescriptionProcessor(PostProcessorApi):
)
return None
def _read_and_crop_snapshot(self, event: Event) -> bytes | None:
"""Read, decode, and crop the snapshot image."""
snapshot_file = os.path.join(CLIPS_DIR, f"{event.camera}-{event.id}.jpg")
if not os.path.isfile(snapshot_file):
logger.error(
f"Cannot load snapshot for {event.id}, file not found: {snapshot_file}"
)
return None
try:
with open(snapshot_file, "rb") as image_file:
snapshot_image = image_file.read()
img = cv2.imdecode(
np.frombuffer(snapshot_image, dtype=np.int8),
cv2.IMREAD_COLOR,
)
# Crop snapshot based on region
# provide full image if region doesn't exist (manual events)
height, width = img.shape[:2]
x1_rel, y1_rel, width_rel, height_rel = event.data.get(
"region", [0, 0, 1, 1]
)
x1, y1 = int(x1_rel * width), int(y1_rel * height)
cropped_image = img[
y1 : y1 + int(height_rel * height),
x1 : x1 + int(width_rel * width),
]
_, buffer = cv2.imencode(".jpg", cropped_image)
return buffer.tobytes()
except Exception:
return None
def _process_genai_description(
self, event: Event, camera_config: CameraConfig, thumbnail
) -> None:
if event.has_snapshot and camera_config.objects.genai.use_snapshot:
snapshot_image = self._read_and_crop_snapshot(event, camera_config)
snapshot_image = self._read_and_crop_snapshot(event)
if not snapshot_image:
return

45
frigate/embeddings/maintainer.py
View File

@ -3,13 +3,10 @@
import base64
import datetime
import logging
import os
import threading
from multiprocessing.synchronize import Event as MpEvent
from typing import Any
import cv2
import numpy as np
from peewee import DoesNotExist
from frigate.comms.detections_updater import DetectionSubscriber, DetectionTypeEnum
@ -35,9 +32,6 @@ from frigate.config.camera.updater import (
CameraConfigUpdateEnum,
CameraConfigUpdateSubscriber,
)
from frigate.const import (
CLIPS_DIR,
)
from frigate.data_processing.common.license_plate.model import (
LicensePlateModelRunner,
)
@ -589,42 +583,3 @@ class EmbeddingMaintainer(threading.Thread):
return
self.embeddings.embed_thumbnail(event_id, thumbnail)
def _read_and_crop_snapshot(self, event: Event, camera_config) -> bytes | None:
"""Read, decode, and crop the snapshot image."""
snapshot_file = os.path.join(CLIPS_DIR, f"{event.camera}-{event.id}.jpg")
if not os.path.isfile(snapshot_file):
logger.error(
f"Cannot load snapshot for {event.id}, file not found: {snapshot_file}"
)
return None
try:
with open(snapshot_file, "rb") as image_file:
snapshot_image = image_file.read()
img = cv2.imdecode(
np.frombuffer(snapshot_image, dtype=np.int8),
cv2.IMREAD_COLOR,
)
# Crop snapshot based on region
# provide full image if region doesn't exist (manual events)
height, width = img.shape[:2]
x1_rel, y1_rel, width_rel, height_rel = event.data.get(
"region", [0, 0, 1, 1]
)
x1, y1 = int(x1_rel * width), int(y1_rel * height)
cropped_image = img[
y1 : y1 + int(height_rel * height),
x1 : x1 + int(width_rel * width),
]
_, buffer = cv2.imencode(".jpg", cropped_image)
return buffer.tobytes()
except Exception:
return None