blakeblackshear.frigate/frigate/data_processing/real_time/license_plate.py
Nicolas Mowen 4b4053d174
Add face and lpr to tracked object update topic (#17940)
* Send tracked object updates for face and license_plate objects

* Update docs

* Add to type enum

* Add camera to object description update

* Formatting

* Consolidate yue-Hant

* Add missing
2025-04-28 16:43:03 -06:00

56 lines
1.8 KiB
Python

"""Handle processing images for face detection and recognition."""
import logging
import numpy as np
from frigate.comms.event_metadata_updater import EventMetadataPublisher
from frigate.comms.inter_process import InterProcessRequestor
from frigate.config import FrigateConfig
from frigate.data_processing.common.license_plate.mixin import (
LicensePlateProcessingMixin,
)
from frigate.data_processing.common.license_plate.model import (
LicensePlateModelRunner,
)
from ..types import DataProcessorMetrics
from .api import RealTimeProcessorApi
logger = logging.getLogger(__name__)
class LicensePlateRealTimeProcessor(LicensePlateProcessingMixin, RealTimeProcessorApi):
def __init__(
self,
config: FrigateConfig,
requestor: InterProcessRequestor,
sub_label_publisher: EventMetadataPublisher,
metrics: DataProcessorMetrics,
model_runner: LicensePlateModelRunner,
detected_license_plates: dict[str, dict[str, any]],
):
self.requestor = requestor
self.detected_license_plates = detected_license_plates
self.model_runner = model_runner
self.lpr_config = config.lpr
self.config = config
self.sub_label_publisher = sub_label_publisher
super().__init__(config, metrics)
def process_frame(
self,
obj_data: dict[str, any],
frame: np.ndarray,
dedicated_lpr: bool | None = False,
):
"""Look for license plates in image."""
self.lpr_process(obj_data, frame, dedicated_lpr)
def handle_request(self, topic, request_data) -> dict[str, any] | None:
return
def expire_object(self, object_id: str):
if object_id in self.detected_license_plates:
self.detected_license_plates.pop(object_id)