Ecovacs in Home Assistant Part8 - Create a complete Home Assistant integration for the Ecovacs X5 Pro(skills)

Control Ecovacs Deebot robot vacuums via the Ecovacs Open Platform AK and a gateway (/robot/skill/*).

 

I'll analyze these https://github.com/mslycn/vacumm-ecovacs-deebot/blob/main/references/api.md#cloudctl-clean to understand the Ecovacs API.I've created a complete Home Assistant integration for the Ecovacs Deetbot X5 Pro.

Repository Structure

custom_components/ecovacs_x5pro/
├── __init__.py (setup integration)
├── manifest.json (metadata)
├── const.py (constants)
├── config_flow.py (configuration UI)
├── api.py (API client)
└── vacuum.py (vacuum entity)

 

api.py    GetWorkState

 

 

config_flow.py 

api.py

 

the actual API returns:

JSON response from api.py: {'msg': 'OK', 'code': 0, 'data': {'code': 0, 'msg': 'success', 'data': {'ctl': {'data': {'ret': 'ok', 'cleanSt': 'h', 'chargeSt': 'charging', 'stationSt': 'dust'}}}}}
JSON response from api.py: {'msg': 'OK', 'code': 0, 'data': {'code': 0, 'msg': 'success', 'data': {'ctl': {'data': {'ret': 'ok', 'cleanSt': 'wash', 'chargeSt': 'charging', 'stationSt': 'i'}}}}}
JSON response from api.py: {'msg': 'OK', 'code': 0, 'data': {'code': 0, 'msg': 'success', 'data': {'ctl': {'data': {'ret': 'ok', 'cleanSt': 'washpause', 'chargeSt': 'charging', 'stationSt': 'i'}}}}

Your API response has one extra level of nesting that wasn't accounted for.So you need to go one level deeper:

ctl = (
    data.get("data", {})      # First data level
        .get("data", {})      # ← SECOND data level (was missing!)
        .get("ctl", {})       # Then ctl
        .get("data", {})      # Then final data
)

 

vacuum.py

当前 vacuum.py 推荐结构（tree）

vacuum.py
│
├── imports
│
├── logger
│
├── SCAN_INTERVAL
│
├── 状态映射
│   └── STATION_MAP
│
├── class EcovacsX5Vacuum(StateVacuumEntity)
│   │
│   ├── __init__
│   │
│   ├── device_info
│   │
│   ├── should_poll
│   │
│   ├── async_update
│   │
│   ├── state
│   │
│   ├── extra_state_attributes
│   │
│   ├── async_start
│   │
│   ├── async_pause
│   │
│   ├── async_return_to_base
│   │
│   └── _send
│
└── async_setup_entry

 

more detail

vacuum.py
│
├── from homeassistant.components.vacuum import ...
├── from datetime import timedelta
├── import aiohttp
├── import logging
│
├── _LOGGER
│
├── SCAN_INTERVAL
│
├── STATION_MAP
│   ├── i → idle
│   ├── w → washing
│   ├── d → drying
│   ├── e → emptying
│   ├── g → going_charging
│   ├── c → charging
│   └── p → paused
│
├── class EcovacsX5Vacuum
│   │
│   ├── __init__(api)
│   │   ├── 保存 api
│   │   ├── 设置 entity 名称
│   │   ├── 设置 unique_id
│   │   ├── 初始化 state
│   │   └── 设置支持功能
│   │
│   ├── device_info
│   │   └── 注册设备到 HA
│   │
│   ├── should_poll
│   │   └── 告诉 HA 需要轮询
│   │
│   ├── async_update
│   │   ├── 调 API
│   │   ├── 获取状态
│   │   ├── 更新 self._state
│   │   └── 更新 self._station
│   │
│   ├── state
│   │   └── 返回 vacuum 状态
│   │
│   ├── extra_state_attributes
│   │   └── 返回 station_state
│   │
│   ├── async_start
│   │   └── Clean s
│   │
│   ├── async_pause
│   │   └── Clean p
│   │
│   ├── async_return_to_base
│   │   └── Charge go
│   │
│   └── _send
│       ├── POST 控制命令
│       ├── 刷新状态
│       └── 更新 UI
│
└── async_setup_entry
    ├── 创建 API
    └── 注册实体

 

vacuum.py

必须提供 device_info（否则没有设备卡片）

@property
def device_info(self):
    return {
        "identifiers": {("ecovacs_x5pro", self.api.name)},
        "name": "Ecovacs X5 Pro",
        "manufacturer": "Ecovacs",
        "model": "X5 Pro",
    }

 

实体必须有 unique_id（否则不会注册）

self._attr_unique_id = f"ecovacs_x5pro_{api.name}"

 

async_add_entities 必须执行，实体创建

async_add_entities([EcovacsX5Vacuum(api)], update_before_add=True)

vacuum.py - async_setup_entry(hass, entry, async_add_entities)

# ---------------- setup ----------------

async def async_setup_entry(hass, entry, async_add_entities):
    """Set up vacuum entity from config entry."""


    async_add_entities(
        [EcovacsX5Vacuum(api)],
        update_before_add=True
    )

 

 

__init__.py - async_setup_entry 必须正确加载平台

await hass.config_entries.async_forward_entry_setups(
    entry,
    ["vacuum", "sensor"]   # 👈 你有哪些就写哪些
)

 

vacuum.py

class EcovacsX5Vacuum(StateVacuumEntity):
    """Ecovacs X5 Pro vacuum entity."""

    def __init__(self, api):

        self._station = "idle"

    @property
    def device_info(self):
        """Return device information."""
        return {
            "identifiers": {("ecovacs_x5pro", self.api.name)},
            "name": "Ecovacs X5 Pro",
            "manufacturer": "Ecovacs",
            "model": "X5 Pro",
        }
    ....

    async def async_update(self):
        """Update the vacuum state."""
    ...
            # Update  State
            if clean_code == "wash":
                self._state = "washing"
    ...


# ---------------- setup ----------------

async def async_setup_entry(hass, entry, async_add_entities):
    """Set up vacuum entity from config entry."""

    from .api import EcovacsAPI

    api = EcovacsAPI(
        entry.data["ak"],
        entry.data["nickname"]
    )

   #  4.3 实体创建
    async_add_entities(
        [EcovacsX5Vacuum(api)],
        update_before_add=True
    )

 

 

 

   @property
   def icon(self):
       return "mdi:robot-vacuum"    # 加 icon，让 UI 更像官方

 

 

Control commands

vacuum.py

run ok

 async def async_start(self):
        """Start the vacuum."""
        await self._send("Clean", {"act": "s"})

    async def async_pause(self):
        """Pause the vacuum."""
        await self._send("Clean", {"act": "p"})

    async def async_return_to_base(self):
        """Return the vacuum to base."""
        await self._send("Charge", {"act": "go"})

    async def _send(self, cmd, data):
        payload = {
            "ak": self.api.ak,
            "nickName": self.api.name,
            "ctl": {
                "cmd": cmd,
                "data": data
            }
        }