ThingsBoard API 사용하기 with Python

본 문서는 Python으로 ThingsBoard API를 사용해 데이터를 받아오는 방법을 담고 있다.

ThingsBoard

ThingsBoard는 IoT 애플리케이션의 빠른 생성 및 확장을 가능하게 하는 Java 기반 오픈 소스 IoT 플랫폼이다. 데이터 수집, 처리, 시각화 및 장치 관리에 사용됩니다. MQTT, CoAP 및 HTTP는 장치 연결을 제공하기 위해 Thingsboard에서 지원하는 표준 프로토콜이며 클라우드 및 로컬 배포를 모두 지원한다. ThingsBorad는 엄청 화려하게(?) 많은 기능들이 갖추어져 있다. 문서화도 엄청 잘 되어 있다.

1. ThingsBoard Swagger-UI

본인의 ThingsBoard에서 URL을 {ThingsBoard URL}/swagger-ui/#/로 바꾸면 Swagger-UI 페이지로 연결된다.

Authorize 버튼을 눌러 로그인을 해야 이용할 수 있다.

그림1

어마어마하게 많은 종류의 API를 테스트할 수 있다. 이 중 로깅되고 있는 데이터를 받아오는 API와 API 사용을 위해 주기적으로 업데이트 해야 하는 JWT Token 생성 API를 사용해보았다.

1-1. 데이터 불러오기

telemetry-controller 섹션에서 Get time-series data 항목이 있다.

그림1

이 중 가장 최신 값을 불러오는 API를 사용해보았다.

그림1

entityType: 보통 ‘DEVICE’ 값이다.
entityId: 장치 아이디
- Home > 개체 > 장치 > 장비 > 장치 아이디 복사

요청 변수를 잘 입력하면 다음과 같은 결과를 얻을 수 있다.

그림1

Curl의 X-Authorization 부분에 JWT Token 값이 필요하다. 이 값은 주기적으로 갱신된다. 따라서 이 값도 API를 사용하여 얻는 것이 좋다.

1-2. JWT Token 불러오기

login-endpoint 섹션에 하나의 API가 있다.

그림1

username: ThingsBoard ID
password: ThingsBoard Password

요청 변수를 잘 입력하면 갱신된 JWT Token을 얻을 수 있다.

그림1

2. Python으로 ThingsBoard API 사용하기

2-1. Import Library

import requests
import json

2-2. JWT Token 가져오기

def get_token():
    login_url = f"{thingsboard_url}/api/auth/login"

    data = {
        "username": 'username',
        "password": 'password'
    }

    response = requests.post(login_url, json=data)
    js = json.loads(response.content)

    token = js['token']
    
    return token

2-3. 데이터 불러오기

Swagger-UI에서 확인하였던 Curl 에 맞추어 header와 요청변수를 맞추어 값을 얻는다. 결과가 바로 DataFrame으로 바꾸기에는 복잡해서 result2dict를 만들어 각 값만 얻어낼 수 있도록 하였다.

DEVICE_KEY = 'DEVICE_KEY'

def result2dict(js):
    key_value_dict = {}

    for key, value in js.items():
        key_value_dict[key] = value[0]['value']

    return key_value_dict

def get_timeseries_data():

    # JWT TOKEN 가져오기
    JWT_TOKEN = get_token()

    headers={
        "accept":"application/json",
        "Authorization":f"Bearer {JWT_TOKEN}"
    }

    #-----전체 값 가져오기
    all_url = f'{thingsboard_url}/api/plugins/telemetry/DEVICE/{DEVICE_KEY}/values/timeseries'
    response = requests.get(url=all_url, headers=headers)
    all_js = json.loads(response.content)
    all_dict = result2dict(all_js)

    #-----특정값들만 가져오기(%2C)로 연결
    values = 'keys=irrigation%2Cdrain%2Ctemperature'
    values_url = f'{thingsboard_url}/api/plugins/telemetry/DEVICE/{DEVICE_KEY}/values/timeseries?{values}'
    values_response = requests.get(values_url, headers=headers)
    values_js = json.loads(values_response.content)
    values_dict = result2dict(values_js)

all_dict와 values_dict를 출력한 결과이다.

# 전체 값
{'temperature': '21.68', 'humidity': '42.3', 'light': '0', 'co2': '403', 'uptime': '2805336', 'firmware': '2.4.0', 'user2': '0', 'serial': '08B61F753CC8', 'irrigation': '5559', 'pcb_temp': '26.1875', 'drain': '3145', 'medium': '16184', 'user1': '0', 'user3': '0'}
# 특정 값
{'irrigation': '5559', 'drain': '3145', 'temperature': '21.68'}