58 lines
1.9 KiB
Python
58 lines
1.9 KiB
Python
import asyncio
|
|
import logging
|
|
from datetime import datetime
|
|
|
|
from xknx import XKNX
|
|
from xknx.io import ConnectionConfig, ConnectionType
|
|
from xknx.devices import Sensor
|
|
|
|
logging.basicConfig(level=logging.INFO)
|
|
|
|
GA_TEMP = "0/0/8" # Außentemperatur
|
|
POLL_SECONDS = 60 # Abfrageintervall
|
|
TIMEOUT_SECONDS = 10 # Antwort-Timeout pro Read
|
|
|
|
async def main():
|
|
connection_config = ConnectionConfig(
|
|
connection_type=ConnectionType.TUNNELING,
|
|
gateway_ip="10.0.0.111",
|
|
gateway_port=3671,
|
|
local_ip="192.168.1.88",
|
|
route_back=True,
|
|
# Optional: festen UDP-Quellport setzen, falls NAT instabil wird
|
|
# local_port=50055,
|
|
)
|
|
|
|
async with XKNX(connection_config=connection_config, daemon_mode=True) as xknx:
|
|
temp = Sensor(
|
|
xknx=xknx,
|
|
name="Aussentemperatur",
|
|
group_address_state=GA_TEMP,
|
|
value_type="temperature",
|
|
sync_state=True,
|
|
)
|
|
xknx.devices.async_add(temp)
|
|
|
|
while True:
|
|
logging.info("Sende GroupValueRead an %s ...", GA_TEMP)
|
|
try:
|
|
try:
|
|
await temp.sync(wait_for_result=True, timeout=TIMEOUT_SECONDS)
|
|
except TypeError:
|
|
await asyncio.wait_for(temp.sync(wait_for_result=True), timeout=TIMEOUT_SECONDS)
|
|
|
|
value = temp.resolve_state()
|
|
ts = datetime.now().isoformat(timespec="seconds")
|
|
if value is None:
|
|
logging.warning("%s Aussentemperatur: None (keine verwertbare Antwort)", ts)
|
|
else:
|
|
logging.info("%s Aussentemperatur: %.2f °C", ts, value)
|
|
|
|
except asyncio.TimeoutError:
|
|
logging.warning("Timeout nach %ss: keine Antwort", TIMEOUT_SECONDS)
|
|
|
|
await asyncio.sleep(POLL_SECONDS)
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|