KNX MQTT Entegrasyonu: Grup Adreslerini MQTT Topic'lerine Köprülemek

Bir KNX MQTT köprüsü, KNX hattındaki telegramları MQTT mesajlarına, MQTT komutlarını da yeniden KNX grup yazmalarına dönüştürür. Bu eğitimde bu köprüyü açık kaynak xknx kütüphanesiyle Python'da kuracak, grup adreslerini okunabilir topic'lere eşleyecek, datapoint tiplerini JSON'a çözecek ve panellerin ve diğer binaların KNX verinizi kullanabilmesi için köprüyü bir ConnectMQ broker'a bağlayacaksınız.

KNX Neden MQTT'ye Köprülenir

KNX oda düzeyinde otomasyonda mükemmeldir. Butonlar, dimmerler, jaluzi aktüatörleri ve varlık dedektörleri grup adresleri üzerinden doğrudan birbirleriyle konuşur; böylece binadaki tüm sunucular kapalı olsa bile lambalar çalışmaya devam eder. Bu merkezi olmayan tasarım, KNX verisinin binadan nadiren çıkmasının da nedenidir: hat, kompakt ikili datapoint tipleriyle (DPT) grup telegramları konuşur ve KNX/IP yönlendirmesi internete geçmeyen multicast kullanır.

Bir MQTT köprüsü, kurulumun çalışma biçimini değiştirmeden kuzey yönlü bir arayüz ekler. İlgilenilen her grup adresi bir topic'e, değerler JSON'a dönüşür ve MQTT broker'a bağlı herhangi bir istemci bunları okuyabilir. Bu, bina otomasyon sistemi MQTT rehberimizde anlatılan entegrasyon katmanının aynısıdır.

İhtiyacınız Olanlar

• LAN üzerinden erişilebilen bir KNX/IP arayüzü veya router. Bir arayüze tunneling en yaygın kurulumdur.
• ETS projesinden, her adresin DPT'sini içeren grup adresi listesi. ETS'niz yoksa KNX entegratöründen bir dışa aktarım isteyin.
• Aynı ağda bir Linux kutusu, Raspberry Pi veya PC.
• Python 3.10 veya daha yenisi.

python -m venv knx-gw
source knx-gw/bin/activate
pip install xknx "paho-mqtt>=2.0"

Çoğu KNX/IP arayüzü yalnızca sınırlı sayıda eşzamanlı tüneli destekler; çoğunlukla bir ile birkaç arasında. ETS, bir görselleştirme ve köprünüz aynı anda bağlanırsa biri reddedilebilir. Başlamadan önce arayüzünüzün kaç tünel sunduğunu kontrol edin.

Adım 1: Grup Adreslerini Eşleyin

KNX grup adresleri 1/2/3 gibi görünür. Bunları doğrudan MQTT topic seviyeleri olarak kullanmak derin hiyerarşiler ve kafa karıştırıcı topic adları yaratır; bu yüzden her birini anlamlı bir ada ve bir DPT'ye eşleyin:

Aynı eşlemeyi köprüye bir Python sözlüğü olarak yazın. Durum ve komut adreslerini ayrı tutmak standart KNX uygulamasıdır ve MQTT'ye de güzel uyar: durum topic'lerini köprü yayınlar, komut topic'lerine abone olunur.

Adım 2: KNX MQTT Köprüsü

Köprü KNX'e tunneling ile bağlanır, tüm telegramları dinler, eşlemede olanları çözer ve yayınlar. Ayrıca .../set topic'lerine abone olur ve bunları grup yazmalarına dönüştürür.

import asyncio
import json
import uuid
from datetime import datetime, timezone

import paho.mqtt.client as mqtt
from xknx import XKNX
from xknx.dpt import DPTArray, DPTBinary, DPTScaling, DPTTemperature
from xknx.io import ConnectionConfig, ConnectionType
from xknx.telegram import GroupAddress, Telegram
from xknx.telegram.apci import GroupValueResponse, GroupValueWrite

KNX_GATEWAY_IP = "192.168.20.10"
BROKER = "broker.connectmq.com"
PORT = 1883                         # özel broker için 8883 + TLS
USERNAME = None                     # "YOUR_USERNAME"
PASSWORD = None                     # "YOUR_PASSWORD"
CLIENT_ID = f"knx-gw-{uuid.uuid4().hex[:8]}"   # "YOUR_CLIENT_ID"
PREFIX = "connectmq/demo/knx-4mq8"

