From a9ad4a285f58c1f91302e0916f533077ea58c0e2 Mon Sep 17 00:00:00 2001 From: xyzm Date: Tue, 16 Sep 2025 23:19:11 +0800 Subject: [PATCH] first commit --- README.md | 128 +++++ custom_components/vivo_cloud/__init__.py | 115 +++++ custom_components/vivo_cloud/config_flow.py | 280 +++++++++++ custom_components/vivo_cloud/const.py | 64 +++ custom_components/vivo_cloud/coordinator.py | 468 ++++++++++++++++++ .../vivo_cloud/device_tracker.py | 298 +++++++++++ custom_components/vivo_cloud/manifest.json | 12 + custom_components/vivo_cloud/sensor.py | 211 ++++++++ custom_components/vivo_cloud/services.yaml | 15 + .../vivo_cloud/translations/en.json | 70 +++ .../vivo_cloud/translations/zh-Hans.json | 70 +++ hacs.json | 7 + 12 files changed, 1738 insertions(+) create mode 100644 README.md create mode 100644 custom_components/vivo_cloud/__init__.py create mode 100644 custom_components/vivo_cloud/config_flow.py create mode 100644 custom_components/vivo_cloud/const.py create mode 100644 custom_components/vivo_cloud/coordinator.py create mode 100644 custom_components/vivo_cloud/device_tracker.py create mode 100644 custom_components/vivo_cloud/manifest.json create mode 100644 custom_components/vivo_cloud/sensor.py create mode 100644 custom_components/vivo_cloud/services.yaml create mode 100644 custom_components/vivo_cloud/translations/en.json create mode 100644 custom_components/vivo_cloud/translations/zh-Hans.json create mode 100644 hacs.json diff --git a/README.md b/README.md new file mode 100644 index 0000000..aaec3e6 --- /dev/null +++ b/README.md @@ -0,0 +1,128 @@ +# Vivo Cloud Device Tracker + +这是一个Home Assistant自定义组件，用于跟踪Vivo Cloud设备的位置。 + +## 功能特点 + +- 支持通过手动输入Cookie或使用CookieCloud获取Cookie +- 自动获取Vivo Cloud设备列表 +- 支持多设备选择和跟踪 +- 定时更新设备位置信息 +- 显示设备信号和电池信息 +- 提供响铃提醒服务 + +## 安装方法 + +1. 将`custom_components/vivo_cloud`目录复制到Home Assistant的`custom_components`目录下 +2. 重启Home Assistant +3. 在"配置" -> "集成"中点击"+"添加集成 +4. 搜索"Vivo Cloud"并点击添加 +5. 按照配置向导完成设置 + +## 配置步骤 + +### 1. Cookie设置方式选择 + +选择获取Cookie的方式： +- **手动输入Cookie**：直接输入从浏览器获取的Cookie +- **使用CookieCloud获取Cookie**：通过CookieCloud服务获取Cookie + +### 2. Cookie配置 + +根据选择的Cookie设置方式，输入相应的信息： +- **手动输入**：输入完整的Cookie字符串 +- **CookieCloud**：输入UUID、密码和服务器地址（可选，默认为https://cookie.nextrt.com） + +### 3. 设备选择 + +从获取的设备列表中选择要跟踪的设备。可以多选。 + +### 4. 扫描间隔设置 + +设置设备位置信息的更新频率（秒），建议值在360-3600秒之间。 + +## 使用方法 + +### 设备跟踪 + +插件会自动为每个选中的设备创建一个设备跟踪实体，显示在Home Assistant的地图上。 + +### 设备信息 + +每个设备跟踪实体包含以下信息： +- 位置信息（经纬度、位置描述） +- 设备状态（在线状态、电池电量） +- 信号信息（运营商、网络类型、信号强度） +- 电池信息（电量、充电状态等） + +### 响铃提醒 + +可以通过服务调用让设备响铃，以便查找设备： + +```yaml +service: vivo_cloud.ring_device +target: + entity_id: device_tracker.vivo_device_123456789 +data: + device_id: "123456789" +``` + +## 获取Cookie的方法 + +### 手动获取Cookie + +1. 在浏览器中登录https://find.vivo.com.cn +2. 打开开发者工具（F12） +3. 切换到"网络"选项卡 +4. 刷新页面或执行设备查找操作 +5. 找到名为"devices"的请求 +6. 在请求头中找到"Cookie"字段，复制完整的Cookie值 + +### 使用CookieCloud + +1. 安装CookieCloud浏览器扩展：[CookieCloud - Microsoft Edge Addons](https://microsoftedge.microsoft.com/addons/detail/bffenpfpjikaeocaihdonmgnjjdpjkeo) + +2. 在浏览器中登录https://find.vivo.com.cn ，注意选择两周免认证 + + 服务器地址：`https://cookie.nextrt.com` + + 用户KEY · UUID：`选择自动生成避免和别人一样` + + 端对端加密密码：`选择自动生成或者自己设置` + + 是否同步Local Storage：`是` + + 同步域名关键词：`vivo.com.cn` + +3. 使用CookieCloud扩展同步Cookie + +4. 在Home Assistant中输入CookieCloud的UUID和密码 + +## 故障排除 + +### 无法连接到Vivo Cloud + +- 检查Cookie是否有效 +- 确保网络连接正常 +- 尝试重新获取Cookie + +### 无法获取设备列表 + +- 确保Cookie中包含必要的权限 +- 检查账号是否有设备绑定 + +### 设备位置不更新 + +- 检查设备是否在线 +- 确保设备开启了位置服务 +- 尝试减小扫描间隔 + +## 注意事项 + +- 插件需要定期更新Cookie，建议使用CookieCloud自动同步 +- 设备位置更新频率不宜过高，以免消耗过多电量 +- 某些设备可能不支持所有功能 + +## 许可证 + +MIT License \ No newline at end of file diff --git a/custom_components/vivo_cloud/__init__.py b/custom_components/vivo_cloud/__init__.py new file mode 100644 index 0000000..a29e245 --- /dev/null +++ b/custom_components/vivo_cloud/__init__.py @@ -0,0 +1,115 @@ +""" +Component to integrate with Vivo Cloud. + +For more details about this component, please refer to +https://bbs.233py.com +""" +import asyncio +import json +import logging +from typing import Dict, Any + +from homeassistant.config_entries import ConfigEntry +from homeassistant.const import CONF_SCAN_INTERVAL +from homeassistant.core import HomeAssistant, ServiceCall +from homeassistant.exceptions import ConfigEntryNotReady +from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed +from homeassistant.helpers.aiohttp_client import async_get_clientsession + +from .const import ( + DOMAIN, + CONF_DEVICES, + CONF_SCAN_INTERVAL, + DEFAULT_SCAN_INTERVAL, + COORDINATOR, + UNDO_UPDATE_LISTENER, + SERVICE_RING_DEVICE, +) + +from .coordinator import VivoCloudDataUpdateCoordinator + +_LOGGER = logging.getLogger(__name__) + +PLATFORMS = ["device_tracker", "sensor"] + + +async def async_setup(hass: HomeAssistant, config: Dict) -> bool: + """Set up configured Vivo Cloud.""" + hass.data[DOMAIN] = {} + return True + + +async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: + """Set up Vivo Cloud as config entry.""" + cookie_method = entry.data.get("cookie_method") + cookie = entry.data.get("cookie") + cookiecloud_uuid = entry.data.get("cookiecloud_uuid") + cookiecloud_password = entry.data.get("cookiecloud_password") + cookiecloud_server = entry.data.get("cookiecloud_server", "https://cookie.nextrt.com") + devices = entry.data.get(CONF_DEVICES, []) + scan_interval = entry.options.get(CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL) + + coordinator = VivoCloudDataUpdateCoordinator( + hass, + cookie_method=cookie_method, + cookie=cookie, + cookiecloud_uuid=cookiecloud_uuid, + cookiecloud_password=cookiecloud_password, + cookiecloud_server=cookiecloud_server, + devices=devices, + scan_interval=scan_interval, + ) + + await coordinator.async_refresh() + + if not coordinator.last_update_success: + raise ConfigEntryNotReady + + undo_listener = entry.add_update_listener(update_listener) + + hass.data[DOMAIN][entry.entry_id] = { + COORDINATOR: coordinator, + UNDO_UPDATE_LISTENER: undo_listener, + } + + hass.async_create_task( + hass.config_entries.async_forward_entry_setups(entry, PLATFORMS) + ) + + # 注册服务 + async def async_ring_device(call: ServiceCall) -> None: + """Handle ring device service call.""" + device_id = call.data.get("device_id") + await coordinator.async_ring_device(device_id) + + hass.services.async_register( + DOMAIN, SERVICE_RING_DEVICE, async_ring_device + ) + + return True + + +async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: + """Unload a config entry.""" + unload_ok = all( + await asyncio.gather( + *[ + hass.config_entries.async_forward_entry_unload(entry, platform) + for platform in PLATFORMS + ] + ) + ) + + hass.data[DOMAIN][entry.entry_id][UNDO_UPDATE_LISTENER]() + + if unload_ok: + hass.data[DOMAIN].pop(entry.entry_id) + # 注销服务 + hass.services.async_remove(DOMAIN, SERVICE_RING_DEVICE) + + return unload_ok + + +async def update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None: + """Update listener.""" + await hass.config_entries.async_reload(entry.entry_id) \ No newline at end of file diff --git a/custom_components/vivo_cloud/config_flow.py b/custom_components/vivo_cloud/config_flow.py new file mode 100644 index 0000000..90f90ee --- /dev/null +++ b/custom_components/vivo_cloud/config_flow.py @@ -0,0 +1,280 @@ +"""Config flow for Vivo Cloud.""" +import logging +import voluptuous as vol +from typing import Any, Dict, Optional + +from homeassistant import config_entries +from homeassistant.const import CONF_NAME +from homeassistant.core import HomeAssistant, callback +from homeassistant.helpers import config_validation as cv +from homeassistant.helpers.aiohttp_client import async_get_clientsession + +from .const import ( + DOMAIN, + CONF_COOKIE_METHOD, + COOKIE_METHOD_MANUAL, + COOKIE_METHOD_COOKIECLOUD, + CONF_COOKIE, + CONF_COOKIECLOUD_UUID, + CONF_COOKIECLOUD_PASSWORD, + CONF_COOKIECLOUD_SERVER, + DEFAULT_COOKIECLOUD_SERVER, + CONF_DEVICES, + CONF_SCAN_INTERVAL, + DEFAULT_SCAN_INTERVAL, + VIVO_CLOUD_API_BASE, + VIVO_CLOUD_API_DEVICES, +) + +_LOGGER = logging.getLogger(__name__) + + +class VivoCloudConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): + """Handle a config flow for Vivo Cloud.""" + + VERSION = 1 + CONNECTION_CLASS = config_entries.CONN_CLASS_CLOUD_POLL + + def __init__(self): + """Initialize.""" + self._cookie_method = None + self._cookie = None + self._cookiecloud_uuid = None + self._cookiecloud_password = None + self._cookiecloud_server = DEFAULT_COOKIECLOUD_SERVER + self._devices = [] + self._selected_devices = [] + self._scan_interval = DEFAULT_SCAN_INTERVAL + + async def async_step_user(self, user_input=None): + """Handle the initial step.""" + errors = {} + + if user_input is not None: + self._cookie_method = user_input[CONF_COOKIE_METHOD] + return await self.async_step_cookie_config() + + data_schema = vol.Schema({ + vol.Required(CONF_COOKIE_METHOD): vol.In({ + COOKIE_METHOD_MANUAL: "手动输入Cookie", + COOKIE_METHOD_COOKIECLOUD: "使用CookieCloud获取Cookie", + }) + }) + + return self.async_show_form( + step_id="user", + data_schema=data_schema, + errors=errors, + ) + + async def async_step_cookie_config(self, user_input=None): + """Handle the cookie configuration step.""" + errors = {} + + if user_input is not None: + if self._cookie_method == COOKIE_METHOD_MANUAL: + self._cookie = user_input[CONF_COOKIE] + if await self._validate_cookie(self._cookie): + return await self.async_step_device_list() + else: + errors["base"] = "invalid_cookie" + else: # COOKIE_METHOD_COOKIECLOUD + self._cookiecloud_uuid = user_input[CONF_COOKIECLOUD_UUID] + self._cookiecloud_password = user_input[CONF_COOKIECLOUD_PASSWORD] + self._cookiecloud_server = user_input.get( + CONF_COOKIECLOUD_SERVER, + DEFAULT_COOKIECLOUD_SERVER + ) + + cookie = await self._get_cookie_from_cookiecloud() + if cookie: + self._cookie = cookie + return await self.async_step_device_list() + else: + errors["base"] = "cookiecloud_error" + + if self._cookie_method == COOKIE_METHOD_MANUAL: + data_schema = vol.Schema({ + vol.Required(CONF_COOKIE): str, + }) + else: # COOKIE_METHOD_COOKIECLOUD + data_schema = vol.Schema({ + vol.Required(CONF_COOKIECLOUD_UUID): str, + vol.Required(CONF_COOKIECLOUD_PASSWORD): str, + vol.Optional(CONF_COOKIECLOUD_SERVER, default=DEFAULT_COOKIECLOUD_SERVER): str, + }) + + return self.async_show_form( + step_id="cookie_config", + data_schema=data_schema, + errors=errors, + ) + + async def async_step_device_list(self, user_input=None): + """Handle the device list step.""" + errors = {} + + if user_input is not None: + # 获取用户选择的设备 + selected_device_ids = user_input[CONF_DEVICES] + + self._selected_devices = [ + device for device in self._devices + if str(device.get("id")) in selected_device_ids + ] + + if self._selected_devices: + return await self.async_step_scan_interval() + else: + errors["base"] = "no_device_selected" + + # 获取设备列表 + if not self._devices: + self._devices = await self._get_device_list() + if not self._devices: + return self.async_abort(reason="cannot_connect") + + # 创建设备选择列表 + device_options = {} + for device in self._devices: + device_id = str(device.get("id")) + device_name = f"{device.get('alias', 'Unknown')} - {device.get('model', 'Unknown')}" + device_options[device_id] = device_name + + # 创建多选表单 + data_schema = vol.Schema( + { + vol.Required(CONF_DEVICES): cv.multi_select(device_options), + } + ) + + return self.async_show_form( + step_id="device_list", + data_schema=data_schema, + errors=errors, + ) + + async def async_step_scan_interval(self, user_input=None): + """Handle the scan interval step.""" + errors = {} + + if user_input is not None: + self._scan_interval = user_input[CONF_SCAN_INTERVAL] + + # 创建配置条目 + data = { + CONF_COOKIE_METHOD: self._cookie_method, + CONF_DEVICES: self._selected_devices, + } + + if self._cookie_method == COOKIE_METHOD_MANUAL: + data[CONF_COOKIE] = self._cookie + else: # COOKIE_METHOD_COOKIECLOUD + data[CONF_COOKIECLOUD_UUID] = self._cookiecloud_uuid + data[CONF_COOKIECLOUD_PASSWORD] = self._cookiecloud_password + data[CONF_COOKIECLOUD_SERVER] = self._cookiecloud_server + + options = { + CONF_SCAN_INTERVAL: self._scan_interval, + } + + return self.async_create_entry( + title="Vivo Cloud", + data=data, + options=options, + ) + + data_schema = vol.Schema({ + vol.Required( + CONF_SCAN_INTERVAL, + default=self._scan_interval + ): vol.All(vol.Coerce(int), vol.Range(min=30, max=3600)), + }) + + return self.async_show_form( + step_id="scan_interval", + data_schema=data_schema, + errors=errors, + ) + + async def _validate_cookie(self, cookie: str) -> bool: + """Validate cookie by trying to get device list.""" + try: + devices = await self._get_device_list(cookie) + return bool(devices) + except Exception: + return False + + async def _get_cookie_from_cookiecloud(self) -> Optional[str]: + """Get cookie from CookieCloud.""" + try: + from .coordinator import VivoCloudDataUpdateCoordinator + + # 创建临时协调器来获取Cookie + temp_coordinator = VivoCloudDataUpdateCoordinator( + self.hass, # type: ignore + cookie_method=COOKIE_METHOD_COOKIECLOUD, + cookiecloud_uuid=self._cookiecloud_uuid, + cookiecloud_password=self._cookiecloud_password, + cookiecloud_server=self._cookiecloud_server, + scan_interval=0, # 不需要定时更新 + ) + + return await temp_coordinator._get_cookie_from_cookiecloud() + except Exception as error: + _LOGGER.error("Error getting cookie from CookieCloud: %s", error) + return None + + async def _get_device_list(self, cookie: str = None) -> list: + """Get device list from Vivo Cloud.""" + try: + from .coordinator import VivoCloudDataUpdateCoordinator + + # 使用提供的Cookie或使用当前Cookie + cookie_to_use = cookie or self._cookie + if not cookie_to_use: + return [] + + # 创建临时协调器来获取设备列表 + temp_coordinator = VivoCloudDataUpdateCoordinator( + self.hass, # type: ignore + cookie_method=COOKIE_METHOD_MANUAL, + cookie=cookie_to_use, + scan_interval=0, # 不需要定时更新 + ) + + return await temp_coordinator._get_device_list() + except Exception: + return [] + + + @staticmethod + @callback + def async_get_options_flow(config_entry): + """Get the options flow for this handler.""" + return VivoCloudOptionsFlow(config_entry) + + +class VivoCloudOptionsFlow(config_entries.OptionsFlow): + """Handle options flow for Vivo Cloud.""" + + def __init__(self, config_entry): + """Initialize options flow.""" + super().__init__(config_entry) + + async def async_step_init(self, user_input=None): + """Manage the options.""" + if user_input is not None: + return self.async_create_entry(title="", data=user_input) + + data_schema = vol.Schema({ + vol.Required( + CONF_SCAN_INTERVAL, + default=self.config_entry.options.get(CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL), + ): vol.All(vol.Coerce(int), vol.Range(min=30, max=3600)), + }) + + return self.async_show_form( + step_id="init", + data_schema=data_schema, + ) \ No newline at end of file diff --git a/custom_components/vivo_cloud/const.py b/custom_components/vivo_cloud/const.py new file mode 100644 index 0000000..67a9752 --- /dev/null +++ b/custom_components/vivo_cloud/const.py @@ -0,0 +1,64 @@ +"""Const file for Vivo Cloud.""" + +DOMAIN = "vivo_cloud" +COORDINATOR = "coordinator" +DATA_LISTENER = "listener" +UNDO_UPDATE_LISTENER = "undo_update_listener" +DEFAULT_SCAN_INTERVAL = 360 +MIN_SCAN_INTERVAL = 180 +SIGNAL_STATE_UPDATED = f"{DOMAIN}.updated" + +# Cookie设置方式 +CONF_COOKIE_METHOD = "cookie_method" +COOKIE_METHOD_MANUAL = "manual" +COOKIE_METHOD_COOKIECLOUD = "cookiecloud" + +# CookieCloud配置 +CONF_COOKIECLOUD_UUID = "cookiecloud_uuid" +CONF_COOKIECLOUD_PASSWORD = "cookiecloud_password" +CONF_COOKIECLOUD_SERVER = "cookiecloud_server" +DEFAULT_COOKIECLOUD_SERVER = "https://cookie.nextrt.com" + +# 手动Cookie配置 +CONF_COOKIE = "cookie" + +# 设备配置 +CONF_DEVICES = "devices" +CONF_SCAN_INTERVAL = "scan_interval" + +# Vivo Cloud API +VIVO_CLOUD_API_BASE = "https://webcloud.vivo.com.cn" +VIVO_CLOUD_API_DEVICES = "/findphone/v3/devices" +VIVO_CLOUD_API_DEVICE_STATUS = "/findphone/devicestatus" +VIVO_CLOUD_API_OPERATE = "/findphone/operate" + +# 服务 +SERVICE_RING_DEVICE = "ring_device" + +# 设备属性 +ATTR_DEVICE_ALIAS = "alias" +ATTR_DEVICE_MODEL = "model" +ATTR_DEVICE_IMEI = "imei" +ATTR_DEVICE_EMMC_ID = "emmc_id" +ATTR_DEVICE_ID = "device_id" +ATTR_DEVICE_ONLINE = "online" +ATTR_DEVICE_PICTURE = "device_model_picture" +ATTR_DEVICE_LOCATION_KEY = "location" +ATTR_LOCATION_DESC = "location_desc" +ATTR_LOCATION_DESC_KEY = "locationDesc" +ATTR_SIGNAL_INFO = "signal_info" +ATTR_BATTERY_INFO = "battery_info" + +# 信号信息属性 +ATTR_SIM_OPERATOR = "sim_operator" +ATTR_SIM_NETWORK_TYPE = "sim_network_type" +ATTR_SIM_SIGNAL_STRENGTH = "sim_signal_strength" +ATTR_SIM_NUMBER1 = "sim_number1" +ATTR_SIM_NUMBER2 = "sim_number2" + +# 电池信息属性 +ATTR_BATTERY_CAPACITY = "battery_capacity" +ATTR_IS_LOW_BATTERY = "is_low_battery" +ATTR_IS_POWER_SAVING_MODE = "is_power_saving_mode" +ATTR_IS_CHARGING = "is_charging" +ATTR_CHARGING_TYPE = "charging_type" \ No newline at end of file diff --git a/custom_components/vivo_cloud/coordinator.py b/custom_components/vivo_cloud/coordinator.py new file mode 100644 index 0000000..64b429d --- /dev/null +++ b/custom_components/vivo_cloud/coordinator.py @@ -0,0 +1,468 @@ +"""Vivo Cloud Data Update Coordinator.""" +import asyncio +import json +import logging +from datetime import timedelta +from typing import Dict, Any, List, Optional + +import aiohttp +import async_timeout + +from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed +from homeassistant.helpers.aiohttp_client import async_get_clientsession +from homeassistant.core import HomeAssistant + +from .const import ( + VIVO_CLOUD_API_BASE, + VIVO_CLOUD_API_DEVICES, + VIVO_CLOUD_API_DEVICE_STATUS, + VIVO_CLOUD_API_OPERATE, + COOKIE_METHOD_MANUAL, + COOKIE_METHOD_COOKIECLOUD, + DEFAULT_COOKIECLOUD_SERVER, + ATTR_DEVICE_ALIAS, + ATTR_DEVICE_MODEL, + ATTR_DEVICE_IMEI, + ATTR_DEVICE_EMMC_ID, + ATTR_DEVICE_ID, + ATTR_DEVICE_ONLINE, + ATTR_DEVICE_PICTURE, + ATTR_SIGNAL_INFO, + ATTR_BATTERY_INFO, + ATTR_SIM_OPERATOR, + ATTR_SIM_NETWORK_TYPE, + ATTR_SIM_SIGNAL_STRENGTH, + ATTR_SIM_NUMBER1, + ATTR_SIM_NUMBER2, + ATTR_DEVICE_LOCATION_KEY, + ATTR_BATTERY_CAPACITY, + ATTR_IS_LOW_BATTERY, + ATTR_IS_POWER_SAVING_MODE, + ATTR_IS_CHARGING, + ATTR_CHARGING_TYPE, +) + +_LOGGER = logging.getLogger(__name__) + + +class VivoCloudDataUpdateCoordinator(DataUpdateCoordinator): + """Class to manage fetching Vivo Cloud data API.""" + + def __init__( + self, + hass: HomeAssistant, + cookie_method, + cookie=None, + cookiecloud_uuid=None, + cookiecloud_password=None, + cookiecloud_server=None, + devices=None, + scan_interval=60, + ): + """Initialize.""" + self.hass = hass + self.cookie_method = cookie_method + self.cookie = cookie + self.cookiecloud_uuid = cookiecloud_uuid + self.cookiecloud_password = cookiecloud_password + self.cookiecloud_server = cookiecloud_server or DEFAULT_COOKIECLOUD_SERVER + self.devices = devices or [] + self.scan_interval = scan_interval + self._session = async_get_clientsession(hass) + self._cookies = {} + self._device_data = {} + self._round_counter = 0 # 添加round计数器，从0开始 + + update_interval = timedelta(seconds=scan_interval) + + super().__init__(hass, _LOGGER, name="Vivo Cloud", update_interval=update_interval) + + async def _async_update_data(self) -> Dict[str, Any]: + """Update data via library.""" + try: + # 获取Cookie + if self.cookie_method == COOKIE_METHOD_COOKIECLOUD: + self.cookie = await self._get_cookie_from_cookiecloud() + if not self.cookie: + raise UpdateFailed("Failed to get cookie from CookieCloud") + + # 解析Cookie + self._cookies = self._parse_cookie(self.cookie) + + # 获取设备数据 + device_data = await self._get_all_devices_data() + if not device_data: + raise UpdateFailed("Failed to get device data") + + return device_data + + except Exception as error: + raise UpdateFailed(f"Error updating Vivo Cloud data: {error}") + + async def _get_cookie_from_cookiecloud(self) -> Optional[str]: + """Get cookie from CookieCloud.""" + try: + # 发送密码到服务端进行解密 + url = f"{self.cookiecloud_server}/get/{self.cookiecloud_uuid}" + + # 准备请求数据 + request_data = { + "password": self.cookiecloud_password + } + + async with async_timeout.timeout(10): + async with self._session.post( + url, + headers={"Content-Type": "application/json"}, + json=request_data + ) as response: + if response.status != 200: + _LOGGER.error("Failed to get data from CookieCloud: %s", response.status) + return None + + data = await response.json() + + if not data: + _LOGGER.error("Invalid response from CookieCloud") + return None + + # 检查是否是错误响应 + if "error" in data: + _LOGGER.error("CookieCloud returned error: %s", data["error"]) + return None + + # 提取Cookie + cookies = [] + vivo_cookies = [] + + # 检查数据结构 + if "cookie_data" in data: + # 使用字典来去重，键为Cookie名称，值为Cookie字符串 + unique_cookies = {} + + for domain, domain_cookies in data["cookie_data"].items(): + # 检查是否是Vivo Cloud相关域名 + is_vivo_domain = "vivo.com.cn" in domain + + if is_vivo_domain: + for cookie in domain_cookies: + if isinstance(cookie, dict) and 'name' in cookie and 'value' in cookie: + cookie_str = f"{cookie['name']}={cookie['value']}" + # 使用字典去重，相同名称的Cookie只保留最后一个 + unique_cookies[cookie['name']] = cookie_str + # 将去重后的Cookie添加到列表中 + cookies = list(unique_cookies.values()) + vivo_cookies = list(unique_cookies.keys()) + elif "cookies" in data: + for cookie in data["cookies"]: + if isinstance(cookie, dict) and 'name' in cookie and 'value' in cookie: + cookie_str = f"{cookie['name']}={cookie['value']}" + cookies.append(cookie_str) + vivo_cookies.append(cookie['name']) + else: + _LOGGER.error("Unknown CookieCloud data structure") + _LOGGER.error("Available keys: %s", list(data.keys())) + return None + + if not cookies: + _LOGGER.error("No Vivo Cloud cookies found in CookieCloud data") + _LOGGER.error("Full CookieCloud response: %s", json.dumps(data, indent=2)) + return None + + # 检查必要的Cookie是否存在 + required_cookies = ["JSESSIONID", "vivo_yun_csrftoken"] + missing_cookies = [cookie for cookie in required_cookies if cookie not in vivo_cookies] + + if missing_cookies: + _LOGGER.warning("Missing potentially required cookies: %s", missing_cookies) + + return "; ".join(cookies) + + except Exception as error: + _LOGGER.error("Error getting cookie from CookieCloud: %s", error) + return None + + + + def _parse_cookie(self, cookie_str: str) -> Dict[str, str]: + """Parse cookie string to dict.""" + cookies = {} + if cookie_str: + for item in cookie_str.split(';'): + if '=' in item: + key, value = item.strip().split('=', 1) + cookies[key] = value + return cookies + + async def _get_all_devices_data(self) -> Dict[str, Any]: + """Get all devices data.""" + try: + # 获取设备列表 + devices = await self._get_device_list() + if not devices: + return {} + + # 获取每个设备的详细信息 + device_data = {} + for device in devices: + if device.get("id") in [d.get("id") for d in self.devices]: + device_info = await self._get_device_info(device) + if device_info: + device_data[str(device.get("id"))] = device_info + + return device_data + + except Exception as error: + _LOGGER.error("Error getting all devices data: %s", error) + return {} + + async def _get_device_list(self) -> List[Dict[str, Any]]: + """Get device list from Vivo Cloud.""" + try: + url = f"{VIVO_CLOUD_API_BASE}{VIVO_CLOUD_API_DEVICES}" + headers = { + "accept": "application/json, text/plain, */*", + "accept-language": "zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6", + "content-type": "application/x-www-form-urlencoded;charset=UTF-8", + "origin": "https://find.vivo.com.cn", + "referer": "https://find.vivo.com.cn/", + "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36 Edg/139.0.0.0", + } + + # 添加Cookie到HTTP头 + if self.cookie: + headers["cookie"] = self.cookie + + data = "round=0" + + async with async_timeout.timeout(10): + async with self._session.post( + url, + headers=headers, + data=data + ) as response: + if response.status != 200: + _LOGGER.error("Failed to get device list: %s", response.status) + return [] + + result = await response.json() + + if result.get("code") != 0: + _LOGGER.error("Error in device list response: %s", result.get("msg")) + + # 检查是否是登录异常 + if "登录异常" in result.get("msg", ""): + _LOGGER.error("Login failed, cookies may be invalid or expired") + + return [] + + return result.get("data", {}).get("myDevices", []) + + except Exception as error: + _LOGGER.error("Error getting device list: %s", error) + return [] + + async def _get_device_info(self, device: Dict[str, Any]) -> Optional[Dict[str, Any]]: + """Get device info from Vivo Cloud.""" + try: + # 先调用 async_update_device_info 让后台对数据进行更新 + device_id = str(device.get("id")) + if device_id: + update_success = await self.async_update_device_info(device_id) + if not update_success: + _LOGGER.warning("Failed to update device info for device %s before getting info", device_id) + + url = f"{VIVO_CLOUD_API_BASE}{VIVO_CLOUD_API_DEVICE_STATUS}" + headers = { + "accept": "application/json, text/plain, */*", + "accept-language": "zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6", + "content-type": "application/x-www-form-urlencoded;charset=UTF-8", + "origin": "https://find.vivo.com.cn", + "referer": "https://find.vivo.com.cn/", + "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36 Edg/139.0.0.0", + } + + # 添加Cookie到HTTP头 + if self.cookie: + headers["cookie"] = self.cookie + + data = ( + f"version={device.get('version', 4)}&" + f"model={device.get('alias', '')}&" + f"imei={device.get('imei', '')}&" + f"round={self._round_counter}&" + f"source=devicepage&" + f"emmcId={device.get('emmcId', '')}" + ) + + # 更新round计数器，当到10000时从0重新开始 + self._round_counter = (self._round_counter + 1) % 10000 + + async with async_timeout.timeout(10): + async with self._session.post( + url, + headers=headers, + data=data + ) as response: + if response.status != 200: + _LOGGER.error("Failed to get device info: %s", response.status) + return None + + result = await response.json() + if result.get("code") != 0: + _LOGGER.error("Error in device info response: %s", result.get("msg")) + return None + + _LOGGER.error("后台返回数据为: %s", result) + data = result.get("data", {}) + device_vo = data.get("deviceVO", {}) + + # 提取设备信息，使用常量定义的键名 + signal_info = data.get("signalInfo", {}) + battery_info = data.get("batteryInfo", {}) + + device_info = { + ATTR_DEVICE_ID: device_vo.get("id"), + ATTR_DEVICE_EMMC_ID: device_vo.get("emmcId"), + ATTR_DEVICE_IMEI: device_vo.get("imei"), + ATTR_DEVICE_MODEL: device_vo.get("model"), + ATTR_DEVICE_ALIAS: device_vo.get("alias"), + ATTR_DEVICE_ONLINE: device_vo.get("online"), + ATTR_DEVICE_PICTURE: device_vo.get("deviceModelPicture"), + ATTR_DEVICE_LOCATION_KEY: device_vo.get("location", {}), + ATTR_SIGNAL_INFO: { + ATTR_SIM_OPERATOR: signal_info.get("simOperator"), + ATTR_SIM_NETWORK_TYPE: signal_info.get("simNetWorkType"), + ATTR_SIM_SIGNAL_STRENGTH: signal_info.get("simSignalStrength"), + ATTR_SIM_NUMBER1: signal_info.get("simNumber1"), + ATTR_SIM_NUMBER2: signal_info.get("simNumber2"), + }, + ATTR_BATTERY_INFO: { + ATTR_BATTERY_CAPACITY: battery_info.get("batteryCapacity"), + ATTR_IS_LOW_BATTERY: battery_info.get("isLowBattery"), + ATTR_IS_POWER_SAVING_MODE: battery_info.get("isPowerSavingMode"), + ATTR_IS_CHARGING: battery_info.get("isCharging"), + ATTR_CHARGING_TYPE: battery_info.get("chargingType"), + }, + } + + return device_info + + except Exception as error: + _LOGGER.error("Error getting device info: %s", error) + return None + + async def async_ring_device(self, device_id: str) -> bool: + """Ring device.""" + try: + # 找到设备 + device = None + for d in self.devices: + if str(d.get("id")) == device_id: + device = d + break + + if not device: + _LOGGER.error("Device not found: %s", device_id) + return False + + url = f"{VIVO_CLOUD_API_BASE}{VIVO_CLOUD_API_OPERATE}" + headers = { + "accept": "application/json, text/plain, */*", + "accept-language": "zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6", + "content-type": "application/x-www-form-urlencoded;charset=UTF-8", + "origin": "https://find.vivo.com.cn", + "referer": "https://find.vivo.com.cn/", + "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36 Edg/139.0.0.0", + } + + # 添加Cookie到HTTP头 + if self.cookie: + headers["cookie"] = self.cookie + + data = ( + f"emmcId={device.get('emmcId', '')}&" + f"imei={device.get('imei', '')}&" + f"version={device.get('version', 4)}&" + f"cmdType=2&" + f"operateVersion=1&" + f"sequenceNo=0" + ) + + async with async_timeout.timeout(10): + async with self._session.post( + url, + headers=headers, + data=data + ) as response: + if response.status != 200: + _LOGGER.error("Failed to ring device: %s", response.status) + return False + + result = await response.json() + if result.get("code") != 0: + _LOGGER.error("Error in ring device response: %s", result.get("msg")) + return False + + return True + + except Exception as error: + _LOGGER.error("Error ringing device: %s", error) + return False + + + async def async_update_device_info(self, device_id: str) -> bool: + """update device.""" + try: + # 更新设备信息 + device = None + for d in self.devices: + if str(d.get("id")) == device_id: + device = d + break + + if not device: + _LOGGER.error("Device not found: %s", device_id) + return False + + url = f"{VIVO_CLOUD_API_BASE}{VIVO_CLOUD_API_OPERATE}" + headers = { + "accept": "application/json, text/plain, */*", + "accept-language": "zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6", + "content-type": "application/x-www-form-urlencoded;charset=UTF-8", + "origin": "https://find.vivo.com.cn", + "referer": "https://find.vivo.com.cn/", + "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36 Edg/139.0.0.0", + } + + # 添加Cookie到HTTP头 + if self.cookie: + headers["cookie"] = self.cookie + + data = ( + f"emmcId={device.get('emmcId', '')}&" + f"imei={device.get('imei', '')}&" + f"version={device.get('version', 4)}&" + f"cmdType=1" + ) + + async with async_timeout.timeout(10): + async with self._session.post( + url, + headers=headers, + data=data + ) as response: + if response.status != 200: + _LOGGER.error("Failed to update device: %s", response.status) + return False + + result = await response.json() + if result.get("code") != 0: + _LOGGER.error("Error in update device response: %s", result.get("msg")) + return False + + return True + + except Exception as error: + _LOGGER.error("Error update device: %s", error) + return False \ No newline at end of file diff --git a/custom_components/vivo_cloud/device_tracker.py b/custom_components/vivo_cloud/device_tracker.py new file mode 100644 index 0000000..560761d --- /dev/null +++ b/custom_components/vivo_cloud/device_tracker.py @@ -0,0 +1,298 @@ +"""Support for the Vivo Cloud device tracking.""" +import logging +import math +from typing import Dict, Any, Optional, Tuple + +from homeassistant.components.device_tracker import SourceType +from homeassistant.components.device_tracker.config_entry import TrackerEntity +from homeassistant.const import ( + ATTR_BATTERY_LEVEL, + ATTR_GPS_ACCURACY, + ATTR_LATITUDE, + ATTR_LONGITUDE, +) +from homeassistant.core import callback +from homeassistant.helpers.dispatcher import async_dispatcher_connect +from homeassistant.helpers.entity import Entity +from homeassistant.core import HomeAssistant + +from .const import ( + DOMAIN, + COORDINATOR, + ATTR_DEVICE_ALIAS, + ATTR_DEVICE_MODEL, + ATTR_DEVICE_IMEI, + ATTR_DEVICE_EMMC_ID, + ATTR_DEVICE_ID, + ATTR_DEVICE_ONLINE, + ATTR_DEVICE_PICTURE, + ATTR_LOCATION_DESC, + ATTR_LOCATION_DESC_KEY, + ATTR_SIGNAL_INFO, + ATTR_BATTERY_INFO, + ATTR_SIM_OPERATOR, + ATTR_SIM_NETWORK_TYPE, + ATTR_SIM_SIGNAL_STRENGTH, + ATTR_SIM_NUMBER1, + ATTR_SIM_NUMBER2, + ATTR_BATTERY_CAPACITY, + ATTR_IS_LOW_BATTERY, + ATTR_IS_POWER_SAVING_MODE, + ATTR_IS_CHARGING, + ATTR_CHARGING_TYPE, +) + +def bd09_to_wgs84(bd_lon: float, bd_lat: float) -> Tuple[float, float]: + """ + 将 BD09 坐标系转换为 WGS-84 坐标系 + + 参数: + bd_lon: BD09 坐标系经度 + bd_lat: BD09 坐标系纬度 + + 返回: + Tuple[float, float]: WGS-84 坐标系的经度和纬度 + """ + # 首先将 BD09 转换为 GCJ-02 + gcj_lon, gcj_lat = bd09_to_gcj02(bd_lon, bd_lat) + + # 然后将 GCJ-02 转换为 WGS-84 + wgs_lon, wgs_lat = gcj02_to_wgs84(gcj_lon, gcj_lat) + + return wgs_lon, wgs_lat + + +def bd09_to_gcj02(bd_lon: float, bd_lat: float) -> Tuple[float, float]: + """ + 将 BD09 坐标系转换为 GCJ-02 坐标系 + + 参数: + bd_lon: BD09 坐标系经度 + bd_lat: BD09 坐标系纬度 + + 返回: + Tuple[float, float]: GCJ-02 坐标系的经度和纬度 + """ + x_pi = math.pi * 3000.0 / 180.0 + + # BD09 坐标系转 GCJ-02 坐标系 + x = bd_lon - 0.0065 + y = bd_lat - 0.006 + z = math.sqrt(x * x + y * y) - 0.00002 * math.sin(y * x_pi) + theta = math.atan2(y, x) - 0.000003 * math.cos(x * x_pi) + + gcj_lon = z * math.cos(theta) + gcj_lat = z * math.sin(theta) + + return gcj_lon, gcj_lat + + +def gcj02_to_wgs84(gcj_lon: float, gcj_lat: float) -> Tuple[float, float]: + """ + 将 GCJ-02 坐标系转换为 WGS-84 坐标系 + + 参数: + gcj_lon: GCJ-02 坐标系经度 + gcj_lat: GCJ-02 坐标系纬度 + + 返回: + Tuple[float, float]: WGS-84 坐标系的经度和纬度 + """ + a = 6378245.0 # 长半轴 + ee = 0.00669342162296594323 # 偏心率平方 + + # GCJ-02 坐标系转 WGS-84 坐标系 + dlat = _transformlat(gcj_lon - 105.0, gcj_lat - 35.0) + dlon = _transformlon(gcj_lon - 105.0, gcj_lat - 35.0) + radlat = gcj_lat / 180.0 * math.pi + magic = math.sin(radlat) + magic = 1 - ee * magic * magic + sqrtmagic = math.sqrt(magic) + dlat = (dlat * 180.0) / ((a * (1 - ee)) / (magic * sqrtmagic) * math.pi) + dlon = (dlon * 180.0) / (a / sqrtmagic * math.cos(radlat) * math.pi) + + wgs_lat = gcj_lat - dlat + wgs_lon = gcj_lon - dlon + + return wgs_lon, wgs_lat + + +def _transformlat(lon: float, lat: float) -> float: + """经度偏移转换""" + ret = -100.0 + 2.0 * lon + 3.0 * lat + 0.2 * lat * lat + \ + 0.1 * lon * lat + 0.2 * math.sqrt(abs(lon)) + ret += (20.0 * math.sin(6.0 * lon * math.pi) + 20.0 * math.sin(2.0 * lon * math.pi)) * 2.0 / 3.0 + ret += (20.0 * math.sin(lat * math.pi) + 40.0 * math.sin(lat / 3.0 * math.pi)) * 2.0 / 3.0 + ret += (160.0 * math.sin(lat / 12.0 * math.pi) + 320 * math.sin(lat * math.pi / 30.0)) * 2.0 / 3.0 + return ret + + +def _transformlon(lon: float, lat: float) -> float: + """纬度偏移转换""" + ret = 300.0 + lon + 2.0 * lat + 0.1 * lon * lon + \ + 0.1 * lon * lat + 0.1 * math.sqrt(abs(lon)) + ret += (20.0 * math.sin(6.0 * lon * math.pi) + 20.0 * math.sin(2.0 * lon * math.pi)) * 2.0 / 3.0 + ret += (20.0 * math.sin(lon * math.pi) + 40.0 * math.sin(lon / 3.0 * math.pi)) * 2.0 / 3.0 + ret += (150.0 * math.sin(lon / 12.0 * math.pi) + 300.0 * math.sin(lon / 30.0 * math.pi)) * 2.0 / 3.0 + return ret + + +_LOGGER = logging.getLogger(__name__) + + +async def async_setup_entry( + hass: HomeAssistant, config_entry, async_add_entities +): + """Configure a dispatcher connection based on a config entry.""" + coordinator = hass.data[DOMAIN][config_entry.entry_id][COORDINATOR] + devices = [] + + # 为每个选中的设备创建跟踪实体 + for device in config_entry.data.get("devices", []): + device_id = str(device.get("id")) + devices.append(VivoDeviceEntity(hass, coordinator, device_id)) + + async_add_entities(devices, True) + + +class VivoDeviceEntity(TrackerEntity, Entity): + """Represent a tracked device.""" + + def __init__(self, hass, coordinator, device_id): + """Set up Vivo device entity.""" + self._hass = hass + self.coordinator = coordinator + self._device_id = device_id + self._device_info = {} + self._name = None + self._unique_id = None + self._icon = "mdi:cellphone" + self._should_poll = False + + async def async_added_to_hass(self): + """Subscribe for update from the coordinator.""" + self.async_on_remove( + self.coordinator.async_add_listener(self.async_write_ha_state) + ) + + @property + def should_poll(self): + """Return the polling requirement of the entity.""" + return self._should_poll + + @property + def unique_id(self): + """Return the unique ID.""" + return f"{DOMAIN}_{self._device_id}" + + @property + def name(self): + """Return the name of the device.""" + device_data = self.coordinator.data.get(self._device_id, {}) + return device_data.get(ATTR_DEVICE_ALIAS, f"Vivo Device {self._device_id}") + + @property + def icon(self): + """Icon to use in the frontend, if any.""" + return self._icon + + @property + def source_type(self): + """Return the source type, eg gps or router, of the device.""" + return SourceType.GPS + + @property + def latitude(self) -> Optional[float]: + """Return latitude value of the device.""" + device_data = self.coordinator.data.get(self._device_id, {}) + location = device_data.get("location", {}) + try: + bd_lat = float(location.get("latitude", 0)) + bd_lon = float(location.get("longitude", 0)) + _LOGGER.debug("Original BD09 coordinates: lon=%s, lat=%s", bd_lon, bd_lat) + # 将 BD09 坐标系转换为 WGS-84 坐标系 + wgs_lon, wgs_lat = bd09_to_wgs84(bd_lon, bd_lat) + _LOGGER.debug("Converted WGS-84 coordinates: lon=%s, lat=%s", wgs_lon, wgs_lat) + return wgs_lat + except (ValueError, TypeError): + return None + + @property + def longitude(self) -> Optional[float]: + """Return longitude value of the device.""" + device_data = self.coordinator.data.get(self._device_id, {}) + location = device_data.get("location", {}) + try: + bd_lat = float(location.get("latitude", 0)) + bd_lon = float(location.get("longitude", 0)) + _LOGGER.debug("Original BD09 coordinates: lon=%s, lat=%s", bd_lon, bd_lat) + # 将 BD09 坐标系转换为 WGS-84 坐标系 + wgs_lon, wgs_lat = bd09_to_wgs84(bd_lon, bd_lat) + _LOGGER.debug("Converted WGS-84 coordinates: lon=%s, lat=%s", wgs_lon, wgs_lat) + return wgs_lon + except (ValueError, TypeError): + return None + + @property + def location_accuracy(self) -> Optional[int]: + """Return the gps accuracy of the device.""" + # Vivo Cloud API没有直接提供精度信息，我们使用一个默认值 + return 100 + + @property + def battery_level(self) -> Optional[int]: + """Return battery value of the device.""" + device_data = self.coordinator.data.get(self._device_id, {}) + battery_info = device_data.get(ATTR_BATTERY_INFO, {}) + try: + return int(battery_info.get(ATTR_BATTERY_CAPACITY, 0)) + except (ValueError, TypeError): + return None + + @property + def device_state_attributes(self) -> Dict[str, Any]: + """Return device specific attributes.""" + device_data = self.coordinator.data.get(self._device_id, {}) + location = device_data.get("location", {}) + signal_info = device_data.get(ATTR_SIGNAL_INFO, {}) + battery_info = device_data.get(ATTR_BATTERY_INFO, {}) + + attrs = { + ATTR_DEVICE_ID: device_data.get(ATTR_DEVICE_ID), + ATTR_DEVICE_ALIAS: device_data.get(ATTR_DEVICE_ALIAS), + ATTR_DEVICE_MODEL: device_data.get(ATTR_DEVICE_MODEL), + ATTR_DEVICE_IMEI: device_data.get(ATTR_DEVICE_IMEI), + ATTR_DEVICE_EMMC_ID: device_data.get(ATTR_DEVICE_EMMC_ID), + ATTR_DEVICE_ONLINE: device_data.get(ATTR_DEVICE_ONLINE), + ATTR_DEVICE_PICTURE: device_data.get(ATTR_DEVICE_PICTURE), + ATTR_LOCATION_DESC: location.get(ATTR_LOCATION_DESC_KEY), + ATTR_SIM_OPERATOR: signal_info.get(ATTR_SIM_OPERATOR), + ATTR_SIM_NETWORK_TYPE: signal_info.get(ATTR_SIM_NETWORK_TYPE), + ATTR_SIM_SIGNAL_STRENGTH: signal_info.get(ATTR_SIM_SIGNAL_STRENGTH), + ATTR_SIM_NUMBER1: signal_info.get(ATTR_SIM_NUMBER1), + ATTR_SIM_NUMBER2: signal_info.get(ATTR_SIM_NUMBER2), + ATTR_BATTERY_CAPACITY: battery_info.get(ATTR_BATTERY_CAPACITY), + ATTR_IS_LOW_BATTERY: battery_info.get(ATTR_IS_LOW_BATTERY), + ATTR_IS_POWER_SAVING_MODE: battery_info.get(ATTR_IS_POWER_SAVING_MODE), + ATTR_IS_CHARGING: battery_info.get(ATTR_IS_CHARGING), + ATTR_CHARGING_TYPE: battery_info.get(ATTR_CHARGING_TYPE), + } + + # 过滤掉None值 + return {k: v for k, v in attrs.items() if v is not None} + + @property + def device_info(self) -> Dict[str, Any]: + """Return the device info.""" + device_data = self.coordinator.data.get(self._device_id, {}) + return { + "identifiers": {(DOMAIN, self._device_id)}, + "name": device_data.get(ATTR_DEVICE_ALIAS, f"[Vivo] {self._device_id}"), + "manufacturer": "Vivo", + "model": device_data.get(ATTR_DEVICE_MODEL, "Unknown"), + "sw_version": "1.0.0", + } + + async def async_ring_device(self) -> bool: + """Ring the device.""" + return await self.coordinator.async_ring_device(self._device_id) \ No newline at end of file diff --git a/custom_components/vivo_cloud/manifest.json b/custom_components/vivo_cloud/manifest.json new file mode 100644 index 0000000..56da5b1 --- /dev/null +++ b/custom_components/vivo_cloud/manifest.json @@ -0,0 +1,12 @@ +{ + "domain": "vivo_cloud", + "name": "Vivo Cloud", + "documentation": "https://bbs.233py.com", + "issue_tracker": "https://bbs.233py.com", + "dependencies": [], + "iot_class": "cloud_polling", + "version": "1.0.0", + "config_flow": true, + "codeowners": ["@xyzmos"], + "requirements": [] +} \ No newline at end of file diff --git a/custom_components/vivo_cloud/sensor.py b/custom_components/vivo_cloud/sensor.py new file mode 100644 index 0000000..8ea6c1e --- /dev/null +++ b/custom_components/vivo_cloud/sensor.py @@ -0,0 +1,211 @@ +"""Support for the Vivo Cloud sensor.""" +import logging +from typing import Dict, Any, Optional + +from homeassistant.components.sensor import SensorEntity, SensorDeviceClass +from homeassistant.const import ( + PERCENTAGE, + SIGNAL_STRENGTH_DECIBELS_MILLIWATT, +) +from homeassistant.core import HomeAssistant, callback +from homeassistant.helpers.dispatcher import async_dispatcher_connect +from homeassistant.helpers.entity import Entity + +from .const import ( + DOMAIN, + COORDINATOR, + ATTR_DEVICE_ALIAS, + ATTR_DEVICE_MODEL, + ATTR_DEVICE_IMEI, + ATTR_DEVICE_EMMC_ID, + ATTR_DEVICE_ID, + ATTR_DEVICE_ONLINE, + ATTR_DEVICE_LOCATION_KEY, + ATTR_DEVICE_PICTURE, + ATTR_LOCATION_DESC, + ATTR_LOCATION_DESC_KEY, + ATTR_SIGNAL_INFO, + ATTR_BATTERY_INFO, + ATTR_SIM_OPERATOR, + ATTR_SIM_NETWORK_TYPE, + ATTR_SIM_SIGNAL_STRENGTH, + ATTR_SIM_NUMBER1, + ATTR_SIM_NUMBER2, + ATTR_BATTERY_CAPACITY, + ATTR_IS_LOW_BATTERY, + ATTR_IS_POWER_SAVING_MODE, + ATTR_IS_CHARGING, + ATTR_CHARGING_TYPE, +) + +_LOGGER = logging.getLogger(__name__) + + +async def async_setup_entry( + hass: HomeAssistant, config_entry, async_add_entities +): + """Configure a dispatcher connection based on a config entry.""" + coordinator = hass.data[DOMAIN][config_entry.entry_id][COORDINATOR] + sensors = [] + + # 为每个选中的设备创建传感器 + for device in config_entry.data.get("devices", []): + device_id = str(device.get("id")) + + # 设备基本信息传感器 + sensors.append(VivoDeviceSensor(hass, coordinator, device_id, "alias", "别名")) + sensors.append(VivoDeviceSensor(hass, coordinator, device_id, "model", "型号")) + sensors.append(VivoDeviceSensor(hass, coordinator, device_id, "imei", "IMEI")) + sensors.append(VivoDeviceSensor(hass, coordinator, device_id, "emmc_id", "EMMC ID")) + sensors.append(VivoDeviceSensor(hass, coordinator, device_id, "online", "在线状态")) + sensors.append(VivoDeviceSensor(hass, coordinator, device_id, "location_desc", "位置描述")) + + # 信号信息传感器 + sensors.append(VivoDeviceSensor(hass, coordinator, device_id, "sim_operator", "运营商")) + sensors.append(VivoDeviceSensor(hass, coordinator, device_id, "sim_network_type", "网络类型")) + sensors.append(VivoDeviceSensor(hass, coordinator, device_id, "sim_signal_strength", "信号强度", SensorDeviceClass.SIGNAL_STRENGTH, SIGNAL_STRENGTH_DECIBELS_MILLIWATT)) + sensors.append(VivoDeviceSensor(hass, coordinator, device_id, "sim_number1", "SIM卡1号码")) + sensors.append(VivoDeviceSensor(hass, coordinator, device_id, "sim_number2", "SIM卡2号码")) + + # 电池信息传感器 + sensors.append(VivoBatterySensor(hass, coordinator, device_id, "battery_capacity", "电池容量", SensorDeviceClass.BATTERY, PERCENTAGE)) + sensors.append(VivoDeviceSensor(hass, coordinator, device_id, "is_low_battery", "低电量")) + sensors.append(VivoDeviceSensor(hass, coordinator, device_id, "is_power_saving_mode", "省电模式")) + sensors.append(VivoDeviceSensor(hass, coordinator, device_id, "is_charging", "充电状态")) + sensors.append(VivoDeviceSensor(hass, coordinator, device_id, "charging_type", "充电类型")) + + async_add_entities(sensors, True) + + +class VivoDeviceSensor(SensorEntity): + """Represent a Vivo device sensor.""" + + def __init__(self, hass, coordinator, device_id, sensor_type, name, device_class=None, unit_of_measurement=None): + """Set up Vivo device sensor.""" + self._hass = hass + self.coordinator = coordinator + self._device_id = device_id + self._sensor_type = sensor_type + self._name = name + self._device_class = device_class + self._unit_of_measurement = unit_of_measurement + self._should_poll = False + + async def async_added_to_hass(self): + """Subscribe for update from the coordinator.""" + self.async_on_remove( + self.coordinator.async_add_listener(self.async_write_ha_state) + ) + + @property + def should_poll(self): + """Return the polling requirement of the entity.""" + return self._should_poll + + @property + def unique_id(self): + """Return the unique ID.""" + return f"{DOMAIN}_{self._device_id}_{self._sensor_type}" + + @property + def name(self): + """Return the name of the sensor.""" + device_data = self.coordinator.data.get(self._device_id, {}) + device_alias = device_data.get(ATTR_DEVICE_ALIAS, f"Vivo Device {self._device_id}") + return f"{device_alias} {self._name}" + + @property + def device_class(self): + """Return the device class of the sensor.""" + return self._device_class + + @property + def unit_of_measurement(self): + """Return the unit of measurement.""" + return self._unit_of_measurement + + @property + def device_info(self): + """Return the device info.""" + return { + "identifiers": {(DOMAIN, self._device_id)}, + "name": f"Vivo Device {self._device_id}", + "manufacturer": "Vivo", + "model": "Unknown", + } + + @property + def state(self): + """Return the state of the sensor.""" + device_data = self.coordinator.data.get(self._device_id, {}) + + if self._sensor_type == "alias": + return device_data.get(ATTR_DEVICE_ALIAS) + elif self._sensor_type == "model": + return device_data.get(ATTR_DEVICE_MODEL) + elif self._sensor_type == "imei": + return device_data.get(ATTR_DEVICE_IMEI) + elif self._sensor_type == "emmc_id": + return device_data.get(ATTR_DEVICE_EMMC_ID) + elif self._sensor_type == "online": + return device_data.get(ATTR_DEVICE_ONLINE) + elif self._sensor_type == "location_desc": + location = device_data.get(ATTR_DEVICE_LOCATION_KEY, {}) + return location.get(ATTR_LOCATION_DESC_KEY) + elif self._sensor_type == "sim_operator": + signal_info = device_data.get(ATTR_SIGNAL_INFO, {}) + return signal_info.get(ATTR_SIM_OPERATOR) + elif self._sensor_type == "sim_network_type": + signal_info = device_data.get(ATTR_SIGNAL_INFO, {}) + return signal_info.get(ATTR_SIM_NETWORK_TYPE) + elif self._sensor_type == "sim_signal_strength": + signal_info = device_data.get(ATTR_SIGNAL_INFO, {}) + return signal_info.get(ATTR_SIM_SIGNAL_STRENGTH) + elif self._sensor_type == "sim_number1": + signal_info = device_data.get(ATTR_SIGNAL_INFO, {}) + return signal_info.get(ATTR_SIM_NUMBER1) + elif self._sensor_type == "sim_number2": + signal_info = device_data.get(ATTR_SIGNAL_INFO, {}) + return signal_info.get(ATTR_SIM_NUMBER2) + elif self._sensor_type == "battery_capacity": + battery_info = device_data.get(ATTR_BATTERY_INFO, {}) + return battery_info.get(ATTR_BATTERY_CAPACITY) + elif self._sensor_type == "is_low_battery": + battery_info = device_data.get(ATTR_BATTERY_INFO, {}) + return battery_info.get(ATTR_IS_LOW_BATTERY) + elif self._sensor_type == "is_power_saving_mode": + battery_info = device_data.get(ATTR_BATTERY_INFO, {}) + return battery_info.get(ATTR_IS_POWER_SAVING_MODE) + elif self._sensor_type == "is_charging": + battery_info = device_data.get(ATTR_BATTERY_INFO, {}) + return battery_info.get(ATTR_IS_CHARGING) + elif self._sensor_type == "charging_type": + battery_info = device_data.get(ATTR_BATTERY_INFO, {}) + return battery_info.get(ATTR_CHARGING_TYPE) + + return None + + @property + def extra_state_attributes(self): + """Return the state attributes.""" + return {} + + +class VivoBatterySensor(VivoDeviceSensor): + """Represent a Vivo device battery sensor with additional attributes.""" + + @property + def extra_state_attributes(self): + """Return the state attributes.""" + device_data = self.coordinator.data.get(self._device_id, {}) + battery_info = device_data.get(ATTR_BATTERY_INFO, {}) + + attrs = { + "is_low_battery": battery_info.get(ATTR_IS_LOW_BATTERY), + "is_power_saving_mode": battery_info.get(ATTR_IS_POWER_SAVING_MODE), + "is_charging": battery_info.get(ATTR_IS_CHARGING), + "charging_type": battery_info.get(ATTR_CHARGING_TYPE), + } + + # 过滤掉None值 + return {k: v for k, v in attrs.items() if v is not None} \ No newline at end of file diff --git a/custom_components/vivo_cloud/services.yaml b/custom_components/vivo_cloud/services.yaml new file mode 100644 index 0000000..35c8098 --- /dev/null +++ b/custom_components/vivo_cloud/services.yaml @@ -0,0 +1,15 @@ +ring_device: + name: Ring Device + description: Ring a Vivo device to find it + target: + entity: + domain: device_tracker + integration: vivo_cloud + fields: + device_id: + name: Device ID + description: The ID of the device to ring + required: true + example: "123456789" + selector: + text: \ No newline at end of file diff --git a/custom_components/vivo_cloud/translations/en.json b/custom_components/vivo_cloud/translations/en.json new file mode 100644 index 0000000..8176651 --- /dev/null +++ b/custom_components/vivo_cloud/translations/en.json @@ -0,0 +1,70 @@ +{ + "config": { + "step": { + "user": { + "title": "Vivo Cloud Setup", + "description": "Choose how to provide your Vivo Cloud cookie", + "data": { + "cookie_method": "Cookie Method" + } + }, + "cookie_config": { + "title": "Cookie Configuration", + "description": "Configure your Vivo Cloud cookie", + "data": { + "cookie": "Cookie", + "cookiecloud_uuid": "CookieCloud UUID", + "cookiecloud_password": "CookieCloud Password", + "cookiecloud_server": "CookieCloud Server" + } + }, + "device_list": { + "title": "Select Devices", + "description": "Select the devices you want to track", + "data": { + "devices": "Devices" + } + }, + "scan_interval": { + "title": "Scan Interval", + "description": "Set how often to update device location (in seconds)", + "data": { + "scan_interval": "Scan Interval" + } + } + }, + "error": { + "invalid_cookie": "Invalid cookie, please check and try again", + "cookiecloud_error": "Failed to get cookie from CookieCloud, please check your UUID and password", + "cannot_connect": "Cannot connect to Vivo Cloud, please check your cookie and network connection", + "no_device_selected": "Please select at least one device" + }, + "abort": { + "already_configured": "Vivo Cloud is already configured", + "cannot_connect": "Cannot connect to Vivo Cloud" + } + }, + "options": { + "step": { + "init": { + "title": "Vivo Cloud Options", + "description": "Configure Vivo Cloud options", + "data": { + "scan_interval": "Scan Interval (seconds)" + } + } + } + }, + "services": { + "ring_device": { + "name": "Ring Device", + "description": "Ring a Vivo device to find it", + "fields": { + "device_id": { + "name": "Device ID", + "description": "The ID of the device to ring" + } + } + } + } +} \ No newline at end of file diff --git a/custom_components/vivo_cloud/translations/zh-Hans.json b/custom_components/vivo_cloud/translations/zh-Hans.json new file mode 100644 index 0000000..dd8465f --- /dev/null +++ b/custom_components/vivo_cloud/translations/zh-Hans.json @@ -0,0 +1,70 @@ +{ + "config": { + "step": { + "user": { + "title": "Vivo Cloud 设置", + "description": "选择如何提供您的 Vivo Cloud Cookie", + "data": { + "cookie_method": "Cookie 方式" + } + }, + "cookie_config": { + "title": "Cookie 配置", + "description": "配置您的 Vivo Cloud Cookie", + "data": { + "cookie": "Cookie", + "cookiecloud_uuid": "CookieCloud UUID", + "cookiecloud_password": "CookieCloud 密码", + "cookiecloud_server": "CookieCloud 服务器" + } + }, + "device_list": { + "title": "选择设备", + "description": "选择您想要跟踪的设备", + "data": { + "devices": "设备" + } + }, + "scan_interval": { + "title": "扫描间隔", + "description": "设置更新设备位置的频率（秒）", + "data": { + "scan_interval": "扫描间隔" + } + } + }, + "error": { + "invalid_cookie": "无效的 Cookie，请检查后重试", + "cookiecloud_error": "从 CookieCloud 获取 Cookie 失败，请检查您的 UUID 和密码", + "cannot_connect": "无法连接到 Vivo Cloud，请检查您的 Cookie 和网络连接", + "no_device_selected": "请至少选择一个设备" + }, + "abort": { + "already_configured": "Vivo Cloud 已经配置", + "cannot_connect": "无法连接到 Vivo Cloud" + } + }, + "options": { + "step": { + "init": { + "title": "Vivo Cloud 选项", + "description": "配置 Vivo Cloud 选项", + "data": { + "scan_interval": "扫描间隔（秒）" + } + } + } + }, + "services": { + "ring_device": { + "name": "响铃提醒", + "description": "让 Vivo 设备响铃以便查找", + "fields": { + "device_id": { + "name": "设备 ID", + "description": "要响铃的设备 ID" + } + } + } + } +} \ No newline at end of file diff --git a/hacs.json b/hacs.json new file mode 100644 index 0000000..051b891 --- /dev/null +++ b/hacs.json @@ -0,0 +1,7 @@ +{ + "name": "Vivo云服务", + "domains": ["device_tracker","sensor"], + "render_readme": true, + "homeassistant": "0.99.9", + "country": ["CN"] + }