Telegram Bot plugin

Subforum for general discussions. Do not dump your questions/problems here, but try to find the subforum where it belongs!

Moderators: leecollings, remb0

Post Reply
User avatar
dyter
Posts: 42
Joined: Thursday 08 February 2018 15:48
Target OS: NAS (Synology & others)
Domoticz version: 8841
Location: Lyon
Contact:

Telegram Bot plugin

Post by dyter »

Hello,

I build a Telegram plugin with ChatGPT in french, for send message from Telegram to Domoticz ;)

You need to rename it to plugin.py and move it in a folder

Code: Select all

# -*- coding: utf-8 -*-

import Domoticz
import urllib.request
import json
import time  # Pour gérer les délais

# Contenu du fichier XML intégré dans le plugin
PLUGIN_XML = """
<plugin key="TelegramBot" name="Telegram Bot" author="Dyter" version="1.0.0" wikilink="https://github.com/your-repo" externallink="https://github.com/your-repo">
    <params>
        <!-- Champ pour le token Telegram -->
        <param field="Mode1" label="Token Telegram" width="300px" required="true" default=""/>
        
        <!-- Champ pour l'intervalle de récupération (en secondes) -->
        <param field="Mode2" label="Intervalle de récupération (secondes)" width="100px" required="true" default="5"/>
        
        <!-- Option pour activer/désactiver les logs de débogage -->
        <param field="Mode6" label="Debug" width="75px">
            <options>
                <option label="True" value="Debug"/>
                <option label="False" value="Normal" default="true"/>
            </options>
        </param>
    </params>
</plugin>
"""

class BasePlugin:
    def __init__(self):
        self.token = ""  # Token Telegram (configuré par l'utilisateur)
        self.update_interval = 5  # Intervalle de récupération en secondes (configuré par l'utilisateur)
        self.last_update_id = None  # Dernier ID de message traité
        self.text_device_id = None  # ID du capteur texte dans Domoticz
        self.debug = False
        self.last_check_time = 0  # Temps de la dernière vérification

    def onStart(self):
        Domoticz.Log("Plugin Telegram Bot démarré")
        self.token = Parameters["Mode1"]  # Récupère le token depuis la configuration
        self.update_interval = int(Parameters["Mode2"])  # Récupère l'intervalle depuis la configuration

        if Parameters["Mode6"] == "Debug":
            self.debug = True
            Domoticz.Debugging(1)
            DumpConfigToLog()
        else:
            Domoticz.Debugging(0)

        # Crée un capteur texte dans Domoticz s'il n'existe pas déjà
        if 1 not in Devices:
            Domoticz.Device(Name="Telegram Messages", Unit=1, TypeName="Text").Create()
            Domoticz.Log("Capteur texte créé.")

        self.text_device_id = 1
        Domoticz.Heartbeat(self.update_interval)

    def onStop(self):
        Domoticz.Log("Plugin Telegram Bot arrêté")

    def onHeartbeat(self):
        current_time = time.time()
        if current_time - self.last_check_time >= self.update_interval:
            self.get_updates()
            self.last_check_time = current_time  # Met à jour le temps de la dernière vérification

    def get_updates(self):
        # Récupère les nouveaux messages via l'API Telegram
        url = f"https://api.telegram.org/bot{self.token}/getUpdates"
        params = {"offset": self.last_update_id + 1 if self.last_update_id else None}  # Pas de timeout
        url_with_params = url + "?" + urllib.parse.urlencode(params)

        try:
            with urllib.request.urlopen(url_with_params) as response:
                if response.status == 200:
                    data = response.read().decode("utf-8")  # Décode la réponse en UTF-8
                    self.handle_telegram_response(data)
                else:
                    if self.debug:
                        Domoticz.Log(f"Erreur lors de la récupération des messages : {response.status}")
        except urllib.error.URLError as e:
            if self.debug:
                Domoticz.Log(f"Erreur lors de la connexion à l'API Telegram : {str(e)}")

    def handle_telegram_response(self, response_data):
        # Gère la réponse de l'API Telegram
        try:
            data = json.loads(response_data)
            if data["ok"] and data["result"]:
                for update in data["result"]:
                    self.last_update_id = update["update_id"]
                    message_text = update["message"]["text"]
                    self.update_text_device(message_text)
        except Exception as e:
            if self.debug:
                Domoticz.Log(f"Erreur lors du traitement de la réponse : {str(e)}")

    def update_text_device(self, message):
        # Met à jour le capteur texte dans Domoticz
        Devices[self.text_device_id].Update(nValue=0, sValue=message)
        if self.debug:
            Domoticz.Log(f"Message reçu : {message}")