# grup adresi -> (topic son eki, dpt türü)
STATE_MAP = {
    "1/1/2": ("f2/office-201/light/state", "switch"),
    "1/1/5": ("f2/office-201/light/brightness-state", "percent"),
    "3/1/1": ("f2/office-201/temperature", "temperature"),
    "4/1/1": ("f2/office-201/presence", "switch"),
}
# topic son eki -> (grup adresi, dpt türü)
COMMAND_MAP = {
    "f2/office-201/light/switch": ("1/1/1", "switch"),
    "f2/office-201/light/brightness": ("1/1/4", "percent"),
}

loop: asyncio.AbstractEventLoop | None = None
xknx: XKNX | None = None
mqttc: mqtt.Client | None = None


def decode(kind, value):
    if kind == "switch" and isinstance(value, DPTBinary):
        return bool(value.value)
    if kind == "percent" and isinstance(value, DPTArray):
        return DPTScaling.from_knx(value)
    if kind == "temperature" and isinstance(value, DPTArray):
        return round(DPTTemperature.from_knx(value), 2)
    return None


def encode(kind, value):
    if kind == "switch":
        return DPTBinary(1 if value in (True, 1, "ON", "on", "true") else 0)
    if kind == "percent":
        return DPTScaling.to_knx(max(0, min(100, float(value))))
    raise ValueError(kind)


async def telegram_received(telegram: Telegram):
    ga = str(telegram.destination_address)
    if ga not in STATE_MAP:
        return
    if not isinstance(telegram.payload, (GroupValueWrite, GroupValueResponse)):
        return
    suffix, kind = STATE_MAP[ga]
    value = decode(kind, telegram.payload.value)
    if value is None:
        return
    payload = {"ts": datetime.now(timezone.utc).isoformat(timespec="seconds"),
               "value": value, "ga": ga, "source": str(telegram.source_address)}
    # paneller son durumu hemen görsün diye retained
    mqttc.publish(f"{PREFIX}/{suffix}", json.dumps(payload), qos=1, retain=True)


async def send_knx(ga, kind, value):
    telegram = Telegram(destination_address=GroupAddress(ga),
                        payload=GroupValueWrite(encode(kind, value)))
    await xknx.telegrams.put(telegram)


def on_connect(client, userdata, flags, reason_code, properties):
    print("MQTT connected:", reason_code)
    client.subscribe(f"{PREFIX}/+/+/+/+/set", qos=1)
    client.subscribe(f"{PREFIX}/+/+/+/set", qos=1)
    client.publish(f"{PREFIX}/gateway/status", "online", qos=1, retain=True)


def on_message(client, userdata, msg):
    suffix = msg.topic[len(PREFIX) + 1:].removesuffix("/set")
    if suffix not in COMMAND_MAP:
        print("Ignored command for", suffix)
        return
    ga, kind = COMMAND_MAP[suffix]
    try:
        value = json.loads(msg.payload)
    except json.JSONDecodeError:
        value = msg.payload.decode()
    if isinstance(value, dict):
        value = value.get("value")
    # paho callback'leri başka bir thread'de çalışır; KNX yazmasını asyncio döngüsünde planla
    asyncio.run_coroutine_threadsafe(send_knx(ga, kind, value), loop)


async def main():
    global loop, xknx, mqttc
    loop = asyncio.get_running_loop()

    mqttc = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID)
    if USERNAME:
        mqttc.username_pw_set(USERNAME, PASSWORD)
        mqttc.tls_set()
    mqttc.will_set(f"{PREFIX}/gateway/status", "offline", qos=1, retain=True)
    mqttc.on_connect = on_connect
    mqttc.on_message = on_message
    mqttc.connect(BROKER, PORT, keepalive=60)
    mqttc.loop_start()

    xknx = XKNX(
        connection_config=ConnectionConfig(
            connection_type=ConnectionType.TUNNELING,
            gateway_ip=KNX_GATEWAY_IP,
        ),
        telegram_received_cb=telegram_received,
    )
    await xknx.start()
    print("KNX connected")
    try:
        await asyncio.Event().wait()   # sonsuza kadar çalış
    finally:
        await xknx.stop()
        mqttc.loop_stop()

asyncio.run(main())

