[Python] Home Assistant Smarthings IR 플러그인 제작

 

저번 게시글에서 Home Assistant 기본 구축과 더불어, 기본적인 대시보드 설정을 해주었다.

이번엔, Smartthings 플러그인이 커버하지 못하는 Galaxy Home mini의 IR 센서 동작들을 구현해보겠다.

 

 

Smartthings Extra

 

플러그인을 제작하기전에 벤치마킹 하기 좋은 플러그인을 발견했다.

Smartthings Extra라는 플러그인이었는데.

Smartthings 공식 플러그인의 OAuth 인증정보를 그대로 활용하는 방식으로 동작했다.

https://github.com/mik-laj/ha_smartthing_extra

GitHub - mik-laj/ha_smartthing_extra

 

이 방식을 그대로 사용하려 했지만, 이 플러그인 역시 deprecated된 IR 동작을 지원하지 않았다.

따라서 필요한 부분만 벤치마킹해서 사용하려고 한다.

 

코드에서 필요한 부분을 찾아보자

아래는 공식 플러그인의 OAuth 클라이언트 저장위치를 찾는 로직이다.

st_entry_id: Optional[str] = None
for entry_id in device_entry.config_entries:
    entry = hass.config_entries.async_get_entry(entry_id)
    if entry and entry.domain == ST_DOMAIN and entry.state == ConfigEntryState.LOADED:
        st_entry_id = entry.entry_id
        break

 

 

그리고 OAuth 클라이언트를 획득하는 로직

entry: ConfigEntry | None = hass.config_entries.async_get_entry(st_entry_id)
if not entry or not entry.runtime_data:
    raise HomeAssistantError("SmartThings entry not available or not loaded.")

client = entry.runtime_data.client

 

 

마지막으로, device id에 명령 전송 로직은 다음과 같았다.

async def _resolve_st_ids(ha_device_id: str) -> Tuple[str, str]:
    dev_reg = dr.async_get(hass)
    device_entry = dev_reg.async_get(ha_device_id)
    if not device_entry:
        raise HomeAssistantError(f"No HA device found: {ha_device_id}")

    st_device_id: Optional[str] = None
    for domain, ident in device_entry.identifiers:
        if domain == ST_DOMAIN:
            st_device_id = ident
            break

    st_entry_id: Optional[str] = None
    for entry_id in device_entry.config_entries:
        entry = hass.config_entries.async_get_entry(entry_id)
        if entry and entry.domain == ST_DOMAIN and entry.state == ConfigEntryState.LOADED:
            st_entry_id = entry.entry_id
            break

    return st_device_id, st_entry_id

 

 

이제 필요한 로직은 파악했으니 커스텀 플러그인 제작을 해보자

 

 

플러그인 구현

 

먼저 통합 정보인 manifest.json을 구현하자

필요한 aiohttp와, 명령 전송을 위한 pysmartthings 패키지를 작성해 주었다.

{
  "domain": "smartthings_ir",
  "name": "SmartThings IR Controller",
  "version": "1.0.0",
  "requirements": ["aiohttp>=3.8.0", "pysmartthings>=0.8.2"],
  "dependencies": [],
  "codeowners": ["@101Sean"],
  "config_flow": true,
  "iot_class": "cloud_polling"
}

 

 

진입점인 __init__.py, Home Assistant가 로드할때 가장 먼저 실행하는 파일이다.

async_setup_entry 메소드로 climate와 media_player를 로드하게 해주었다. (필요한 악세서리)

PLATFORMS = [Platform.CLIMATE, Platform.MEDIA_PLAYER]

async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
    hass.data.setdefault(DOMAIN, {})
    hass.data[DOMAIN][entry.entry_id] = entry.data
    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
    return True

async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
    if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
        hass.data[DOMAIN].pop(entry.entry_id)
    return unload_ok

 

 

이제 Extra의 핵심 로직을 구현할 api.py를 구현하자

