科沃斯扫地机器人接入Home Assistant（mqtt 编程源代码解析）
在 Home Assistant 官方内置的科沃斯依赖库 deebot_client 中，底层代码采用的是（Capability 模块化）设计；为 X5 PRO 手写一个专属的 mxse7w.py 并不需要从零去写网络协议或解密通信，只需要按照它的结构规范进行配置即可。
Table of Contents
根据实战总结，科沃斯 X5 PRO（mxse7w）的整机骨架、风量四档（Quiet/Normal/Max/Max+）和地图协议版本与 X8 PRO 完全一致，但它的三档水量调节和拖布检测逻辑则与 X2 一致。
因此，最完美的“手写”方案是组合 X8 PRO 的框架与 X2 的水量模块。
源代码分析
https://github.com/DeebotUniverse/client.py/blob/dev/deebot_client/hardware/lr4qcs.py
https://github.com/DeebotUniverse/client.py/blob/dev/deebot_client/hardware/p1jij8.py
true
https://github.com/DeebotUniverse/client.py/blob/dev/deebot_client/hardware/mxse7w.py
在面向对象的软件设计中，StaticDeviceInfo这段代码的本质是一个“静态能力描述器”（Capability Descriptor）。它没有包含任何复杂的业务执行逻辑，而是通过纯粹的数据结构化配置.
def get_device_info()树形拓扑图
[StaticDeviceInfo] (根节点 Root)
│
[Capabilities] (分支节点)
┌───────────────────┼───────────────────┬───────────────────┐
▼ ▼ ▼ ▼
[clean] [map] [settings] [station] (子树/分支)
├── action ├── cached_info ├── advanced_mode ├── action
├── continuous ├── changed ├── carpet_boost ├── auto_empty
├── count ├── info ├── child_lock └── state
├── log ├── major ├── ota
├── preference ├── minor ├── true_detect
└── work_mode ├── multi_state ├── voice_assistant
├── position └── volume
├── relocation
├── rooms
├── set
└── trace树形结构的角色分工
这个树形结构划分为三个层级：
1. 根节点 (Root Node) —— StaticDeviceInfo
整个结构的入口。它除了包含“能力树”之外，还定义了树的全局环境参数，比如控制数据交换格式的叶子节点 DataType.JSON。
2. 分支节点 (Branch Nodes) —— 核心功能分类
即 Capabilities 下的第一层属性，如 clean（清扫）、map（地图）、settings（系统设置）、station（基站）。
设计目的： 在软件工程中这叫高内聚（High Cohesion）。它将数十个零散的底层传感器和执行机构，按照人类的认知习惯归类成几个核心大组，方便上层应用（如 Home Assistant）按组调用。
3. 叶子节点 (Leaf Nodes) —— 能力
树结构的最末端，也就是真正干活的逻辑单元。每个叶子节点本身又是另一个微型组合对象：
例如叶子节点 child_lock（童锁），它包含了：
事件（Event）： ChildLockEvent
获取状态（Get）： [GetChildLock()]
设置状态（Set）： SetChildLock
当需要给扫地机器人增加配置一项“能力”时，需在底层绑定了三样东西：属性对象（Capability）、发送指令（Commands）、接收事件（Events）
┌──────────────────────────────────────────────┐
│ get_device_info() 静态能力描述 │
│ (声明 DEEBOT X5 PRO OMNI 认识哪些参数和动作) │
└──────────────────────────────────────────────┘
│
├─► 1. 属性实体 (Capability) ───► 例如: fan_speed (风速)
│
├─► 2. 发送指令 (Commands) ─────► 例如: GetFanSpeed(), SetFanSpeed
│
└─► 3. 接收事件 (Events) ───────► 例如: FanSpeedEvent (云端/机器状态主动推送)如：配置风扇速度能力
fan_speed=CapabilitySetTypes(
event=FanSpeedEvent,
get=[GetFanSpeed()],
set=SetFanSpeed,
types=(
FanSpeedLevel.QUIET,
FanSpeedLevel.NORMAL,
FanSpeedLevel.MAX,
FanSpeedLevel.MAX_PLUS,
),
),
完整源代码（mxse7w.py）
https://github.com/DeebotUniverse/client.py/blob/dev/deebot_client/hardware/mxse7w.py
"""Deebot DEEBOT X5 PRO OMNI Capabilities."""
from __future__ import annotations
# 从 deebot_client 库导入各种“能力（Capability）”基类与设备类型定义
from deebot_client.capabilities import (
Capabilities,
CapabilityClean,
CapabilityCleanAction,
CapabilityCustomCommand,
CapabilityEvent,
CapabilityExecute,
CapabilityExecuteTypes,
CapabilityLifeSpan,
CapabilityMap,
CapabilitySet,
CapabilitySetEnable,
CapabilitySettings,
CapabilitySetTypes,
CapabilityStation,
CapabilityStats,
CapabilityWater,
DeviceType,
)
# 导入控制基站动作和 JSON 序列化的相关指令
from deebot_client.commands import StationAction
from deebot_client.commands.json import station_action
from deebot_client.commands.json.advanced_mode import GetAdvancedMode, SetAdvancedMode
from deebot_client.commands.json.auto_empty import GetAutoEmpty, SetAutoEmpty
from deebot_client.commands.json.battery import GetBattery
from deebot_client.commands.json.carpet import (
GetCarpetAutoFanBoost,
SetCarpetAutoFanBoost,
)
from deebot_client.commands.json.charge import Charge
from deebot_client.commands.json.charge_state import GetChargeState
from deebot_client.commands.json.child_lock import GetChildLock, SetChildLock
from deebot_client.commands.json.clean import CleanAreaV2, CleanV2
from deebot_client.commands.json.clean_count import GetCleanCount, SetCleanCount
from deebot_client.commands.json.clean_logs import GetCleanLogs
from deebot_client.commands.json.clean_preference import (
GetCleanPreference,
SetCleanPreference,
)
from deebot_client.commands.json.continuous_cleaning import (
GetContinuousCleaning,
SetContinuousCleaning,
)
from deebot_client.commands.json.custom import CustomCommand
from deebot_client.commands.json.efficiency import GetEfficiencyMode, SetEfficiencyMode
from deebot_client.commands.json.error import GetError
from deebot_client.commands.json.fan_speed import GetFanSpeed, SetFanSpeed
from deebot_client.commands.json.life_span import GetLifeSpan, ResetLifeSpan
from deebot_client.commands.json.map import (
GetCachedMapInfo,
GetMajorMap,
GetMapInfoV2,
GetMapSetV2,
GetMapTrace,
GetMinorMap,
SetMajorMap,
)
from deebot_client.commands.json.multimap_state import (
GetMultimapState,
SetMultimapState,
)
from deebot_client.commands.json.network import GetNetInfo
from deebot_client.commands.json.ota import GetOta, SetOta
from deebot_client.commands.json.play_sound import PlaySound
from deebot_client.commands.json.pos import GetPos
from deebot_client.commands.json.relocation import SetRelocationState
from deebot_client.commands.json.stats import GetStats, GetTotalStats
from deebot_client.commands.json.sweep_mode import GetSweepMode, SetSweepMode
from deebot_client.commands.json.true_detect import GetTrueDetect, SetTrueDetect
from deebot_client.commands.json.voice_assistant_state import (
GetVoiceAssistantState,
SetVoiceAssistantState,
)
from deebot_client.commands.json.volume import GetVolume, SetVolume
from deebot_client.commands.json.water_info import GetWaterInfo, SetWaterInfo
from deebot_client.commands.json.work_mode import GetWorkMode, SetWorkMode
from deebot_client.commands.json.work_state import GetWorkState
from deebot_client.const import DataType
# 导入当扫地机状态改变时，通知客户端的“事件（Event）”和枚举定义
from deebot_client.events import (
AdvancedModeEvent,
AvailabilityEvent,
BatteryEvent,
CachedMapInfoEvent,
CarpetAutoFanBoostEvent,
ChildLockEvent,
CleanCountEvent,
CleanLogEvent,
CleanPreferenceEvent,
ContinuousCleaningEvent,
CustomCommandEvent,
EfficiencyModeEvent,
ErrorEvent,
FanSpeedEvent,
FanSpeedLevel,
LifeSpan,
LifeSpanEvent,
MajorMapEvent,
MapChangedEvent,
MapTraceEvent,
MultimapStateEvent,
NetworkInfoEvent,
OtaEvent,
PositionsEvent,
ReportStatsEvent,
RoomsEvent,
StateEvent,
StationEvent,
StatsEvent,
SweepModeEvent,
TotalStatsEvent,
TrueDetectEvent,
VoiceAssistantStateEvent,
VolumeEvent,
WorkMode,
WorkModeEvent,
auto_empty,
water_info,
)
from deebot_client.events.auto_empty import AutoEmptyEvent
from deebot_client.events.efficiency_mode import EfficiencyMode
from deebot_client.models import StaticDeviceInfo
def get_device_info() -> StaticDeviceInfo:
"""获取 DEEBOT X5 PRO OMNI 模型支持的所有静态设备信息与能力映射。"""
return StaticDeviceInfo(
DataType.JSON, # 该机型通信采用 JSON 数据格式
Capabilities(
device_type=DeviceType.VACUUM, # 设备类型：吸尘/扫地机器人
# --- 基础状态能力 ---
availability=CapabilityEvent(
AvailabilityEvent, [GetBattery(is_available_check=True)]
), # 可用性状态（通过检查电池状态来判断设备是否在线）
battery=CapabilityEvent(BatteryEvent, [GetBattery()]), # 电池电量状态
charge=CapabilityExecute(Charge), # 充电控制指令（回充）
# --- 清扫控制能力 ---
clean=CapabilityClean(
action=CapabilityCleanAction(command=CleanV2, area=CleanAreaV2), # 全局清扫与选区清扫
continuous=CapabilitySetEnable(
ContinuousCleaningEvent,
[GetContinuousCleaning()],
SetContinuousCleaning,
), # 断点续扫（获取/设置/事件）
count=CapabilitySet(CleanCountEvent, [GetCleanCount()], SetCleanCount), # 清扫次数（例如：清扫1次或2次）
log=CapabilityEvent(CleanLogEvent, [GetCleanLogs()]), # 清扫日志/历史记录
preference=CapabilitySetEnable(
CleanPreferenceEvent, [GetCleanPreference()], SetCleanPreference
), # 清扫偏好设置
work_mode=CapabilitySetTypes(
event=WorkModeEvent,
get=[GetWorkMode()],
set=SetWorkMode,
types=(
WorkMode.MOP, # 仅拖地
WorkMode.MOP_AFTER_VACUUM, # 先吸后拖
WorkMode.VACUUM, # 仅吸地
WorkMode.VACUUM_AND_MOP, # 吸拖一体
),
), # 工作模式切换
),
# --- 自定义指令与错误处理 ---
custom=CapabilityCustomCommand(
event=CustomCommandEvent, get=[], set=CustomCommand
), # 透传/自定义底层指令能力
error=CapabilityEvent(ErrorEvent, [GetError()]), # 故障与报错事件
# --- 风扇风速设置 ---
fan_speed=CapabilitySetTypes(
event=FanSpeedEvent,
get=[GetFanSpeed()],
set=SetFanSpeed,
types=(
FanSpeedLevel.QUIET, # 静音
FanSpeedLevel.NORMAL, # 标准
FanSpeedLevel.MAX, # 强效
FanSpeedLevel.MAX_PLUS, # 超强
),
),
# --- 耗材与维护生命周期 ---
life_span=CapabilityLifeSpan(
types=(
LifeSpan.BRUSH, # 滚刷
LifeSpan.FILTER, # 尘盒滤网
LifeSpan.HAND_FILTER, # 手持/精细滤网
LifeSpan.SIDE_BRUSH, # 边刷
LifeSpan.UNIT_CARE, # 整机保养
LifeSpan.CLEANING_SOLUTION, # 清洁液
LifeSpan.SEWAGE_BOX, # 污水箱维护
),
event=LifeSpanEvent,
get=[
GetLifeSpan(
[
LifeSpan.BRUSH,
LifeSpan.FILTER,
LifeSpan.HAND_FILTER,
LifeSpan.SIDE_BRUSH,
LifeSpan.CLEANING_SOLUTION,
LifeSpan.SEWAGE_BOX,
]
)
],
reset=ResetLifeSpan, # 重置耗材计时指令
),
# --- 地图相关能力 ---
map=CapabilityMap(
cached_info=CapabilityEvent(CachedMapInfoEvent, [GetCachedMapInfo()]), # 缓存的地图信息
changed=CapabilityEvent(MapChangedEvent, []), # 地图更新/改变事件
info=CapabilityExecute(GetMapInfoV2), # 获取地图基础详情
major=CapabilitySet(MajorMapEvent, [GetMajorMap()], SetMajorMap), # 主地图管理
minor=CapabilityExecute(GetMinorMap), # 次要/多楼层地图管理
multi_state=CapabilitySetEnable(
MultimapStateEvent, [GetMultimapState()], SetMultimapState
), # 多地图管理开关状态
position=CapabilityEvent(PositionsEvent, [GetPos()]), # 机器人在地图上的实时坐标位置
relocation=CapabilityExecute(SetRelocationState), # 重新定位指令
rooms=CapabilityEvent(RoomsEvent, [GetCachedMapInfo()]), # 地图中的房间/分区信息
set=CapabilityExecute(GetMapSetV2), # 地图配置设置
trace=CapabilityEvent(MapTraceEvent, [GetMapTrace()]), # 清扫轨迹/移动路径
),
# --- 网络与声音 ---
network=CapabilityEvent(NetworkInfoEvent, [GetNetInfo()]), # Wi-Fi 及网络状态信息
play_sound=CapabilityExecute(PlaySound), # 寻找地宝/播放提示音
# --- 高级常规设置 (Settings) ---
settings=CapabilitySettings(
advanced_mode=CapabilitySetEnable(
AdvancedModeEvent, [GetAdvancedMode()], SetAdvancedMode
), # 高级模式开关
carpet_auto_fan_boost=CapabilitySetEnable(
CarpetAutoFanBoostEvent,
[GetCarpetAutoFanBoost()],
SetCarpetAutoFanBoost,
), # 地毯自动增压吸力开关
child_lock=CapabilitySetEnable(
ChildLockEvent, [GetChildLock()], SetChildLock
), # 童锁开关
efficiency_mode=CapabilitySetTypes(
event=EfficiencyModeEvent,
get=[GetEfficiencyMode()],
set=SetEfficiencyMode,
types=(
EfficiencyMode.ENERGY_EFFICIENT_MODE, # 节能模式
EfficiencyMode.STANDARD_MODE, # 标准模式
),
), # 清扫效率/能耗模式
ota=CapabilitySetEnable(OtaEvent, [GetOta()], SetOta), # 固件在线升级状态与设置
sweep_mode=CapabilitySetEnable(
SweepModeEvent, [GetSweepMode()], SetSweepMode
), # 扫地模式设置
true_detect=CapabilitySetEnable(
TrueDetectEvent, [GetTrueDetect()], SetTrueDetect
), # TrueDetect 3D 结构光避障功能开关
voice_assistant=CapabilitySetEnable(
VoiceAssistantStateEvent,
[GetVoiceAssistantState()],
SetVoiceAssistantState,
), # YIKO 语音助手状态与开关
volume=CapabilitySet(VolumeEvent, [GetVolume()], SetVolume), # 播报音量大小设置
),
# --- 工作与基站状态监控 ---
state=CapabilityEvent(StateEvent, [GetChargeState(), GetWorkState()]), # 机器人的主状态（充电中/工作中）
station=CapabilityStation(
action=CapabilityExecuteTypes(
station_action.StationAction, types=(StationAction.EMPTY_DUSTBIN,)
), # 控制基站执行动作（例如：手动控制集尘）
auto_empty=CapabilitySetTypes(
event=AutoEmptyEvent,
get=[GetAutoEmpty()],
set=SetAutoEmpty,
types=(
auto_empty.Frequency.AUTO, # 自动集尘
auto_empty.Frequency.SMART, # 智能集尘
),
), # 自动集尘频率设置
state=CapabilityEvent(StationEvent, [GetWorkState()]), # 全能基站自身的工作状态
),
# --- 清扫统计数据 ---
stats=CapabilityStats(
clean=CapabilityEvent(StatsEvent, [GetStats()]), # 当前/单次清扫统计（面积、时间）
report=CapabilityEvent(ReportStatsEvent, []), # 数据上报事件
total=CapabilityEvent(TotalStatsEvent, [GetTotalStats()]), # 历史累计清扫统计
),
# --- 水箱与拖地模组控制 ---
water=CapabilityWater(
amount=CapabilitySetTypes(
event=water_info.WaterAmountEvent,
get=[GetWaterInfo()],
set=SetWaterInfo,
types=(
water_info.WaterAmount.LOW, # 低水量
water_info.WaterAmount.MEDIUM, # 中水量
water_info.WaterAmount.HIGH, # 高水量
),
), # 拖地出水量调节
mop_attached=CapabilityEvent(
water_info.MopAttachedEvent, [GetWaterInfo()]
), # 检测拖地支架/抹布是否已安装
),
),
)
为扫地机器人添加基站烘干能力（Mop Drying Capability）
要为这款扫地机器人添加基站烘干能力（Mop Drying Capability），我们需要遵循代码原有的声明式树形结构。首先要确定属于哪个分支节点 (Branch Nodes)。其次在这个分支节点下添加一项能力。
具体到本例，烘干能力是在 station（基站）分支节点下，扩展一个新的叶子节点（能力）。
通过分析科沃斯底层的通用 JSON 协议结构，添加此功能通常需要以下 4 个步骤：
step 1. 修改步骤与核心代码
1. 导入必要的控制指令与事件（假设命名）
在文件顶部的 import 区域，需要引入负责获取/设置烘干状态的指令，以及对应的事件类和烘干状态枚举（具体类名需与你的 deebot_client 库底层保持一致，以下为标准命名示例）：
# 1. 导入烘干相关的 JSON 指令 (根据实际库路径调整)
from deebot_client.commands.json.drying import GetDryingState, SetDryingState
# 2. 导入烘干相关的事件与状态枚举
from deebot_client.events import DryingEvent
from deebot_client.events.drying import DryingState # 通常包含：DRYING, STOPPED 等step 2. 在 station 子树中挂载新能力
找到 station=CapabilityStation(...) 所在的位置，我们看到已经有清空基站能力；
我们需要做的，是在内部添加一个名为 drying（或 mop_drying）的叶子节点（能力）；
这个能力使用 CapabilitySetTypes 或 CapabilitySetEnable 进行包装，代码需包括：属性对象（Capability）、发送指令（Commands）、接收事件（Events）三个部分。
station=CapabilityStation(
action=CapabilityExecuteTypes(
station_action.StationAction, types=(StationAction.EMPTY_DUSTBIN,)
),
auto_empty=CapabilitySetTypes(
event=AutoEmptyEvent,
get=[GetAutoEmpty()],
set=SetAutoEmpty,
types=(
auto_empty.Frequency.AUTO,
auto_empty.Frequency.SMART,
),
),
# ─── 新增：基站抹布烘干能力 ──────────────────────────────────
drying=CapabilitySetTypes(
event=DryingEvent, # 监听烘干状态变化的事件
get=[GetDryingState()], # 初始化或主动获取烘干状态的指令
set=SetDryingState, # 控制开启/关闭烘干的指令
types=(
DryingState.STOPPED, # 停止烘干
DryingState.DRYING, # 正在烘干
DryingState.RAPID_DRYING, # 快速烘干（若硬件支持）
),
),
# ───────────────────────────────────────────────────────────
state=CapabilityEvent(StationEvent, [GetWorkState()]),
),
step 3.结构变化对比
旧的station（基站）分支节点： 新的station（基站）分支节点（添加烘干后）：
[station] [station]
├── action ├── action
├── auto_empty ├── auto_empty
└── state ├── drying ◄── [新加入的叶子节点]
└── state
step 4.调用方式
上层应用（如 Home Assistant 的实体组件）可以直接对基站烘干进行双向控制了：
主动触发/停止烘干：
device.capabilities.station.drying.set(DryingState.DRYING)
前端 UI ：
上层应用会自动读取 types 里的枚举，在 App 或智能家居面板上自动生成对应的“开始烘干 / 停止烘干”下拉菜单或开关。
Comments
Be the first to post a comment