root
135 lines
2.9 KiB
Python
135 lines
2.9 KiB
Python
#!/usr/bin/env python3
|
|
import logging
|
|
import subprocess
|
|
from enocean.communicators.serialcommunicator import SerialCommunicator
|
|
|
|
# -----------------------------
|
|
# Konfiguration
|
|
# -----------------------------
|
|
|
|
# -----------------------------
|
|
# EnOcean
|
|
# -----------------------------
|
|
ENOCEAN_PORT = "/dev/ttyS4"
|
|
SENDER_ID = "FE:D0:A6:44"
|
|
|
|
# -----------------------------
|
|
# KNXD
|
|
# -----------------------------
|
|
KNXD_IP = "ip:172.20.21.168"
|
|
GA_FRONT = "11/1/14"
|
|
GA_REAR = "11/1/15"
|
|
|
|
# -----------------------------
|
|
# Logging
|
|
# -----------------------------
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s %(levelname)s: %(message)s"
|
|
)
|
|
|
|
log = logging.getLogger("enocean-knx")
|
|
|
|
|
|
# -----------------------------
|
|
# KNX Senden über knxd
|
|
# -----------------------------
|
|
|
|
def knx_write(ip: str, ga: str, state: bool):
|
|
"""
|
|
Sendet Boolean an KNX über knxd (groupswrite).
|
|
"""
|
|
value = "1" if state else "0"
|
|
|
|
try:
|
|
subprocess.run(
|
|
["/usr/lib/knxd/groupswrite", ip, ga, value],
|
|
check=True,
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL
|
|
)
|
|
log.info(f"KNX → {ga} = {value}")
|
|
|
|
except Exception as e:
|
|
log.error(f"KNX send failed: {e}")
|
|
|
|
|
|
# -----------------------------
|
|
# EnOcean Verarbeitung
|
|
# -----------------------------
|
|
|
|
last_button = None # "front" | "rear"
|
|
|
|
|
|
def process_packet(packet):
|
|
global last_button
|
|
|
|
try:
|
|
sender = packet.sender_hex
|
|
|
|
if sender != SENDER_ID:
|
|
return
|
|
|
|
# Rohdaten
|
|
data = packet.data
|
|
|
|
if not data:
|
|
return
|
|
|
|
event = data[1]
|
|
|
|
# -------------------------
|
|
# Fronttür (Button BO)
|
|
# -------------------------
|
|
if event == 0x70:
|
|
last_button = "front"
|
|
knx_write(KNXD_IP, GA_FRONT, True)
|
|
log.info("Front door OPEN")
|
|
|
|
elif event == 0x00 and last_button == "front":
|
|
knx_write(KNXD_IP, GA_FRONT, False)
|
|
log.info("Front door CLOSED")
|
|
|
|
# -------------------------
|
|
# Hecktür (Button BI)
|
|
# -------------------------
|
|
elif event == 0x50:
|
|
last_button = "rear"
|
|
knx_write(KNXD_IP, GA_REAR, True)
|
|
log.info("Rear door OPEN")
|
|
|
|
elif event == 0x00 and last_button == "rear":
|
|
knx_write(KNXD_IP, GA_REAR, False)
|
|
log.info("Rear door CLOSED")
|
|
|
|
except Exception as e:
|
|
log.error(f"Packet error: {e}")
|
|
|
|
|
|
# -----------------------------
|
|
# Main Loop
|
|
# -----------------------------
|
|
|
|
def main():
|
|
log.info("Starting EnOcean → KNX gateway")
|
|
|
|
communicator = SerialCommunicator(port=ENOCEAN_PORT)
|
|
communicator.start()
|
|
|
|
try:
|
|
while True:
|
|
packet = communicator.receive.get()
|
|
process_packet(packet)
|
|
|
|
except KeyboardInterrupt:
|
|
log.info("Stopping gateway")
|
|
|
|
finally:
|
|
communicator.stop()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|
|
|