공식 플러그인의 OAuth 인증을 재사용 하고, device_id 변환과 capability 매핑을 할 것이다.

 

device_id를 Smartthings UUID로 변환하는 메소드이다.

기기 정보 조회 > UUID 추출 > Entry ID를 찾아 변환 하는 순서이다.

async def _resolve_st_ids(self, ha_device_id: str) -> Tuple[str, str]:
    dev_reg = dr.async_get(self.hass)
    device_entry = dev_reg.async_get(ha_device_id)
    if not device_entry:
        raise HomeAssistantError(f"No HA device found: {ha_device_id}")

    st_device_id: Optional[str] = None
    for domain, ident in device_entry.identifiers:
        if domain == ST_DOMAIN:
            st_device_id = ident
            break
    if not st_device_id:
        raise HomeAssistantError("Device is not linked to SmartThings.")

    st_entry_id: Optional[str] = None
    for entry_id in device_entry.config_entries:
        entry = self.hass.config_entries.async_get_entry(entry_id)
        if entry and entry.domain == ST_DOMAIN and entry.state == ConfigEntryState.LOADED:
            st_entry_id = entry.entry_id
            break
    if not st_entry_id:
        raise HomeAssistantError("No loaded SmartThings config entry for this device.")

    return st_device_id, st_entry_id

 

 

다음으로는, 공식 플러그인에서 OAuth 인증된 API 클라이언트를 획득하는 메소드이다.

async def _get_client(self, st_entry_id: str):
    entry = self.hass.config_entries.async_get_entry(st_entry_id)
    if not entry or not entry.runtime_data:
        raise HomeAssistantError("SmartThings entry not available or not loaded.")
    return entry.runtime_data.client

 

 

마지막으로는, 실제 Smartthings 기기에 명령 전송하는 메소드이다.

special_caps 매핑으로, deprecated된 IR 명령들도 사용할 수 있게 해주었다.

async def command(
    self,
    ha_device_id: str,
    capability: str,
    command: str,
    arguments: Optional[list] = None,
    component: str = "main",
) -> bool:
    try:
        st_device_id, st_entry_id = await self._resolve_st_ids(ha_device_id)
        client = await self._get_client(st_entry_id)

        special_caps = {
            "STATELESS_AUDIO_VOLUME_BUTTON": "statelessAudioVolumeButton",
            "STATELESS_CHANNEL_BUTTON": "statelessChannelButton",
            "STATELESS_AUDIO_MUTE_BUTTON": "statelessAudioMuteButton",
            "STATELESS_POWER_TOGGLE_BUTTON": "statelessPowerToggleButton",
        }

        upper_input = capability.upper()
        if upper_input in special_caps:
            mapped_capability = special_caps[upper_input]
        else:
            mapped_capability = getattr(Capability, upper_input, capability)

        mapped_command = getattr(Command, command.upper(), command)

        _LOGGER.debug(
            "Command -> device=%s, capability=%s, command=%s, args=%s",
            st_device_id, mapped_capability, mapped_command, arguments
        )

        await client.execute_device_command(
            device_id=st_device_id,
            capability=mapped_capability,
            command=mapped_command,
            component=component,
            argument=arguments or [],
        )

        return True

    except Exception as e:
        _LOGGER.error("Command failed: %s", e)
        return False

 

 

이제 에어컨을 제어 할 climate.py를 구현하자.

IR로 제어하는 에어컨을 Home Assistant의 Climate 엔티티로 변환 할 것이다.

https://www.home-assistant.io/integrations/climate/

Climate

 

기본 기기 정보 저장을 할 클래스를 선언하고

