Telegram integration

[maxdd]

Hello,
I was looking for the possibility to integrate frigate events and streaming links with telegram.
I honestly would like to avoid Home Automation simply because i'm not using it extensively to justitfy its deployment.
I was wondering whether, given it is just a chat_id and a bot_token, the integration can be directly implemented here.
If not, i was planning to setup an http api to telegram sender, in that case would it make sense to send out an event api request every 1/2m to check whether there was a new event and reconstruct the url to snapshots/record?

[NickM-27]

Personally, I'd imagine frigate will just implement notifications directly. Supporting each service is a lot to provide to users and will bring even more requests of "use this service too"

There are many options besides home assistant like node red or even a simple local script to handle this via MQTT.

[maxdd]

I'm not really sure why I would want to go with a polling method when frigate has built-in notification.
I will stick with my implementation for now which is 20 lines of code

[NickM-27]

not sure what you mean by polling method, if you are referring to frigate-notify then that project works with the same MQTT topics that most other notification implementations work with.

[alexbelgium]

I integrated a decent implementation (not async though, but it can be done if needed) of telegram which sends the snapshots and a caption with the url of the video, the implementation expects a configuration as such

telegram:
  enabled: True
  bot_token: redacted
  bot_chatID: redacted
  host: 'http://192.168.1.52:5000' # to get direct link of event/clip in the caption of the photo
  topics:
  - events # send notification when events payload is dispatched
  zones:
  - zone_0 # send notification when zone is zone_0
  time_range: '21:00,06:30' # (start,end) send notification only in this window e.g. now >= start or now <= end

If you feel like it could make sense to have a look let me know

That's exactly what I was looking for :) did you describe somewhere your solution? Thanks!

[maxdd]

I didn't try on new frigate, this was working on the older versions

diff --git a/Makefile b/Makefile
index df40d7df..9b7071f7 100644
--- a/Makefile
+++ b/Makefile
@@ -18,7 +18,7 @@ version:
        echo 'VERSION = "$(VERSION)-$(COMMIT_HASH)"' > frigate/version.py
 
 local: version
-       docker buildx build --target=frigate --tag frigate:latest --load --file docker/main/Dockerfile .
+       docker buildx build --target=frigate --tag frigate_mod:latest --load --file docker/main/Dockerfile .
 
 amd64:
        docker buildx build --platform linux/amd64 --target=frigate --tag $(IMAGE_REPO):$(VERSION)-$(COMMIT_HASH) --file docker/main/Dockerfile .
diff --git a/frigate/app.py b/frigate/app.py
index 4a3cf48d..50c5f6b8 100644
--- a/frigate/app.py
+++ b/frigate/app.py
@@ -20,6 +20,7 @@ from playhouse.sqliteq import SqliteQueueDatabase
 from frigate.comms.dispatcher import Communicator, Dispatcher
 from frigate.comms.inter_process import InterProcessCommunicator
 from frigate.comms.mqtt import MqttClient