# Fonction pour retourner le contenu XML
def onGetXML():
    return PLUGIN_XML

# Crée une instance du plugin
global _plugin
_plugin = BasePlugin()

# Fonctions requises par Domoticz
def onStart():
    global _plugin
    _plugin.onStart()

def onStop():
    global _plugin
    _plugin.onStop()

def onHeartbeat():
    global _plugin
    _plugin.onHeartbeat()

# Fonction pour afficher la configuration dans les logs
def DumpConfigToLog():
    for x in Parameters:
        if Parameters[x] != "":
            Domoticz.Debug(f"'{x}': '{str(Parameters[x])}'")
    Domoticz.Debug(f"Nombre de dispositifs : {str(len(Devices))}")
    for x in Devices:
        Domoticz.Debug(f"Dispositif : {str(x)} - {str(Devices[x])}")
        Domoticz.Debug(f"ID du dispositif : '{str(Devices[x].ID)}'")
        Domoticz.Debug(f"Nom du dispositif : '{Devices[x].Name}'")
        Domoticz.Debug(f"Valeur nValue : {str(Devices[x].nValue)}")
        Domoticz.Debug(f"Valeur sValue : '{Devices[x].sValue}'")
Last edited by dyter on Wednesday 22 January 2025 7:57, edited 1 time in total.
Domoticz v2024.1 on Synology Docker host, RFLink, ESPEasy wemos D1, Ecodevices, relays x8, dahua doorbell, reolink cameras ....
User avatar
waltervl
Posts: 5397
Joined: Monday 28 January 2019 18:48
Target OS: Linux
Domoticz version: 2024.7
Location: NL
Contact:

Re: Telegram Bot

Post by waltervl »

How does this work? What kind of message should be sent to Domoticz?

I have by the way an python Telegram bot that runs a service and you can get status of Domoticz devices and send commands to Domoticz. https://github.com/waltervl/dynamicTelegramBot
Domoticz running on Udoo X86 (on Ubuntu)
Devices/plugins: ZigbeeforDomoticz (with Xiaomi, Ikea, Tuya devices), Nefit Easy, Midea Airco, Omnik Solar, Goodwe Solar
User avatar
dyter
Posts: 42
Joined: Thursday 08 February 2018 15:48
Target OS: NAS (Synology & others)
Domoticz version: 8841
Location: Lyon
Contact:

Re: Telegram Bot

Post by dyter »

Ok I make this script for very simple use and beginner. The script capture message from telegram and send it to a text device (automatically created at fist start if you "Accept new devices"). The API and timer push is configurable in domoticz hardware interface, and no additional installation is required.

I make a personal lua script for swap router or estimate battery charging, devices status or toggle, Gemini IA ...
This is an exemple because you can program everything you want:

Code: Select all

commandArray = {}

