[HA] Gruppenmonitor

[henfri]

Hallo,

ich würde mir Verbesserungen bei den Filtermöglichkeiten des Gruppenmonitor wünschen.
Beispiel: 1.0.1 zeigt 1.0.1*
Das kann praktisch sein, aber es kann auch ungewollt sein. Deshalb wäre mein Vorschlag, dass man es beeinflussen kann. Z.B. indem man explizit den Stern hinzufügen muss.

Zudem wäre es super, wenn man nach mehreren GAs filtern kann (ODER-Verknüpft).

Das Sahnehäubchen wäre, wenn man die Filter speichern kann.

Gruß,
Hendrik

[gbglace]

Dann Stelle doch im GitHub nen Festurerequest.

[henfri]

Kann ich gerne machen - wäre auch nicht mein Erster :-)
Aber es macht ja schon Sinn, das erstmal zu sammeln und zu konsolidieren - gerade wo die Entwickler hier anwesend sind.

Gruß,
Hendrik

[henfri]

Hallo,

meti ich habe ja schon einen Feature-Request erstellt:
https://github.com/XKNX/knx-frontend/issues/184

Besteht hier eine Chance für eine Umsetzung?
Ich habe mir das selbst einmal angesehen. Den Python-Teil könnte ich. Aber das Frontend - und prinzipiell das Entwickeln in HA - ist eine (zu) große Hürde.

Der Leidensdruck ist aber hoch. Es gibt aktuell m.W. kein Tool, mit dem ich einfach sagen kann: Sag mal, was in den letzten Stunden auf dem Kanal von dem Aktor so passiert ist.

Ich überlege sonst, selbst ein kleines Kommandozeilen-tool zu schreiben.
Die Busmonitor-Daten werden nicht irgendwo gespeichert, oder?
Wäre es möglich, diese für eine konfigurierbare Zeit (z.b. letzte 24h, Woche, ...) auf der Platte zu speichern?

Gruß,
Hendrik

[meti]

ich habe ja schon einen Feature-Request erstellt
Besteht hier eine Chance für eine Umsetzung?

Hab ich gesehen. Die Chance besteht immer, allerdings - zumindest in absehbarer Zeit - nicht von mir. Wenn du dich selbst dran trauen möchtest, bin ich allerdings gern behilflich... es hat ja jeder mal irgendwo angefangen - das HA KNX Panel war auch meine erste Berührung mit Frontends.

Ich überlege sonst, selbst ein kleines Kommandozeilen-tool zu schreiben.
Die Busmonitor-Daten werden nicht irgendwo gespeichert, oder?
Wäre es möglich, diese für eine konfigurierbare Zeit (z.b. letzte 24h, Woche, ...) auf der Platte zu speichern?

Die Daten werden nur beim Runterfahren von HA gespeichert - um sie beim Hochfahren wiederherstellen zu können.
Ich würde für solche Spezialanwendungen vielleicht wirklich eher auf ein eigenes Tool zurückgreifen. Von einer xknx-Instanz kannst du recht einfach fertig dekodierte Telegramme bekommen die du ganz nach Bedarf in irgendwelche Dateien oder Datenbanken speichern kannst.

[henfri]

Hallo,

so, mal ein Teaser:
https://photos.google.com/photo/AF1Q...ELbvqQhLYvungo

meti
ich habe dazu passend einen Logger geschrieben.
Wie kann ich mit xknx die payload mit hilfe der knxproj dekodieren?

Übrigens war Gemini fest davon überzeugt, dass daemon_mode=True als Parameter von xknx.start() und nicht beim initialisieren der Klasse hinzugefügt werden muss. War das mal so, oder haluziniert Gemini?

