LG ThinQ(씽큐) 플랫폼 API 공개 및 OpenAPI 사용해보기 (스마트솔루션 API)

LG전자가 스마트홈 플랫폼 ThinQ의 웹 기반 API를 전격적으로 공개했다

https://live.lge.co.kr/2412-lg-api/

“누구나 손쉽게 스마트한 공간 구현”LG전자, 스마트홈 플랫폼 씽큐 API 전면 개방 - LG전자 뉴스

원래는 B2B 기반으로 LG전자와 파트너십을 맺은 기업의 개발자만 접근할 수 있었는데, 이젠 개인도 누구나 손쉽게 API를 통해 ThinQ 지원 LG전자 기기의 IoT 제어 솔루션을 만들수 있게 됐다

 

기존에 ThinQ RestAPI를 사용하는 방법에 대해 내가 포스팅한 블로그를 다시 되짚어보니, AWS의 MQTT 브로커 토픽을 구독하기까지 그 과정이 상당히 난해했는데, 이번에 공개된 API를 보니 정말 간단하게 GET/POST로 기기의 상태 조회 및 제어가 가능한 것을 알 수 있었다

https://yogyui.tistory.com/408

LG ThinQ REST API::파이썬 연동

본 글에서는 이번에 공개된 LG 스마트솔루션 API를 사용하는 방법에 대한 간단한 예제 코드를 다뤄보도록 한다

1. 사이트 접속 및 로그인

아래 사이트에 접속해 본인의 ThinQ 계정으로 로그인

LG 스마트솔루션 API 개발자 링크: https://smartsolution.developer.lge.com/ko/cloud/landing

::: LG | ThinQ Developer Site :: ThinQ.Cloud Home

2. OpenAPI 명세서(specification) 확인

사이트에서 API 명세를 json 포맷으로 다운받을 수 있다

openapi.json

0.20MB

 

명세는 웹사이트에서 HTML로 더 보기 쉽게 정리되어 있으니 사이트 문서를 참고하는 것을 추천

 

3. 액세스 토큰 발급

ThinQ API 사용시 필요한 것은 액세스 토큰이 전부이다 (본인의 ThinQ 계정으로 발급받은 토큰)

 

아래 사이트에서 토큰을 발급받을 수 있다

PAT 생성 사이트: https://connect-pat.lgthinq.com/login

Personal Access Tokens Publishing

 

ThinQ 계정으로 로그인한 뒤, PAT 항목에서 '새 토큰 만들기' 클릭

 

토큰 이름과 범위를 체크한 뒤 '토큰 만들기' 클릭

(별다른 이유가 없다면 일단 범위는 모두 체크해준다)

 

발급된 토큰 문자열 (thinqpat_ 접두어로 시작)을 메모장에 복사-붙여넣기 해둔다

(웹사이트를 통해 언제든 다시 확인할 수 있다)

 

4. ThinQ API 호출 예시 (python)

Python의 requests 모듈을 활용해 ThinQ OpenAPI를 호출하고 결과를 활용하는 방법에 대해 예제 코드를 몇가지 작성해봤다

4.1. 디바이스 목록 조회

본인이 ThinQ 플랫폼에 등록해놓은 IoT 기기들의 목록을 리스트업해보자

 

우선 디바이스 조회/제어 관련 시 사용되는 헤더 딕셔너리를 생성하는 함수를 만들어주자

필수로 요구되는 헤더 키(key)는 Authorization, x-message-id, x-country, x-client-id, x-api-key 가 있으며, 각 키들의 값(value)들은 API 명세에 따라 기입될 수 있도록 다음과 같이 작성해주자

import base64
import requests

thinq_api_url = "https://api-kic.lgthinq.com"

def generate_api_header() -> dict:
    message_id = "yogyui_thinq_api_tester"
    message_id_encoded = base64.urlsafe_b64encode(bytes(message_id, "UTF-8"))
    
    pat_token = "https://connect-pat.lgthinq.com 에서 발급받은 액세스 토큰"
    return {
        "Authorization": "Bearer " + pat_token,
        "x-message-id": message_id_encoded.decode()[:22],
        "x-country": "KR",
        "x-client-id": "yogyui-thinq-api-tester-1234",
        "x-api-key": "v6GFvkweNo7DK7yD3ylIZ9w52aKBU0eJ7wLXkSR3"
    }

