Eine Torsteuerung auf einem RaspiZero mit Telegram Bot Benachrichtigungen

Stellt hier eure Projekte vor.
Internetseiten, Skripte, und alles andere bzgl. Python.
Antworten
Rudi42
User
Beiträge: 1
Registriert: Donnerstag 8. Oktober 2026, 20:47

Hallo.

Ich habe hier einen Menge elektrischer Tore und Türen (Türsummer). Dafür waren mir einzelne oder auch nur zusätzliche Handfernbedienungen zu umständlich. Daher hatte ich den Plan das per Smartphone steuern zu können.

Features:
- einfache Torsteuerung per Webseite (simple Knöpfe)
- Statusanzeige, sofern konfiguriert (Symbol im Button)
- Steuerung einzelner Relais per GPIO
- Status per GPIO
- Steuerung und Status per Homeatic (CCU3)
- Benachrichtigung an User per Telegram Bot
- Die einzelnen "Knöpfe" können auch per HTML GET Aufruf in andere Dienste / Programme integriert werden (vermutlich auch Alexa,..)

Screenshot:
Bild

Hier erstmal nur der Python Code:

Code: Alles auswählen

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

import sys
import time
import json
import requests
import telegram
import uvicorn
import RPi.GPIO as GPIO

from xml.dom.minidom import parseString
from jinja2 import Environment, FileSystemLoader, select_autoescape

from fastapi import FastAPI, Request, Response
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import HTMLResponse, PlainTextResponse, RedirectResponse

# -------------------------------------------------------------------------------------

class Relay:
    def __init__(self, name, link, color, position, pinSwitch, pinState, stateEnabled,
                 stateHM, stateInverted, toggleTime, msgUsers, msgText, datapoint=""):
        self.name = name
        self.link = link
        self.color = color
        self.position = position
        self.pinSwitch = pinSwitch
        self.pinState = pinState
        self.stateEnabled = stateEnabled
        self.stateHM = stateHM
        self.datapoint = datapoint
        self.stateInverted = stateInverted
        self.toggleTime = toggleTime
        self.msgUsers = msgUsers
        self.msgText = msgText

# -------------------------------------------------------------------------------------

class Macro:
    def __init__(self, name, link, color, position, actions):
        self.name = name
        self.link = link
        self.color = color
        self.position = position
        self.actions = actions

# -------------------------------------------------------------------------------------

class TelegrammUser:
    def __init__(self, name, myId, silent):
        self.name = name
        self.id = myId
        self.silent = silent

# -------------------------------------------------------------------------------------

class Config:
    def __init__(self):
        try:
            with open("config.json", encoding="utf-8-sig") as configFile:
                config = json.load(configFile)

        except IOError:
            print('Could not read file "config.json"!')
            sys.exit(1)

        except json.JSONDecodeError:
            print("config.json: decode error!")
            sys.exit(1)

        # config
        self.ip = config["config"]["IP"]
        self.port = config["config"]["PORT"]
        self.stateUpdate = config["config"]["STATEUPDATE"]
        self.title = config["config"]["TITLE"]
        self.tableX = config["config"]["TABLE_X"]
        self.tableY = config["config"]["TABLE_Y"]
        self.telegrammBotId = config["config"]["TELEGRAMM_BOT_ID"]
        self.ccu_url = config["config"]["CCU_URL"]

        # Telegram users
        self.telegrammUsers = []

        for item in config["config"]["TELEGRAMM_USERS"]:
            user = TelegrammUser(item["name"], item["id"], item["silent"])
            self.telegrammUsers.append(user)

        # relays
        self.relais = []
        for item in config["relais"]:
            relay = Relay(item["button_name"], item["link"], item["button_color"],
                          item["button_grid_position"], item["relais_pin"],
                          item["relais_state"], item["state_enabled"], item["state_isHM"],
                          item["state_inverted"], item["relais_time"], 
                          item["msg_user"].split(","), item["msg_text"])

            if relay.stateHM:
                relay.datapoint = item["datapoint"]
            self.relais.append(relay)

        # macros
        self.macros = []
        for item in config["macros"]:
            macro = Macro(item["button_name"], item["link"], item["button_color"],
                          item["button_grid_position"], item["actions"].split(","))
            self.macros.append(macro)

    def isValidUserId(self, name):
        for user in self.telegrammUsers:
            if user.name == name and user.id:
                return True
        return False

    def getUserByName(self, name):
        for user in self.telegrammUsers:
            if user.name == name and user.id:
                return user
        return None

