Das deutsche Python-Forum

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()