Apple 홈킷 - 외부 대기오염(Air-quality) 센서 추가하기

며칠전에 공공데이터포털에서 Open API로 대기오염정보를 가져오는 글을 포스트했다

공공데이터포털::대기오염정보 조회 (REST API)

공공데이터포털::대기오염정보 조회 (REST API)

여러 가지 서비스를 제공받을 수 있는데, 이 중 "측정소별 실시간 측정정보" 서비스를 통해 특정 측정소에서 1시간 주기로 측정된 미세먼지(PM10)/초미세먼지(PM2.5)/아황산가스/일산화탄소/오존/이산화질소 농도값을 얻을 수 있다

 

공짜로 공기질 측정 센서를 가진 것과 마찬가지로 활용할 수 있는데, 내가 이제껏 사용해온 homebridge의 mqttthing 플러그인에도 마침 Air Quality Sensor 액세서리 항목이 있어서 별도로 복잡한 구현없이 기존의 홈네트워크 시스템에 통합할 수 있을 것 같아 한 번 시도해봤다

https://github.com/arachnetech/homebridge-mqttthing/blob/master/docs/Accessories.md#air-quality-sensor

 

전체 코드를 공유하면 글이 너무 길어지니, 자세한 구현 방식은 기존글들을 참고하기 바란다

광교아이파크::Bestin ↔ Apple HomeKit 연동 Summary (1)
광교아이파크::Bestin - Apple 홈킷 연동 소스코드 GitHub 업로드

공공데이터포털::대기오염정보 조회 (REST API)

※ 공공데이터포털(data.go.kr)에 회원 가입 후 해당 API의 활용 신청을 해야한다

 

아래 구현 항목들은 모두 깃헙 저장소에 신규로 커밋해두었다

1. Air quality sensor 클래스 추가

Open API로 공기질 측정값을 가져오는 클래스를 다음과 같이 추가해준다

class AirqualitySensor(Device):
    """
    공공데이터포털 - 대기오염정보
    """
    _api_key: str = ''
    _obs_name: str = ''
    _last_query_time: datetime.datetime = None

    def __init__(self, **kwargs):
        self._measure_data = {
            'khaiGrade': -1,
            'so2Value': 0.0,
            'coValue': 0.0,
            'o3Value': 0.0,
            'no2Value': 0.0,
            'pm10Value': 0.0,
            'pm25Value': 0.0,
        }
        super().__init__('Airquality', **kwargs)

    def setApiParams(self, api_key: str, obs_name: str):
        self._api_key = api_key
        self._obs_name = obs_name

    def refreshData(self):
        if self._last_query_time is None:
            call_api = True
        else:
            tmdiff = datetime.datetime.now() - self._last_query_time
            if tmdiff.seconds > 3600:
                call_api = True
            else:
                call_api = False

        if call_api:
            url_base = "http://apis.data.go.kr/B552584/ArpltnInforInqireSvc"
            url_spec = "getMsrstnAcctoRltmMesureDnsty"
            url = url_base + "/" + url_spec
            api_key_decode = requests.utils.unquote(self._api_key, encoding='utf-8')
            params = {
                "serviceKey": api_key_decode,
                "returnType": "xml",
                "stationName": self._obs_name,
                "dataTerm": "DAILY",
                "ver": "1.3",
                "numOfRows": 1,
                "pageNo": 1
            }
            response = requests.get(url, params=params)
            if response.status_code == 200:
                xml = BeautifulSoup(response.text.replace('\n', ''), "lxml")
                result_code = xml.find('resultcode').text
                result_msg = xml.find('resultmsg').text
                if result_code == '00':
                    items = xml.findAll("item")
                    if len(list(items)) >= 1:
                        item = items[0]
                        self._measure_data['dataTime'] = item.find('dataTime'.lower()).text
                        self._measure_data['so2Value'] = float(item.find('so2Value'.lower()).text)
                        self._measure_data['coValue'] = float(item.find('coValue'.lower()).text)
                        self._measure_data['o3Value'] = float(item.find('o3Value'.lower()).text)
                        self._measure_data['no2Value'] = float(item.find('no2Value'.lower()).text)
                        self._measure_data['pm10Value'] = float(item.find('pm10Value'.lower()).text)
                        self._measure_data['pm25Value'] = float(item.find('pm25Value'.lower()).text)
                        self._measure_data['khaiValue'] = float(item.find('khaiValue'.lower()).text)
                        self._measure_data['khaiGrade'] = int(item.find('khaiGrade'.lower()).text)
                        self._last_query_time = datetime.datetime.now()
                else:
                    writeLog(f"API Error ({result_code, result_msg})", self)
                    self.clearData()
            else:
                writeLog(f"Request GET Error ({response.status_code})", self)
                self.clearData()

    def publish_mqtt(self):
        self.refreshData()
        obj = {
            "timestamp": self._measure_data.get('dataTime'),
            "grade": self._measure_data.get('khaiGrade'),
            "so2": self._measure_data.get('so2Value'),
            "co": self._measure_data.get('coValue'),
            "o3": self._measure_data.get('o3Value'),
            "no2": self._measure_data.get('no2Value'),
            "pm10": self._measure_data.get('pm10Value'),
            "pm25": self._measure_data.get('pm25Value')
        }
        if self.mqtt_client is not None:
            self.mqtt_client.publish(self.mqtt_publish_topic, json.dumps(obj), 1)