# -------------------------------------------------------------------------------------

class Controller:
    def __init__(self):
        self.config = Config()
        self.templates = Environment(loader=FileSystemLoader("templates"), autoescape=select_autoescape(["html", "xml"]))
        self._initGPIO()

    def _render(self, template, **kwargs):
        tmpl = self.templates.get_template(template)
        return tmpl.render(**kwargs)

    def _getItems(self):
        table = ""
        for i in range(int(self.config.tableX * self.config.tableY)):
            table += self._getEntry(i)
        return table

    def _getEntry(self, pos):
        # search relays
        for relay in self.config.relais:
            if relay.position == pos:
                return self._render("item.html", btn_link=relay.link, btn_color=relay.color,
                                    ip=self.config.ip, port=self.config.port, btn_name=relay.name)

        # search macros
        for macro in self.config.macros:
            if macro.position == pos:
                return self._render("item.html", btn_link=macro.link, btn_color=macro.color,
                                    ip=self.config.ip, port=self.config.port,btn_name=macro.name)
        return self._render("item_empty.html")

    def _stateJavaScript(self):
        counter = 0
        script = ""

        for relay in self.config.relais:
            if relay.stateEnabled:
                script += ('\n\t\t\t\t\t\tsetClassOnState(''document.getElementById("' + relay.link + '"), pinStates[' + str(counter) + "]);")
            counter += 1
        return script

    def _handleMessage(self, relay, request: Request):
        if not self.config.telegrammBotId:
            return

        try:
            bot = telegram.Bot(token=self.config.telegrammBotId)
            for user in relay.msgUsers:
                if not self.config.isValidUserId(user):
                    continue
                myUser = self.config.getUserByName(user)
                if myUser is None:
                    continue
                msg = relay.msgText
                bot.send_message(chat_id=myUser.id, text=msg, disable_notification=myUser.silent)
        except telegram.error.TelegramError as e:
            print("Telegram error occurred ({})!".format(str(e)))

    def _stateGPIO(self, relay):
        if not relay.stateEnabled:
            return "unknown"
        state = GPIO.input(relay.pinState)
        if relay.stateInverted:
            state = not state
        if state:
            return "open"
        return "closed"

    def _initGPIO(self):
        GPIO.setwarnings(False)
        GPIO.cleanup()
        GPIO.setmode(GPIO.BCM)
        for item in self.config.relais:
            self._initGPIOEntry(item)

    def _initGPIOEntry(self, relay):
        if relay.pinSwitch >= 0:
            GPIO.setup(relay.pinSwitch, GPIO.OUT)
            GPIO.output(relay.pinSwitch, True)

        if relay.stateEnabled and relay.pinState >= 0:
            GPIO.setup(relay.pinState, GPIO.IN, pull_up_down=GPIO.PUD_UP)

    def _switchGPIO(self, relay, request: Request):
        if relay.pinSwitch == -1:
            return "OK"

        self._handleMessage(relay, request)
        if relay.toggleTime == -1.0:
            GPIO.output(relay.pinSwitch, not GPIO.input(relay.pinSwitch))
            return "OK"

        GPIO.output(relay.pinSwitch, False)
        time.sleep(relay.toggleTime)
        GPIO.output(relay.pinSwitch, True)
        return "OK"

    def _getHmState(self, relay):
        if relay.stateEnabled:
            try:
                url = self.config.ccu_url.format(relay.datapoint)
                r = requests.get(url, timeout=5)
                dom = parseString(r.content)
                state = bool(int(dom.getElementsByTagName("datapoint")[0].getAttribute("value")))
                if relay.stateInverted:
                    state = not state
                if state:
                    return "open"
                return "closed"
            except Exception:
                pass
        return "unknown"

    def mainPage(self):
        return self._render("template.html", page_title=self.config.title, tableX=self.config.tableX,
                            tableY=self.config.tableY, ip=self.config.ip, port=self.config.port,
                            update_cycle=self.config.stateUpdate, script_part=self._stateJavaScript(),
                            btn_table=self._getItems())

    def states(self):
        currentStates = ""
        for item in self.config.relais:
            if item.stateHM:
                currentStates += (item.color + "_" + self._getHmState(item) + ",")
            else:
                currentStates += (item.color + "_" + self._stateGPIO(item) + ",")
        return currentStates[:-1]

    def reinit(self):
        self.config = Config()

    def action(self, page, request: Request):
        # relay action
        for relay in self.config.relais:
            if page == relay.link:
                return self._switchGPIO(relay, request)

        # macro action
        for macro in self.config.macros:
            if page == macro.link:
                for myAction in macro.actions:
                    self.action(myAction, request)
                return "OK"
        return "NOK"

