Zum Hauptinhalt springen

Auf Daten zugreifen

In diesem Tutorial holen Sie ein Access-Token, senden Ihre erste GraphQL-Abfrage und lesen dann Runtime-Entities, folgen Assoziationen und fragen Zeitreihen ab: Rohwerte, Summen je Tag und Monat sowie vorab verdichtete Rollups. Alle Beispiele lesen nur.

Die Beispiele verwenden das Energiegemeinschafts-Modell aus Tutorial 3, weil es Stammdaten und Zeitreihen hat. Die Abfragemuster sind für jedes Construction Kit dieselben.

Dauer: etwa 60 Minuten.

Der Endpunkt​

Jeder Tenant hat einen GraphQL-Endpunkt am Asset Repository:

https://assets.<your-domain>/tenants/<tenant>/GraphQL

Senden Sie Abfragen als POST mit einem JSON-Body {"query": "...", "variables": {...}} und dem Header Authorization: Bearer <access-token>. Verwenden Sie den Pfad genau wie gezeigt, mit /tenants/ vor der Tenant-Id; die Schreibweise GraphQL ist die in dieser Dokumentation durchgängig verwendete.

Zum interaktiven Erkunden stellt derselbe Dienst einen Playground mit Schema-Browser und Autovervollständigung unter https://assets.<your-domain>/tenants/<tenant>/graphql/playground bereit.

Schritt 1: Ein Access-Token holen​

Schnellstart: die octo-cli-Anmeldung wiederverwenden​

Wenn Sie Tutorial 1 gefolgt sind, hält octo-cli bereits ein Token für Ihren Kontext. Für Experimente können Sie es aus der Kontextdatei lesen:

octo-cli --context <context-name> -c AuthStatus > /dev/null # refreshes the token if needed
export OCTO_TOKEN=$(jq -r '.Contexts["<context-name>"].Authentication.AccessToken' ~/.octo-cli/contexts.json)
export OCTO_GRAPHQL=https://assets.<your-domain>/tenants/<tenant>/GraphQL

Das Token ist etwa eine Stunde gültig. Es trägt die Rollen Ihres Benutzers; ein Skript, das es verwendet, kann also genau das, was Sie im Studio können.

Eigene Anwendung: Authorization Code mit PKCE​

Eine Web- oder Desktop-Anwendung, die im Namen eines Benutzers handelt, verwendet den OpenID-Connect-Flow Authorization Code mit PKCE. Ihre Administration registriert dafür einen OAuth-Client im Tenant (siehe Eine neue Webanwendung registrieren) und gewährt ihm den Scope octo_api (oder octo_api.read_only für eine Anwendung, die nur liest).

Der Ablauf:

  1. Ihre App erzeugt einen zufälligen code_verifier und dessen SHA-256-Hash, die code_challenge.
  2. Sie leitet den Browser zum Authorize-Endpunkt des Identity Service, mit acr_values=tenant:<tenant>, damit sich die Anmeldeseite des richtigen Tenants öffnet.
  3. Der Benutzer meldet sich an, der Browser kehrt mit einem code zu Ihrer redirect_uri zurück.
  4. Ihre App tauscht code plus code_verifier am Token-Endpunkt gegen Tokens.

Der Identity Service veröffentlicht alle Endpunkte in seinem Discovery-Dokument unter https://connect.<your-domain>/.well-known/openid-configuration. Eine minimale Python-Implementierung für ein lokales Werkzeug, nur mit der Standardbibliothek:

pkce_login.py
import base64, hashlib, http.server, json, os, secrets, urllib.parse, urllib.request, webbrowser

ISSUER = "https://connect.<your-domain>"
CLIENT_ID = "<your-client-id>" # registered with redirect URI http://localhost:8765/callback
TENANT = "<tenant>"
REDIRECT_URI = "http://localhost:8765/callback"

verifier = secrets.token_urlsafe(64)
challenge = base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()).rstrip(b"=").decode()
state = secrets.token_urlsafe(16)

params = {
"response_type": "code",
"client_id": CLIENT_ID,
"redirect_uri": REDIRECT_URI,
"scope": "openid profile email role octo_api offline_access",
"acr_values": f"tenant:{TENANT}",
"code_challenge": challenge,
"code_challenge_method": "S256",
"state": state,
}
webbrowser.open(f"{ISSUER}/connect/authorize?{urllib.parse.urlencode(params)}")