Burada birkaç ayrıntı önemlidir. İki abonelik, önekin dört ve üç seviye altındaki komut topic'lerini kapsar; bu da COMMAND_MAP içindeki adlarla eşleşir. Köprü ayrıca bilmediği her grup adresini yok sayar; böylece bir topic'teki hata asla rastgele bir adrese yazamaz. Son olarak run_coroutine_threadsafe gereklidir, çünkü paho mesajları kendi ağ thread'inde iletirken xknx asyncio içinde çalışır.

Başlangıçta İlk Durumları Okumak

KNX cihazları yalnızca bir şey değiştiğinde telegram gönderir; bu yüzden yeni başlatılmış bir köprü binanın mevcut durumu hakkında hiçbir şey bilmez. Retained MQTT mesajları bunu paneller için gizler, ancak uzun bir kesintiden sonra retained değerler eskimiş olabilir. Çözüm, KNX bağlantısı kurulduktan sonra her durum adresine bir kez GroupValueRead göndermektir. ETS'de okuma bayrağı (read flag) ayarlanmış aktüatörler bir GroupValueResponse ile yanıt verir; telegram_received hem yazmaları hem yanıtları kabul ettiği için köprü bunu zaten işler.

from xknx.telegram.apci import GroupValueRead

async def read_all_states():
    for ga in STATE_MAP:
        await xknx.telegrams.put(Telegram(destination_address=GroupAddress(ga),
                                          payload=GroupValueRead()))
        await asyncio.sleep(0.2)   # hatta nazik davran

await xknx.start() satırının hemen ardından await read_all_states() çağırın. Okumalar arasındaki kısa duraklama önemlidir: bir KNX bükümlü çift hattının bant genişliği sınırlıdır ve onlarca okumayı aynı anda göndermek butonlardan gelen gerçek telegramları geciktirebilir. Bir durum adresi hiç yanıt vermiyorsa, büyük olasılıkla aktüatörün iletişim nesnesinde okuma bayrağı ayarlanmamıştır; köprüde geçici çözüm üretmek yerine entegratörden bunu ETS'de etkinleştirmesini isteyin.

Adım 3: Köprüyü Test Edin

Yayınlanan durumları izleyin:

mosquitto_sub -h broker.connectmq.com -p 1883 -i knx-view-4mq8 -t "connectmq/demo/knx-4mq8/#" -v

201 numaralı ofisteki duvar anahtarına basın. Şunu görmelisiniz:

connectmq/demo/knx-4mq8/f2/office-201/light/state {"ts": "2026-10-16T09:02:11+00:00", "value": true, "ga": "1/1/2", "source": "1.1.12"}

Şimdi lambayı MQTT'den açın ve yüzde 40'a kısın:

mosquitto_pub -h broker.connectmq.com -p 1883 -i knx-cmd-4mq8 -q 1 \
  -t "connectmq/demo/knx-4mq8/f2/office-201/light/switch/set" -m '{"value": "ON"}'
mosquitto_pub -h broker.connectmq.com -p 1883 -i knx-cmd-4mq8 -q 1 \
  -t "connectmq/demo/knx-4mq8/f2/office-201/light/brightness/set" -m '{"value": 40}'

Herkese açık MQTT test broker'ı bu ilk tur için uygundur, ancak herkes herhangi bir herkese açık topic altında okuma ve yayın yapabilir. Herkese açık ortamda test ederken gerçek bir ofisi değil, bir laboratuvar KNX setini kullanın.

Adım 4: Özel Bir Broker'a Karşı Çalıştırın

Canlı bir kurulum için ConnectMQ portalında bir broker oluşturun ve kimlik bilgilerini kullanın:

BROKER = "broker.connectmq.com"   # veya broker'ınız için gösterilen host
PORT = 8883
USERNAME = "YOUR_USERNAME"
PASSWORD = "YOUR_PASSWORD"
CLIENT_ID = "YOUR_CLIENT_ID"
PREFIX = "acme/ist01/hq"

Ardından köprüyü yeniden başlatmalardan sonra tekrar çalışması için bir systemd servisi olarak kurun:

sudo tee /etc/systemd/system/knx-mqtt.service > /dev/null <<'EOF'
[Unit]
Description=KNX MQTT bridge
After=network-online.target
Wants=network-online.target

[Service]
WorkingDirectory=/opt/knx-gw
ExecStart=/opt/knx-gw/bin/python /opt/knx-gw/bridge.py
Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target
EOF
sudo systemctl daemon-reload
sudo systemctl enable --now knx-mqtt

Yönetilen bir MQTT broker ile sahada bakımını yaptığınız tek süreç köprünün kendisidir.