# -------------------------------------------------------------------------------------

ctrl = Controller()
app = FastAPI(title="Door Control", docs_url=None, redoc_url=None, openapi_url=None)
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_credentials=False,
                   allow_methods=["GET"], allow_headers=["*"])

# -------------------------------------------------------------------------------------
# Routes
# -------------------------------------------------------------------------------------

@app.get("/", response_class=HTMLResponse)
def main_page():
    return HTMLResponse(content=ctrl.mainPage())

@app.get("/states", response_class=PlainTextResponse)
def states():
    return ctrl.states()

@app.get("/reinit")
def reinit():
    ctrl.reinit()
    return RedirectResponse(url="/", status_code=302)

@app.get("/{page}", response_class=PlainTextResponse)
def action(page: str, request: Request):
    return ctrl.action(page, request )

@app.api_route("/health", methods=["GET", "HEAD"])
def health():
    return Response(status_code=200)

# -------------------------------------------------------------------------------------

if __name__ == "__main__":
    try:
        uvicorn.run(app, host=ctrl.config.ip, port=int(ctrl.config.port),
            # Important for Raspberry Pi Zero:
            workers=1,
            # pure Python implementation
            loop="asyncio", http="h11",
            # avoid unnecessary logging
            access_log = False, log_level="warning"
        )

    finally:
        GPIO.cleanup()
Und hier noch der Link zu allem (enthält auch eine Beispiel "config.json", die html Templates und Bilder): https://www.chaos-thor.net/files/upload ... oorCtrl.7z

Ich hoffe externe Links sind erlaubt?

Bin für Anregungen, Verbesserungen oder auch nur Kommentare immer offen.


MfG Rudi
Nach oben
Sirius3
User
Beiträge: 18431
Registriert: Sonntag 21. Oktober 2012, 17:20

UTF8-Codierung ist der default, braucht also nicht angegeben werden. Beim Import benutzt man `as` zum Umbenennen, GPIO wird aber gar nicht umbenannt.
Statt minidom benutzt man üblicherweise ElementTree.
Ein Kommentar, der nur aus Minuszeichen besteht, ist nicht sehr informativ.
Für die Klassen benutzt man Dataclasses.
Variabelnamen und Funktionen werden in Python nach Konvention klein_mit_unterstrich geschrieben.
Strings stückelt man nicht mit + zusammen, sondern benutzt `join`, oder Formatstrings.
Wenn man einen Index in einer for-Schleife braucht, benutzt man enumerate.

Das ganze könnte also so aussehen:

Code: Alles auswählen

import sys
import time
import json
from dataclasses import dataclass
import xml.etree.ElementTree as et

import requests
import telegram
import uvicorn
from fastapi import FastAPI, Request, Response
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import HTMLResponse, PlainTextResponse, RedirectResponse
from jinja2 import Environment, FileSystemLoader, select_autoescape
from RPi import GPIO

CONFIG_FILENAME = "config.json"

@dataclass
class Relay:
    link: str
    button_name: str
    button_color: str
    button_grid_position: int
    relais_pin: int
    relais_state: int
    state_enabled: bool
    state_isHM: bool
    datapoint: str
    state_inverted: bool
    relais_time: float
    msg_user: str
    msg_text: str


@dataclass
class Macro:
    link: str
    button_name: str
    button_color: str
    button_grid_position: int
    actions: str