class Callback(http.server.BaseHTTPRequestHandler):
def do_GET(self):
query = urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query)
assert query["state"][0] == state, "state mismatch"
self.server.code = query["code"][0]
self.send_response(200); self.end_headers()
self.wfile.write(b"Login complete - you can close this window.")

server = http.server.HTTPServer(("localhost", 8765), Callback)
server.handle_request()

token_request = urllib.parse.urlencode({
"grant_type": "authorization_code",
"client_id": CLIENT_ID,
"code": server.code,
"redirect_uri": REDIRECT_URI,
"code_verifier": verifier,
}).encode()
tokens = json.load(urllib.request.urlopen(f"{ISSUER}/connect/token", data=token_request))
access_token = tokens["access_token"] # keep it in memory, never log it

Für Browser-Anwendungen verwenden Sie statt handgeschriebenem Code eine bewährte OIDC-Bibliothek (zum Beispiel angular-auth-oidc-client oder oidc-client-ts). Die Hintergründe beschreibt Authentifizierung.

Dienste ohne Benutzer: Client Credentials​

Ein Hintergrunddienst oder ein Gerät verwendet den Flow Client Credentials mit eigener Client-Id und eigenem Secret. Das behandelt Tutorial 5.

Schritt 2: Ihre erste Abfrage​

curl -s "$OCTO_GRAPHQL" \
-H "Authorization: Bearer $OCTO_TOKEN" \
-H "Content-Type: application/json" \
-d '{"query":"{ runtime { energyCommunityCustomer(first: 2) { totalCount items { rtId rtWellKnownName } } } }"}'
{
"data": {
"runtime": {
"energyCommunityCustomer": {
"totalCount": 45,
"items": [
{ "rtId": "6ac75edd1f682dc27a1b1d9f", "rtWellKnownName": "customer-018" },
{ "rtId": "6ac75ede1f682dc27a1b1e12", "rtWellKnownName": "customer-041" }
]
}
}
}
}

Das Wurzelfeld runtime enthält ein typisiertes Feld pro CK-Typ des Tenants. Sein Name ist die CK-Typ-Id in camelCase ohne Trennzeichen: aus EnergyCommunity/Customer wird energyCommunityCustomer, aus Basic.Energy/EnergyMeasurement wird basicEnergyEnergyMeasurement.

Schritt 3: Runtime-Entities, Filter und Paging​

Attribute auswählen​

Typisierte Felder bieten die Attribute des Typs als GraphQL-Felder an, Records als verschachtelte Objekte:

query Customers {
runtime {
energyCommunityCustomer(first: 10) {
totalCount
items {
rtId
customerNumber
contact { firstName lastName companyName }
}
}
}
}

Filtern​

fieldFilter filtert nach Attributwerten. Mehrere Filter werden mit UND verknüpft:

query ConsumptionAnchors {
runtime {
basicEnergyEnergyMeasurement(
first: 50
fieldFilter: [{ attributePath: "obisCode", operator: EQUALS, comparisonValue: "1-1:1.9.0 G.01" }]
) {
totalCount
items { rtId rtWellKnownName obisCode }
}
}
}

Alle Operatoren, die Textsuche und die Sortierung beschreibt Suche und Filter.

Große Ergebnisse seitenweise lesen​

Connections werden mit first (Seitengröße) und after (Cursor) geblättert. Lesen Sie pageInfo.endCursor und übergeben Sie ihn als after, bis hasNextPage den Wert false hat:

query Page($after: String) {
runtime {
basicEnergyEnergyMeasurement(first: 500, after: $after) {
pageInfo { hasNextPage endCursor }
items { rtId obisCode }
}
}
}

Setzen Sie immer first. Ohne Seitengröße liefert eine Abfrage alles, was sie findet, und das ist bei großen Tenants langsam.

Schritt 4: Assoziationen folgen​

