Develop Tuya-Compatible Home Assistant Drivers

Home Assistant Core is developed based on Python and can run on ODROID, Raspberry Pi, ASUS Tinker board, Intel NUC, Windows, macOS, Linux, and more.
We have set up the development environment on Raspberry Pi by using Home Assistant Core. For more information, see Set up Home Assistant Development Environment on Raspberry Pi.

1. Download Tuya official Home Assistant integration.

2. Go to the Home Assistant configuration directory, for example, ~/.homeassistant for macOS. For more information about directory configuration, see Access Home Assistant through Samba.

3. Create a folder named custom_components in the configuration directory and a subfolder named tuya.

   

4. Extract the Tuya integration to the tuya folder.

5. Open the configuration.yaml file with a text editor. See Register with Tuya IoT Platform to get the required information and complete related fields, as shown below.

   

Once the configuration is completed, you can run Home Assistant to call aysnc_setup in the __init__.py file to start the plugin initialization process.

__init__.async_setup

async def async_setup(hass, config):
    """Set up the Tuya integration."""

    conf = config.get(DOMAIN) ## Read information from configuration.yaml
    print('Tuya async setup conf %s \n' % conf)
    if conf is not None: 
        async def flow_init() -> Any:
            try:   ## Go to config_flow.py to proceed with initialization
                result = await hass.config_entries.flow.async_init(
                    DOMAIN, context={"source": SOURCE_IMPORT}, data=conf
                )
            except Exception as inst:
                print(inst.args)
            print("Tuya async setup flow_init")
            return result
        # await hass.async_add_executor_job(
        hass.async_create_task(
            flow_init()
        )

    return True

TuyaConfigFlow.async_step_user

In the __init__.py file, execute async_setup until TuyaConfigFlow.async_step_import and then execute async_step_user.

    async def async_step_user(self, user_input=None, is_import=False):
        _LOGGER.debug('TuyaConfigFlow.async_step_user start, is_import=', is_import)

        errors = {}
        if user_input is not None:
			## Try to log in with the configuration information
            response = await self.hass.async_add_executor_job(self._try_login, user_input)
            if response.get('success', False):
                return self.async_create_entry(
                    title=user_input[CONF_USERNAME],
                    data=user_input,
                )
            else:
                errors['base'] = 'code={}, msg={}'.format(response.get('code', 0), response.get('msg', ''))
                if is_import == True:
                    return self.async_abort(reason=errors['base'])
			
		## If no information has been configured, set a pop-up window to guide the user to configure information.
        return self.async_show_form(
            step_id='user',
            data_schema=DATA_SCHEMA_USER,
            errors=errors,
        )

__init__.async_setup_entry

async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
    print("tuya.__init__.async_setup_entry-->", entry.data)

	## Execute _init_tuya_sdk for initialization
    success = await hass.async_add_executor_job(_init_tuya_sdk, hass, entry.data)
    if not success:
        return False

    ## Initialize Home Assistant devices according to the tuya->ha type mapping table.
    for platform in set(TUYA_TYPE_TO_HA.values()):
        print("tuya async platform-->", platform)
        hass.async_create_task(
            hass.config_entries.async_forward_entry_setup(
                entry, platform
            )
        )

    return True

This section describes how to develop a driver for sockets.

Prerequisites

Before development, go to the Home Assistant developer website to find an entity that fits the socket, such as the switch. Follow the Switch Entity to develop a driver for sockets.

Procedure

1. Create a new switch.py file in the custom_components/tuya directory.

   

2. Create a new async_setup_entry method for initialization.
   This method reads haDevices from the cache. The switch is the matched type so we create a new device entity as follows.

   async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry, async_add_entities):
   print('tuya.switch.async_setup_entry')
   
   haDevices = []
   for haDevice in hass.data[DOMAIN]['haDevices']:
       if haDevice.platform == 'switch':
           haDevices.append(haDevice)
   async_add_entities(haDevices)
   

3. Create a new TuyaHaSwitch as the defined entity for Tuya sockets in Home Assistant. Based on the Switch Entity, implement the required Properties and Methods, such as turn_on and turn_off for switch’s Methods and is_on for Properties.

   Sample code:

   class TuyaHaSwitch(TuyaHaDevice, SwitchEntity):
   """Tuya Switch Device."""
   
   platform = 'switch'
   
   # ToggleEntity
   
   @property
   def is_on(self) -> bool:
       """Return true if switch is on."""
       return self.tuyaDevice.status.get(DPCODE_SWITCH, False)
   
   def turn_on(self, **kwargs: Any) -> None:
       """Turn the switch on."""
       self.tuyaDeviceManager.publishCommands(self.tuyaDevice.id, [{'code': DPCODE_SWITCH, 'value': True}])
   
   def turn_off(self, **kwargs: Any) -> None:
       """Turn the device off."""
       self.tuyaDeviceManager.publishCommands(self.tuyaDevice.id, [{'code': DPCODE_SWITCH, 'value': False}])