Regenerate genai tracked object descriptions (#13930)

* add genai to frigateconfig

* add regenerate button if genai is enabled

* add endpoint and new zmq pub/sub model

* move publisher to app

* dont override

* logging

* debug timeouts

* clean up

* clean up

* allow saving of empty description

* ensure descriptions can be empty

* update search detail when results change

* revalidate explore page on focus

* global mutate hook

* description websocket hook and dispatcher

* revalidation and mutation

* fix merge conflicts

* update tests

* fix merge conflicts

* fix response message

* fix response message

* fix fastapi

* fix test

* remove log

* json content

* fix content response

* more json content fixes

* another one
This commit is contained in:
Josh Hawkins
2024-09-24 08:14:51 -06:00
committed by GitHub
parent cffc431bf0
commit ecbf0410eb
14 changed files with 274 additions and 19 deletions
+44
View File
@@ -0,0 +1,44 @@
"""Facilitates communication between processes."""
import logging
from enum import Enum
from typing import Optional
from .zmq_proxy import Publisher, Subscriber
logger = logging.getLogger(__name__)
class EventMetadataTypeEnum(str, Enum):
all = ""
regenerate_description = "regenerate_description"
class EventMetadataPublisher(Publisher):
"""Simplifies receiving event metadata."""
topic_base = "event_metadata/"
def __init__(self, topic: EventMetadataTypeEnum) -> None:
topic = topic.value
super().__init__(topic)
class EventMetadataSubscriber(Subscriber):
"""Simplifies receiving event metadata."""
topic_base = "event_metadata/"
def __init__(self, topic: EventMetadataTypeEnum) -> None:
topic = topic.value
super().__init__(topic)
def check_for_update(
self, timeout: float = None
) -> Optional[tuple[EventMetadataTypeEnum, any]]:
return super().check_for_update(timeout)
def _return_object(self, topic: str, payload: any) -> any:
if payload is None:
return (None, None)
return (EventMetadataTypeEnum[topic[len(self.topic_base) :]], payload)