Assoziationen erscheinen als Navigationsfelder, benannt nach der Assoziationsrolle, zum Beispiel facilities an einem Kunden oder children an einer Anlage. Da das Ziel einer Rolle mehrere Typen haben kann, verlangt ein Navigationsfeld die Liste der Zieltypen (ckTypeIds) und liefert eine Union. Wählen Sie die Felder je Typ mit einem Inline-Fragment (... on <GraphQLTypeName>):

query CustomerTree {
runtime {
energyCommunityCustomer(first: 2) {
items {
customerNumber
facilities(ckTypeIds: ["Basic.Energy/OperatingFacility"]) {
items {
... on BasicEnergyOperatingFacility {
name
children(ckTypeIds: ["EnergyCommunity/Consumer", "EnergyCommunity/Producer"]) {
items {
... on EnergyCommunityConsumer { ckTypeId meteringPointNumber partitionFactor }
... on EnergyCommunityProducer { ckTypeId meteringPointNumber productionType }
}
}
}
}
}
}
}
}
}
{
"customerNumber": "20041",
"facilities": {
"items": [
{
"name": "PV plant 041",
"children": {
"items": [
{
"ckTypeId": "EnergyCommunity/Producer",
"meteringPointNumber": "AT0099990000090012000000000000009",
"productionType": "SOLAR"
}
]
}
}
]
}
}

Die Gegenrichtung funktioniert genauso: Ein Energiemessungs-Anker erreicht seinen Zählpunkt über parent(ckTypeIds: [...]). Abfragen über beliebige Assoziationsrollen und transitive Beziehungen beschreibt Assoziationen.

Schritt 5: Stream Data abfragen​

Zeitreihen werden aus einem Archiv gelesen, adressiert über dessen rtId. Die Archive Ihres Tenants finden Sie im Studio unter Archive, oder Sie fragen GraphQL, welche Archivfamilie (das Basisarchiv und alle seine Rollups) welchen Zeitraum abdeckt:

query Coverage {
streamData {
coverageFor(rtId: "<archive-rtId>") {
archiveRtId
rtWellKnownName
status
isBase
bucketSizeMs
bucketAlignment
storedFunctions
availableFrom
availableTo
}
}
}

Ad-hoc-Abfragen laufen über streamData.transientStreamDataQuery, das vier Abfragearten bietet: simple (Zeilen), aggregation, groupingAggregation und downsampling. Das Ergebnis enthält die Spaltenliste in columns und die Daten in rows, einer geblätterten Connection von Zeilen mit cells.

Rohwerte (15-Minuten-Fenster)​

Das Rohdatenarchiv der Energiegemeinschaft ist ein Time-Range-Archiv: Jede Zeile deckt ein Fenster [window_start, window_end) ab, hier 15 Minuten. rtIds beschränkt die Abfrage auf die Anker, die Sie interessieren, arg.from/arg.to auf einen Zeitraum (UTC):

query RawValues($archive: OctoObjectId!, $anchor: OctoObjectId!, $from: DateTime!, $to: DateTime!) {
streamData {
transientStreamDataQuery {
simple(
archiveRtId: $archive
rtIds: [$anchor]
columnPaths: ["window_start", "window_end", "amount.value", "amount.unit", "obisCode", "dataQuality"]
arg: { from: $from, to: $to, queryMode: DEFAULT }
sortOrder: [{ attributePath: "window_start", sortOrder: ASCENDING }]
) {
items {
columns { attributePath }
rows(first: 96) {
totalCount
pageInfo { hasNextPage endCursor }
items { rtId timestamp cells { items { attributePath value } } }
}
}
}
}
}
}

Mit from = 2026-10-05T22:00:00Z und to = 2026-10-06T22:00:00Z (ein Kalendertag in Wien, Sommerzeit) hat das Ergebnis totalCount: 96, eine Zeile pro Viertelstunde:

{
"rtId": "6ac75edd1f682dc27a1b1dba",
"timestamp": "2026-10-05T22:15:00Z",
"cells": {
"items": [
{ "attributePath": "window_start", "value": "2026-10-05T22:00:00Z" },
{ "attributePath": "window_end", "value": "2026-10-05T22:15:00Z" },
{ "attributePath": "amount.value", "value": 0.065513 },
{ "attributePath": "amount.unit", "value": "KWh" },
{ "attributePath": "obisCode", "value": "1-1:1.9.0 G.01" },
{ "attributePath": "dataQuality", "value": "L1" }
]
}
}

