Hass/vivo_cloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

first commit

Browse Source
This commit is contained in:
zimonianhua
2025-09-16 23:19:11 +08:00
commit a9ad4a285f
12 changed files with 1738 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+128
View File
@@ -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
custom_components/vivo_cloud/__init__.py
+115
View File
@@ -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)
custom_components/vivo_cloud/config_flow.py
+280
View File
@@ -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,
)
custom_components/vivo_cloud/const.py
+64
View File
@@ -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"
custom_components/vivo_cloud/coordinator.py
+468
View File
@@ -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
custom_components/vivo_cloud/device_tracker.py
+298
View File
@@ -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)
custom_components/vivo_cloud/manifest.json
+12
View File
@@ -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": []
}
custom_components/vivo_cloud/sensor.py
+211
View File
@@ -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}
custom_components/vivo_cloud/services.yaml
+15
View File
@@ -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:
custom_components/vivo_cloud/translations/en.json
+70
View File
@@ -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"
}
}
}
}
}
custom_components/vivo_cloud/translations/zh-Hans.json
+70
View File
@@ -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"
}
}
}
}
}
hacs.json
+7
View File
@@ -0,0 +1,7 @@
{
"name": "Vivo云服务",
"domains": ["device_tracker","sensor"],
"render_readme": true,
"homeassistant": "0.99.9",
"country": ["CN"]
}