Zum Hauptinhalt springen

REST API Basics

Einloggen

Um sich bei der optiCLOUD API zu autorisieren, senden wir eine POST-Request an den /api/noauth/authorization Endpunkt. Dabei muss der Anfragekörper folgende JSON-Felder enthalten:

  1. username: Unser Nutzername bzw. unsere Email
  2. password: Unser Passwort

Zurück erhalten wir eine Liste an Tenants, zu denen wir Zugriff haben.
Mit der authorityID des Tenants, den wir Nutzen wollen, senden wir eine POST-Request an den api/auth/login Endpunkt. Diesmal muss der Anfragekörper sowohl Nutzername und Passwort, als auch unsere gewählte authorityID als Feld enthalten.
Der Endpunkt antwortet dann mit unserem Session Key bzw. token, den wir ab sofort zur Autorisierung sämtlicher Aktionen nutzen müssen.

In Python sieht das ganze aus wie folgt:

import requests
import os

# enter information
BASEURL = "https://test.opticloud.io"
CREDENTIALS = {"username":"$YOUR_USERNAME$", "password":"\"$YOUR_PASSWORD$"}
TENANT = "$YOUR_TENANT$"

# leave empty
TENANT_INFO = None
TOKENS = None
HEADER = None

# Login to Tenant
def login(baseurl, credentials, tenant):
auth_response = requests.post(baseurl + "/api/noauth/authorization", json=credentials, verify=True)
auth_response.raise_for_status()
for i in auth_response.json():
try:
if i["tenant"]["name"] == tenant:
global TENANT_INFO
TENANT_INFO = i
credentials["authorityId"] = i["authorityId"]["id"]
except KeyError:
continue
if credentials["authorityId"]:
login_response = requests.post(baseurl + "/api/auth/login", json=credentials, verify=True)
login_response.raise_for_status()
global TOKENS
TOKENS = login_response.json()
global HEADER
HEADER = {"X-Authorization": f"Bearer {TOKENS['token']}"}
return True
else:
return False

Geräteliste abrufen

Der api/tenants/{tenant_id}/devices Endpunkt kann verwendet werden, um eine Liste von Geräten eines Tenants zu erhalten. Dazu senden wir einfach eine GET-Request mit den URL-Parametern page=0 und limit=-1. Und schon erhalten wir eine Liste aller Geräte, samt IDs zurück.

Das ganze in Python:

# Get Devices from Tenant
def get_devices(baseurl, header, tenant_id):
parameters = {
"page": 0,
"limit":-1
}
devices_response = requests.get(baseurl + f"/api/tenants/{tenant_id}/devices", headers=header, params=parameters, verify=True)
devices_response.raise_for_status()
devices_list = devices_response.json()["data"]
return devices_list

OSF-Dateien eines Gerätes abfragen

Mithilfe einer device-id aus dem vorherigen Schritt, können wir mit einer GET-Request vom api/devices/{device_id}/data-files/original Endpoint die OSF-Dateien eines Geräts auflisten lassen. Dazu müssen wir vier URL Parameter angeben:

  1. startTime: Der Anfang des Zeitbereichs, von dem wir Dateien auflisten wollen. (Unix Zeitstempel mit Millisekunden-Genauigkeit)
  2. endTime: Das Ende des Zeitbereichs, von dem wir Dateien auflisten wollen. (Unix Zeitstempel mit Millisekunden-Genauigkeit)
  3. limit: Die maximale Anzahl an Dateien, die aufgelistet werden sollen. (-1 für kein Limit)
  4. page=0

In Python sieht funktioniert dies wie folgt:

# get processed osf files (data-files/original) works only if files were processed by OptiCloud
def get_processed_osf_files(baseurl,header, device_id, start_time, end_time):
parameters = {
"startTime": start_time,
"endTime": end_time,
"limit":-1,
"page": 0
}
osf_files_response = requests.get(baseurl + f"/api/devices/{device_id}/data-files/original", headers=header, params=parameters, verify=True)
osf_files_response.raise_for_status()
osf_file_list = osf_files_response.json()["data"]
return osf_file_list