Spaltenpfade sind die CK-Attributpfade in camelCase (amount.value, obisCode); Time-Range-Archive ergänzen die Fensterspalten window_start und window_end. Zeilen werden mit rows(first:, after:) geblättert, genau wie Runtime-Connections.

Summen: SUM je Register für einen Tag​

groupingAggregation aggregiert auf dem Server. Diese Abfrage summiert alle 15-Minuten-Werte eines Tages je OBIS-Code, über alle Anker des Tenants:

query DailyTotalsByObis($archive: OctoObjectId!, $from: DateTime!, $to: DateTime!) {
streamData {
transientStreamDataQuery {
groupingAggregation(
archiveRtId: $archive
groupByColumnPaths: ["obisCode"]
columnPaths: [{ attributePath: "amount.value", aggregationType: SUM }]
arg: { from: $from, to: $to, queryMode: DEFAULT }
) {
items {
columns { attributePath }
rows { items { cells { items { attributePath value } } } }
}
}
}
}
}

Die Ergebnisspalten sind obisCode und amountvalue_sum, zum Beispiel:

obisCodeamountvalue_sum (kWh)
1-1:1.9.0 G.01431,36
1-1:2.9.0 G.01474,27
1-1:2.9.0 G.01T474,01
1-1:2.9.0 G.03207,63
1-1:2.9.0 P.01T266,38

Tutorial 3 erklärt, was diese Register bedeuten.

Rollups: SUM je Tag oder Monat, ohne Rohdaten anzufassen​

Ein Jahr 15-Minuten-Werte zu summieren heißt, etwa 35.000 Zeilen pro Anker zu lesen. Rollup-Archive enthalten die Summen bereits. In einem Rollup heißen die aggregierten Spalten <column>_<function>, zum Beispiel amountvalue_sum und dataquality_max. Welches logische Attribut ein Rollup verdichtet, liefert streamData.rollupQueryMetadata(rtId:).

Tageswerte eines Ankers aus dem Tages-Rollup:

query DailyRollup($archive: OctoObjectId!, $anchor: OctoObjectId!, $from: DateTime!, $to: DateTime!) {
streamData {
transientStreamDataQuery {
simple(
archiveRtId: $archive
rtIds: [$anchor]
columnPaths: ["window_start", "window_end", "amountvalue_sum", "dataquality_max"]
arg: { from: $from, to: $to, queryMode: DEFAULT }
sortOrder: [{ attributePath: "window_start", sortOrder: ASCENDING }]
) {
items { rows(first: 31) { totalCount items { cells { items { attributePath value } } } } }
}
}
}
}

Kalender-Rollups sind an einer Referenzzeitzone ausgerichtet (zum Beispiel Europe/Vienna), ein „Tag“ beginnt also um lokale Mitternacht: 22:00Z im Sommer, 23:00Z im Winter. Übergeben Sie from und to auf diesen Grenzen.

Für Monatssummen einer Gruppe von Ankern fragen Sie das Monats-Rollup mit der Liste der Anker-Ids ab und gruppieren nach dem Fensterbeginn, wie es das Python-Beispiel unten tut.

Schritt 6: Ein vollständiges Python-Beispiel​

Das Skript berechnet je Monat zwei Kennzahlen einer Energiegemeinschaft, die Tutorial 3 erklärt: den Anteil des Verbrauchs, der aus der Gemeinschaft gedeckt wurde, und den Anteil der angebotenen Erzeugung, der nicht verteilt wurde. Es kombiniert alles aus diesem Tutorial: eine gefilterte Runtime-Abfrage mit Paging, um die Anker jedes Registers zu finden, und eine gruppierte Aggregation über das Monats-Rollup.

monthly_kpis.py
"""Monthly coverage ratio and surplus ratio of an energy community.

export OCTO_GRAPHQL=https://assets.<your-domain>/tenants/<tenant>/GraphQL
export OCTO_TOKEN=... # see step 1
python monthly_kpis.py 2026-06 2026-09
"""
import os
import sys
from datetime import datetime
from zoneinfo import ZoneInfo