+from frigate.comms.telegram import TelegramClient
 from frigate.comms.ws import WebSocketClient
 from frigate.config import BirdseyeModeEnum, FrigateConfig
 from frigate.const import (
@@ -407,7 +408,8 @@ class FrigateApp:
 
         if self.config.mqtt.enabled:
             comms.append(MqttClient(self.config))
-
+        if self.config.telegram.enabled:
+            comms.append(TelegramClient(self.config))
         comms.append(WebSocketClient(self.config))
         comms.append(self.inter_process_communicator)
 
diff --git a/frigate/config.py b/frigate/config.py
index 2e8b2570..8332818f 100644
--- a/frigate/config.py
+++ b/frigate/config.py
@@ -148,6 +148,20 @@ class MqttConfig(FrigateBaseModel):
             raise ValueError("Password must be provided with username.")
         return v
 
+class TelegramConfig(FrigateBaseModel):
+    enabled: bool = Field(title="Enable Telegram Communication.", default=False)
+    bot_token: str = Field(default="", title="Bot Token")
+    bot_chatID: str = Field(default="", title="Bot ChatID")
+    topics: List[str] = Field(
+        default_factory=list,
+        title="List of topics to be notified in telegram.",
+    )
+    zones: List[str] = Field(
+        default_factory=list,
+        title="List of zones to be notified in telegram.",
+    )
+    host: str = Field(default="", title="Hostname for URL")
+    time_range: str = Field(default="", title="Range for notification to be sent (start,end)")
 
 class ZoomingModeEnum(str, Enum):
     disabled = "disabled"
@@ -1052,6 +1066,8 @@ def verify_autotrack_zones(camera_config: CameraConfig) -> ValueError | None:
 
 class FrigateConfig(FrigateBaseModel):
     mqtt: MqttConfig = Field(title="MQTT Configuration.")
+    telegram: TelegramConfig = Field(
+        default_factory=TelegramConfig, title="Telegram Configuration.")
     database: DatabaseConfig = Field(
         default_factory=DatabaseConfig, title="Database configuration."
     )

frigate/comms/telegram.py

import logging
from typing import Any, Callable

import requests
import json
import datetime

from frigate.comms.dispatcher import Communicator
from frigate.config import FrigateConfig

logger = logging.getLogger(__name__)

class TelegramClient(Communicator):  # type: ignore[misc]
    """Frigate wrapper for telegram client."""

    def __init__(self, config: FrigateConfig) -> None:
        self.config = config
        self.telegram_config = config.telegram

    def subscribe(self, receiver: Callable) -> None:
        """Wrapper for allowing dispatcher to subscribe."""
        self._dispatcher = receiver
        
    def stop(self) -> None:
        logger.info("Nothing to stop")

    def publish(self, topic: str, payload: Any, retain: bool = False) -> None:
        """Wrapper for publishing when client is in valid state."""
        time_range = self.telegram_config.time_range.split(',') 
        now = datetime.datetime.now().time()
        start = datetime.datetime.strptime(time_range[0], '%H:%M').time()
        end = datetime.datetime.strptime(time_range[1], '%H:%M').time()

        if now >= start or now <= end:
            logger.info("time to send notification")
            if topic in self.telegram_config.topics:
                logger.info("topic is valid")
                json_payload = json.loads(payload)
                if json_payload["type"] == "end":
                    logger.info("type is end")
                    valid_zone = False
                    for zone in self.telegram_config.zones:
                        if zone in json_payload["before"]["current_zones"]:
                            logger.info("zone is fine")
                            valid_zone = True
                            break

                    if valid_zone:
                        event_id = json_payload["before"]["id"]
                        camera = json_payload["before"]["camera"]

                        img = open(f'/media/frigate/clips/{camera}-{event_id}.jpg', 'rb')
                        caption = f'{self.telegram_config.host}/api/events/{event_id}/clip.mp4'

                        #url = f'https://api.telegram.org/bot{self.telegram_config.bot_token}/sendPhoto?chat_id={self.telegram_config.bot_chatID}&parse_mode=HTML&caption=Notification of event<a href="{caption}">Clip</a>'
                        #message_url = f'https://api.telegram.org/bot{self.telegram_config.bot_token}/sendMessage?chat_id={self.telegram_config.bot_chatID}&text={message}'
                        url = f'https://api.telegram.org/bot{self.telegram_config.bot_token}/sendPhoto?chat_id={self.telegram_config.bot_chatID}&parse_mode=MarkdownV2&caption=Notification of event\n[Clip]({caption})'
                        
                        try: 
                            response = requests.post(url, files={'photo': img}, timeout=2)
                            logger.info(response.json())
                            #response = requests.get(message_url, timeout=2)
                            #logger.info(response.json())
                        except requests.exceptions.Timeout:
                            print("Telegram API timed out")

[alexbelgium]

Thanks very much!

[maxdd]

Polling via web-api is the other available method of frigate-notify.
Mqtt is not ideal, we would have to pass through two network protocols to send a single message.