기존에 구현해둔 Device 클래스를 상속해서 다른 객체(전등, 환기, 밸브 등)와 유사하게 사용할 수 있다

공기질 측정값들은 1시간에 1번씩 업데이트되기 때문에, 쿼리 명령(publish_mqtt)이 들어왔을 때 최근 조회 시간과 현재 시간과의 차이가 1시간(3600초) 이내라면 이전의 측정값을 보낼 수 있도록 구현했다

 

위 코드에서 mqtt publish 시 json으로 보내는 요소들의 상세 정보는 다음과 같다

• timestamp: API에서 제공하는 데이터 측정 시간, 문자열(%Y-%m-%d %H:%M)
• grade: 통합대기환경지수, 정수 (1: 좋음, 2: 보통, 3: 나쁨, 4: 매우나쁨)
• so2: 아황산가스(이산화황) 농도, 실수 (단위: ppm)
• co: 일산화탄소 농도, 실수 (단위: ppm)
• o3: 오존 농도, 실수 (단위: ppm)
• no2: 이산화질소 농도, 실수 (단위: ppm)
• pm10: 미세먼지(PM10) 농도, 실수 (단위: ug/㎥)
• pm25: 초미세먼지(PM2.5) 농도, 실수 (단위: ug/㎥)

2. Home 클래스에 공기질 센서 인스턴스 추가

위에서 구현한 AirQualitySensor 클래스의 인스턴스를 기존 Home 클래스에 추가해주자

class Home:
    ##
    airquality: AirqualitySensor
    ##
    
    def __init__(self, room_info: List = None, name: str = 'Home', init_service: bool = True):
        ##
        self.airquality = AirqualitySensor(mqtt_client=self.mqtt_client)
        ##
        self.device_list.append(self.airquality)
    
    def load_config(self, filepath: str):
        ##
        node = root.find('airquality')
        mqtt_node = node.find('mqtt')
        self.airquality.mqtt_publish_topic = mqtt_node.find('publish').text
        apikey = node.find('apikey').text
        obsname = node.find('obsname').text
        self.airquality.setApiParams(apikey, obsname)

device_list 리스트에 인스턴스를 포함했기 때문에, 나머지 모듈들과 마찬가지로 쓰레드(ThreadMonitoring)에서 일정 간격으로 mqtt publish를 수행할 수 있다

다른 모듈과 마찬가지로 공기질 센서도 로컬의 XML 파일에서 관련 설정들을 불러오게 만들었는데, 공공데이터포털의 API Key 문자열과 측정소 위치를 편하게 수정할 수 있도록 구현했다 (하드코딩은 죄악이다)

 

Device 상속 객체 하나만 추가하면 나머지는 수정 사항이 거의 없다!

이거슨 바로 모듈화 설계의 힘

3. Config XML 포맷 추가

위에서 구현한 대로 XML에도 다음과 같이 공기질 센서와 관련된 항목을 추가해주면 된다

    <airquality>
        <apikey>your api key from data portal</apikey>
        <obsname>name of observatory</obsname>
        <mqtt>
            <publish>home/ipark/airquality/state</publish>
            <subscribe>home/ipark/airquality/command</subscribe>
        </mqtt>
    </airquality>