OSF-Datei herunterladen

Um Dateien aus der soeben angefragten Liste herunterzuladen, müssen wir erst mit einer POST Request zum /api/device/{device_id}/data-files/CLIENT_SCOPE/download Endpunkt unseren Downloadwunsch anmelden. Dafür senden wir im Body der Anfrage eine JSON-Liste mit dem Namen der Datei, die wir herunterladen wollen.
Zurück erhalten wir einen Token, mit dem wir eine GET Request an /api/device/{device_id}/data-files/CLIENT_SCOPE/token/{download_token} senden können, um unsere Datei als Bytestring zurückzuerhalten.

Das ganze in Python implementiert:

def download_processed_osf_file(baseurl, header, device_id, filename, savedir):
prepared_filename = [filename]
download_response = requests.post(baseurl + f"/api/device/{device_id}/data-files/CLIENT_SCOPE/download", headers=header, json=prepared_filename, verify=True)
download_response.raise_for_status()
download_token = download_response.text

download_file_response = requests.get(baseurl + f"/api/device/{device_id}/data-files/CLIENT_SCOPE/token/{download_token}", headers=header, verify=True)
download_file_response.raise_for_status()
truncated_filename = str(filename).split("/")[-1]
save_path = os.path.join(savedir, truncated_filename)
with open(save_path, "wb") as f:
f.write(download_file_response.content)
return save_path
info

Falls mehrere Dateien angefragt worden, handelt bei dem heruntergeladenem Bytestring nicht um eine .osfz Datei, sondern um ein Zip-Archiv. Dieses muss dementsprechend vor der weiteren Verarbeitung erstmal entpackt werden.

Letzte Telemetriedaten von Gerät abrufen

Wenn wir von einem Gerät nur die neusten Daten brauchen, können wir diese mit einer GET-Request an den /api/plugins/telemetry/DEVICE/{device_id}/values/timeseries Endpunkt abrufen.
Dabei müssen wir auch zwei URL-Parameter, startTs und endTs, angeben. Diese werden normalerweise dafür gebraucht, um Telemetriedaten aus einer bestimmten Zeitspanne abzufragen, aber da wir nur die neuesten Werte brauchen, stellen wir beide auf die aktuelle Zeit (als Unix Zeitstempel mit Millisekunden-Genauigkeit).
Nach erfolgreicher Anfrage liefert uns der Endpunkt eine Liste aller Kanäle mit ihren entsprechenden neuesten Werten und Zeitstempeln.

In Python kann dies beispielsweise so implementiert werden:

def get_latest_telemetry(baseurl, header, device_id):
parameters = {
"startTs": time.time(),
"endTs": time.time()
}
latest_telemetry = requests.get(baseurl + f"/api/plugins/telemetry/DEVICE/{device_id}/values/timeseries",
parameters, headers=header, verify=True)
latest_telemetry.raise_for_status()

return latest_telemetry.json()

Gerätekommunikation abrufen

Um die Kommunikationsprotokolle von optiMEAS Geräten (PING, ATTN, RETURN_FILE etc.) abzurufen, kann eine GET-Request an den /api/devices/{deviceId}/communication/data Endpunkt gesendet werden. Dabei müssen wir vier URL-Parameter angeben:

  1. startTime: Der Anfang des Zeitbereichs, von dem wir Dateien auflisten wollen. (Unix Zeitstempel mit Millisekunden-Genauigkeit)
  2. endTime: Das Ende des Zeitbereichs, von dem wir Dateien auflisten wollen. (Unix Zeitstempel mit Millisekunden-Genauigkeit)
  3. limit: Die maximale Anzahl an Dateien, die aufgelistet werden sollen (maximal 2147483647).
  4. page: immer 0

Eine beispielhafte Implementierung in Python:

# Gets device communication between the given timestamps
def get_device_communication(baseurl, header, deviceId, startTs, endTs):
parameters = {
"startTime": startTs,
"endTime": endTs,
"limit": 2147483647,
"page": 0
}