import requests

ENDPOINT = os.environ["OCTO_GRAPHQL"]
TOKEN = os.environ["OCTO_TOKEN"]
VIENNA = ZoneInfo("Europe/Vienna")
MONTHLY_ROLLUP = "<monthly-rollup-rtId>" # e.g. the archive "energy-measurements-monthly"

REGISTERS = {
"consumption": "1-1:1.9.0 G.01",
"self_coverage": "1-1:2.9.0 G.03",
"offered": "1-1:2.9.0 G.01T",
"surplus": "1-1:2.9.0 P.01T",
}

ANCHORS_QUERY = """
query Anchors($obis: SimpleScalar!, $after: String) {
runtime {
basicEnergyEnergyMeasurement(
first: 500, after: $after,
fieldFilter: [{ attributePath: "obisCode", operator: EQUALS, comparisonValue: $obis }]
) {
pageInfo { hasNextPage endCursor }
items { rtId }
}
}
}
"""

MONTHLY_QUERY = """
query Monthly($archive: OctoObjectId!, $rtIds: [OctoObjectId], $from: DateTime!, $to: DateTime!) {
streamData {
transientStreamDataQuery {
groupingAggregation(
archiveRtId: $archive
rtIds: $rtIds
groupByColumnPaths: ["window_start"]
columnPaths: [{ attributePath: "amountvalue_sum", aggregationType: SUM }]
arg: { from: $from, to: $to, queryMode: DEFAULT }
) {
items { rows { items { cells { items { attributePath value } } } } }
}
}
}
}
"""


def graphql(query: str, variables: dict) -> dict:
response = requests.post(
ENDPOINT,
json={"query": query, "variables": variables},
headers={"Authorization": f"Bearer {TOKEN}"},
timeout=60,
)
response.raise_for_status() # 401, 400 (invalid query), 5xx
body = response.json()
if body.get("errors"): # execution errors arrive with HTTP 200!
raise RuntimeError(body["errors"][0]["message"])
return body["data"]


def anchors_for(obis: str) -> list[str]:
ids, after = [], None
while True:
page = graphql(ANCHORS_QUERY, {"obis": obis, "after": after})
conn = page["runtime"]["basicEnergyEnergyMeasurement"]
ids += [item["rtId"] for item in conn["items"]]
if not conn["pageInfo"]["hasNextPage"]:
return ids
after = conn["pageInfo"]["endCursor"]


def month_start(year: int, month: int) -> str:
"""Local midnight in Europe/Vienna, expressed in UTC."""
return datetime(year, month, 1, tzinfo=VIENNA).astimezone(ZoneInfo("UTC")).isoformat()


def monthly_sums(rt_ids: list[str], first: str, last: str) -> dict[str, float]:
y1, m1 = map(int, first.split("-"))
y2, m2 = map(int, last.split("-"))
y2, m2 = (y2 + 1, 1) if m2 == 12 else (y2, m2 + 1) # exclusive upper bound
data = graphql(MONTHLY_QUERY, {
"archive": MONTHLY_ROLLUP,
"rtIds": rt_ids,
"from": month_start(y1, m1),
"to": month_start(y2, m2),
})
result = {}
rows = data["streamData"]["transientStreamDataQuery"]["groupingAggregation"]["items"][0]["rows"]["items"]
for row in rows:
cells = {c["attributePath"]: c["value"] for c in row["cells"]["items"]}
local = datetime.fromisoformat(cells["window_start"].replace("Z", "+00:00")).astimezone(VIENNA)
result[local.strftime("%Y-%m")] = cells["amountvalue_sum_sum"]
return result


def main() -> None:
first, last = sys.argv[1], sys.argv[2]
sums = {name: monthly_sums(anchors_for(obis), first, last) for name, obis in REGISTERS.items()}
print(f"{'month':8} {'consumption':>12} {'self-cov.':>10} {'coverage':>9} {'surplus %':>9}")
for month in sorted(sums["consumption"]):
consumption = sums["consumption"].get(month, 0.0)
covered = sums["self_coverage"].get(month, 0.0)
offered = sums["offered"].get(month, 0.0)
surplus = sums["surplus"].get(month, 0.0)
coverage = covered / consumption if consumption else 0.0
surplus_ratio = surplus / offered if offered else 0.0
print(f"{month:8} {consumption:12.1f} {covered:10.1f} {coverage:9.1%} {surplus_ratio:9.1%}")