<apikey> 태그에는 자신이 공공데이터포털에서 발급받은 API Key를 "인코딩된 형태"로 집어넣고, <obsname> 태그에는 자신의 집 근처 관측소의 관측소 명을 집어넣으면 된다

관측소명은 https://airkorea.or.kr/web/stationInfo?pMENU_NO=93 에서 확인할 수 있다

4. Homebridge 액세서리 추가

마지막으로 Homebridge - mqttthing 플러그인을 사용하는 공기질센서 액세서리를 추가해주자

{
    "accessory": "mqttthing",
    "type": "airQualitySensor",
    "name": "Air Quality Sensor (Outer)",
    "url": "mosquitto broker url",
    "username": "mosquitto auth id",
    "password": "mosquitto auth password",
    "topics": 
    {
        "getAirQuality": {
            "topic": "home/ipark/airquality/state",
            "apply": "return JSON.parse(message).grade;"
        },
        "getPM10Density": {
            "topic": "home/ipark/airquality/state",
            "apply": "return JSON.parse(message).pm10;"
        },
        "getPM2_5Density": {
            "topic": "home/ipark/airquality/state",
            "apply": "return JSON.parse(message).pm25;"
        },
        "getOzoneDensity": {
            "topic": "home/ipark/airquality/state",
            "apply": "return JSON.parse(message).o3;"
        },
        "getNitrogenDioxideDensity": {
            "topic": "home/ipark/airquality/state",
            "apply": "return JSON.parse(message).no2;"
        },
        "getSulphurDioxideDensity": {
            "topic": "home/ipark/airquality/state",
            "apply": "return JSON.parse(message).so2;"
        },
        "getCarbonMonoxideLevel": {
            "topic": "home/ipark/airquality/state",
            "apply": "return JSON.parse(message).co;"
        }
    },
    "airQualityValues": [-1, 0, 1, 2, 3, 4],
    "history": false,
    "room2": false,
    "logMqtt": true
}

API에서 제공되는 항목에 대해서만 topic을 추가하고, airQualityValues는 1 ~ 4 까지가 실제 의미있는 값이니 가이드라인에 따라 위와 같이 구현해주면 된다 (-1은 unknown, 0은 매우 좋음을 뜻하는데 둘 다 API 호출 결과로는 나오지 않는 값들)

5. 동작 확인

Homebridge를 재시작하고 액세서리가 제대로 등록되었는지 확인하자

액세서리 많다 많아... 든든허다

 

Apple 디바이스의 Home 앱에서도 확인해보자

상단 탭에 기존에 있던 Weather Station 아래에 먼지바람 모양과 함께 '좋음' 이라는 문구가 추가된 것이 보인다 (먼지바람 아이콘 너무 귀여운거 아니냐 ㅎㅎ)

 

탭을 터치해보면

기후 탭에 Air Quality Sensor (Outer) 액세서리가 추가된 것을 볼 수 있다

(homebridge config 파일에서 기입한 'name' 항목대로 표기)

 

액세서리를 터치해보면

공기청정도, 미세먼지 농도, 오존 농도, 이산화질소 농도, 이산화황 농도, 일산화탄소 수준이 각각 표기되는 것을 알 수 있다 (아쉽게도 홈앱 특성상 정수형으로 truncation되어서 나오는 바람에 소수점 아래는 확인할 수가 없다 ㅠㅠ)

 

글쓰는 시점에 에어코리아 홈페이지에서 '광교동' 측정소의 측정값을 보면

이산화질소 0.03, 일산화탄소 0.7, 오존 0.008, 아황산가스 0.003인 것을 볼 수 있다

 

homebridge 로그를 보면

Open API로 얻은 값을 제대로 보내고 있는 것을 확인할 수 있다 (뭔가 아쉽 ㅎㅎ)

사실 미세먼지 농도는 API의 단위가 ug/㎥이라 전송할 때 1000으로 나눠서 보내야되는데 (플러그인의 단위는 ug/㎥), 위 예시처럼 1000ug 미만이면 결국 표기되는건 0이기 때문에 굳이 나누지않고 보내기로 했다

나머지 측정값도 mg 단위로 보고싶다면 publish_mqtt 메서드에서 농도값을 1000 곱해서 보내주게 수정하면 된다 

 

끝~!