※ Authorization 값 작성 시 PAT 토큰값 앞에 "Bearer " 문자열을 붙여줘야 한다

 

디바이스 목록은 base url에 /devices 문자열을 붙인 뒤 GET 메서드로 API를 호출하면 된다

(한국 기준 url은 https://api-kic.lgthinq.com/devices)

response = requests.get(
    thinq_api_url + "/devices", 
    headers=generate_api_header()
)

응답코드가 200이어야 API가 제대로 호출된 것 (오류 발생 시 400 혹은 401 코드가 응답될 수 있다)

In [1]: print(response.status_code, response.reason)
200 OK

 

응답 내용(content)는 bytes 객체로 반환되며, 일반 문자열(str)로 디코딩해서 읽어보자

In [2]: response.content.decode()

등록된 디바이스들은 "response" 키에 리스트 형태의 값으로 반환된다

각각의 디바이스 데이터는 deviceId, deviceInfo 2개의 키-값 형태의 딕셔너리로 구성되며, deviceInfo는 다시 deviceType, modelName, alias, reportable 4개의 키-값 형태의 딕셔너리 형태로 구성되어 있다

이 중 deviceId가 실제 개별 디바이스 조회/제어에 사용되는 값이다

디바이스 목록의 위계구조는 아래와 같다

┌ messageId
├ timestamp
└ response
    ├ deviceId
    └ deviceInfo    
        ├ deviceType
        ├  modelName  
        ├ alias
        └ reportable

(각 제품별로 고유한 값이므로 외부에 유출되지 않도록 주의~)

 

문자열로 작성된 딕셔너리 데이터를 파싱해서 디바이스 리스트 정보가 담긴 pandas 데이터프레임을 만드는 함수를 간단히 작성해봤다

import json
import pandas as pd

def get_device_list() -> pd.DataFrame:
    response = requests.get(
        thinq_api_url + "/devices", 
        headers=generate_api_header()
    )
    
    if not response.status_code == 200:
        raise ValueError(f"API Error {response.status_code} ({response.reason})")
    
    content = response.content.decode()
    obj = json.loads(content)
    dev_list = obj.get('response', [])
    temp = []
    for d in dev_list:
        info = d.get('deviceInfo')
        temp.append({
            'deviceId': d.get('deviceId'),
            'deviceType': info.get('deviceType'),
            'modelName': info.get('modelName'),
            'alias': info.get('alias'),
            'reportable': info.get('reportable'),
        })
    
    return pd.DataFrame(temp)

 

데이터프레임 생성 후 눈으로 확인해보자

dev_list_df = get_device_list()

나의 경우 집에 있는 총 10개의 LG전자 ThinQ 지원 제품이 모두 정상적으로 조회되는 것을 확인할 수 있다

4.2. 디바이스 프로파일 조회

개별 디바이스가 어떤 상태값 카테고리를 가지고 있는지, 제어 시 필요한 페이로드는 무엇인지, 지원되는 알림 기능은 무엇인지 등의 프로파일을 조회할 수 있다 (이를 정확히 알아야 조회/제어 어플리케이션을 제대로 만들 수 있다)

 

헤더는 동일하고 url에 /profile 문자열이 붙는다

url에 조회하고자 하는 기기의 deviceId 값을 기입해야 하는 것에 주의

(한국 기준 url은 https://api-kic.lgthinq.com/devices/{device_id}/profile)

device_id = "xxxxxxxxxx" # 디바이스 목록 조회시 얻은 'deviceId'값
response = requests.get(
    thinq_api_url + f"/devices/{device_id}/profile", 
    headers=generate_api_header()
)
obj = json.loads(response.content.decode())
print(obj)

 

거실에 놓고 사용중인 공기청정기 (AS300DNPA)의 프로파일을 조회해보니 다음과 같은 딕셔너리 데이터를 얻을 수 있었다

 

프로파일의 위계구조는 아래와 같다

┌ messageId
├ timestamp
└ response
    ├ notification
        ├ ...
    └ property    
        ├ ...

 

해당 기기가 상태 조회/제어한 특성(프로퍼티, property)와 알림(notification) 가능한 항목들을 확인 가능하다

예를 들어 공기청정기의 알림 항목의 경우 푸시(push) 알림이 가능하며, 

In [1]: obj.get('response').get('notification')
Out[1]: 
{'push': ['TIME_TO_CHANGE_FILTER',
  'POLLUTION_IS_HIGH',
  'LACK_OF_WATER',
  'TIME_TO_CLEAN_FILTER']}

필터 교체 (TIME_TO_CHANGE_FILTER), 공기오염도 높음(POLLUTION_IS_HIGH), 필터 청소 (TIME_TO_CLEAN_FILTER), 물 부족(LACK_OF_WATER) 4가지 종류의 푸시 알림을 사용할 수 있음을 알 수 있다 (그냥 공기청정기인데 물부족 알림은 왜 있는지 의문 ㅋㅋㅋ, 가습 공기청정기랑 딱히 구분하지는 않는 것 같다)

 

또한 프로퍼티의 경우 기기 종류별로 다양한 항목을 가지게 되는데, 아래는 공기청정기가 가지는 특성을 나열해본 것이다

특성의 이름이 직관적이기 때문에 어플리케이션 작성 시 큰 어려움은 없을 것으로 기대된다

In [2]: obj.get('response').get('property')
Out[2]: 
{
  'airPurifierJobMode': {
    'currentJobMode': {
      'type': 'enum',
      'mode': ['r', 'w'],
      'value': {
        'r': ['DUAL_CLEAN', 'AUTO', 'CIRCULATOR', 'CLEAN'],
        'w': ['DUAL_CLEAN', 'AUTO', 'CIRCULATOR', 'CLEAN']
      }
    }
  },
  'operation': {
    'airPurifierOperationMode': {
      'type': 'enum',
      'mode': ['r', 'w'],
      'value': {
        'r': ['POWER_ON', 'POWER_OFF'], 
        'w': ['POWER_ON', 'POWER_OFF']
      }
    }
  },
  'timer': {
    'absoluteHourToStart': {
      'type': 'number', 
      'mode': ['r', 'w']
    },
    'absoluteMinuteToStart': {
      'type': 'number', 
      'mode': ['r', 'w']
    },
    'absoluteStartTimer': {
      'type': 'enum',
      'mode': ['r', 'w'],
      'value': {
        'r': ['SET', 'UNSET'], 
        'w': ['UNSET']
      }
    },
    'absoluteHourToStop': {
      'type': 'number', 
      'mode': ['r', 'w']
    },
    'absoluteMinuteToStop': {
      'type': 'number', 
      'mode': ['r', 'w']
    },
    'absoluteStopTimer': {
      'type': 'enum',
      'mode': ['r', 'w'],
      'value': {
        'r': ['SET', 'UNSET'], 
        'w': ['UNSET']
      }
    }
  },
  'sleepTimer': {
    'relativeHourToStop': {
      'type': 'number', 'mode': ['r', 'w']
    },
    'relativeMinuteToStop': {
      'type': 'number', 
      'mode': ['r']
    },
    'relativeStopTimer': {
      'type': 'enum',
      'mode': ['r', 'w'],
      'value': {
        'r': ['SET', 'UNSET'], 
        'w': ['UNSET']
      }
    }
  },
  'airFlow': {
    'windStrength': {
      'type': 'enum',
      'mode': ['r', 'w'],
      'value': {
        'r': ['AUTO', 'POWER', 'MID', 'HIGH', 'LOW'],
        'w': ['AUTO', 'POWER', 'MID', 'HIGH', 'LOW']
      }
    }
  },
  'airQualitySensor': {
    'oder': {
      'type': 'number', 
      'mode': ['r']
    },
    'odor': {
      'type': 'number', 
      'mode': ['r']
    },
    'odorLevel': {
      'type': 'enum',
      'mode': ['r'],
      'value': {
        'r': ['INVALID', 'WEAK', 'NORMAL', 'STRONG', 'VERY_STRONG']
      }
    },
    'PM1': {
      'type': 'number', 
      'mode': ['r']
    },
    'PM2': {
      'type': 'number', 
      'mode': ['r']
    },
    'PM10': {
      'type': 'number', 
      'mode': ['r']
    },
    'totalPollution': {
      'type': 'number', 
      'mode': ['r']
    },
    'totalPollutionLevel': {
      'type': 'enum',
      'mode': ['r'],
      'value': {
        'r': ['INVALID', 'GOOD', 'NORMAL', 'BAD', 'VERY_BAD']
      }
    },
    'monitoringEnabled': {
      'type': 'enum',
      'mode': ['r'],
      'value': {
        'r': ['ON_WORKING', 'ALWAYS']
      }
    }
  },
  'filterInfo': {
    'filterRemainPercent': {
      'type': 'number', 
      'mode': ['r']
    }
  }
}

 

각각의 특성은 type, mode, value(type이 enum)일 경우 키-값 쌍을 갖는 것을 알 수 있다

mode가 'r'만 있을 경우 현재 상태 조회만 가능한 특성이며, 'r', 'w'가 모두 있는 경우 외부에서 사용자에 의해 제어가 가능한 항목인 것을 가리킨다

※ API 명세서에서는 각 기기별 특성 명세를 확인할 수 있다

4.3. 디바이스 상태 조회

디바이스 프로파일에서 조회된 특성(프로퍼티) 중 'read' 가능한 항목의 현재 상태값을 조회할 수 있다

 

헤더는 동일하며 url에 /state 문자열이 붙는다

(한국 기준 url은 https://api-kic.lgthinq.com/devices/{device_id}/state)

device_id = "xxxxxxxxxx" # 디바이스 목록 조회시 얻은 'deviceId'값
response = requests.get(
    thinq_api_url + f"/devices/{device_id}/state", 
    headers=generate_api_header()
)
obj = json.loads(response.content.decode())
print(obj)

 

조회 가능한 '프로퍼티'의 현재 값을 간단하게 API를 통해 읽을 수 있다

In [1]:obj.get('response')
Out[1]: 
{
  'airPurifierJobMode': {
    'currentJobMode': 'AUTO'
  },
  'operation': {
    'airPurifierOperationMode': 'POWER_ON'
  },
  'timer': {
    'absoluteStartTimer': 'UNSET', 
    'absoluteStopTimer': 'UNSET'
  },
  'sleepTimer': {
    'relativeStopTimer': 'UNSET'
  },
  'airFlow': {
    'windStrength': 'AUTO'
  },
  'airQualitySensor': {
    'PM1': 8,
    'PM2': 8,
    'PM10': 10,
    'oder': 1,
    'odor': 1,
    'odorLevel': 'WEAK',
    'totalPollution': 1,
    'totalPollutionLevel': 'GOOD',
    'monitoringEnabled': 'ALWAYS'
  },
  'filterInfo': {
    'filterRemainPercent': 38
  }
}

4.4. 디바이스 제어 (상태 변경)

디바이스 프로파일에서 조회된 특성(프로퍼티) 중 'write' 가능한 항목의 상태값을 변경(제어)할 수 있다

 

헤더는 동일하며 url에 /control 문자열이 붙는다

(한국 기준 url은 https://api-kic.lgthinq.com/devices/{device_id}/control)

또한 제어 데이터를 보내는 방식이기 때문에 POST 메서드를 사용해야 하며, API 호출 시 페이로드로 json 포맷을 보내줘야하는데, python requests에서는 아래와 같이 구현해주면 된다

device_id = "xxxxxxxxxx" # 디바이스 목록 조회시 얻은 'deviceId'값

payload = {
    "operation": {
        "airPurifierOperationMode": "POWER_OFF"
    }
}

response = requests.post(
    thinq_api_url + f"/devices/{device_id}/control", 
    headers=generate_api_header(),
    json=payload
)
obj = json.loads(response.content.decode())

5. 데모

거실 공기청정기의 전원을 껐다 켜는 제어 API를 호출하면서 상태 조회 API 호출 후 그 결과를 로그로 찍어보는 간단한 데모 코드를 작성해봤다

우선 위에서 작성한 코드들은 아래와 같이 전부 함수로 만들어줬다

import json
import base64
import requests
import pandas as pd

thinq_api_url = "https://api-kic.lgthinq.com"

def generate_api_header() -> dict:
    message_id = "yogyui_thinq_api_tester"
    message_id_encoded = base64.urlsafe_b64encode(bytes(message_id, "UTF-8"))
    
    pat_token = "https://connect-pat.lgthinq.com 에서 발급받은 액세스 토큰"
    return {
        "Authorization": "Bearer " + pat_token,
        "x-message-id": message_id_encoded.decode()[:22],
        "x-country": "KR",
        "x-client-id": "yogyui-thinq-api-tester-1234",
        "x-api-key": "v6GFvkweNo7DK7yD3ylIZ9w52aKBU0eJ7wLXkSR3"
    }

def get_device_list() -> pd.DataFrame:
    response = requests.get(
        thinq_api_url + "/devices", 
        headers=generate_api_header()
    )
    
    if not response.status_code == 200:
        raise ValueError(f"API Error {response.status_code} ({response.reason})")
    
    content = response.content.decode()
    obj = json.loads(content)
    dev_list = obj.get('response', [])
    temp = []
    for d in dev_list:
        info = d.get('deviceInfo')
        temp.append({
            'deviceId': d.get('deviceId'),
            'deviceType': info.get('deviceType'),
            'modelName': info.get('modelName'),
            'alias': info.get('alias'),
            'reportable': info.get('reportable'),
        })
    
    return pd.DataFrame(temp)

def query_device_profile(device_id: str) -> dict:
    response = requests.get(
        thinq_api_url + f"/devices/{device_id}/profile", 
        headers=generate_api_header()
    )
    obj = json.loads(response.content.decode())
    return obj.get('response')

def query_device_state(device_id: str) -> dict:
    response = requests.get(
        thinq_api_url + f"/devices/{device_id}/state", 
        headers=generate_api_header()
    )
    obj = json.loads(response.content.decode())
    return obj.get('response')

def control_device(device_id: str, payload: dict) -> bool:
    response = requests.post(
        thinq_api_url + f"/devices/{device_id}/control", 
        headers=generate_api_header(),
        json=payload
    )
    return response.status_code == 200

 

그리고 5초 간격으로 공기청정기의 전원 (operation - airPurifierOperationMode 프로퍼티)을 켜고 끄는 작업을 수행해봤다

if __name__ == '__main__':
    import time
    device_id = "xxxxxxxxxxxxxxxxxxxxxxxxxxxx"
    
    response = control_device(device_id,
        {"operation": {"airPurifierOperationMode": "POWER_ON"}}
    )

    response = query_device_state(device_id)
    print(response.get('operation').get('airPurifierOperationMode'))
    
    time.sleep(5)
    
    response = control_device(device_id,
        {"operation": {"airPurifierOperationMode": "POWER_OFF"}}
    )

    response = query_device_state(device_id)
    print(response.get('operation').get('airPurifierOperationMode'))

 

아래와 같이 현재 상태가 ON - OFF로 정상적으로 변경되는 것을 확인할 수 있다 (실제 기기 전원도 정상적으로 켜지고 꺼졌다)

6. TODO

이제 OpenAPI로 액세스 토큰(PAT)만 있으면 손쉽게 기기 상태 조회 및 제어가 가능하기 때문에 기존에 복잡하게 구현해뒀던 ThinQ 관련 코드를 간소화할 수 있을 것 같다

또한 polling 방식으로 조회하는 것도 좋지만 상태 변경에 대한 알림(notification)을 어플리케이션이 받는 것도 중요하기 때문에 다음 글에서는 푸시(Push)와 이벤트(Event) API 활용 방법에 대해서도 자세히 알아보도록 한다