Weitere Schnittstellen

Neben der REST-Schnittstelle nimmt die Plattform Feld- und Zeitreihendaten über zwei Wege entgegen:

• MQTT – Geräte/Publisher senden Messwerte an einen MQTT-Broker; der mqtt-client der Plattform abonniert diesen und leitet die Daten weiter.
• OPC-UA – die Plattform (opc-ua-client) verbindet sich aktiv mit einem von Ihnen bereitgestellten OPC-UA-Server und liest dessen Variablen-Knoten.

Beide Wege münden in denselben internen Kafka-Ingestion-Topic und werden von dort in die Zeitreihen-Datenbank geschrieben:

MQTT-Publisher   ──►  MQTT-Broker  ──►  mqtt-client  ──┐
                                                       ├──►  Kafka (Ingestion-Topic)  ──►  timeseries-ingestion-consumer  ──►  Zeitreihen-DB
OPC-UA-Server    ◄──  opc-ua-client ─────────────────┘

Betrieblicher Kontext

Die Anbindung erfolgt in Abstimmung mit dem Plattform-Betreiber: Broker-Zugang und Topic-/Tenant-Zuordnung (MQTT) bzw. die Aktivierung der OPC-UA-Knoten werden plattformseitig konfiguriert. Die internen Komponenten (mqtt-client, opc-ua-client, Kafka, ts-service) validieren keine JWTs — sie sind nur im internen Netz erreichbar. Sie als Entwickler stellen die Datenquelle bereit (MQTT-Publisher bzw. OPC-UA-Server) im hier beschriebenen Format.

MQTT

Broker-Verbindung

