Home Assistant configuration.yaml – automatisation ONVIF + KNX

# configuration.yaml

knx:
  # KNX/IP interface on the network
  tunneling:
    host: 192.168.1.10      # KNX/IP gateway IP (e.g. MDT SCN-IP000.03)
    port: 3671

# ONVIF camera added via UI: Settings → Devices → Add Integration → ONVIF
# HA creates binary_sensor.camera_entrance_motion automatically

# automations.yaml

- alias: "ONVIF Entrance Motion → KNX DALI Scene 3"
  trigger:
    - platform: state
      entity_id: binary_sensor.camera_entrance_motion
      to: "on"
  action:
    - service: knx.send
      data:
        address: "3/0/3"     # DALI scene GA — scene 3 (entrance full)
        payload: 3            # DPT 5.001 scene number
        response: false
    - service: knx.send
      data:
        address: "3/0/10"    # DALI brightness GA
        payload: 255          # 100% brightness
        response: false

- alias: "ONVIF Motion OFF → KNX DALI return to standby (5 min)"
  trigger:
    - platform: state
      entity_id: binary_sensor.camera_entrance_motion
      to: "off"
      for: "00:05:00"
  action:
    - service: knx.send
      data:
        address: "3/0/10"
        payload: 77           # 30% brightness (standby)

Python — Événement de mouvement ONVIF vers télégramme KNX

import asyncio
from onvif import ONVIFCamera
from xknx import XKNX
from xknx.dpt import DPTBinary

CAMERA_IP = "192.168.1.64"
CAMERA_PORT = 80
CAMERA_USER = "admin"
CAMERA_PASS = "yourpassword"

KNX_GATEWAY_IP = "192.168.1.10"
KNX_MOTION_GA = "8/0/1"   # Motion alarm group address (DPT 1.001)
KNX_SCENE_GA  = "3/0/3"   # DALI scene recall GA

async def main():
    # Connect to ONVIF camera
    cam = ONVIFCamera(CAMERA_IP, CAMERA_PORT, CAMERA_USER, CAMERA_PASS)
    await cam.update_xaddrs()
    events_service = cam.create_events_service()

    # Create pull-point subscription
    pullpoint = await events_service.CreatePullPointSubscription()
    manager = cam.create_pullpoint_service(pullpoint)

    # Connect to KNX/IP gateway
    xknx = XKNX(address_format="3/3/4")
    xknx.knx_ip_interface.gateway_ip = KNX_GATEWAY_IP
    await xknx.start()

    print(f"Polling ONVIF events from {CAMERA_IP}...")

    while True:
        try:
            messages = await manager.PullMessages({
                "MessageLimit": 100,
                "Timeout": "PT1S",  # 1-second long-poll timeout
            })
            for msg in messages.NotificationMessage or []:
                topic = str(msg.Topic._value_1)
                # Filter for motion detection topic
                if "MotionDetection" in topic or "CellMotionDetector" in topic:
                    data = msg.Message._value_1.Data
                    simple_items = data.SimpleItem or []
                    for item in simple_items:
                        if item.Name in ("IsMotion", "State", "Value"):
                            is_active = str(item.Value).lower() in ("true", "1")
                            # Send KNX telegram
                            await xknx.telegrams.put(
                                xknx.telegram_queue.telegram_received_cb
                            )
                            # Write DPT 1.001 to motion alarm GA
                            from xknx.telegram import Telegram, TelegramDirection
                            from xknx.telegram.address import GroupAddress
                            t = Telegram(
                                destination_address=GroupAddress(KNX_MOTION_GA),
                                payload=DPTBinary(1 if is_active else 0),
                            )
                            await xknx.telegrams.put(t)
                            print(f"KNX {KNX_MOTION_GA} = {1 if is_active else 0}")
        except Exception as e:
            print(f"Poll error: {e}")
            await asyncio.sleep(2)

asyncio.run(main())