if devicechanged['Telegram Bot'] then
    last_message = otherdevices['Telegram Bot']

    -- Extraction des paramètres du message
    amperage_match = last_message:match("(%d+)A")
    percentage_match = last_message:match("(%d+)%%")
    tarif_match = last_message:gsub("\\u20ac", "€"):match("(%d*%.?%d+)€")

    -- Variables initiales avec priorisation des valeurs extraites
    start_batt = tonumber(otherdevices['Batterie Tesla'])
    end_batt = tonumber(percentage_match) or 85
    capacity_kwh = 70 -- capacité batterie incluant la dégradation
    avg_voltage = 230
    ampere = tonumber(amperage_match) or 28
    tarif = tonumber(tarif_match) or 0.2516 -- tarif normal
    tarif_heures_creuses = 0.1296 -- tarif heures creuses
    efficiency = (ampere > 23) and 90 or 80 -- Ajustement de l'efficacité selon l'ampérage

    -- Fonction pour estimer le temps de charge
    function estimateChargeTime(start_batt, end_batt, ampere)
        power_kw = ampere * avg_voltage / 1000
        energy_needed_kwh = ((end_batt - start_batt) / 100 * capacity_kwh)/(efficiency/100)
        return math.floor((energy_needed_kwh / power_kw) * 3600)
    end

    -- Calcul du temps de charge & coûts
    charge_time_sec = estimateChargeTime(start_batt, end_batt, ampere)
    finish_time_sec = os.time({year=2024, month=12, day=21, hour=22, min=0, sec=0}) + charge_time_sec
    start_time_sec = os.time({year=2024, month=12, day=21, hour=6}) - charge_time_sec - 2700 -- 45 minutes avant
    hours = math.floor(charge_time_sec / 3600)
    minutes = math.floor((charge_time_sec % 3600) / 60)
    charge_cost = tarif * energy_needed_kwh
    charge_cost_hc = tarif_heures_creuses * energy_needed_kwh

    -- Réponse en fonction du message
    if last_message:find("22H") then
        commandArray['SendNotification'] = "Avec un démarrage de la charge à 22H00 à " .. ampere .. "A, la charge atteindra " .. end_batt .. "% à " .. os.date("%H:%M", finish_time_sec) .. ". Coût estimé : " .. string.format("%.2f", charge_cost_hc) .. " €#####telegram"
    elseif last_message:find("6H") then
        commandArray['SendNotification'] = "Pour finir la charge à " .. end_batt .. "% avec " .. ampere .. "A à 6H00, la charge commencera vers " .. os.date("%H:%M", start_time_sec) .. " (environ 45 mins avant la fin). Coût estimé : " .. string.format("%.2f", charge_cost_hc) .. " €#####telegram"
    elseif amperage_match or percentage_match or tarif_match then
        commandArray['SendNotification'] = "Estimation du temps de charge de la Tesla à " .. end_batt .. "% avec " .. ampere .. "A : " .. hours .. "h " .. minutes .. "m restants. Coût estimé : " .. string.format("%.2f", charge_cost) .. " €#####telegram"
    elseif otherdevices[last_message] then
        commandArray['SendNotification'] = last_message .. " = " .. otherdevices[last_message] .. '#####telegram'
    elseif otherdevices[nom] and valeur then
        if valeur == "on" then valeur = "On"
        elseif valeur == "off" then valeur = "Off"
        elseif valeur:find("evel") then
            levelNumber = valeur:match("evel%s*(%d+)") or "0"
        valeur = "Set Level " .. levelNumber
        end
        commandArray[nom]=valeur
        commandArray['SendNotification'] = nom .. " = " .. valeur .. "#####telegram"
    elseif last_message:find("ascul") then
        current_default = io.popen("ip route show default 2>/dev/null"):read("*a")
        if not current_default:find("192.168.0.1") then
            os.execute("ip route del default via 192.168.0.254 2>/dev/null")
            os.execute("ip route add default via 192.168.0.1 2>/dev/null")
            commandArray['SendNotification'] = "bascule secours 4G vers Livebox OK.#####telegram"
        elseif not current_default:find("192.168.0.254") then
            os.execute("ip route del default via 192.168.0.1 2>/dev/null")
            os.execute("ip route add default via 192.168.0.254 2>/dev/null")
            commandArray['SendNotification'] = "bascule livebox vers secours 4G OK.#####telegram"
        else
            commandArray['SendNotification'] = "bascule du routeur secours vers principal impossible.#####telegram"
        end
    else
        commandArray['SendNotification'] = "Veuillez envoyer une commande valide (Bascul, 28A, 90%, 22H, 6H ou un device).#####telegram"
        -- GOOGLE AI STUDIO API 
        --API_URL = "https://generativelanguage.googleapis.com/v1beta/models/gemini-1.5-flash:generateContent?key=YOUR TOKEN"
        --cmd = 'curl -s -X POST -H "Content-Type: application/json" -d \'{"contents":[{"parts":[{"text":"' .. last_message .. '"}]}]}\' "' .. API_URL .. '"'
        --handle = io.popen(cmd)
        --response = handle:read("*a")
        --handle:close()
        --text = response:match('"text":%s*"(.-)"') text = text:gsub("\\n", "") text = text:gsub("\\t", "")
        --commandArray['SendNotification'] = text .. "#####telegram"
    end

end
return commandArray
Domoticz v2024.1 on Synology Docker host, RFLink, ESPEasy wemos D1, Ecodevices, relays x8, dahua doorbell, reolink cameras ....
Post Reply

Who is online

Users browsing this forum: No registered users and 0 guests