Parameter	Wert
Protokoll	MQTT v5 (Eclipse Paho)
Broker-URL	tcp://<host>:<port> (Standard-Port 1883)
Authentifizierung	Benutzername / Passwort
QoS	konfigurierbar (Standard 0)
Verschlüsselung	im Auslieferungszustand kein TLS (tcp://)

Die Zugangsdaten und der/die abonnierten Topics werden vom Betreiber vergeben. Der mqtt-client abonniert die pro Tenant konfigurierten Topics (Wildcards möglich, z. B. int/#); das Topic bestimmt die Tenant-Zuordnung. Die Identität des Messwerts (welches Signal) steckt nicht im Topic, sondern im Feld metric der Payload.

Payload-Format

Jede MQTT-Nachricht enthält einen Datenpunkt als JSON:

{
  "timestamp": "2026-07-22T09:15:00Z",
  "metric": "water.north.level",
  "value": 621.4
}

Feld	Typ	Beschreibung
timestamp	String	Zeitstempel im ISO-8601-Format (UTC/Offset, z. B. 2026-07-22T09:15:00Z). Muss strikt Instant-parsbar sein.
metric	String	Frei wählbarer Metrik-Name (die externe Zeitreihen-ID). Wird pro Tenant beim ersten Datenpunkt automatisch als Metrik mit Herkunft MQTT angelegt; ein Signal verweist danach auf diese Metrik.
value	Number	Numerischer Messwert (numerischer String wird ebenfalls akzeptiert; nicht-numerische Werte werden verworfen). Darf nur fehlen, wenn status die Severity Bad trägt. Ein Bad-Punkt wird immer ohne Wert mit seinem Status gespeichert; eine mitgeschickte Zahl wird verworfen.
status	Number	Optional: der Status des Werts als OPC-UA-StatusCode, vorzeichenlos (2150694912 für Bad_NoCommunication) oder vorzeichenbehaftet geschrieben. Fehlt das Feld, gilt der Wert als Good. Die Bits 12 und 13 gehören der Plattform und werden verworfen.

Ist an der Metrik eine Verdichtung eingestellt (Signal- oder Metrikverwaltung), verdichtet der mqtt-client die Werte, bevor sie auf Kafka gehen: Auf dem Ingestion-Topic erscheinen dann nur die verdichteten Punkte mit dem Beginn ihres Intervalls als Zeitstempel und einem Status nach OPC UA Part 13, nicht die einzelnen Nachrichten. Ausnahme ist Bei Änderung: Eine Änderung über das Totband geht sofort und mit ihrem eigenen Zeitstempel hinaus, nur das Lebenszeichen trägt den Intervallbeginn. Ein Intervall wird nach seinem Ende plus einer Karenzzeit (MQTT_COMPRESSION_GRACE, Grundeinstellung 30 s) geschlossen; danach eintreffende Werte des Intervalls werden gezählt (hetida.mqtt.compression.late), aber nicht geschrieben. Eine schweigende Metrik wird nur ein Intervall lang fortgeschrieben: das Intervall ihres letzten Werts und das folgende, danach nichts bis zum nächsten Wert. Fällt er in das Intervall danach, geht es ohne Bruch weiter; kommt er später, beginnt die Verdichtung neu, und sein Intervall wird aus den eigenen Werten gemeldet – Minimum und Maximum vollständig, der Mittelwert ab dem ersten Wert und als Uncertain_DataSubNormal (die Zeit davor ist unbekannt), der Wert am Rasterpunkt erst ab dem nächsten Intervall. Die offenen Intervalle sichert der mqtt-client alle MQTT_COMPRESSION_CHECKPOINT_INTERVAL_MS (10 s) und beim geordneten Herunterfahren in der Plattform-Datenbank; ein Absturz verliert höchstens die Werte der letzten 10 s. Eine Nachricht ohne lesbaren timestamp oder mit einem mehr als 5 Minuten in der Zukunft wird roh durchgereicht.

Beispiel (Python)

Mit aiomqtt (pip install aiomqtt):

import asyncio
import json
from datetime import datetime, timezone

import aiomqtt


async def main() -> None:
    async with aiomqtt.Client(
        hostname="<broker-host>",
        port=1883,
        username="<mqtt-user>",
        password="<mqtt-pass>",
    ) as client:
        payload = json.dumps({
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "metric": "water.north.level",
            "value": 621.4,
        })
        # Das Topic muss unter einem abonnierten (Wildcard-)Topic liegen, z. B. int/#
        await client.publish("int/local/test", payload)


if __name__ == "__main__":
    asyncio.run(main())

OPC-UA

Anders als bei MQTT verbindet sich die Plattform aktiv mit Ihrem OPC-UA-Server und liest dessen Variablen-Knoten (überwiegend per Subscription, optional mit historischem Nachladen via HistoryRead).

Was Sie bereitstellen müssen

• Einen erreichbaren opc.tcp://<host>:<port>-Endpunkt.
• Variablen-Knoten mit numerischen Werten unterhalb der konfigurierten Root Nodes.

Sicherheitsprofil

Der Client verbindet sich aktuell ausschließlich anonym und unverschlüsselt (SecurityPolicy.None, AnonymousProvider). Ihr OPC-UA-Server muss daher einen Endpunkt ohne Security-Policy und mit anonymem Zugriff anbieten. Für den Einsatz über nicht-vertrauenswürdige Netze ist eine abgesicherte Transportstrecke (z. B. VPN) einzuplanen.

Server-Konfiguration

Die anzubindenden Server pflegen Sie in der Plattform selbst, unter Administration → OPC-UA-Server. Änderungen wirken ohne Neustart: der opc-ua-client gleicht seine Verbindungen im Minutentakt mit der Konfiguration ab.

Je Server hinterlegen Sie:

Feld	Beschreibung
Name	Bezeichnung innerhalb Ihres Mandanten, jederzeit änderbar
Adresse und Port	Der opc.tcp://<host>-Endpunkt und sein Port
Metrik-Muster	Vorlage für den Metrik-Namen. Platzhalter: NODE_ID/ID, BROWSE_NAME, DISPLAY_NAME, UUID (z. B. ergibt opcua.NODE_ID.BROWSE_NAME → opcua.<NodeId>.<BrowseName>). Pflichtangabe und danach nicht mehr änderbar, weil es die Metrik-Namen der Knoten bestimmt. Der Name entsteht erst, wenn ein Variablenknoten einem Signal zugeordnet wird — nicht beim Browsen; ein nicht zugeordneter Knoten hat keine Metrik. Einzige Ausnahme: ein aus einer älteren Fassung übernommener Server, dem noch keines hinterlegt ist — dort tragen Sie es einmalig nach, andernfalls sucht der Server nicht nach Knoten
In Betrieb	Außer Betrieb genommene Server werden getrennt; Knoten, Signale und Zeitreihen bleiben erhalten
Nach neuen Knoten suchen	Ob und wie oft der Adressraum unterhalb der Root Nodes durchsucht wird
Suchzeitplan	Wann gesucht wird, als Quartz-Cron-Ausdruck in der Zeitzone des Mandanten: Sekunde Minute Stunde Tag Monat Wochentag, genau eines von Tag und Wochentag ist ? (0 30 2 * * ? sucht täglich um 2:30 Uhr). Unterschiedliche Uhrzeiten entzerren mehrere Server gegeneinander
Historische Daten	Voreinstellung für das Nachladen per HistoryRead samt Startzeitpunkt. Als Startzeitpunkt sind ein ISO-8601-Zeitstempel (2026-01-01T00:00:00Z) oder das Schlüsselwort now zulässig; ohne Angabe gilt now
Nachladezeitplan	Wann fehlende Werte nachgeholt werden, ebenfalls als Quartz-Cron-Ausdruck. Voreinstellung: stündlich um halb

Je Server pflegen Sie außerdem die Root Nodes, ab denen die Traversierung des Adressraums startet. Sie tragen nur ihre Knoten-Kennung — ob, ab wann und wann historische Daten geholt werden, entscheiden Sie einmal am Server.

Lücken im Datenstrom

Steht Ihr Server still — Wartung, Netzausfall, nicht erreichbar —, liefert die Subscription die versäumten Werte später nicht nach. Ist der Abruf historischer Daten eingeschaltet, schließt die Plattform solche Lücken selbst: Sie merkt sich je Knoten, bis zu welchem Zeitpunkt die Historie abgeholt ist, und holt regelmäßig nach, was seither dazugekommen ist. Ein wiederholter Abruf richtet keinen Schaden an, weil bereits gespeicherte, unveränderte Werte verworfen werden.

Wann nachgeladen wird, bestimmen Sie je Server über den Nachladezeitplan — etwa nachts oder außerhalb einer Schicht, wenn die Anlage tagsüber nicht zusätzlich belastet werden soll. Ein Lauf ist zusätzlich in Menge und Dauer begrenzt; läuft er in diese Grenze, hört er zwischen zwei Zeitscheiben auf und der nächste setzt genau dort fort. Es geht dabei nichts verloren, ein Nachlauf wird nur später fertig.

Vier Eigenschaften sollten Sie dabei kennen:

• Der Nachlauf endet einige Minuten vor der Gegenwart — die neuesten Werte kommen ohnehin über die Subscription. Eine beendete Lücke ist daher wenige Stunden später geschlossen, nicht sofort.
• Nachgeladen werden kann nur, was Ihr OPC-UA-Server noch vorhält — und in der Auflösung, in der er es vorhält. Viele Server führen ihre Historie gröber als sie Werte über die Subscription meldet; eine geschlossene Lücke kann daher weniger Werte enthalten als ein gleich langer Zeitraum im Livebetrieb. In einer Messung an einer Anlage waren es rund 6 Werte je Stunde gegenüber 42 im Livebetrieb. Die Lücke ist damit gefüllt, aber nicht ununterscheidbar vom Normalbetrieb.
• Reicht die Aufbewahrung des Servers nicht so weit zurück wie die Lücke, bleibt sie bestehen. Aus demselben Grund reicht der Nachlauf von sich aus höchstens 60 Tage zurück (Betreibereinstellung): Was älter ist, holt die Plattform nicht mehr aus der Anlage, sondern erwartet es aus dem Archiv per Dateiimport.
• Je seltener der Zeitplan, desto später ist eine Lücke geschlossen. Ein nächtlicher Lauf schont die Anlage, hält eine tagsüber entstandene Lücke aber bis zur Nacht offen.

Ohne eingeschalteten historischen Abruf bleiben Lücken dauerhaft bestehen.

Ob eine Verbindung tatsächlich zustande kommt, sehen Sie in der Übersicht in der Spalte Verbindung — mit dem letzten Fehler und dem Zeitpunkt der letzten Knotensuche. Die Spalte Historie vollständig bis daneben zeigt, wie weit der Nachlauf gekommen ist: den ältesten Stand über alle Knoten des Servers, und im Tooltip den letzten Nachlauf, die Zahl der noch nie abgeholten Knoten und die der Knoten, für die der Server keine Historie führt. Je Knoten steht derselbe Stand in der Knotenliste in der Spalte Historie abgeholt bis; wer dort sortiert, sieht zuerst die Knoten, deren Lücke noch offen ist. Vor dem Speichern können Sie die eingegebenen Parameter über Verbindung testen prüfen.

Betrieb

Der opc-ua-client ist auf genau eine Instanz ausgelegt. Mehrere Instanzen würden dieselben Knoten doppelt abonnieren und damit doppelte Werte erzeugen.

Die Aktivierung und Metrik-Zuordnung einzelner Knoten erfolgt anschließend über die OPC-UA-Endpunkte der REST-Schnittstelle (Tag opc-ua, JWT-geschützt, administrativ) — z. B. Server auflisten, Objekt-/ Variablen-Knoten lesen und Knoten aktiv schalten. Ein Signal wird einem Variablenknoten zugeordnet, indem es mit opcUaNodeId (der Id des gebrowsten Knotens) angelegt wird; die Plattform bildet dann aus dem Metrik-Muster des Servers den Metrik-Namen, legt die Metrik mit Herkunft OPC_UA an und verknüpft Knoten und Signal damit. Diese Endpunkte konfigurieren die Anbindung; sie sind kein Ingestion-Weg für Nutzdaten.

Das interne Kafka-Ingestion-Format

Sowohl mqtt-client als auch opc-ua-client erzeugen dieselbe JSON-Hülle und senden sie an den Ingestion-Topic (Umgebungsvariable KAFKA_TOPIC_TIMESERIES_INGESTION, Standard het.ingestion.timeseries). Von dort schreibt der timeseries-ingestion-consumer die Punkte weg.

Intern

Kafka ist ein internes Transportformat ohne externe Authentifizierung. In der Regel produzieren Sie nicht direkt nach Kafka, sondern über MQTT/OPC-UA. Das Format ist hier zum Verständnis und zur Fehlersuche dokumentiert.

{
  "tenantId": "customer",
  "topic": "int/local/test",
  "originalPayload": "{\"timestamp\":\"2026-07-22T09:15:00Z\",\"metric\":\"water.north.level\",\"value\":621.4}",
  "payloads": [
    { "timestamp": "2026-07-22T09:15:00Z", "metric": "water.north.level", "value": "621.4" }
  ]
}

Feld	Beschreibung
tenantId	Ziel-Tenant (aus Topic-Zuordnung bzw. OPC-UA-Server-Konfiguration)
topic	Ursprungs-Topic (nur informativ)
origin	Herkunft der Daten: MQTT (mqtt-client), OPC_UA (opc-ua-client), IMPORT (data-fetcher). Optional — fehlt das Feld, gilt UNKNOWN.
originalPayload	Die unveränderte Original-Nachricht (nur informativ)
payloads[]	Ein oder mehrere Datenpunkte (timestamp, metric, value, optional status). MQTT liefert genau einen, OPC-UA batcht mehrere.

Der Consumer legt pro Tenant und metric automatisch eine Metrik mit der gelieferten Herkunft an und schreibt die Punkte über ts-service. Die Herkunft einer Metrik wird einmal gesetzt; nur UNKNOWN wird beim ersten Datenpunkt mit bekannter Herkunft nachgetragen, eine abweichende Herkunft wird protokolliert und ignoriert. Der value wird dabei zu einer Zahl (Double) konvertiert; der timestamp muss strikt ISO-8601 sein. status ist der OPC-UA-StatusCode des Punkts; der Consumer speichert ihn ohne die Plattformbits 12/13 – siehe Status eines Datenpunkts. Fehlt er, gilt Good. Ein Punkt mit Severity Bad darf ohne value kommen und wird immer ohne Wert gespeichert, auch wenn eine Zahl mitkommt.

Benachrichtigungen (Notifications)

Ein zweiter interner Kafka-Topic (KAFKA_TOPIC_NOTIFICATION, Standard het.notification) transportiert Benachrichtigungen, die der notification-consumer in die Datenbank schreibt. Die Nutzlast enthält ein notification-Objekt (subject, text, entryTimestamp, acknowledgmentType, …) sowie recipientUsernames. Produzenten sind interne Plattform-Komponenten; für Anwendungen sind Benachrichtigungen über die REST-Schnittstelle (Tag notifications) zugänglich.

Nicht für externe Dateneinspeisung vorgesehen

Zur Klarstellung: Es existieren zwei REST-Schreibpfade für Zeitreihen, die intern bzw. für den hetida designer bestimmt sind und nicht als Geräte-Ingestion-API dienen:

• ts-service POST /timeseries – interner Bulk-Insert (kein Auth, benötigt eine bereits existierende metricId); wird ausschließlich vom timeseries-ingestion-consumer aufgerufen.
• core-backend POST /api/adapter/timeseries – Bestandteil des hetida-designer-Adapters (JWT-geschützt, administrative Rollen); dient der Datenanbindung des Designers, nicht der Geräte-Einspeisung.

Für die Einspeisung von Feld-/Sensordaten nutzen Sie daher MQTT oder OPC-UA wie oben beschrieben.