communication_response = requests.get(baseurl + f"/api/devices/{deviceId}/communication/data", headers=header, params=parameters)
communication_response.raise_for_status()
return communication_response.json()

Mit einem Gerät kommunizieren

Dateiliste anfragen (LIST_FILES)

Um eine Liste von Dateien zu erhalten, die auf einem Gerät liegen, können wir eine POST-Request an den /api/devices/{deviceId}/request-list-Endpunkt senden. Dabei müssen wir im Anfragekörper folgende Parameter übergeben:

  1. startTime: Anfang des Dateisuchzeitraums. Im ISO 8601 Format anzugeben (z.B. 2025-06-15 00:51+01:00).
  2. endTime: Ende des Dateisuchzeitraums. Im ISO 8601 Format anzugeben (z.B. 2026-04-26 23:59+01:00).
  3. tag: Datei-tag, nach dem das Gerät suchen soll (meist preview).

Als Beispiel in Python:

# Request a list of files on the target device
def request_list(baseurl, header, deviceId, startTime, endTime, tag):
request_body = {
"startTime": startTime,
"endTime": endTime,
"tag": tag
}

list_request_response = requests.post(baseurl + f"/api/devices/{deviceId}/request-list",
headers=header, json=request_body)
list_request_response.raise_for_status()

Dateiupload anfragen (REQUEST_FILE)

Wenn mithilfe des LIST_FILES-Kommandos Dateien gefunden wurden, die auf dem Gerät, aber nicht auf dem Server vorhanden sind, kann mithilfe einer POST-Request an den /api/devices/{deviceId}/request-file-Endpunkt die entsprechende Datei angefordert werden.
Dazu müssen wir lediglich im Anfragekörper folgende Parameter übergeben:

  1. name: Der Name der Datei, samt absolutem Pfad (z.B. /sde/preview/20250615/20250615_133751.osfz)
  2. tag: Der Tag der Datei (meist preview).

In Python lässt sich dies wie folgt implementieren:

# Request device to upload named file to server
def request_file(baseurl, header, deviceId, fileName, fileTag):
request_body = [{
"name": fileName,
"tag": fileTag
}]

file_request_response = requests.post(baseurl + f"/api/devices/{deviceId}/request-file",
headers=header, json=request_body)
file_request_response.raise_for_status()

Gerät neustarten (REBOOT)

Wir können einem optiMEAS Gerät den Befehl zum Neustarten geben, indem wir eine POST-Request an den /api/devices/{deviceId}/reboot Endpunkt senden.

In Python beispielsweise:

def reboot_device(baseurl, header, deviceId):
reboot_response = requests.post(baseurl + f"/api/devices/{deviceId}/reboot", headers=header)
reboot_response.raise_for_status()

Benutzerdefinierte Kommandos

Mithilfe einer POST-Request zum /api/devices/{deviceId}/commands Endpunkt, können benutzerdefinierte Kommandos an das Gerät versendet werden. Dieser Endpunkt entspricht dem "Befehl senden"-Knopf unter dem Menüpunkt "Management".
Dementsprechend müssen wir neben dem frei wählbaren Körper der Anfrage auch den URL-Parameter method übergeben. Dieser kann die Werte SCPI_APP, SCPI_DEVICE_MANAGER oder BASH annehmen.

Das folgende Python-Beispiel sendet ein Kommando, um auf dem Gerät das Unix-Programm uptime ausführen zu lassen:

def send_uptime_command(baseurl, header, deviceId):
parameters = {
"method": "BASH"
}
request_body = "uptime"

uptime_response = requests.post(baseurl + f"/api/devices/{deviceId}/commands", params=parameters, json=request_body,
headers=header)
uptime_response.raise_for_status()

Ausloggen

Um uns am Ende der Sitzung wieder auszuloggen, müssen wir einfach nur den /api/auth/logout Endpunkt aufrufen, wie folgt:

def logout(baseurl, header):
logout_response = requests.post(baseurl + "/api/auth/logout", headers=header, verify=True)
logout_response.raise_for_status()
return logout_response.status_code