if __name__ == "__main__":
main()

Beispielausgabe:

month consumption self-cov. coverage surplus %
2026-06 12940.8 7950.0 61.4% 61.7%
2026-07 13372.1 8093.9 60.5% 61.4%
2026-08 13372.1 7623.0 57.0% 60.2%
2026-09 12737.3 6720.7 52.8% 58.1%

Typische Fallen​

GraphQL-Fehler kommen mit HTTP 200. Nur Transportprobleme erzeugen einen HTTP-Fehlerstatus: 401 bei fehlendem oder abgelaufenem Token, 400 bei einer Abfrage, die nicht zum Schema passt. Eine gültige Abfrage, die bei der Ausführung scheitert (ein unbekannter Spaltenpfad, eine fehlende Berechtigung auf einem Feld, eine serverseitige Ausnahme), liefert HTTP 200 mit einem errors-Array und null-Daten für den gescheiterten Teil:

{
"errors": [{ "message": "Invalid column paths: timeRange.from", "extensions": { "code": "ASSET1004" } }],
"data": { "streamData": { "transientStreamDataQuery": { "simple": null } } }
}

Prüfen Sie immer errors im Antwort-Body, wie es graphql() im Beispiel tut.

Ohne attributeNames kommen alle Attribute. Die generische Abfrage runtimeEntities liefert Attribute als Name/Wert-Paare über attributes(attributeNames: [...]). Lassen Sie das Argument weg, erhalten Sie alle Attribute der Entity, einschließlich Zugangsdaten, die an Konfigurations-Entities gespeichert sind. Übergeben Sie immer eine explizite, nicht leere Liste. Die Namen werden nur in camelCase erkannt (["host"] funktioniert, ["Host"] liefert nichts), und wenn Sie ein Record-Attribut lesen, geben Sie auch die Unterattribute des Records an. Typisierte Abfragen wie energyCommunityCustomer { contact { ... } } haben dieses Problem nicht, weil Sie jedes Feld explizit auswählen. Siehe Secret-Attribute.

Den exakten Endpunktpfad verwenden. Er lautet /tenants/<tenant>/GraphQL auf dem Host des Asset Repository, nicht auf dem Studio-Host und nicht ohne /tenants/.

Zeitzonen. Archive speichern UTC. Kalender-Rollups werden um lokale Mitternacht ihrer Referenzzeitzone geschnitten. Rechnen Sie lokale Tages- oder Monatsgrenzen vor der Abfrage in UTC um, wie es month_start() tut.

Angeschnittene Fenster. Eine Time-Range-Abfrage liefert jedes Fenster, das [from, to) überlappt. Legen Sie from und to auf das Fensterraster (volle Viertelstunden beim Rohdatenarchiv, lokale Mitternacht bei Tages-Rollups), sonst zählt ein nur teilweise enthaltenes Fenster mit seinem vollen Wert.

Typnamen. Das typisierte GraphQL-Feld ist camelCase (basicEnergyOperatingFacility), der in Fragmenten verwendete GraphQL-Typ ist PascalCase (BasicEnergyOperatingFacility), und Filter sowie ckTypeIds verwenden die CK-Typ-Id (Basic.Energy/OperatingFacility).

Zusammenfassung​

  • Ein Token holen Sie für Experimente aus octo-cli, für eine eigene benutzerbezogene App über Authorization Code + PKCE oder für einen Dienst über Client Credentials.
  • runtime hat ein typisiertes Feld pro CK-Typ, mit fieldFilter, Paging und Navigationsfeldern für Assoziationen.
  • streamData.transientStreamDataQuery liest Rohzeilen, aggregiert und gruppiert auf dem Server; Rollup-Archive liefern Tages- oder Monatssummen direkt.
  • Prüfen Sie das errors-Array auch dann, wenn der HTTP-Status 200 ist.

Weiter: Energiedaten verstehen erklärt das Modell und die Register, die Sie gerade abgefragt haben.