@dataclas
class TelegrammUser:
    name: str
    id: str
    silent: bool


class Config:
    def __init__(self):
        try:
            with open(CONFIG_FILENAME, "rb") as file:
                config = json.load(file)

        except IOError:
            print(f'Could not read file "{CONFIG_FILENAME}"!')
            sys.exit(1)

        except json.JSONDecodeError:
            print(f"{CONFIG_FILENAME}: decode error!")
            sys.exit(1)

        # config
        self.ip = config["config"]["IP"]
        self.port = config["config"]["PORT"]
        self.state_update = config["config"]["STATEUPDATE"]
        self.title = config["config"]["TITLE"]
        self.table_x = config["config"]["TABLE_X"]
        self.table_y = config["config"]["TABLE_Y"]
        self.telegramm_bot_id = config["config"]["TELEGRAMM_BOT_ID"]
        self.ccu_url = config["config"]["CCU_URL"]
        self.telegramm_users = [
            TelegrammUser(**item)
            for item in config["config"]["TELEGRAMM_USERS"]
        ]
        self.relais = [
            Relay(**item)
            for item in config["relais"]
        ]
        self.macros = [
            Macro(**item)
            for item in config["macros"]
        ]

    def is_valid_user_id(self, name):
        return any(
            user.name == name and user.id
            for user in self.telegramm_users
        )

    def get_user_by_name(self, name):
        for user in self.telegramm_users:
            if user.name == name and user.id:
                return user
        return None


class Controller:
    def __init__(self):
        self.config = Config()
        self.templates = Environment(
            loader=FileSystemLoader("templates"),
            autoescape=select_autoescape(["html", "xml"]),
        )
        self._init_gpio()

    def _render(self, template, **kwargs):
        tmpl = self.templates.get_template(template)
        return tmpl.render(**kwargs)

    def _get_items(self):
        return [
            self._get_entry(i)
            for i in range(self.config.table_x * self.config.table_y)
        ] 

    def _get_entry(self, pos):
        for item in self.config.relais + self.config.macros:
            if item.button_grid_position == pos:
                return item
        return None

    def _state_javascript(self):
        return "".join(
            f'\n\t\t\t\t\t\tsetClassOnState(document.getElementById("{relay.link}"), pinStates[{index}]);'
            for index, relay in enumerate(self.config.relais)
            if relay.state_enabled
        )

    def _handle_message(self, relay):
        if not self.config.telegramm_bot_id:
            return

        try:
            bot = telegram.Bot(token=self.config.telegramm_bot_id)
            for user_name in relay.msg_user:
                user = self.config.get_user_by_name(user_name)
                if user:
                    bot.send_message(
                        chat_id=user.id,
                        text=relay.msg_text,
                        disable_notification=user.silent
                    )
        except telegram.error.TelegramError as e:
            print(f"Telegram error occurred ({e})!")

    def _state_gpio(self, relay):
        if not relay.state_enabled:
            return "unknown"
        state = GPIO.input(relay.relais_state)
        if relay.state_inverted:
            state = not state
        return "open" if state else "closed"

    def _init_gpio(self):
        GPIO.setwarnings(False)
        GPIO.cleanup()
        GPIO.setmode(GPIO.BCM)
        for item in self.config.relais:
            self._init_gpio_entry(item)

    def _init_gpio_entry(self, relay):
        if relay.relais_pin >= 0:
            GPIO.setup(relay.relais_pin, GPIO.OUT)
            GPIO.output(relay.relais_pin, True)

        if relay.state_enabled and relay.relais_state >= 0:
            GPIO.setup(relay.relais_state, GPIO.IN, pull_up_down=GPIO.PUD_UP)

    def _switch_gpio(self, relay):
        if relay.relais_pin == -1:
            return "OK"

        self._handle_message(relay)
        if relay.toggle_time == -1.0:
            GPIO.output(relay.relais_pin, not GPIO.input(relay.relais_pin))
            return "OK"

        GPIO.output(relay.relais_pin, False)
        time.sleep(relay.toggleTime)
        GPIO.output(relay.relais_pin, True)
        return "OK"

    def _get_hm_state(self, relay):
        if relay.state_enabled:
            try:
                url = self.config.ccu_url.format(relay.datapoint)
                response = requests.get(url, timeout=5)
                root = et.fromstring(response.content)
                state = bool(root.find('.//datapoint').attrib['value'])
                if relay.state_inverted:
                    state = not state
                return "open" if state else "closed"
            except Exception:
                pass
        return "unknown"

    def main_page(self):
        return self._render(
            "template.html",
            page_title=self.config.title,
            tableX=self.config.table_x,
            tableY=self.config.table_y,
            ip=self.config.ip,
            port=self.config.port,
            update_cycle=self.config.state_update,
            script_part=self._state_javascript(),
            btn_table=self._get_items(),
        )

    def states(self):
        return ",".join(
            f"{item.button_color}_{self._get_hm_state(item) if item.state_isHM else self._state_gpio(item)}"
            for item in self.config.relais
        )

    def reinit(self):
        self.config = Config()

    def action(self, page):
        # relay action
        for relay in self.config.relais:
            if page == relay.link:
                return self._switch_gpio(relay)

        # macro action
        for macro in self.config.macros:
            if page == macro.link:
                for action in macro.actions:
                    self.action(action)
                return "OK"
        return "NOK"