Gruß,
Hendrik

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
Ein Python-Tool zum Loggen des KNX-Busverkehrs in eine rotierende Log-Datei.
Dekodiert Payloads mithilfe einer ETS-Projektdatei.
Alte Logs werden um Mitternacht automatisch mit gzip komprimiert.
"""

import asyncio
import logging
import sys
from logging.handlers import TimedRotatingFileHandler
from datetime import datetime
import gzip

from xknx import XKNX
from xknx.io import ConnectionConfig, ConnectionType
from xknx.telegram import Telegram

# --- Konfiguration ---
# Tragen Sie hier den Namen Ihrer ETS-Projektdatei ein
ETS_PROJECT_FILE = "mein_projekt.knxproj"
LOG_FILE = "knx_bus.log"
# --- Ende Konfiguration ---


# --- Konfiguration des Loggings ---

# 1. Rotator-Klasse anpassen, um .gz Suffix zu verwenden
class GzipTimedRotatingFileHandler(TimedRotatingFileHandler):
    """Handler for rotating logs with gzip compression."""
    def rotator(self, source: str, dest: str) -> None:
        """Compress the source file into the destination file."""
        with open(source, "rb") as sf:
            # The destination file name is without .gz, so we add it
            with gzip.open(f"{dest}.gz", "wb") as df:
                df.writelines(sf)
        super().deleteRolledFile(source)

# 2. Logger einrichten
logger = logging.getLogger("knx_logger")
logger.setLevel(logging.INFO)

formatter = logging.Formatter('%(message)s')

# Handler erstellen, der um Mitternacht rotiert und alte Logs komprimiert
handler = GzipTimedRotatingFileHandler(
    LOG_FILE,
    when="midnight",
    interval=1,
    backupCount=30,
    encoding='utf-8'
)
handler.namer = lambda name: name.replace(f".{LOG_FILE}", "") + ".gz"
handler.setFormatter(formatter)
logger.addHandler(handler)

# --- Ende der Logging-Konfiguration ---


def log_knx_telegram(telegram: Telegram) -> None:
    """
    Wird für jedes empfangene KNX-Telegramm aufgerufen, formatiert die Nachricht
    und schreibt sie in die Log-Datei.
    """
    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
    
    payload_str = str(telegram.payload.value) if telegram.payload.value is not None else str(telegram.payload)

    # BEHOBEN: TypeError durch explizite Umwandlung der Adressen in Strings
    log_message = (
        f"{now} | "
        f"Quelle: {str(telegram.source_address):<10} | "
        f"Ziel: {str(telegram.destination_address):<15} | "
        f"Wert: {payload_str}"
    )
    logger.info(log_message)


async def main() -> None:
    """
    Hauptfunktion: Initialisiert XKNX, lädt die Projektdatei, verbindet sich
    und wartet auf Telegramme.
    """
    print(f"Starte den KNX Bus-Logger... Log-Ausgabe in: {LOG_FILE}")
    print("Drücken Sie STRG+C, um das Programm zu beenden.")
    
    logging.getLogger("xknx.log").setLevel(logging.WARNING)
    logging.getLogger("xknx.knx").setLevel(logging.WARNING)
    logging.getLogger("xknx.telegram").setLevel(logging.WARNING)

    connection_config = ConnectionConfig(
        connection_type=ConnectionType.TUNNELING,
        gateway_ip="192.168.177.9",
        gateway_port=3671,
    )
    # BEHOBEN: daemon_mode hier entfernt
    xknx = XKNX(
        telegram_received_cb=log_knx_telegram,
        #connection_config=connection_config,
        daemon_mode=True
    )
    


    try:
        # BEHOBEN: daemon_mode an der korrekten Stelle in start()
        await xknx.start()
    finally:
        print("\nLogger wird beendet, räume auf...")
        if xknx.started:
            await xknx.stop()


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        # Fängt STRG+C ab, damit das Programm nicht mit einem Fehler abbricht.
        # Der `finally`-Block in main() hat das Aufräumen bereits erledigt.
        print("\nProgramm wurde durch Benutzer beendet.")
        sys.exit(0)​

[meti]

Wie kann ich mit xknx die payload mit hilfe der knxproj dekodieren?

Wenn du das Projekt mit xknxproject parsed, kannst dir das bei der HA Integration abschauen. https://github.com/home-assistant/co...ct.py#L96-L104 und evtl. https://github.com/home-assistant/co...egrams.py#L110

Vor Kurzem war das noch etwas anders, da wurde das im Nachhinein dekodiert - kann man auch machen: https://github.com/home-assistant/co...a39b89e876L117

[henfri]

Hallo,

danke dir.
Sehe ich es richtig, dass xknx alleine das nicht kann, sondern ich mir die project.py von HA borgen muss?

from .project import KNXProject

Gruß,
Hendrik

[meti]

Nein. Du brauchst xknxproject https://github.com/XKNX/xknxproject um das Projekt zu parsen und lädst das Ergebnis dann in `xknx.group_address_dpt` bzw. suchst dir halt die Namen der GAs und Geräte raus.

[henfri]

Hallo,

Wenn ich das wörtlich interpretiere, meinst du:

knxproj: XKNXProj = XKNXProj(
path="path/to/your/file.knxproj",
)​
KNXProject = knxproj.parse()
xknx.group_address_dpt.set(KNXProject)

Richtig?

Gruß,
Hendrik

[meti]

Nein, xknx kann ein KNXProject dict nicht direkt benutzen. Schau dir doch die Type-Annotations von `xknx.group_address_dpt.set` an und die Beispiele die ich oben schon verlinkt hab.

[henfri]

Hallo,

du überschätzt meine Fähigkeiten. Type-Annotations habe ich heute zum ersten mal gehört.

Mir fehlt in den Beispielen halt der Einstieg:

        if project := data or await self._store.async_load():
            self.devices = project["devices"]
            self.info = project["info"]
            GroupAddress.address_format = self.get_address_format()
            xknx.group_address_dpt.clear()
            xknx_ga_dict: dict[DeviceAddressableType, DPTType] = {}

            for ga_model in project["group_addresses"].values():
                ga_info = _create_group_address_info(ga_model)
                self.group_addresses[ga_info.address] = ga_info
                if (dpt_model := ga_model.get("dpt")) is not None:
                    xknx_ga_dict[ga_model["address"]] = dpt_model

            xknx.group_address_dpt.set(xknx_ga_dict)​

Dabei kommt "project" aus

Store[KNXProjectModel](hass, STORAGE_VERSION, STORAGE_KEY)

oder

 project = await self.hass.async_add_executor_job(_parse_project)

Und da komme ich halt nicht weiter.
ist dieses project ein KNXProject?

Gruß,
Hendrik

[meti]

Das project da ist was bei `XKNXProj.parse()` rauskommt. Ein TypedDict - KNXProject.
Ich hab dir mal das Monitor-Example upgedated - da siehst du wie man das nutzen könnte: https://github.com/XKNX/xknx/blob/ma...ram_monitor.py

[henfri]

Danke. Das hat geholfen

https://knx-user-forum.de/forum/%C3%...llung-knx-lens