class SmartThingsIRClimate(ClimateEntity):
    _attr_temperature_unit = UnitOfTemperature.CELSIUS
    _attr_min_temp = 18
    _attr_max_temp = 30
    _attr_target_temperature_step = 1
    _attr_supported_features = ClimateEntityFeature.TARGET_TEMPERATURE | ClimateEntityFeature.FAN_MODE
    _attr_hvac_modes = [HVACMode.OFF, HVACMode.COOL, HVACMode.DRY, HVACMode.AUTO]
    _attr_fan_modes = ["low", "medium", "high"]

    def __init__(self, hass, device_id, name):
        self.hass = hass
        self._device_id = device_id
        self._attr_name = name
        self._attr_unique_id = f"smartthings_ir_climate_{device_id}"
        self._attr_target_temperature = 24
        self._attr_hvac_mode = HVACMode.OFF
        self._attr_fan_mode = "medium"
        self._api = SmartThingsIRAPI(hass)

 

 

에어컨 모드 변경 메소드 부분을 구현했다.

에어컨의 모드(냉방, 제습, 자동)등과 on, off를 매핑

async def async_set_hvac_mode(self, hvac_mode: str) -> None:
    if hvac_mode == HVACMode.OFF:
        await self._send_command("switch", "off")
    else:
        await self._send_command("switch", "on")
        mode_map = {
            HVACMode.COOL: "cool",
            HVACMode.DRY: "dry",
            HVACMode.AUTO: "auto"
        }
        await self._send_command("AIR_CONDITIONER_MODE", "SET_AIR_CONDITIONER_MODE", [mode_map.get(hvac_mode, "cool")])
    self._attr_hvac_mode = hvac_mode
    self.async_write_ha_state()

 

 

다음으로는 온도 설정 메소드

async def async_set_temperature(self, **kwargs) -> None:
    temp = kwargs.get(ATTR_TEMPERATURE)
    if temp is not None:
        self._attr_target_temperature = temp
        await self._send_command("THERMOSTAT_COOLING_SETPOINT", "SET_COOLING_SETPOINT", [temp])
        self.async_write_ha_state()

 

 

그리고 팬 속도 변경 메소드까지 구현 하면, 동작 구현은 끝이다.

async def async_set_fan_mode(self, fan_mode: str) -> None:
    self._attr_fan_mode = fan_mode
    await self._send_command("AIR_CONDITIONER_FAN_MODE", "SET_FAN_MODE", [fan_mode])
    self.async_write_ha_state()

 

 

마지막으로 API 호출을 위한 메소드 까지 구현

capability, command, argument로 Smartthings API 규약에 맞게 구현했다.

async def _send_command(self, capability, command, arguments=None):
    await self._api.command(
        ha_device_id=self._device_id,
        capability=capability,
        command=command,
        arguments=arguments or [],
    )

 

 

이제, TV와 Settop을 매핑할 media_player.py이다.

IR로 제어하는 두 기기를 Home Assistant의 Media Player 엔티티로 변환할 것이다.

https://www.home-assistant.io/integrations/media_player/

Media player

 

역시, 기본적인 기기 정보를 저장 할 클래스를 생성

ON/OFF의 구분은 필요 없어서 TURN_ON은 제거 했다.

class SmartThingsIRMediaPlayer(MediaPlayerEntity):
    _attr_supported_features = (
        MediaPlayerEntityFeature.TURN_OFF |
        MediaPlayerEntityFeature.VOLUME_STEP |
        MediaPlayerEntityFeature.VOLUME_MUTE |
        MediaPlayerEntityFeature.NEXT_TRACK |
        MediaPlayerEntityFeature.PREVIOUS_TRACK
    )
    _attr_assumed_state = True

    def __init__(self, hass, device_id, name, device_type):
        self.hass = hass
        self._device_id = device_id
        self._attr_name = name
        self._attr_device_class = "tv"
        self._attr_unique_id = f"smartthings_ir_media_player_{device_id}"
        self._attr_state = MediaPlayerState.ON
        self._device_type = device_type
        self._api = SmartThingsIRAPI(hass)

 

 

전원 토글을 위한 메소드를 구현하고

