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:
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()
Ich hoffe externe Links sind erlaubt?
Bin für Anregungen, Verbesserungen oder auch nur Kommentare immer offen.
MfG Rudi