controller = Controller()
app = FastAPI(
    title="Door Control",
    docs_url=None,
    redoc_url=None,
    openapi_url=None,
)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=False,
    allow_methods=["GET"],
    allow_headers=["*"],
)

@app.get("/", response_class=HTMLResponse)
def main_page():
    return HTMLResponse(content=controller.main_page())

@app.get("/states", response_class=PlainTextResponse)
def states():
    return controller.states()

@app.get("/reinit")
def reinit():
    controller.reinit()
    return RedirectResponse(url="/", status_code=302)

@app.get("/{page}", response_class=PlainTextResponse)
def action(page: str):
    return controller.action(page)

@app.api_route("/health", methods=["GET", "HEAD"])
def health():
    return Response(status_code=200)

def main():
    try:
        uvicorn.run(
            app,
            host=controller.config.ip,
            port=int(controller.config.port),
            # Important for Raspberry Pi Zero:
            workers=1,
            # pure Python implementation
            loop="asyncio",
            http="h11",
            # avoid unnecessary logging
            access_log=False,
            log_level="warning"
        )
    finally:
        GPIO.cleanup()

if __name__ == "__main__":
    main()
Und hier noch der Teil vom Template, der das Rendern der Buttons in einer Datei integriert:

Code: Alles auswählen

	<body onload="updateStates();">
		<iframe style="display:none;" name="dummy"></iframe>
		<div class="grid">{{btn_table|safe}}
{%for item in btn_table %}
    {%if item %}
	<input id="{{item.link}}" type="button" class='button {{item.button_color}}' onclick="window.frames['dummy'].location.href='/{{item.link}}';" value='{{item.button_name}}' />
	{%else%}
	<input type="button" class='button black' onclick='' value='' disabled/>
    {%endif%}
{%endfor%}
		</div>
	</body>[/html]
Nach oben
Sirius3
User
Beiträge: 18431
Registriert: Sonntag 21. Oktober 2012, 17:20

Für Icons benutzt man am besten svg:

Code: Alles auswählen

				--icon_open:   url("data:image/svg+xml,%3Csvg xmlns='http://www.w3.org/2000/svg' viewBox='0 0 100 91.5'%3E%3Cpath d='m13 45.5-4.6 4.1L0 37.4 50 0l50 37.4-8.4 12.2-5.6-4.1v46h-8v-46H21v46h-8z'/%3E%3C/svg%3E") no-repeat 6% 6%;
				--icon_closed: url("data:image/svg+xml,%3Csvg xmlns='http://www.w3.org/2000/svg' viewBox='0 0 100 91.5'%3E%3Cpath d='m13 45.5-4.6 4.1L0 37.4 50 0l50 37.4-8.4 12.2-5.6-4.1v46h-8v-46H21v46h-8zm14 36h45v10H27zm0-5.5V66h45v10zm0-25.5h45v10H27z'/%3E%3C/svg%3E") no-repeat 6% 6%;
Nach oben
Antworten

Zurück zu „Showcase“