Köprüyü Güvende Tutmak

Köprü KNX kurulumuna açılan bir bağlantı tutar; bu yüzden ona diğer bina kontrolörleri gibi davranın. Onu ayrılmış bir cihazda veya VLAN'da çalıştırın, işletim sistemini güncel tutun ve MQTT parolasını betiğin içinde değil, yalnızca servis kullanıcısının okuyabildiği bir ortam dosyasında saklayın. KNX kurulumunuz KNX Secure kullanıyorsa, xknx ETS'den dışa aktarılan keyring ile güvenli tunneling kullanabilir; arayüzde güvenliği devre dışı bırakmak yerine entegratörden kimlik bilgilerini isteyin.

Sorun Giderme

1. Tünel bağlantısı reddedildi. Arayüzdeki tüm tüneller kullanımda. ETS'yi veya başka bir görselleştirmeyi kapatın ya da daha fazla tüneli olan bir arayüz kullanın.
2. Hiç telegram gelmiyor. Arayüz grup telegramlarını filtreliyor olabilir. ETS'de filtre tablosunu kontrol edin veya bunun yerine bir KNX/IP router'a bağlanın.
3. Değerler anlamsız çözülüyor. Eşlemenizdeki DPT yanlış. Ölçekleme değeri (DPT 5) olarak çözülen 2 baytlık bir float (DPT 9) anlamsız sayılar üretir.
4. Köprü yeniden başlatıldıktan sonra durumlar eksik. KNX telegramları yalnızca değişiklikte gönderir. Başlangıçta her durum adresi için bir GroupValueRead gönderin veya retained MQTT mesajlarına güvenin.
5. Yinelenen komutlar. QoS 1 bir mesajı iki kez iletebilir; bir lambayı iki kez açmak zararsızdır, ancak toggle (aç/kapa geçişi) anlamından kaçının ve her zaman mutlak değerler gönderin.

Bilmeye Değer Alternatifler

Görsel bir aracı tercih ediyorsanız, bir KNX node paketiyle Node-RED aynı eşlemeyi yapabilir; Home Assistant'ın da otomasyonlar aracılığıyla MQTT'ye yayın yapabilen yerel bir KNX entegrasyonu vardır. Yukarıdaki Python köprüsü en şeffaf seçenektir ve sürüm kontrolüne alınması kolaydır. xknx.github.io adresindeki resmi xknx belgeleri desteklenen tüm DPT'leri listeler.

Bu köprünün üzerinde aydınlatma sahneleri için MQTT ile akıllı aydınlatma yazısına bakın; aynı bina BACnet HVAC da çalıştırıyorsa bunu BACnet'ten MQTT'ye gateway ile birleştirin. IoT otomasyon rehberimiz ilgili projeleri bir araya getiriyor.

KNX Kurulumunuzu ConnectMQ'ya Bağlayın

app.connectmq.com adresinde kredi kartı gerekmeden ücretsiz bir Serverless broker oluşturun; köprü için kimlik bilgileriniz birkaç dakika içinde hazır olur. ConnectMQ içindeki canlı topic gezgini her grup adresini değiştiği anda gösterir; bu da devreye almayı ve arıza bulmayı çok hızlandırır.

Sık Sorulan Sorular

Bir KNX/IP router'a mı ihtiyacım var, yoksa bir arayüz yeterli mi?

Tek bir köprü için tunneling destekli bir IP arayüzü yeterlidir. Birden fazla uygulamanın eşzamanlı erişime ihtiyaç duyduğu durumlarda router yardımcı olur, çünkü yönlendirme sınırlı sayıda tünel yerine multicast kullanır.

MQTT çökerse KNX kurulumu çalışmaya devam eder mi?

Evet. KNX cihazları grup adresleri üzerinden doğrudan iletişim kurar; köprü yalnızca görünürlük ve uzaktan kontrol ekler, yerel anahtarlama asla ona bağlı değildir.

Durum topic'leri retained olmalı mı?

Evet. KNX değerleri yalnızca değişiklikte gönderir; bu yüzden retained MQTT mesajları bir panelin bağlandıktan hemen sonra mevcut durumu göstermesini sağlar.

MQTT'den herhangi bir grup adresine gönderim yapabilir miyim?

Yapmamalısınız. Köprüde açık bir komut eşlemesi tutun; böylece MQTT'den yalnızca onaylanmış adreslere yazılabilir.