async def async_turn_off(self):
    await self._send_command("STATELESS_POWER_TOGGLE_BUTTON", "SET_BUTTON", ["powerToggle"])
    self._attr_state = MediaPlayerState.ON
    self.async_write_ha_state()

 

 

볼륨 조절도 구현

async def async_volume_up(self):
    await self._send_command("STATELESS_AUDIO_VOLUME_BUTTON", "SET_BUTTON", ["volumeUp"])

async def async_volume_down(self):
    await self._send_command("STATELESS_AUDIO_VOLUME_BUTTON", "SET_BUTTON", ["volumeDown"])

 

 

채널 변경은, Media Player의 트랙 변경에다가 매핑을 해주었다.

async def async_media_next_track(self):
    await self._send_command("STATELESS_CHANNEL_BUTTON", "SET_BUTTON", ["channelUp"])

async def async_media_previous_track(self):
    await self._send_command("STATELESS_CHANNEL_BUTTON", "SET_BUTTON", ["channelDown"])

 

 

동일하게 실제 API로 전송하는 메소드까지 구현 완료

async def _send_command(self, capability, command, arguments=None):
    await self._api.command(
        ha_device_id=self._device_id,
        capability=capability,
        command=command,
        arguments=arguments or [],
    )

 

 

이제 device_id가 필요한데, 공식 플러그인이 어떤 id를 사용중인지 직접 알 수 없어서

extra 플러그인의 방식을 착안해서 서비스를 하나 만드려고 한다.

 

services.py 생성 후, 공식 플러그인의 기기 선택과, UUID 변환/확인을 구현했다.

async def async_register_services(hass: HomeAssistant) -> None:
    
    async def handle_show_device_id(call: ServiceCall) -> None:
        ha_device_id = call.data.get("device_id")
        
        if not ha_device_id:
            raise HomeAssistantError("device_id is required")
        
        api = SmartThingsIRAPI(hass)
        st_device_id, st_entry_id = await api._resolve_st_ids(ha_device_id)
    
    hass.services.async_register(DOMAIN, "show_device_id", handle_show_device_id)


async def async_unregister_services(hass: HomeAssistant) -> None:
    hass.services.async_remove(DOMAIN, "show_device_id")

 

 

그리고 서비스 틀까지 정의 하면 끝

show_device_id:
  name: Show Device UUID
  description: Display the SmartThings UUID for a selected device
  fields:
    device_id:
      name: Device
      description: Select a SmartThings device
      required: true
      selector:
        device:
          filter:
            - integration: smartthings

 

 

마지막으로 config_flow.py를 구현해서,

통합구성요소를 추가할 때 나오는 설정 화면까지 만들었다. (코드생략)

https://github.com/101Sean/hass_smartthings_ir

 

 

플러그인 등록

 

이제 완성된 플러그인을 HACS를 통해 저장소 추가를 해보자

 

HACS > 우측 상단 메뉴 > Custom repositories 메뉴 선택

Repository 추가와 타입은 통합으로 추가

 

추가 후, 통합구성요소에서 검색하면 등록한 플러그인이 나타난다.

 

이제 설정 > 개발자 도구 > 동작

services.py에서 구현한 show_device_id 선택

 

기기 선택 후, YAML 모드를 클릭하면 device_id를 볼 수 있다.

 

이제 기기 추가에 device_id를 넣고 등록하면 끝

 

사용중인 디바이스 3개를 정상적으로 등록 했다.

 

마지막으로 대시보드에 타일로 추가 완료

 

 

이제 Home Assistant에서도 Smartthings의 모든 동작을 사용 할 수 있게 되었다.

차근차근 Homebridge의 플러그인들을 옮기면서 의존도를 줄여 나가도록 하겠다.

 

 

원본 게시글

https://blog.naver.com/101artspace/224331267850

[Python] Home Assistant Smarthings IR 플러그인 제작