Added Wyze

This commit is contained in:
2026-06-12 09:59:05 -04:00
parent c156180e03
commit 8843bf77a5
61 changed files with 8331 additions and 1 deletions
+215
View File
@@ -0,0 +1,215 @@
"""The Wyze Home Assistant Integration."""
from __future__ import annotations
import logging
from aiohttp.client_exceptions import ClientConnectorError
from homeassistant.config_entries import ConfigEntry, ConfigEntryNotReady, SOURCE_IMPORT
from homeassistant.const import CONF_USERNAME, CONF_PASSWORD
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.check_config import HomeAssistantConfig
from homeassistant.components import bluetooth
from wyzeapy import Wyzeapy
from wyzeapy.exceptions import AccessTokenError
from wyzeapy.wyze_auth_lib import Token
from .const import (
DOMAIN,
CONF_CLIENT,
ACCESS_TOKEN,
REFRESH_TOKEN,
REFRESH_TIME,
WYZE_NOTIFICATION_TOGGLE,
BULB_LOCAL_CONTROL,
DEFAULT_LOCAL_CONTROL,
KEY_ID,
API_KEY,
)
from .coordinator import WyzeLockBoltCoordinator
from .token_manager import TokenManager
PLATFORMS = [
"light",
"switch",
"lock",
"climate",
"alarm_control_panel",
"sensor",
"siren",
"cover",
"number",
"button",
"camera",
] # Fixme: Re add scene
_LOGGER = logging.getLogger(__name__)
# noinspection PyUnusedLocal
async def async_setup(
hass: HomeAssistant, config: HomeAssistantConfig, discovery_info=None
):
# pylint: disable=unused-argument
"""Set up the WyzeApi domain."""
if hass.config_entries.async_entries(DOMAIN):
_LOGGER.debug(
"Nothing to import from configuration.yaml, loading from Integrations",
)
return True
# noinspection SpellCheckingInspection
domainconfig = config.get(DOMAIN)
# pylint: disable=logging-not-lazy
_LOGGER.debug(
"Importing config information for %s from configuration.yml"
% domainconfig[CONF_USERNAME]
)
if hass.config_entries.async_entries(DOMAIN):
_LOGGER.debug("Found existing config entries")
for entry in hass.config_entries.async_entries(DOMAIN):
if entry:
entry_data = entry.as_dict().get("data")
hass.config_entries.async_update_entry(
entry,
data=entry_data,
)
break
else:
_LOGGER.debug("Creating new config entry")
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data={
CONF_USERNAME: domainconfig[CONF_USERNAME],
CONF_PASSWORD: domainconfig[CONF_PASSWORD],
ACCESS_TOKEN: domainconfig[ACCESS_TOKEN],
REFRESH_TOKEN: domainconfig[REFRESH_TOKEN],
REFRESH_TIME: domainconfig[REFRESH_TIME],
KEY_ID: domainconfig[KEY_ID],
API_KEY: domainconfig[API_KEY],
},
)
)
return True
# noinspection DuplicatedCode
async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
"""Set up Wyze Home Assistant Integration from a config entry."""
hass.data.setdefault(DOMAIN, {})
key_id = config_entry.data.get(KEY_ID)
api_key = config_entry.data.get(API_KEY)
client = await Wyzeapy.create()
token = None
if config_entry.data.get(ACCESS_TOKEN):
token = Token(
config_entry.data.get(ACCESS_TOKEN),
config_entry.data.get(REFRESH_TOKEN),
float(config_entry.data.get(REFRESH_TIME)),
)
a_tkn_manager = TokenManager(hass, config_entry)
client.register_for_token_callback(a_tkn_manager.token_callback)
# We should probably try/catch here to invalidate the login credentials and throw a notification if we cannot get
# a login with the token
try:
await client.login(
config_entry.data.get(CONF_USERNAME),
config_entry.data.get(CONF_PASSWORD),
key_id,
api_key,
token,
)
except ClientConnectorError as e:
raise ConfigEntryNotReady("Unable to login due to network issues.") from e
except AccessTokenError as e:
_LOGGER.error(
"Wyzeapi: Could not login. Please re-login through integration configuration"
)
_LOGGER.error(e)
raise ConfigEntryAuthFailed("Unable to login, please re-login.") from None
hass.data[DOMAIN][config_entry.entry_id] = {
CONF_CLIENT: client,
"key_id": KEY_ID,
"api_key": API_KEY,
"coordinators": {},
}
await setup_coordinators(hass, config_entry, client)
options_dict = {
BULB_LOCAL_CONTROL: config_entry.options.get(
BULB_LOCAL_CONTROL, DEFAULT_LOCAL_CONTROL
)
}
hass.config_entries.async_update_entry(config_entry, options=options_dict)
await hass.config_entries.async_forward_entry_setups(config_entry, PLATFORMS)
mac_addresses = await client.unique_device_ids
mac_addresses.add(WYZE_NOTIFICATION_TOGGLE)
hms_service = await client.hms_service
hms_id = hms_service.hms_id
if hms_id is not None:
mac_addresses.add(hms_id)
device_registry = dr.async_get(hass)
for device in dr.async_entries_for_config_entry(
device_registry, config_entry.entry_id
):
for identifier in device.identifiers:
# domain has to remain here. If it is removed the integration will remove all entities for not being in
# the mac address list each boot.
domain, mac = identifier
if mac not in mac_addresses:
_LOGGER.warning(
"%s is not in the mac_addresses list, removing the entry", mac
)
device_registry.async_remove_device(device.id)
return True
async def options_update_listener(hass: HomeAssistant, config_entry: ConfigEntry):
"""Handle options update."""
_LOGGER.debug("Updated options")
entry_data = config_entry.as_dict().get("data")
hass.config_entries.async_update_entry(
config_entry,
data=entry_data,
)
_LOGGER.debug("Reload entry: " + config_entry.entry_id)
await hass.config_entries.async_reload(config_entry.entry_id)
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
async def setup_coordinators(
hass: HomeAssistant, config_entry: ConfigEntry, client: Wyzeapy
):
"""Set up coordinators for Wyze devices that require Bluetooth."""
# Check if Bluetooth is active and functioning
if bluetooth.async_scanner_count(hass, connectable=True) == 0:
_LOGGER.info(
"Bluetooth is not active or no scanners available. Skipping WyzeLockBoltCoordinator setup."
)
return
lock_service = await client.lock_service
for lock in await lock_service.get_locks():
if lock.product_model == "YD_BT1":
coordinators = hass.data[DOMAIN][config_entry.entry_id].setdefault(
"coordinators", {}
)
coordinators[lock.mac] = WyzeLockBoltCoordinator(hass, lock_service, lock)
await coordinators[lock.mac].update_lock_info()
@@ -0,0 +1,166 @@
"""
This module handles the Wyze Home Monitoring system
"""
import logging
from datetime import timedelta
from typing import Optional, Callable, List, Any
from aiohttp.client_exceptions import ClientConnectionError
from homeassistant.components.alarm_control_panel import (
AlarmControlPanelState,
AlarmControlPanelEntity,
AlarmControlPanelEntityFeature,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_ATTRIBUTION
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from wyzeapy import Wyzeapy, HMSService
from wyzeapy.services.hms_service import HMSMode
from wyzeapy.exceptions import AccessTokenError, ParameterError, UnknownApiError
from .token_manager import token_exception_handler
from homeassistant.helpers.entity import DeviceInfo
from .const import DOMAIN, CONF_CLIENT
_LOGGER = logging.getLogger(__name__)
ATTRIBUTION = "Data provided by Wyze"
SCAN_INTERVAL = timedelta(seconds=15)
@token_exception_handler
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: Callable[[List[Any], bool], None],
):
"""
This function sets up the integration
:param hass: Reference to the HomeAssistant instance
:param config_entry: Reference to the config entry we are setting up
:param async_add_entities:
"""
_LOGGER.debug("""Creating new WyzeApi Home Monitoring System component""")
client: Wyzeapy = hass.data[DOMAIN][config_entry.entry_id][CONF_CLIENT]
hms_service = await client.hms_service
if await hms_service.has_hms:
async_add_entities([WyzeHomeMonitoring(hms_service)], True)
class WyzeHomeMonitoring(AlarmControlPanelEntity):
"""
A representation of the Wyze Home Monitoring system that works for wyze
"""
DEVICE_MODEL = "HMS"
MANUFACTURER = "WyzeLabs"
NAME = "Wyze Home Monitoring System"
AVAILABLE = True
_attr_has_entity_name = True
_attr_name = None
def __init__(self, hms_service: HMSService):
self._attr_unique_id = hms_service.hms_id
self._hms_service = hms_service
self._state = AlarmControlPanelState.DISARMED
self._server_out_of_sync = False
@property
def alarm_state(self) -> str:
return self._state
# NotImplemented Methods
def alarm_arm_vacation(self, code: Optional[str] = None) -> None:
raise NotImplementedError
def alarm_arm_night(self, code: Optional[str] = None) -> None:
raise NotImplementedError
def alarm_trigger(self, code: Optional[str] = None) -> None:
raise NotImplementedError
def alarm_arm_custom_bypass(self, code: Optional[str] = None) -> None:
raise NotImplementedError
# Implemented Methods
@token_exception_handler
async def async_alarm_disarm(self, code: Optional[str] = None) -> None:
"""Send disarm command."""
try:
await self._hms_service.set_mode(HMSMode.DISARMED)
except (AccessTokenError, ParameterError, UnknownApiError) as err:
raise HomeAssistantError(f"Wyze returned an error: {err.args}") from err
except ClientConnectionError as err:
raise HomeAssistantError(err) from err
else:
self._state = "disarmed"
self._server_out_of_sync = True
@token_exception_handler
async def async_alarm_arm_home(self, code: Optional[str] = None) -> None:
try:
await self._hms_service.set_mode(HMSMode.HOME)
except (AccessTokenError, ParameterError, UnknownApiError) as err:
raise HomeAssistantError(f"Wyze returned an error: {err.args}") from err
except ClientConnectionError as err:
raise HomeAssistantError(err) from err
else:
self._state = "armed_home"
self._server_out_of_sync = True
@token_exception_handler
async def async_alarm_arm_away(self, code: Optional[str] = None) -> None:
try:
await self._hms_service.set_mode(HMSMode.AWAY)
except (AccessTokenError, ParameterError, UnknownApiError) as err:
raise HomeAssistantError(f"Wyze returned an error: {err.args}") from err
except ClientConnectionError as err:
raise HomeAssistantError(err) from err
else:
self._state = "armed_away"
self._server_out_of_sync = True
@property
def supported_features(self) -> int:
return (
AlarmControlPanelEntityFeature.ARM_HOME
| AlarmControlPanelEntityFeature.ARM_AWAY
)
@property
def device_info(self) -> DeviceInfo:
return DeviceInfo(
identifiers={(DOMAIN, self.unique_id)},
name=self.NAME,
manufacturer=self.MANUFACTURER,
model=self.DEVICE_MODEL,
)
@property
def extra_state_attributes(self) -> dict | None:
"""Return device attributes of the entity."""
return {ATTR_ATTRIBUTION: ATTRIBUTION, "mac": self.unique_id}
@token_exception_handler
async def async_update(self) -> None:
"""Update the entity with data from the Wyze servers"""
if not self._server_out_of_sync:
state = await self._hms_service.update(self._hms_service.hms_id)
if state is HMSMode.DISARMED:
self._state = AlarmControlPanelState.DISARMED
elif state is HMSMode.AWAY:
self._state = AlarmControlPanelState.ARMED_AWAY
elif state is HMSMode.HOME:
self._state = AlarmControlPanelState.ARMED_HOME
elif state is HMSMode.CHANGING:
self._state = AlarmControlPanelState.DISARMED
else:
_LOGGER.warning(f"Received {state} from server")
self._server_out_of_sync = False
+224
View File
@@ -0,0 +1,224 @@
"""
This module describes the connection between Home Assistant and Wyze for the Sensors
"""
import logging
import time
from typing import Callable, List, Any
from homeassistant.components.binary_sensor import (
BinarySensorEntity,
BinarySensorDeviceClass,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_ATTRIBUTION
from homeassistant.core import HomeAssistant
from wyzeapy import Wyzeapy, CameraService, SensorService
from wyzeapy.services.camera_service import Camera
from wyzeapy.services.sensor_service import Sensor
from wyzeapy.types import DeviceTypes
from .token_manager import token_exception_handler
from .const import DOMAIN, CONF_CLIENT
_LOGGER = logging.getLogger(__name__)
ATTRIBUTION = "Data provided by Wyze"
@token_exception_handler
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: Callable[[List[Any], bool], None],
):
"""
This function sets up the config entry for use in Home Assistant
:param hass: Home Assistant instance
:param config_entry: The current config_entry
:param async_add_entities: This function adds entities to the config_entry
:return:
"""
_LOGGER.debug("""Creating new WyzeApi binary sensor component""")
client: Wyzeapy = hass.data[DOMAIN][config_entry.entry_id][CONF_CLIENT]
sensor_service = await client.sensor_service
camera_service = await client.camera_service
cameras = [
WyzeCameraMotion(camera_service, camera)
for camera in await camera_service.get_cameras()
]
sensors = [
WyzeSensor(sensor_service, sensor)
for sensor in await sensor_service.get_sensors()
]
async_add_entities(cameras, True)
async_add_entities(sensors, True)
class WyzeSensor(BinarySensorEntity):
"""
A representation of the WyzeSensor for use in Home Assistant
"""
def __init__(self, sensor_service: SensorService, sensor: Sensor):
"""Initializes the class"""
self._sensor_service = sensor_service
self._sensor = sensor
self._last_event = int(str(int(time.time())) + "000")
async def async_added_to_hass(self) -> None:
"""Registers for updates when the entity is added to Home Assistant"""
await self._sensor_service.register_for_updates(
self._sensor, self.process_update
)
async def async_will_remove_from_hass(self) -> None:
await self._sensor_service.deregister_for_updates(self._sensor)
def process_update(self, sensor: Sensor):
"""
This function processes an update for the Wyze Sensor
:param sensor: The sensor with the updated values
"""
self._sensor = sensor
self.schedule_update_ha_state()
@property
def device_info(self):
return {
"identifiers": {(DOMAIN, self._sensor.mac)},
"name": self.name,
"manufacturer": "WyzeLabs",
"model": self._sensor.product_model,
}
@property
def available(self) -> bool:
return True
@property
def name(self):
"""Return the display name of this switch."""
return self._sensor.nickname
@property
def should_poll(self) -> bool:
return False
@property
def is_on(self):
"""Return true if sensor detects motion"""
return self._sensor.detected
@property
def unique_id(self):
return "{}-motion".format(self._sensor.mac)
@property
def extra_state_attributes(self):
"""Return device attributes of the entity."""
return {
ATTR_ATTRIBUTION: ATTRIBUTION,
"device model": self._sensor.product_model,
"mac": self.unique_id,
}
@property
def device_class(self):
# pylint: disable=R1705
if self._sensor.type is DeviceTypes.MOTION_SENSOR:
return BinarySensorDeviceClass.MOTION
elif self._sensor.type is DeviceTypes.CONTACT_SENSOR:
return BinarySensorDeviceClass.DOOR
else:
raise RuntimeError(
f"The device type {self._sensor.type} is not supported by this class"
)
class WyzeCameraMotion(BinarySensorEntity):
"""
A representation of the Wyze Camera for use as a binary sensor in Home Assistant
"""
_is_on = False
_last_event = time.time() * 1000
def __init__(self, camera_service: CameraService, camera: Camera):
self._camera_service = camera_service
self._camera = camera
@property
def device_info(self):
return {
"identifiers": {(DOMAIN, self._camera.mac)},
"name": self.name,
"manufacturer": "WyzeLabs",
"model": self._camera.product_model,
}
@property
def available(self) -> bool:
return self._camera.available
@property
def name(self):
"""Return the display name of this switch."""
return self._camera.nickname
@property
def should_poll(self) -> bool:
return False
@property
def is_on(self):
"""Return true if the binary sensor is on"""
return self._is_on
@property
def unique_id(self):
return "{}-motion".format(self._camera.mac)
@property
def extra_state_attributes(self):
"""Return device attributes of the entity."""
return {
ATTR_ATTRIBUTION: ATTRIBUTION,
"device model": self._camera.product_model,
"mac": self.unique_id,
}
@property
def device_class(self):
return BinarySensorDeviceClass.MOTION
async def async_added_to_hass(self) -> None:
await self._camera_service.register_for_updates(
self._camera, self.process_update
)
async def async_will_remove_from_hass(self) -> None:
await self._camera_service.deregister_for_updates(self._camera)
@token_exception_handler
def process_update(self, camera: Camera) -> None:
"""
Is called by the update worker for events to update the values in this sensor
:param camera: An updated version of the current camera
"""
self._camera = camera
if camera.last_event_ts > self._last_event:
self._is_on = True
self._last_event = camera.last_event_ts
else:
self._is_on = False
self._last_event = camera.last_event_ts
self.schedule_update_ha_state()
+317
View File
@@ -0,0 +1,317 @@
"""Platform for button integration."""
from collections.abc import Callable
import logging
from typing import Any
from aiohttp.client_exceptions import ClientConnectionError
from wyzeapy import Wyzeapy
from wyzeapy.services.irrigation_service import Irrigation, IrrigationService, Zone
from wyzeapy.services.switch_service import Switch
from homeassistant.components.button import ButtonDeviceClass, ButtonEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_registry import EntityCategory
from .const import CONF_CLIENT, DOMAIN, RESET_BUTTON_PRESSED
from .token_manager import token_exception_handler
_LOGGER = logging.getLogger(__name__)
ATTRIBUTION = "Data provided by Wyze"
OUTDOOR_PLUGS = ["WLPPO"]
@token_exception_handler
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: Callable[[list[Any], bool], None],
) -> None:
"""This function sets up the config entry.
:param hass: The Home Assistant Instance
:param config_entry: The current config entry
:param async_add_entities: This function adds entities to the config entry
:return:
"""
_LOGGER.debug("""Creating new Wyze button component""")
client: Wyzeapy = hass.data[DOMAIN][config_entry.entry_id][CONF_CLIENT]
irrigation_service = await client.irrigation_service
switch_service = await client.switch_service
# Get all irrigation devices
irrigation_devices = await irrigation_service.get_irrigations()
# Create a button entity for each zone in each irrigation device
buttons = []
for device in irrigation_devices:
# Update the device to get its zones
device = await irrigation_service.update(device)
# Add a button entity for each enabled zone in the irrigation device
buttons.extend(
[
WyzeIrrigationZoneButton(irrigation_service, device, zone)
for zone in device.zones
if zone.enabled
]
)
# Add a stop all schedules button for each irrigation device, not each zone
buttons.append(WyzeIrrigationStopAllButton(irrigation_service, device))
plugs = await switch_service.get_switches()
buttons.extend(
[
WyzePowerSensorResetButton(plug)
for plug in plugs
if plug.product_model in OUTDOOR_PLUGS
]
)
async_add_entities(buttons, True)
class WyzeIrrigationZoneButton(ButtonEntity):
"""Representation of a Wyze Irrigation Zone Button."""
_attr_has_entity_name = True
def __init__(
self, irrigation_service: IrrigationService, irrigation: Irrigation, zone: Zone
) -> None:
"""Initialize the irrigation zone button."""
self._irrigation_service = irrigation_service
self._device = irrigation
self._zone = zone
@property
def name(self) -> str:
"""Return the name of the zone."""
return f"{self._zone.name}"
@property
def unique_id(self) -> str:
"""Return a unique ID for the zone."""
return f"Start {self._device.mac}-zone-{self._zone.zone_number}"
@property
def device_info(self) -> DeviceInfo:
"""Return device information about this entity."""
return DeviceInfo(
identifiers={(DOMAIN, self._device.mac)},
name=self._device.nickname,
manufacturer="WyzeLabs",
model=self._device.product_model,
connections={(dr.CONNECTION_NETWORK_MAC, self._device.mac)},
)
@property
def device_class(self) -> str:
"""Return the device class of the button."""
return ButtonDeviceClass.RESTART
@property
def icon(self) -> str:
"""Return the icon for the zone start button."""
return "mdi:sprinkler"
@property
def extra_state_attributes(self) -> dict[str, Any]:
"""Return entity specific state attributes."""
return {
"zone_number": self._zone.zone_number,
"zone_id": self._zone.zone_id,
"enabled": self._zone.enabled,
"quickrun_duration": self._zone.quickrun_duration,
}
async def async_press(self) -> None:
"""Start the zone with its quickrun duration.
This method is called when the button is pressed in Home Assistant.
It starts the irrigation zone for the duration specified in the corresponding number entity.
The number entity is found using the exact unique_id pattern that follows the format:
{device.mac}-zone-{zone.zone_number}-quickrun-duration
Raises:
HomeAssistantError: If the zone cannot be started or the number entity is invalid.
"""
try:
# The quickrun duration field doesnt exist in the Wyze API
# It has been created in Home Assistant as a number entity
# to conveniently trigger a zone start with a specific duration
# Get the device registry and find the device
device_registry = dr.async_get(self.hass)
device = device_registry.async_get_device(
identifiers={(DOMAIN, self._device.mac)}
)
if not device:
raise HomeAssistantError(f"Device not found for MAC {self._device.mac}")
# Get the entity registry
entity_registry = er.async_get(self.hass)
# Find the matching number entity using the zone number and device MAC
# The number entities have unique_id pattern: {device.mac}-zone-{zone.zone_number}-quickrun-duration
expected_unique_id = (
f"{self._device.mac}-zone-{self._zone.zone_number}-quickrun-duration"
)
matching_entity = None
for entity_id, entity in entity_registry.entities.items():
if (
entity.device_id == device.id
and entity.platform == DOMAIN
and entity_id.startswith("number.")
and entity.unique_id == expected_unique_id
):
matching_entity = entity_id
break
_LOGGER.debug(
"Looking for number entity with unique_id: %s", expected_unique_id
)
_LOGGER.debug("Found matching entity: %s", matching_entity)
if not matching_entity:
raise HomeAssistantError(
f"No number entity found for zone {self._zone.name} (zone {self._zone.zone_number}, device: {self._device.mac})"
)
# Get the current state of the number entity
state = self.hass.states.get(matching_entity)
if state is None or state.state in ["unavailable", "unknown"]:
raise HomeAssistantError(
f"Number entity {matching_entity} is unavailable or unknown"
)
# Convert duration from minutes to seconds
try:
duration_minutes = float(state.state)
if duration_minutes <= 0:
raise ValueError("Duration must be greater than 0")
duration_seconds = int(duration_minutes * 60)
except ValueError as err:
raise HomeAssistantError(
f"Invalid duration {state.state} for {matching_entity}"
) from err
_LOGGER.debug(
"Starting zone %s (zone %s) for %s minutes (%s seconds)",
self._zone.name,
self._zone.zone_number,
duration_minutes,
duration_seconds,
)
# Start the zone with the specified duration
await self._irrigation_service.start_zone(
self._device, self._zone.zone_number, duration_seconds
)
except HomeAssistantError as err:
_LOGGER.error("Failed to start zone %s: %s", self._zone.name, err)
raise
except Exception as err:
_LOGGER.error("Unexpected error starting zone %s: %s", self._zone.name, err)
raise HomeAssistantError(
f"Failed to start zone {self._zone.name}: {err}"
) from err
class WyzeIrrigationStopAllButton(ButtonEntity):
"""Representation of a Wyze Irrigation Stop All Schedules Button."""
_attr_name = "Stop All Zones"
def __init__(
self, irrigation_service: IrrigationService, irrigation: Irrigation
) -> None:
"""Initialize the irrigation stop all button."""
self._irrigation_service = irrigation_service
self._device = irrigation
@property
def unique_id(self) -> str:
"""Return a unique ID for the button."""
return f"Stop All {self._device.mac}"
@property
def device_info(self) -> DeviceInfo:
"""Return device information about this entity."""
return DeviceInfo(
identifiers={(DOMAIN, self._device.mac)},
name=self._device.nickname,
manufacturer="WyzeLabs",
model=self._device.product_model,
serial_number=self._device.sn,
connections={(dr.CONNECTION_NETWORK_MAC, self._device.mac)},
)
@property
def device_class(self) -> str:
"""Return the device class of the button."""
return ButtonDeviceClass.RESTART
@property
def icon(self) -> str:
"""Return the icon for the stop all button."""
return "mdi:octagon"
async def async_press(self) -> None:
"""Stop all running irrigation schedules.
This method is called when the button is pressed in Home Assistant.
It will stop all running irrigation schedules for the device.
Raises:
HomeAssistantError: If the schedules cannot be stopped.
"""
try:
await self._irrigation_service.stop_running_schedule(self._device)
except ClientConnectionError as err:
raise HomeAssistantError(f"Failed to stop schedules: {err}") from err
except Exception as err:
_LOGGER.error("Error stopping schedules: %s", err)
raise HomeAssistantError(f"Failed to stop schedules: {err}") from err
class WyzePowerSensorResetButton(ButtonEntity):
"""Wyze Power Sensor Reset Button."""
_attr_has_entity_name = True
_attr_should_poll = False
_attr_entity_category = EntityCategory.CONFIG
_attr_name = "Energy Usage Reset"
def __init__(self, switch: Switch) -> None:
"""Initialize a power sensor reset button."""
self._switch = switch
@property
def device_info(self) -> DeviceInfo:
"""Return device information about this entity."""
return DeviceInfo(
identifiers={(DOMAIN, self._switch.mac)},
name=self._switch.nickname,
)
@property
def unique_id(self) -> str:
"""Create a unique ID for the button."""
return f"{self._switch.mac} Reset button"
async def async_press(self) -> None:
"""Reset the sensor usage."""
async_dispatcher_send(
self.hass,
f"{RESET_BUTTON_PRESSED}-{self._switch.mac}",
self._switch,
)
+521
View File
@@ -0,0 +1,521 @@
"""Wyze Camera integration for Home Assistant."""
import base64
import json
import asyncio
from dataclasses import asdict
from collections.abc import Callable
from typing import Any
import logging
import uuid
import re
from homeassistant.config_entries import ConfigEntry
from homeassistant.components.camera import Camera as CameraEntity, CameraEntityFeature
from homeassistant.components.camera.webrtc import (
WebRTCClientConfiguration,
WebRTCSendMessage,
WebRTCAnswer,
WebRTCCandidate,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.util.ssl import get_default_context
from propcache.api import cached_property
from webrtc_models import RTCConfiguration, RTCIceCandidateInit, RTCIceServer
from websockets.asyncio.client import connect as websocket_connect
from wyzeapy import Wyzeapy, CameraService
from wyzeapy.services.camera_service import Camera
from .const import CAMERA_UPDATED, CONF_CLIENT, DOMAIN
from .token_manager import token_exception_handler
_LOGGER = logging.getLogger(__name__)
@token_exception_handler
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: Callable[[list[Any], bool], None],
) -> None:
"""This function sets up the config entry.
:param hass: The Home Assistant Instance
:param config_entry: The current config entry
:param async_add_entities: This function adds entities to the config entry
:return:
"""
_LOGGER.debug("Creating new Wyze camera component")
client: Wyzeapy = hass.data[DOMAIN][config_entry.entry_id][CONF_CLIENT]
camera_service = await client.camera_service
camera_devices = await camera_service.get_cameras()
# Create a camera entity for each camera device
cameras = []
for device in camera_devices:
# Update the device to get its zones
device = await camera_service.update(device)
cameras.extend([WyzeCamera(camera_service, device)])
for camera in cameras:
# Pre-seed the ICE server config by fetching it during setup, so the frontend can collect ICE servers before the offer
try:
await camera.config_fetch()
except Exception as e:
# Don't block startup if the config fetch fails, but log the error
_LOGGER.warning(
"Error fetching WebRTC session configuration for camera %s: %s",
camera.name,
e,
)
_LOGGER.debug("Wyze camera component setup complete")
async_add_entities(cameras, True)
class WyzeCamera(CameraEntity):
"""Representation of a Wyze Camera."""
def __init__(self, camera_service: CameraService, camera: Camera):
"""Initialize the camera."""
super().__init__()
self._camera_service = camera_service
self._camera = camera
self.name = camera.nickname
self._attr_unique_id = camera.mac
self.brand = "Wyze"
self.model = camera.product_model
self.supported_features = CameraEntityFeature.STREAM
self._webrtc_provider = None
self.sessions: dict[str, WyzeCameraWebRTCSession] = {}
self._pending_candidates: dict[str, list[RTCIceCandidateInit]] = {}
# Always holds an in-flight Task[dict] for the next config fetch.
# _async_get_webrtc_client_configuration reads the result when ready;
# async_handle_async_webrtc_offer awaits it to guarantee a fresh config.
self._cached_config: dict | None = None
self._config_task: asyncio.Task | None = None
async def config_fetch(self) -> None:
"""Fetch the WebRTC session configuration for this camera and cache it for future use."""
self._cached_config = await self._camera_service.get_stream_info(self._camera)
_LOGGER.debug(
"Initial fetch of WebRTC session configuration complete for camera %s",
self.name,
)
@cached_property
def device_info(self) -> DeviceInfo | None:
"""Return the device info."""
return {
"identifiers": {(DOMAIN, self._camera.mac)},
"name": self._camera.nickname,
"manufacturer": "WyzeLabs",
"model": self._camera.product_model,
}
@property
def available(self) -> bool:
"""Return if the camera is available."""
return self._camera.available
@cached_property
def is_streaming(self) -> bool:
"""Return True if the camera is currently streaming."""
return self._camera.on
@callback
def handle_camera_update(self, camera: Camera) -> None:
"""Update the camera whenever there is an update."""
self._camera = camera
self.async_write_ha_state()
async def async_added_to_hass(self) -> None:
"""Listen for camera updates."""
self.async_on_remove(
async_dispatcher_connect(
self.hass,
f"{CAMERA_UPDATED}-{self._camera.mac}",
self.handle_camera_update,
)
)
@property
def is_on(self) -> bool:
"""Return True if the camera is currently on."""
return self._camera.on
async def async_turn_on(self) -> None:
"""Turn the camera on."""
await self._camera_service.turn_on(self._camera)
async def async_turn_off(self) -> None:
"""Turn the camera off."""
await self._camera_service.turn_off(self._camera)
async def async_disable_motion_detection(self) -> None:
"""Disable motion detection."""
await self._camera_service.turn_off_motion_detection(self._camera)
async def async_enable_motion_detection(self) -> None:
"""Enable motion detection."""
await self._camera_service.turn_on_motion_detection(self._camera)
@property
def motion_detection_enabled(self) -> bool | None:
"""Return True if motion detection is enabled, False if disabled, or None if unknown/not supported."""
motion = getattr(self._camera, "motion", None)
if isinstance(motion, bool):
return motion
# Some Wyze camera models / API responses don't expose motion state.
# Return None so HA omits/marks the attribute as unknown instead of crashing.
return None
async def async_camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return bytes of camera image.
Currently not implemented"""
return None
def _async_get_webrtc_client_configuration(self) -> WebRTCClientConfiguration:
"""Return the WebRTC client configuration for this camera, including ICE servers."""
# This shouldn't happen, but throw an error if we don't have a config ready yet
if self._cached_config is None:
raise HomeAssistantError("WebRTC session configuration not available yet")
config = self._cached_config
ice_servers = []
for server in config.get("ice_servers", []):
_LOGGER.debug("Adding ICE server for camera %s: %s", self.name, server)
ice_servers.append(
RTCIceServer.from_dict(
{
"urls": server["url"],
"username": server["username"],
"credential": server["credential"],
}
)
)
_LOGGER.debug("ICE servers for camera %s: %s", self.name, ice_servers)
configuration = RTCConfiguration(ice_servers=ice_servers)
return WebRTCClientConfiguration(
configuration=configuration, data_channel="data"
)
async def async_handle_async_webrtc_offer(
self, offer_sdp: str, session_id: str, send_message: WebRTCSendMessage
) -> None:
"""Handle an incoming WebRTC offer from the frontend."""
_LOGGER.debug(
"Handling WebRTC offer for camera %s with session ID %s",
self.name,
session_id,
)
# Always fetch a truly fresh config so the signaling URL and ICE servers
# are never stale — KVS signed URLs are single-use and short-lived.
config = await self._camera_service.get_stream_info(self._camera)
# Update cached config with the new ICE servers
self._cached_config = config
_LOGGER.debug("Fresh config for offer on camera %s: %s", self.name, config)
self.sessions[session_id] = WyzeCameraWebRTCSession(
session_id, self, send_message, config
)
await self.sessions[session_id].send_offer(offer_sdp)
pending = self._pending_candidates.pop(session_id, None)
if pending:
_LOGGER.debug(
"Flushing %d buffered ICE candidates for camera %s session %s",
len(pending),
self.name,
session_id,
)
for cand in pending:
await self.sessions[session_id].send_candidate(cand)
async def async_on_webrtc_candidate(
self, session_id: str, candidate: RTCIceCandidateInit
) -> None:
"""Handle an incoming ICE candidate for a WebRTC session."""
if session_id not in self.sessions:
self._pending_candidates.setdefault(session_id, []).append(candidate)
_LOGGER.debug(
"Buffered ICE candidate for camera %s session %s (session not ready yet)",
self.name,
session_id,
)
return
await self.sessions[session_id].send_candidate(candidate)
def close_webrtc_session(self, session_id: str) -> None:
"""Close a WebRTC session and clean up resources."""
_LOGGER.debug("Closing WebRTC session %s", session_id)
self._pending_candidates.pop(session_id, None)
if session_id in self.sessions:
session = self.sessions[session_id]
session.close_connection()
del self.sessions[session_id]
class WyzeCameraWebRTCSession:
"""Represents a WebRTC session for a Wyze camera."""
def __init__(
self,
session_id: str,
camera: WyzeCamera,
callback: WebRTCSendMessage,
config: dict,
):
self.session_id = session_id
self.camera = camera
self.websocket = None # This will hold the WebSocket connection
self.camera_service = None
self.callback = callback
self.close = None
self.lock = asyncio.Lock()
self.task = None
self.config = config
self.sdp_offer = None
self.sdp_answer = None
# Set once connect() succeeds; send_candidate waits on this instead of reconnecting
self._connected = asyncio.Event()
async def connect(self):
"""Establish the WebSocket connection to the KVS signaling URL.
This is called lazily from send_offer() to ensure we have the latest config
and don't connect too early before the offer is ready."""
# The signaling_url from get_stream_info() is often *double*-percent-encoded
# (e.g. "%253A" instead of "%3A"). We must NOT fully URL-decode it because
# that can change SigV4 canonical encoding and make KVS reject the handshake.
# Instead, only "undouble" percent-escapes by converting "%25xx" -> "%xx",
# leaving "%3A", "%2F", etc. intact.
signaling_url = self.config["signaling_url"]
for _ in range(3):
if "%25" not in signaling_url:
break
signaling_url = signaling_url.replace("%25", "%")
self.websocket = await websocket_connect(
signaling_url, ssl=get_default_context(), logger=_LOGGER
)
_LOGGER.debug(
"WebSocket connection established for camera %s with session ID %s",
self.camera.name,
self.session_id,
)
self._connected.set()
asyncio.create_task(self.run_loop())
async def send_offer(self, offer_sdp: str):
"""Send an SDP offer to the Kinesis Video Streams signaling channel."""
async with self.lock:
if self.websocket is None:
_LOGGER.debug("Connecting to websocket from send_offer")
await self.connect()
if self.websocket is None:
raise ConnectionError("WebSocket connection not established")
# Create an offer for Kinesis
self.sdp_offer = offer_sdp
offer = {"type": "offer", "sdp": offer_sdp}
payload = {
"action": "SDP_OFFER",
"recipientClientId": "ada06f08-87f4-4e13-b699-e82db8517ae5",
"messagePayload": base64.b64encode(
json.dumps(offer, separators=(",", ":")).encode()
).decode(),
"correlationId": str(uuid.uuid4()),
}
str_payload = json.dumps(payload)
_LOGGER.debug(
"Sending SDP offer for camera %s with session ID %s, %s",
self.camera.name,
self.session_id,
str_payload,
)
await self.websocket.send(str_payload)
async def send_candidate(self, candidate: RTCIceCandidateInit):
"""Send an ICE candidate to the Kinesis Video Streams signaling channel."""
# Take RTCIceCandidateInit, convert it to the format in the messagePayload above, and send it to the client using the callback
# Wait for send_offer to establish the connection — never reconnect (KVS URLs are single-use)
try:
await asyncio.wait_for(self._connected.wait(), timeout=10.0)
except asyncio.TimeoutError as exc:
raise ConnectionError(
"WebSocket connection not established within timeout"
) from exc
if self.websocket is None:
raise ConnectionError("WebSocket connection not established")
candidate_dict = asdict(candidate)
candidate_payload = {
"candidate": candidate_dict["candidate"],
"sdpMid": candidate_dict["sdp_mid"],
"sdpMLineIndex": candidate_dict["sdp_m_line_index"],
"usernameFragment": candidate_dict["user_fragment"],
}
match = re.search(r"ufrag (\w{4})", candidate_payload["candidate"])
if match is not None:
candidate_payload["usernameFragment"] = match.group(1)
payload = {
"action": "ICE_CANDIDATE",
"recipientClientId": "ada06f08-87f4-4e13-b699-e82db8517ae5",
"messagePayload": base64.b64encode(
json.dumps(candidate_payload, separators=(",", ":")).encode()
).decode(),
}
str_payload = json.dumps(payload)
_LOGGER.debug(
"Sending ICE candidate for camera %s with session ID %s: %s",
self.camera.name,
self.session_id,
str_payload,
)
await self.websocket.send(str_payload)
def close_connection(self):
"""Close the WebSocket connection to the Kinesis Video Streams signaling channel."""
if self.close is not None:
self.close()
def force_correct_sdp_answer(self) -> None:
"""Force the sdp response to have the valid answer.
The Kinesis WebRTC Stream responses to certain offers do not
follow the spec defined in https://www.ietf.org/rfc/rfc3264.txt
An offer of recvonly must be answered with sendonly or inactive.
"""
_LOGGER.debug("Attempt to fix sdp answer...")
if isinstance(self.sdp_answer, str) and isinstance(self.sdp_offer, str):
sdp_kinds = ["audio", "video", "application"]
sdp_directions = ["sendrecv", "sendonly", "recvonly", "inactive"]
sdp_pattern = (
"m=(?P<kind>{})(.|\n)+?a=(?P<direction>{})(\r|\n|\r\n)".format(
"|".join(sdp_kinds), "|".join(sdp_directions)
)
)
sdp_direction_offers = re.finditer(sdp_pattern, self.sdp_offer)
for offer in sdp_direction_offers:
sdp_answers = re.finditer(sdp_pattern, self.sdp_answer)
for answer in sdp_answers:
if (
offer.group("kind") == answer.group("kind")
and offer.group("direction") == "recvonly"
and answer.group("direction") == "sendrecv"
):
correct_answer = re.sub(
"a=sendrecv", "a=sendonly", answer.group(0)
)
_LOGGER.debug("Replacing answer with: %s", str(correct_answer))
self.sdp_answer = self.sdp_answer.replace(
answer.group(0), correct_answer
)
async def run_loop(self):
"""Listen for messages from the Kinesis Video Streams signaling channel and handle them appropriately."""
if self.websocket is None:
raise ConnectionError("WebSocket connection not established")
loop = asyncio.get_running_loop()
def close():
if self.websocket is not None:
return loop.create_task(self.websocket.close())
return None
self.close = close
_LOGGER.debug(
"run_loop starting for camera %s session %s",
self.camera.name,
self.session_id,
)
try:
async for message in self.websocket:
if len(message) == 0:
_LOGGER.debug(
"Received empty message (type=%s) for camera %s session %s",
type(message).__name__,
self.camera.name,
self.session_id,
)
continue
_LOGGER.debug(
"Received message for camera %s with session ID %s: %s",
self.camera.name,
self.session_id,
message,
)
try:
data = json.loads(message)
except json.JSONDecodeError as e:
_LOGGER.error(
"Failed to decode JSON message for camera %s with session ID %s: %s",
self.camera.name,
self.session_id,
e,
)
continue
match data.get("messageType"):
case "ICE_CANDIDATE":
# Decode messagePayload (base64 JSON) → RTCIceCandidateInit → WebRTCCandidate
# KVS uses camelCase keys; map them to RTCIceCandidateInit's snake_case fields
candidate_str = base64.b64decode(
data["messagePayload"]
).decode()
candidate_data = json.loads(candidate_str)
rtccandidate = RTCIceCandidateInit(
candidate=candidate_data["candidate"],
sdp_mid=candidate_data.get("sdpMid"),
sdp_m_line_index=candidate_data.get("sdpMLineIndex"),
user_fragment=candidate_data.get("usernameFragment"),
)
self.callback(WebRTCCandidate(candidate=rtccandidate))
case "SDP_ANSWER":
# Decode messagePayload (base64 JSON with "type"/"sdp" keys) → extract sdp string
answer_str = base64.b64decode(data["messagePayload"]).decode()
try:
answer_obj = json.loads(answer_str)
sdp = answer_obj.get("sdp", answer_str)
except json.JSONDecodeError:
sdp = answer_str
self.sdp_answer = sdp
self.force_correct_sdp_answer()
self.callback(WebRTCAnswer(answer=self.sdp_answer))
case "STATUS_RESPONSE" | "GO_AWAY" | "RECONNECT_ICE_SERVER":
_LOGGER.debug(
"KVS control message '%s' for session %s: %s",
data.get("messageType"),
self.session_id,
data,
)
case other:
_LOGGER.debug(
"Unhandled KVS message type '%s' for session %s: %s",
other,
self.session_id,
data,
)
except Exception as e:
_LOGGER.error(
"run_loop error for camera %s session %s: %s",
self.camera.name,
self.session_id,
e,
exc_info=True,
)
_LOGGER.debug(
"run_loop exited for camera %s session %s",
self.camera.name,
self.session_id,
)
+391
View File
@@ -0,0 +1,391 @@
"""Platform for light integration."""
import logging
# Import the device class from the component that you want to support
from datetime import timedelta
from typing import List, Optional, Callable, Any
from aiohttp.client_exceptions import ClientConnectionError
from homeassistant.components.climate import (
ClimateEntity,
ClimateEntityFeature,
HVACAction,
HVACMode,
)
from homeassistant.components.climate.const import (
FAN_AUTO,
FAN_ON,
PRESET_HOME,
PRESET_AWAY,
PRESET_SLEEP,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import UnitOfTemperature
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import device_registry as dr
from wyzeapy import Wyzeapy, ThermostatService
from wyzeapy.exceptions import AccessTokenError, ParameterError, UnknownApiError
from wyzeapy.services.thermostat_service import (
Thermostat,
TemperatureUnit,
Preset,
FanMode,
HVACState,
HVACMode as WyzeHVACMode,
)
from .token_manager import token_exception_handler
from .const import DOMAIN, CONF_CLIENT
_LOGGER = logging.getLogger(__name__)
ATTRIBUTION = "Data provided by Wyze"
SCAN_INTERVAL = timedelta(seconds=30)
@token_exception_handler
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: Callable[[List[Any], bool], None],
):
"""
This function sets up the config entry so that it is available to Home Assistant
:param hass: The Home Assistant instance
:param config_entry: The current config entry
:param async_add_entities: A function to add entities
:return:
"""
_LOGGER.debug("""Creating new WyzeApi thermostat component""")
client: Wyzeapy = hass.data[DOMAIN][config_entry.entry_id][CONF_CLIENT]
thermostat_service = await client.thermostat_service
thermostats = [
WyzeThermostat(thermostat_service, thermostat)
for thermostat in await thermostat_service.get_thermostats()
]
async_add_entities(thermostats, True)
class WyzeThermostat(ClimateEntity):
"""
This class defines a representation of a Wyze Thermostat that can be used for Home Assistant
"""
# pylint: disable=R0902
_server_out_of_sync = False
def __init__(self, thermostat_service: ThermostatService, thermostat: Thermostat):
self._thermostat_service = thermostat_service
self._thermostat = thermostat
def set_temperature(self, **kwargs) -> None:
raise NotImplementedError
def set_humidity(self, humidity: int) -> None:
raise NotImplementedError
def set_fan_mode(self, fan_mode: str) -> None:
raise NotImplementedError
def set_hvac_mode(self, hvac_mode: str) -> None:
raise NotImplementedError
def set_swing_mode(self, swing_mode: str) -> None:
raise NotImplementedError
def set_preset_mode(self, preset_mode: str) -> None:
raise NotImplementedError
def turn_aux_heat_on(self) -> None:
raise NotImplementedError
def turn_aux_heat_off(self) -> None:
raise NotImplementedError
@property
def current_temperature(self) -> float:
return self._thermostat.temperature
@property
def current_humidity(self) -> Optional[int]:
return self._thermostat.humidity
@property
def temperature_unit(self) -> str:
# if self._thermostat.temp_unit == TemperatureUnit.FAHRENHEIT:
return UnitOfTemperature.FAHRENHEIT
# return TEMP_CELSIUS
@property
def unit_of_measurement(self) -> str:
if self._thermostat.temp_unit == TemperatureUnit.FAHRENHEIT:
return UnitOfTemperature.FAHRENHEIT
return UnitOfTemperature.CELSIUS
@property
def hvac_mode(self) -> str:
# pylint: disable=R1705
if self._thermostat.hvac_mode == WyzeHVACMode.AUTO:
return HVACMode.AUTO
elif self._thermostat.hvac_mode == WyzeHVACMode.HEAT:
return HVACMode.HEAT
elif self._thermostat.hvac_mode == WyzeHVACMode.COOL:
return HVACMode.COOL
else:
return HVACMode.OFF
@property
def hvac_modes(self) -> List[str]:
return [HVACMode.AUTO, HVACMode.HEAT, HVACMode.COOL, HVACMode.OFF]
@property
def target_temperature_high(self) -> Optional[float]:
return self._thermostat.cool_set_point
@property
def target_temperature_low(self) -> Optional[float]:
return self._thermostat.heat_set_point
@property
def preset_mode(self) -> Optional[str]:
match self._thermostat.preset:
case Preset.HOME:
return PRESET_HOME
case Preset.AWAY:
return PRESET_AWAY
case Preset.SLEEP:
return PRESET_SLEEP
case _:
raise NotImplementedError
@property
def preset_modes(self) -> Optional[List[str]]:
return [PRESET_HOME, PRESET_AWAY, PRESET_SLEEP]
@property
def is_aux_heat(self) -> Optional[bool]:
raise NotImplementedError
@property
def fan_mode(self) -> Optional[str]:
if self._thermostat.fan_mode == FanMode.AUTO:
return FAN_AUTO
else:
return FAN_ON
@property
def fan_modes(self) -> Optional[List[str]]:
return [FAN_AUTO, FAN_ON]
@property
def swing_mode(self) -> Optional[str]:
raise NotImplementedError
@property
def swing_modes(self) -> Optional[str]:
raise NotImplementedError
@property
def hvac_action(self) -> str:
# pylint: disable=R1705
if self._thermostat.hvac_state == HVACState.IDLE:
return HVACAction.IDLE
elif self._thermostat.hvac_state == HVACState.HEATING:
return HVACAction.HEATING
elif self._thermostat.hvac_state == HVACState.COOLING:
return HVACAction.COOLING
else:
return HVACAction.OFF
@token_exception_handler
async def async_set_temperature(self, **kwargs) -> None:
target_temp_low = kwargs["target_temp_low"]
target_temp_high = kwargs["target_temp_high"]
try:
# Always send setpoints unconditionally rather than guarding against
# the cached heat_set_point/cool_set_point value. The equality check
# causes a silent no-op whenever the cache is stale (e.g. after a mode
# change resets the physical device to firmware defaults, or after a
# Wyze cloud/app-driven reset), leaving the physical thermostat at the
# wrong setpoint while HA and the Wyze cloud both report the correct one.
# The _server_out_of_sync flag compounds this by delaying the cache
# refresh by up to 60 s after any command. The underlying API call
# (set_iot_prop_by_topic) is idempotent, so always sending is safe.
# See: https://github.com/SecKatie/ha-wyzeapi/issues/813
await self._thermostat_service.set_heat_point(
self._thermostat, int(target_temp_low)
)
self._thermostat.heat_set_point = int(target_temp_low)
await self._thermostat_service.set_cool_point(
self._thermostat, int(target_temp_high)
)
self._thermostat.cool_set_point = int(target_temp_high)
except (AccessTokenError, ParameterError, UnknownApiError) as err:
raise HomeAssistantError(f"Wyze returned an error: {err.args}") from err
except ClientConnectionError as err:
raise HomeAssistantError(err) from err
else:
self._server_out_of_sync = True
self.async_schedule_update_ha_state()
async def async_set_humidity(self, humidity: int) -> None:
raise NotImplementedError
@token_exception_handler
async def async_set_fan_mode(self, fan_mode: str) -> None:
try:
if fan_mode == FAN_ON:
await self._thermostat_service.set_fan_mode(
self._thermostat, FanMode.ON
)
self._thermostat.fan_mode = FanMode.ON
elif fan_mode == FAN_AUTO:
await self._thermostat_service.set_fan_mode(
self._thermostat, FanMode.AUTO
)
self._thermostat.fan_mode = FanMode.AUTO
except (AccessTokenError, ParameterError, UnknownApiError) as err:
raise HomeAssistantError(f"Wyze returned an error: {err.args}") from err
except ClientConnectionError as err:
raise HomeAssistantError(err) from err
else:
self._server_out_of_sync = True
self.async_schedule_update_ha_state()
@token_exception_handler
async def async_set_hvac_mode(self, hvac_mode: str) -> None:
try:
if hvac_mode == HVACMode.OFF:
await self._thermostat_service.set_hvac_mode(
self._thermostat, WyzeHVACMode.OFF
)
self._thermostat.hvac_mode = HVACMode.OFF
elif hvac_mode == HVACMode.HEAT:
await self._thermostat_service.set_hvac_mode(
self._thermostat, WyzeHVACMode.HEAT
)
self._thermostat.hvac_mode = HVACMode.HEAT
elif hvac_mode == HVACMode.COOL:
await self._thermostat_service.set_hvac_mode(
self._thermostat, WyzeHVACMode.COOL
)
self._thermostat.hvac_mode = HVACMode.COOL
elif hvac_mode == HVACMode.AUTO:
await self._thermostat_service.set_hvac_mode(
self._thermostat, WyzeHVACMode.AUTO
)
self._thermostat.hvac_mode = HVACMode.AUTO
except (AccessTokenError, ParameterError, UnknownApiError) as err:
raise HomeAssistantError(f"Wyze returned an error: {err.args}") from err
except ClientConnectionError as err:
raise HomeAssistantError(err) from err
else:
self._server_out_of_sync = True
self.async_schedule_update_ha_state()
async def async_set_swing_mode(self, swing_mode: str) -> None:
raise NotImplementedError
@token_exception_handler
async def async_set_preset_mode(self, preset_mode: str) -> None:
try:
if preset_mode == PRESET_SLEEP:
await self._thermostat_service.set_preset(
self._thermostat, Preset.SLEEP
)
self._thermostat.preset = Preset.SLEEP
elif preset_mode == PRESET_AWAY:
await self._thermostat_service.set_preset(self._thermostat, Preset.AWAY)
self._thermostat.preset = Preset.AWAY
elif preset_mode == PRESET_HOME:
await self._thermostat_service.set_preset(self._thermostat, Preset.HOME)
self._thermostat.preset = Preset.HOME
except (AccessTokenError, ParameterError, UnknownApiError) as err:
raise HomeAssistantError(f"Wyze returned an error: {err.args}") from err
except ClientConnectionError as err:
raise HomeAssistantError(err) from err
else:
self._server_out_of_sync = True
self.async_schedule_update_ha_state()
async def async_turn_aux_heat_on(self) -> None:
raise NotImplementedError
async def async_turn_aux_heat_off(self) -> None:
raise NotImplementedError
@property
def supported_features(self) -> int:
return (
ClimateEntityFeature.TARGET_TEMPERATURE_RANGE
| ClimateEntityFeature.FAN_MODE
| ClimateEntityFeature.PRESET_MODE
)
@property
def device_info(self) -> dict:
return {
"identifiers": {(DOMAIN, self._thermostat.mac)},
"name": self._thermostat.nickname,
"connections": {
(
dr.CONNECTION_NETWORK_MAC,
self._thermostat.mac,
)
},
"manufacturer": "WyzeLabs",
"model": self._thermostat.product_model,
}
@property
def should_poll(self) -> bool:
return False
@property
def name(self) -> str:
"""Return the display name of this lock."""
return self._thermostat.nickname
@property
def unique_id(self) -> str:
return self._thermostat.mac
@property
def available(self) -> bool:
"""Return the connection status of this light"""
return self._thermostat.available
@token_exception_handler
async def async_update(self) -> None:
"""
This function updates the state of the Thermostat
:return: None
"""
if not self._server_out_of_sync:
self._thermostat = await self._thermostat_service.update(self._thermostat)
else:
self._server_out_of_sync = False
@callback
def async_update_callback(self, thermostat: Thermostat):
"""Update the thermostat's state."""
self._thermostat = thermostat
self.async_schedule_update_ha_state()
async def async_added_to_hass(self) -> None:
"""Subscribe to update events."""
self._thermostat.callback_function = self.async_update_callback
self._thermostat_service.register_updater(self._thermostat, 30)
await self._thermostat_service.start_update_manager()
return await super().async_added_to_hass()
async def async_will_remove_from_hass(self) -> None:
self._thermostat_service.unregister_updater(self._thermostat)
+182
View File
@@ -0,0 +1,182 @@
"""Config flow for Wyze Home Assistant Integration integration."""
from __future__ import annotations
import logging
from typing import Any, Optional
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_USERNAME, CONF_PASSWORD, CONF_ACCESS_TOKEN
from homeassistant.core import callback
from homeassistant.exceptions import HomeAssistantError
from wyzeapy import Wyzeapy, exceptions
from .const import (
DOMAIN,
ACCESS_TOKEN,
REFRESH_TOKEN,
REFRESH_TIME,
BULB_LOCAL_CONTROL,
DEFAULT_LOCAL_CONTROL,
KEY_ID,
API_KEY,
)
_LOGGER = logging.getLogger(__name__)
STEP_USER_DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_USERNAME): str,
vol.Required(CONF_PASSWORD): str,
vol.Required(KEY_ID): str,
vol.Required(API_KEY): str,
}
)
STEP_2FA_DATA_SCHEMA = vol.Schema({CONF_ACCESS_TOKEN: str})
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Wyze Home Assistant Integration."""
VERSION = 1
CONNECTION_CLASS = config_entries.CONN_CLASS_CLOUD_POLL
client: Wyzeapy = None
user_params = {}
def __init__(self):
"""Initialize."""
self.email = None
self.password = None
self.key_id = None
self.api_key = None
async def get_client(self):
if not self.client:
self.client = await Wyzeapy.create()
async def async_step_user(
self, user_input: Optional[dict[str, any]] = None
) -> dict[str, Any]:
"""Handle the initial step."""
await self.get_client()
if user_input is None:
return self.async_show_form(
step_id="user", data_schema=STEP_USER_DATA_SCHEMA
)
errors = {}
# noinspection PyBroadException
try:
await self.client.login(
user_input[CONF_USERNAME],
user_input[CONF_PASSWORD],
user_input[KEY_ID],
user_input[API_KEY],
)
except CannotConnect:
errors["base"] = "cannot_connect"
except exceptions.AccessTokenError:
errors["base"] = "invalid_auth"
except exceptions.TwoFactorAuthenticationEnabled:
self.user_params[CONF_USERNAME] = user_input[CONF_USERNAME]
self.user_params[CONF_PASSWORD] = user_input[CONF_PASSWORD]
self.user_params[KEY_ID] = user_input[KEY_ID]
self.user_params[API_KEY] = user_input[API_KEY]
return await self.async_step_2fa()
else:
if self.hass.config_entries.async_entries(DOMAIN):
for entry in self.hass.config_entries.async_entries(DOMAIN):
self.hass.config_entries.async_update_entry(entry, data=user_input)
await self.hass.config_entries.async_reload(entry.entry_id)
return self.async_abort(reason="reauth_successful")
else:
return self.async_create_entry(title="", data=user_input)
return self.async_show_form(
step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
)
async def async_step_2fa(
self, user_input: Optional[dict[str, Any]] = None
) -> dict[str, Any]:
if user_input is None:
return self.async_show_form(step_id="2fa", data_schema=STEP_2FA_DATA_SCHEMA)
errors = {}
try:
token = await self.client.login_with_2fa(
user_input[CONF_ACCESS_TOKEN],
)
except exceptions.LoginError:
errors["base"] = "invalid_auth"
else:
self.user_params[ACCESS_TOKEN] = token.access_token
self.user_params[REFRESH_TOKEN] = token.refresh_token
self.user_params[REFRESH_TIME] = token.refresh_time
if self.hass.config_entries.async_entries(DOMAIN):
for entry in self.hass.config_entries.async_entries(DOMAIN):
self.hass.config_entries.async_update_entry(
entry, data=self.user_params
)
await self.hass.config_entries.async_reload(entry.entry_id)
return self.async_abort(reason="reauth_successful")
else:
return self.async_create_entry(title="", data=self.user_params)
return self.async_show_form(
step_id="2fa", data_schema=STEP_2FA_DATA_SCHEMA, errors=errors
)
async def async_step_import(self, import_config):
"""Import a config entry from configuration.yaml."""
return await self.async_step_user(import_config)
async def async_step_reauth(self, user_input=None):
"""Perform reauth upon an API authentication error."""
if user_input is None:
return self.async_show_form(
step_id="reauth_confirm",
data_schema=vol.Schema({}),
)
return await self.async_step_user()
@staticmethod
@callback
def async_get_options_flow(
config_entry: config_entries.ConfigEntry,
) -> OptionsFlowHandler:
"""Create the Wyze options flow."""
return OptionsFlowHandler()
class OptionsFlowHandler(config_entries.OptionsFlow):
"""Handle an option flow for Wyze."""
async def async_step_init(self, user_input=None):
"""Handle options flow."""
if user_input is not None:
return self.async_create_entry(title="", data=user_input)
data_schema = vol.Schema(
{
vol.Optional(
BULB_LOCAL_CONTROL,
default=self.config_entry.options.get(
BULB_LOCAL_CONTROL, DEFAULT_LOCAL_CONTROL
),
): bool
}
)
return self.async_show_form(step_id="init", data_schema=data_schema)
class CannotConnect(HomeAssistantError):
"""Error to indicate we cannot connect."""
class InvalidAuth(HomeAssistantError):
"""Error to indicate there is invalid auth."""
+28
View File
@@ -0,0 +1,28 @@
"""Constants for the Wyze Home Assistant Integration integration."""
DOMAIN = "wyzeapi"
CONF_CLIENT = "wyzeapi_client"
ACCESS_TOKEN = "access_token"
REFRESH_TOKEN = "refresh_token"
REFRESH_TIME = "refresh_time"
KEY_ID = "key_id"
API_KEY = "api_key"
WYZE_NOTIFICATION_TOGGLE = f"{DOMAIN}.wyze.notification.toggle"
LOCK_UPDATED = f"{DOMAIN}.lock_updated"
CAMERA_UPDATED = f"{DOMAIN}.camera_updated"
LIGHT_UPDATED = f"{DOMAIN}.light_updated"
COVER_UPDATED = f"{DOMAIN}.cover_updated"
RESET_BUTTON_PRESSED = f"{DOMAIN}.reset_button_pressed"
# EVENT NAMES
WYZE_CAMERA_EVENT = "wyze_camera_event"
BULB_LOCAL_CONTROL = "bulb_local_control"
DEFAULT_LOCAL_CONTROL = True
# Yunding (YD) is the provider for Wyze Lock Bolt
YDBLE_LOCK_STATE_UUID = "00002220-0000-6b63-6f6c-2e6b636f6f6c"
YDBLE_UART_RX_UUID = "6e400003-b5a3-f393-e0a9-e50e24dcca9e"
YDBLE_UART_TX_UUID = "6e400002-b5a3-f393-e0a9-e50e24dcca9e"
+196
View File
@@ -0,0 +1,196 @@
import asyncio
import binascii
import logging
from datetime import datetime, timedelta
from typing import Dict
from bleak import BleakClient
from bleak_retry_connector import establish_connection
from homeassistant.components import bluetooth
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import PlatformNotReady
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from wyzeapy.services.lock_service import LockService, Lock
from .const import YDBLE_LOCK_STATE_UUID, YDBLE_UART_RX_UUID, YDBLE_UART_TX_UUID
from .token_manager import token_exception_handler
from .ydble_utils import (
decrypt_ecb,
pack_l1,
pack_l2_dict,
pack_l2_lock_unlock,
parse_l1,
parse_l2_dict,
)
_LOGGER = logging.getLogger(__name__)
class WyzeLockBoltCoordinator(DataUpdateCoordinator):
"""Manages fetching data from BLE periodically."""
def __init__(
self, hass: HomeAssistant, lock_service: LockService, lock: Lock
) -> None:
"""Initialize the coordinator."""
super().__init__(
hass,
_LOGGER,
name="Wyze Lock State Updater",
update_interval=timedelta(seconds=300),
)
self._lock_service = lock_service
self._lock = lock
# The `mac` in the original response should be UUID.
# The actual MAC address should be retrieved from another API.
self._uuid = lock.mac
self._mac = None
self._bleak_client = None
self._current_command = None
@token_exception_handler
async def update_lock_info(self):
self._lock = await self._lock_service.update(self._lock)
mac = self._lock.raw_dict["hardware_info"]["mac"]
# The mac is stored reverse ordered and no colon, e.g. mac="ab8967452301"
self._mac = ":".join(mac[i - 2 : i] for i in range(12, 0, -2)).upper()
async def _async_update_data(self):
"""Fetch the latest data from BLE device."""
# Skip if running a command
if self._current_command:
return self.data
client = await self._get_ble_client()
if client is None:
raise UpdateFailed(
f"Could not find BLE device {self._lock.nickname} with address {self._mac}. Device may not be in range."
)
try:
value = await client.read_gatt_char(YDBLE_LOCK_STATE_UUID)
return self._parse_state(value)
finally:
await self._disconnect()
async def lock_unlock(self, command="lock"):
if self._current_command:
self.async_update_listeners()
raise Exception(f"Waiting for {self._current_command} command to complete")
self._current_command = command
self.async_update_listeners()
client = await self._get_ble_client()
if client is None:
raise Exception(
f"Could not find BLE device {self._lock.nickname} with address {self._mac}. Device may not be in range."
)
# disconnect in 10 seconds in case of error
asyncio.create_task(self._disconnect(delay=10))
context = {"command": command, "stage": 0}
async def _handle_uart_rx_context(sender, data):
await self._handle_uart_rx(sender, data, client, context)
await client.start_notify(YDBLE_UART_RX_UUID, _handle_uart_rx_context)
await client.start_notify(YDBLE_LOCK_STATE_UUID, self._handle_state)
await self._request_challenge(client)
async def _request_challenge(self, client: BleakClient):
l2_content = pack_l2_dict(0x91, 0, {10: b"\x27"})
req = pack_l1(0, 1, l2_content)
await client.write_gatt_char(YDBLE_UART_TX_UUID, req, response=False)
async def _send_lock_unlock(self, client: BleakClient, challenge, command):
l2_content = pack_l2_lock_unlock(
self._lock.ble_id, self._lock.ble_token, challenge, command
)
req = pack_l1(0, 2, l2_content)
await client.write_gatt_char(YDBLE_UART_TX_UUID, req, response=False)
async def _send_ack(self, client: BleakClient, seq_no: int):
req = pack_l1(0x08, seq_no, b"")
await client.write_gatt_char(YDBLE_UART_TX_UUID, req, response=False)
async def _handle_state(self, sender, data: bytearray):
self.data = self._parse_state(data)
self._current_command = None
self.async_update_listeners()
def _parse_state(self, state_data):
data = decrypt_ecb(self._uuid[-16:].lower(), state_data)
result = {
"state": data[0],
"timestamp": datetime.fromtimestamp(int.from_bytes(data[1:5])),
}
return result
async def _handle_uart_rx(
self, sender, data: bytearray, client: BleakClient, context: Dict
):
# Process for unfinished data
if "l1_unfinished" in context:
data = context["l1_unfinished"] + data
del context["l1_unfinished"]
l2_data, l1_flags, seq_no, remain = parse_l1(data)
if remain:
context["l1_unfinished"] = data
return
# Process messages
if context["stage"] == 0:
# Ack for request chanllenge
if seq_no == 1 and l1_flags == 0x48:
context["stage"] = 1
return
if context["stage"] == 1:
if l1_flags == 0x40:
# Process L2 dict
cmd, l2_flags, l2_dict = parse_l2_dict(l2_data)
if cmd == 0x86 and 0xD2 in l2_dict:
# Got generated chanllenge
challenge = l2_dict[0xD2]
await self._send_ack(client, seq_no=seq_no)
await self._send_lock_unlock(client, challenge, context["command"])
context["stage"] = 2
return
if context["stage"] == 2:
# Ack for send_lock_unlock
if seq_no == 2 and l1_flags == 0x48:
context["stage"] = 3
return
if context["stage"] == 3:
if l1_flags == 0x40:
cmd, l2_flags, l2_dict = parse_l2_dict(l2_data)
if cmd == 0x04:
await self._send_ack(client, seq_no=seq_no)
return
_LOGGER.warning(
f"Unexpected message: stage={context['stage']}"
f" flags={l1_flags:01x}, seq_no={seq_no:02x},"
f" l2_data={binascii.hexlify(l2_data)}"
)
async def _get_ble_client(self) -> BleakClient | None:
if not self._bleak_client or not self._bleak_client.is_connected:
if not self._mac:
raise PlatformNotReady("Not initialized")
ble_device = bluetooth.async_ble_device_from_address(
self.hass, self._mac, connectable=True
)
if ble_device is None:
return None
self._bleak_client = await establish_connection(
BleakClient, ble_device, ble_device.address
)
return self._bleak_client
async def _disconnect(self, delay=0):
await asyncio.sleep(delay)
if self._bleak_client and self._bleak_client.is_connected:
await self._bleak_client.disconnect()
self._current_command = None
self.async_update_listeners()
+152
View File
@@ -0,0 +1,152 @@
"""Platform for the cover integration."""
from abc import ABC
import logging
from typing import Any, Callable, List
from wyzeapy import Wyzeapy, CameraService
from wyzeapy.services.camera_service import Camera
from wyzeapy.exceptions import AccessTokenError, ParameterError, UnknownApiError
from wyzeapy.types import DeviceTypes
import homeassistant.components.cover
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_ATTRIBUTION
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers import device_registry as dr
from homeassistant.exceptions import HomeAssistantError
from homeassistant.components.cover import CoverDeviceClass, CoverEntityFeature
from .const import CAMERA_UPDATED, CONF_CLIENT, DOMAIN
from .token_manager import token_exception_handler
_LOGGER = logging.getLogger(__name__)
ATTRIBUTION = "Data provided by Wyze"
@token_exception_handler
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: Callable[[List[Any], bool], None],
) -> None:
"""
This function sets up the config_entry
:param hass: Home Assistant instance
:param config_entry: The current config_entry
:param async_add_entities: This function adds entities to the config_entry
:return:
"""
_LOGGER.debug("""Creating new WyzeApi cover component""")
client: Wyzeapy = hass.data[DOMAIN][config_entry.entry_id][CONF_CLIENT]
camera_service = await client.camera_service
cameras: List[Camera] = await camera_service.get_cameras()
garages = []
for camera in cameras:
if camera.device_params["dongle_product_model"] == "HL_CGDC":
garages.append(WyzeGarageDoor(camera_service, camera))
async_add_entities(garages, True)
class WyzeGarageDoor(homeassistant.components.cover.CoverEntity, ABC):
"""Representation of a Wyze Garage Door."""
_attr_device_class = CoverDeviceClass.GARAGE
_attr_supported_features = CoverEntityFeature.OPEN | CoverEntityFeature.CLOSE
_attr_has_entity_name = True
def __init__(self, camera_service: CameraService, camera: Camera):
"""Initialize a Wyze garage door."""
self._camera = camera
if self._camera.type not in [DeviceTypes.CAMERA]:
raise HomeAssistantError(f"Invalid device type: {self._camera.type}")
self._camera_service = camera_service
self._available = self._camera.available
@property
def device_info(self):
"""Return device information about this entity."""
return {
"identifiers": {(DOMAIN, self._camera.mac)},
"name": f"{self._camera.nickname}",
"connections": {(dr.CONNECTION_NETWORK_MAC, self._camera.mac)},
}
@property
def extra_state_attributes(self):
"""Return device attributes of the entity."""
return {
ATTR_ATTRIBUTION: ATTRIBUTION,
"device model": f"{self._camera.product_model}.{self._camera.device_params['dongle_product_model']}",
}
@property
def should_poll(self) -> bool:
return False
@token_exception_handler
async def async_open_cover(self, **kwargs):
"""Open the cover."""
try:
await self._camera_service.garage_door_open(self._camera)
except (AccessTokenError, ParameterError, UnknownApiError) as err:
raise HomeAssistantError(f"Wyze returned an error: {err.args}") from err
except Exception as err:
raise HomeAssistantError(err) from err
else:
self._camera.garage = True
self.async_write_ha_state()
@token_exception_handler
async def async_close_cover(self, **kwargs):
"""Close the cover."""
try:
await self._camera_service.garage_door_close(self._camera)
except (AccessTokenError, ParameterError, UnknownApiError) as err:
raise HomeAssistantError(f"Wyze returned an error: {err.args}") from err
except Exception as err:
raise HomeAssistantError(err) from err
else:
self._camera.garage = False
self.async_write_ha_state()
@property
def is_closed(self):
"""Return if the cover is closed."""
return not self._camera.garage
@property
def available(self):
"""Return the connection status of this cover."""
return self._camera.available
@property
def unique_id(self):
"""Define a unique id for this entity."""
return f"{self._camera.mac}_GarageDoor"
@property
def name(self):
"""Return the name of the garage door."""
return "Garage Door"
async def async_added_to_hass(self) -> None:
self.async_on_remove(
async_dispatcher_connect(
self.hass,
f"{CAMERA_UPDATED}-{self._camera.mac}",
self.handle_camera_update,
)
)
@callback
def handle_camera_update(self, camera: Camera) -> None:
"""Update the cover whenever there is an update"""
self._camera = camera
self.async_write_ha_state()
+493
View File
@@ -0,0 +1,493 @@
"""Platform for light integration."""
from collections.abc import Callable
from datetime import timedelta
import logging
from typing import Any
from aiohttp.client_exceptions import ClientConnectionError
from wyzeapy import BulbService, CameraService, Wyzeapy
from wyzeapy.exceptions import AccessTokenError, ParameterError, UnknownApiError
from wyzeapy.services.bulb_service import Bulb
from wyzeapy.services.camera_service import Camera
from wyzeapy.types import DeviceTypes, PropertyIDs
from wyzeapy.utils import create_pid_pair
from homeassistant.components.light import (
ATTR_BRIGHTNESS,
ATTR_COLOR_TEMP_KELVIN,
ATTR_EFFECT,
ATTR_HS_COLOR,
ColorMode,
LightEntity,
LightEntityFeature,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.dispatcher import (
async_dispatcher_connect,
async_dispatcher_send,
)
import homeassistant.util.color as color_util
from .const import (
BULB_LOCAL_CONTROL,
CAMERA_UPDATED,
CONF_CLIENT,
DOMAIN,
LIGHT_UPDATED,
)
from .token_manager import token_exception_handler
_LOGGER = logging.getLogger(__name__)
ATTRIBUTION = "Data provided by Wyze"
SCAN_INTERVAL = timedelta(seconds=30)
EFFECT_MODE = "effects mode"
EFFECT_SUN_MATCH = "sun match"
EFFECT_SHADOW = "shadow"
EFFECT_LEAP = "leap"
EFFECT_FLICKER = "flicker"
@token_exception_handler
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: Callable[[list[Any], bool], None],
) -> None:
"""Set up the entities in the config entry."""
_LOGGER.debug("""Creating new WyzeApi light component""")
client: Wyzeapy = hass.data[DOMAIN][config_entry.entry_id][CONF_CLIENT]
camera_service = await client.camera_service
bulb_service = await client.bulb_service
lights = [
WyzeLight(bulb_service, light, config_entry)
for light in await bulb_service.get_bulbs()
]
for camera in await camera_service.get_cameras():
if camera.product_model == "HL_BC":
# Wyze Bulb Cam has integrated light
lights.append(WyzeCamerafloodlight(camera, camera_service, "bulbcam"))
elif (
camera.product_model == "WYZE_CAKP2JFUS"
and camera.device_params["dongle_product_model"] == "HL_CFL"
or camera.product_model in ("LD_CFP", "HL_CFL2") # Floodlight v2
):
lights.append(WyzeCamerafloodlight(camera, camera_service, "floodlight"))
elif (
camera.product_model in ("WYZE_CAKP2JFUS", "HL_CAM4")
) and camera.device_params[
"dongle_product_model"
] == "HL_CAM3SS": # Cam v3 with lamp socket accessory
lights.append(WyzeCamerafloodlight(camera, camera_service, "lampsocket"))
elif (
camera.product_model == "AN_RSCW"
): # Battery cam pro (integrated spotlight)
lights.append(WyzeCamerafloodlight(camera, camera_service, "spotlight"))
async_add_entities(lights, True)
class WyzeLight(LightEntity):
"""Representation of a Wyze Bulb."""
_just_updated = False
_attr_should_poll = False
def __init__(self, bulb_service: BulbService, bulb: Bulb, config_entry) -> None:
"""Initialize a Wyze Bulb."""
self._bulb = bulb
self._device_type = DeviceTypes(self._bulb.product_type)
self._config_entry = config_entry
self._local_control = config_entry.options.get(BULB_LOCAL_CONTROL)
if self._device_type not in [
DeviceTypes.LIGHT,
DeviceTypes.MESH_LIGHT,
DeviceTypes.LIGHTSTRIP,
]:
raise AttributeError("Device type not supported")
self._bulb_service = bulb_service
self._attr_min_color_temp_kelvin = (
1800
if self._device_type in [DeviceTypes.MESH_LIGHT, DeviceTypes.LIGHTSTRIP]
else 2700
)
self._attr_max_color_temp_kelvin = 6500
self._attr_name = self._bulb.nickname
self._attr_unique_id = self._bulb.mac
self._attr_supported_features = LightEntityFeature.EFFECT
@property
def device_info(self):
"""Return the device info."""
return {
"identifiers": {(DOMAIN, self._bulb.mac)},
"name": self._bulb.nickname,
"connections": {
(
dr.CONNECTION_NETWORK_MAC,
self._bulb.mac,
)
},
"manufacturer": "WyzeLabs",
"model": self._bulb.product_model,
}
@token_exception_handler
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on the light."""
options = []
self._local_control = self._config_entry.options.get(BULB_LOCAL_CONTROL)
if kwargs.get(ATTR_BRIGHTNESS) is not None:
brightness = round((kwargs.get(ATTR_BRIGHTNESS) / 255) * 100)
options.append(create_pid_pair(PropertyIDs.BRIGHTNESS, str(brightness)))
_LOGGER.debug("Setting brightness to %s", brightness)
_LOGGER.debug("Options: %s", options)
self._bulb.brightness = brightness
if (
self._bulb.sun_match
): # Turn off sun match if we're changing anything other than brightness
if any([kwargs.get(ATTR_COLOR_TEMP_KELVIN, kwargs.get(ATTR_HS_COLOR))]):
options.append(create_pid_pair(PropertyIDs.SUN_MATCH, str(0)))
self._bulb.sun_match = False
_LOGGER.debug("Turning off sun match")
if kwargs.get(ATTR_COLOR_TEMP_KELVIN) is not None:
_LOGGER.debug("Setting color temp")
color_temp = kwargs.get(ATTR_COLOR_TEMP_KELVIN)
options.append(create_pid_pair(PropertyIDs.COLOR_TEMP, str(color_temp)))
if self._device_type in [DeviceTypes.MESH_LIGHT, DeviceTypes.LIGHTSTRIP]:
options.append(
create_pid_pair(PropertyIDs.COLOR_MODE, str(2))
) # Put bulb in White Mode
self._bulb.color_mode = "2"
self._bulb.color_temp = color_temp
self._bulb.color = color_util.color_rgb_to_hex(
*color_util.color_temperature_to_rgb(color_temp)
)
if kwargs.get(ATTR_HS_COLOR) is not None and (
self._device_type is DeviceTypes.MESH_LIGHT
or self._device_type is DeviceTypes.LIGHTSTRIP
):
_LOGGER.debug("Setting color")
color = color_util.color_rgb_to_hex(
*color_util.color_hs_to_RGB(*kwargs.get(ATTR_HS_COLOR))
)
options.extend(
[
create_pid_pair(PropertyIDs.COLOR, str(color)),
create_pid_pair(
PropertyIDs.COLOR_MODE, str(1)
), # Put bulb in Color Mode
]
)
self._bulb.color = color
self._bulb.color_mode = "1"
if kwargs.get(ATTR_EFFECT) is not None:
if kwargs.get(ATTR_EFFECT) == EFFECT_SUN_MATCH:
_LOGGER.debug("Setting Sun Match")
options.append(create_pid_pair(PropertyIDs.SUN_MATCH, str(1)))
self._bulb.sun_match = True
else:
if (
self._bulb.type is DeviceTypes.MESH_LIGHT
): # Handle mesh light effects
self._local_control = False
options.append(create_pid_pair(PropertyIDs.COLOR_MODE, str(3)))
self._bulb.color_mode = "3"
if kwargs.get(ATTR_EFFECT) == EFFECT_SHADOW:
_LOGGER.debug("Setting Shadow Effect")
options.append(
create_pid_pair(PropertyIDs.LIGHTSTRIP_EFFECTS, str(1))
)
self._bulb.effects = "1"
elif kwargs.get(ATTR_EFFECT) == EFFECT_LEAP:
_LOGGER.debug("Setting Leap Effect")
options.append(
create_pid_pair(PropertyIDs.LIGHTSTRIP_EFFECTS, str(2))
)
self._bulb.effects = "2"
elif kwargs.get(ATTR_EFFECT) == EFFECT_FLICKER:
_LOGGER.debug("Setting Flicker Effect")
options.append(
create_pid_pair(PropertyIDs.LIGHTSTRIP_EFFECTS, str(3))
)
self._bulb.effects = "3"
_LOGGER.debug("Turning on light")
try:
await self._bulb_service.turn_on(self._bulb, self._local_control, options)
except (AccessTokenError, ParameterError, UnknownApiError) as err:
raise HomeAssistantError(f"Wyze returned an error: {err.args}") from err
except ClientConnectionError as err:
raise HomeAssistantError(err) from err
else:
self._bulb.on = True
self._just_updated = True
self.async_schedule_update_ha_state()
@token_exception_handler
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off the light."""
self._local_control = self._config_entry.options.get(BULB_LOCAL_CONTROL)
try:
await self._bulb_service.turn_off(self._bulb, self._local_control)
except (AccessTokenError, ParameterError, UnknownApiError) as err:
raise HomeAssistantError(f"Wyze returned an error: {err.args}") from err
except ClientConnectionError as err:
raise HomeAssistantError(err) from err
else:
self._bulb.on = False
self._just_updated = True
self.async_schedule_update_ha_state()
@property
def supported_color_modes(self):
"""Return the supported color modes."""
if self._bulb.type in [DeviceTypes.MESH_LIGHT, DeviceTypes.LIGHTSTRIP]:
return {ColorMode.COLOR_TEMP, ColorMode.HS}
return {ColorMode.COLOR_TEMP}
@property
def color_mode(self):
"""Return the current color mode."""
if self._bulb.type is DeviceTypes.LIGHT:
return ColorMode.COLOR_TEMP
return ColorMode.COLOR_TEMP if self._bulb.color_mode == "2" else ColorMode.HS
@property
def available(self):
"""Return the connection status of this light."""
return self._bulb.available
@property
def hs_color(self):
"""Return the HS color."""
return color_util.color_RGB_to_hs(
*color_util.rgb_hex_to_rgb_list(self._bulb.color)
)
@property
def extra_state_attributes(self):
"""Return device attributes of the entity."""
dev_info = {}
# noinspection DuplicatedCode
if self._bulb.device_params.get("ip"):
dev_info["IP"] = str(self._bulb.device_params.get("ip"))
if self._bulb.device_params.get("rssi"):
dev_info["RSSI"] = str(self._bulb.device_params.get("rssi"))
if self._bulb.device_params.get("ssid"):
dev_info["SSID"] = str(self._bulb.device_params.get("ssid"))
dev_info["Sun Match"] = self._bulb.sun_match
dev_info["local_control"] = (
self._local_control and not self._bulb.cloud_fallback
)
if self._device_type is DeviceTypes.LIGHTSTRIP and self._bulb.color_mode == "3":
if self._bulb.effects == "1":
dev_info["effect_mode"] = "Shadow"
elif self._bulb.effects == "2":
dev_info["effect_mode"] = "Leap"
elif self._bulb.effects == "3":
dev_info["effect_mode"] = "Flicker"
if (
self._device_type is DeviceTypes.MESH_LIGHT
or self._device_type is DeviceTypes.LIGHTSTRIP
):
if self._bulb.color_mode == "1":
dev_info["mode"] = "Color"
elif self._bulb.color_mode == "2":
dev_info["mode"] = "White"
elif self._bulb.color_mode == "3":
dev_info["mode"] = "Effect"
return dev_info
@property
def brightness(self):
"""Return the brightness of the light."""
return round(self._bulb.brightness * 2.55, 1)
@property
def color_temp_kelvin(self):
"""Return the color temp in Kelvin."""
return self._bulb.color_temp
@property
def effect_list(self):
"""Return the list of effects."""
if self._device_type is DeviceTypes.LIGHTSTRIP:
return [EFFECT_SHADOW, EFFECT_LEAP, EFFECT_FLICKER, EFFECT_SUN_MATCH]
return [EFFECT_SUN_MATCH]
@property
def is_on(self):
"""Return true if light is on."""
return self._bulb.on
@token_exception_handler
async def async_update(self):
"""Update the lock to be up to date with the Wyze Servers."""
if not self._just_updated:
self._bulb = await self._bulb_service.update(self._bulb)
else:
self._just_updated = False
@callback
def async_update_callback(self, bulb: Bulb):
"""Update the bulb's state."""
self._bulb = bulb
self._local_control = self._config_entry.options.get(BULB_LOCAL_CONTROL)
async_dispatcher_send(self.hass, f"{LIGHT_UPDATED}-{self._bulb.mac}", bulb)
self.async_schedule_update_ha_state()
async def async_added_to_hass(self) -> None:
"""Subscribe to update events."""
self._bulb.callback_function = self.async_update_callback
self._bulb_service.register_updater(self._bulb, 30)
await self._bulb_service.start_update_manager()
return await super().async_added_to_hass()
async def async_will_remove_from_hass(self) -> None:
"""Unregister the updater."""
self._bulb_service.unregister_updater(self._bulb)
class WyzeCamerafloodlight(LightEntity):
"""Representation of a Wyze Camera floodlight."""
_available: bool
_just_updated = False
_attr_should_poll = False
def __init__(
self, camera: Camera, camera_service: CameraService, light_type: str
) -> None:
"""Initialize the camera floodlight."""
self._device = camera
self._service = camera_service
self._light_type = light_type
self._attr_unique_id = f"{self._device.mac}-{self._light_type}"
self._is_on = self._device.floodlight
@token_exception_handler
async def async_turn_on(self, **kwargs) -> None:
"""Turn the floodlight on."""
try:
await self._service.floodlight_on(self._device)
except (AccessTokenError, ParameterError, UnknownApiError) as err:
raise HomeAssistantError(f"Wyze returned an error: {err.args}") from err
except ClientConnectionError as err:
raise HomeAssistantError(err) from err
else:
self._is_on = True
self._just_updated = True
self.async_schedule_update_ha_state()
@token_exception_handler
async def async_turn_off(self, **kwargs):
"""Turn the floodlight off."""
try:
await self._service.floodlight_off(self._device)
except (AccessTokenError, ParameterError, UnknownApiError) as err:
raise HomeAssistantError(f"Wyze returned an error: {err.args}") from err
except ClientConnectionError as err:
raise HomeAssistantError(err) from err
else:
self._is_on = False
self._just_updated = True
self.async_schedule_update_ha_state()
@property
def is_on(self):
"""Return true if floodlight is on."""
return self._device.floodlight
@property
def name(self) -> str:
"""Return the device name."""
light_type_names = {
"lampsocket": "Lamp Socket",
"floodlight": "Floodlight",
"spotlight": "Spotlight",
"bulbcam": "Light",
}
return (
f"{self._device.nickname} {light_type_names.get(self._light_type, 'Light')}"
)
@property
def device_info(self):
"""Return the device info."""
return {
"identifiers": {(DOMAIN, self._device.mac)},
"name": self._device.nickname,
"connections": {
(
dr.CONNECTION_NETWORK_MAC,
self._device.mac,
)
},
"manufacturer": "WyzeLabs",
"model": self._device.product_model,
}
@callback
def handle_camera_update(self, camera: Camera) -> None:
"""Update the camera object whenever there is an update."""
self._device = camera
self.async_write_ha_state()
async def async_added_to_hass(self) -> None:
"""Add listener on startup."""
self.async_on_remove(
async_dispatcher_connect(
self.hass,
f"{CAMERA_UPDATED}-{self._device.mac}",
self.handle_camera_update,
)
)
@property
def icon(self):
"""Return the icon to use in the frontend."""
icons = {
"lampsocket": "mdi:lightbulb",
"floodlight": "mdi:track-light",
"spotlight": "mdi:spotlight",
"bulbcam": "mdi:lightbulb",
}
return icons.get(self._light_type, "mdi:lightbulb")
@property
def color_mode(self):
"""Return the color mode."""
return ColorMode.ONOFF
@property
def supported_color_modes(self):
"""Return the supported color mode."""
return ColorMode.ONOFF
+270
View File
@@ -0,0 +1,270 @@
#!/usr/bin/python3
"""Platform for light integration."""
from abc import ABC
from datetime import timedelta
import logging
from typing import Any, Callable, List
from aiohttp.client_exceptions import ClientConnectionError
from wyzeapy import LockService, Wyzeapy
from wyzeapy.services.lock_service import Lock
from wyzeapy.types import DeviceTypes
from wyzeapy.exceptions import AccessTokenError, ParameterError, UnknownApiError
import homeassistant.components.lock
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_ATTRIBUTION
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from homeassistant.helpers import device_registry as dr
from homeassistant.exceptions import HomeAssistantError
from .const import CONF_CLIENT, DOMAIN, LOCK_UPDATED
from .token_manager import token_exception_handler
_LOGGER = logging.getLogger(__name__)
ATTRIBUTION = "Data provided by Wyze"
SCAN_INTERVAL = timedelta(seconds=10)
MAX_OUT_OF_SYNC_COUNT = 5
@token_exception_handler
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: Callable[[List[Any], bool], None],
) -> None:
"""
This function sets up the config_entry
:param hass: Home Assistant instance
:param config_entry: The current config_entry
:param async_add_entities: This function adds entities to the config_entry
:return:
"""
_LOGGER.debug("""Creating new WyzeApi lock component""")
client: Wyzeapy = hass.data[DOMAIN][config_entry.entry_id][CONF_CLIENT]
lock_service = await client.lock_service
all_locks = await lock_service.get_locks()
locks = [
WyzeLock(lock_service, lock)
for lock in all_locks
if lock.product_model != "YD_BT1"
]
lock_bolts = []
coordinators = hass.data[DOMAIN][config_entry.entry_id].get("coordinators", {})
for lock in all_locks:
if lock.product_model == "YD_BT1":
coordinator = coordinators.get(lock.mac)
if coordinator is None:
_LOGGER.warning(
"No coordinator found for Lock Bolt %s (%s), skipping",
lock.nickname,
lock.mac,
)
continue
lock_bolts.append(WyzeLockBolt(coordinator))
async_add_entities(locks + lock_bolts, True)
class WyzeLock(homeassistant.components.lock.LockEntity, ABC):
"""Representation of a Wyze Lock."""
def __init__(self, lock_service: LockService, lock: Lock):
"""Initialize a Wyze lock."""
self._lock = lock
if self._lock.type not in [DeviceTypes.LOCK]:
raise AttributeError("Device type not supported")
self._lock_service = lock_service
self._out_of_sync_count = 0
@property
def device_info(self):
return {
"identifiers": {(DOMAIN, self._lock.mac)},
"name": self._lock.nickname,
"connections": {
(
dr.CONNECTION_NETWORK_MAC,
self._lock.mac,
)
},
"manufacturer": "WyzeLabs",
"model": self._lock.product_model,
}
def lock(self, **kwargs):
raise NotImplementedError
def unlock(self, **kwargs):
raise NotImplementedError
@property
def should_poll(self) -> bool:
return False
@token_exception_handler
async def async_lock(self, **kwargs):
_LOGGER.debug("Turning on lock")
try:
await self._lock_service.lock(self._lock)
except (AccessTokenError, ParameterError, UnknownApiError) as err:
raise HomeAssistantError(f"Wyze returned an error: {err.args}") from err
except ClientConnectionError as err:
raise HomeAssistantError(err) from err
else:
self._lock.unlocked = False
self.async_schedule_update_ha_state()
@token_exception_handler
async def async_unlock(self, **kwargs):
try:
await self._lock_service.unlock(self._lock)
except (AccessTokenError, ParameterError, UnknownApiError) as err:
raise HomeAssistantError(f"Wyze returned an error: {err.args}") from err
except ClientConnectionError as err:
raise HomeAssistantError(err) from err
else:
self._lock.unlocked = True
self.async_schedule_update_ha_state()
@property
def is_locked(self):
return not self._lock.unlocked
@property
def name(self):
"""Return the display name of this lock."""
return self._lock.nickname
@property
def unique_id(self):
return self._lock.mac
@property
def available(self):
"""Return the connection status of this lock"""
return self._lock.available
@property
def extra_state_attributes(self):
"""Return device attributes of the entity."""
dev_info = {
ATTR_ATTRIBUTION: ATTRIBUTION,
"door_open": self._lock.door_open,
}
# Add the lock battery value if it exists
if self._lock.raw_dict.get("power"):
dev_info["lock_battery"] = str(self._lock.raw_dict.get("power"))
# Add the keypad's battery value if it exists
if self._lock.raw_dict.get("keypad", {}).get("power"):
dev_info["keypad_battery"] = str(
self._lock.raw_dict.get("keypad", {}).get("power")
)
return dev_info
@property
def supported_features(self):
return None
@token_exception_handler
async def async_update(self):
"""
This function updates the entity
"""
lock = await self._lock_service.update(self._lock)
if (
lock.unlocked == self._lock.unlocked
or self._out_of_sync_count >= MAX_OUT_OF_SYNC_COUNT
):
self._lock = lock
self._out_of_sync_count = 0
else:
self._out_of_sync_count += 1
@callback
def async_update_callback(self, lock: Lock):
"""Update the switch's state."""
self._lock = lock
async_dispatcher_send(
self.hass,
f"{LOCK_UPDATED}-{self._lock.mac}",
lock,
)
self.async_schedule_update_ha_state()
async def async_added_to_hass(self) -> None:
"""Subscribe to update events."""
self._lock.callback_function = self.async_update_callback
self._lock_service.register_updater(self._lock, 10)
await self._lock_service.start_update_manager()
return await super().async_added_to_hass()
async def async_will_remove_from_hass(self) -> None:
self._lock_service.unregister_updater(self._lock)
class WyzeLockBolt(CoordinatorEntity, homeassistant.components.lock.LockEntity):
def __init__(self, coordinator):
super().__init__(coordinator)
self._lock = coordinator._lock
@property
def name(self):
"""Return the display name of this lock."""
return self._lock.nickname
@property
def unique_id(self):
return self._lock.mac
@property
def device_info(self):
return {
"identifiers": {(DOMAIN, self._lock.mac)},
"name": self._lock.nickname,
"connections": {
(
dr.CONNECTION_NETWORK_MAC,
self.coordinator._mac,
),
("uuid", self.coordinator._uuid),
("serial_number", self._lock.raw_dict["hardware_info"]["sn"]),
},
"manufacturer": "WyzeLabs",
"model": self._lock.product_model,
}
@property
def is_locked(self):
return self.coordinator.data["state"] == 1
async def async_lock(self, **kwargs):
return await self.coordinator.lock_unlock(command="lock")
async def async_unlock(self, **kwargs):
return await self.coordinator.lock_unlock(command="unlock")
@property
def is_locking(self, **kwargs):
return self.coordinator._current_command == "lock"
@property
def is_unlocking(self, **kwargs):
return self.coordinator._current_command == "unlock"
@property
def state_attributes(self):
return {"last_operated": self.coordinator.data["timestamp"]}
+18
View File
@@ -0,0 +1,18 @@
{
"domain": "wyzeapi",
"name": "Wyze",
"codeowners": [
"@SecKatie"
],
"config_flow": true,
"dependencies": ["bluetooth_adapters"],
"documentation": "https://github.com/SecKatie/ha-wyzeapi#readme",
"iot_class": "cloud_polling",
"issue_tracker": "https://github.com/SecKatie/ha-wyzeapi/issues",
"loggers": ["custom_components.wyzeapi"],
"requirements": [
"wyzeapy>=0.5.33,<0.6",
"websockets"
],
"version": "0.1.37"
}
+150
View File
@@ -0,0 +1,150 @@
"""Platform for number integration."""
import logging
from typing import Any, Callable, List
from homeassistant.components.number import RestoreNumber
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers import device_registry as dr
from wyzeapy import Wyzeapy
from wyzeapy.services.irrigation_service import IrrigationService, Irrigation, Zone
from .const import DOMAIN, CONF_CLIENT
from .token_manager import token_exception_handler
_LOGGER = logging.getLogger(__name__)
@token_exception_handler
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: Callable[[List[Any], bool], None],
) -> None:
"""Set up the WyzeApi number platform."""
_LOGGER.debug("Creating new WyzeApi number component")
client: Wyzeapy = hass.data[DOMAIN][config_entry.entry_id][CONF_CLIENT]
irrigation_service = await client.irrigation_service
# Get all irrigation devices
irrigation_devices = await irrigation_service.get_irrigations()
# Create a number entity for each zone in each irrigation device
entities = []
for device in irrigation_devices:
# Update the device to get its zones
device = await irrigation_service.update(device)
for zone in device.zones:
if zone.enabled:
entities.append(
WyzeIrrigationQuickrunDuration(irrigation_service, device, zone)
)
async_add_entities(entities, True)
class WyzeIrrigationQuickrunDuration(RestoreNumber):
"""Representation of a Wyze Irrigation Zone Quickrun Duration."""
_attr_has_entity_name = True
def __init__(
self, irrigation_service: IrrigationService, irrigation: Irrigation, zone: Zone
) -> None:
"""Initialize the irrigation zone quickrun duration."""
self._irrigation_service = irrigation_service
self._device = irrigation
self._zone = zone
@property
def name(self) -> str:
"""Return the name of the zone quickrun duration."""
return f"{self._zone.name}"
@property
def unique_id(self) -> str:
"""Return a unique ID for the zone quickrun duration."""
return f"{self._device.mac}-zone-{self._zone.zone_number}-quickrun-duration"
@property
def device_info(self) -> DeviceInfo:
"""Return device information about this entity."""
return DeviceInfo(
identifiers={(DOMAIN, self._device.mac)},
name=self._device.nickname,
manufacturer="WyzeLabs",
model=self._device.product_model,
serial_number=self._device.sn,
connections={(dr.CONNECTION_NETWORK_MAC, self._device.mac)},
)
@property
def native_value(self) -> float:
"""Return the current value in minutes."""
return float(self._zone.quickrun_duration) / 60.0
@property
def native_min_value(self) -> float:
"""Return the minimum value in minutes."""
return 1.0 # 1 minute
@property
def native_max_value(self) -> float:
"""Return the maximum value in minutes."""
return 180.0 # 3 hours in minutes
@property
def native_step(self) -> float:
"""Return the step value in minutes."""
return 1.0 # 1 minute steps
@property
def mode(self) -> str:
"""Return the mode of the number entity."""
return "box"
@property
def native_unit_of_measurement(self) -> str:
"""Return the unit of measurement."""
return "min"
@property
def icon(self) -> str:
"""Return the icon for the quickrun duration number."""
return "mdi:timer"
async def async_set_native_value(self, value: float) -> None:
"""Set the value in minutes."""
# Convert minutes to seconds for the API
seconds = int(value * 60)
await self._irrigation_service.set_zone_quickrun_duration(
self._device, self._zone.zone_number, seconds
)
self._zone.quickrun_duration = seconds
self.async_write_ha_state()
async def _async_load_value(self) -> None:
"""Load the value from Home Assistant state or update from irrigation service."""
# Try to get the last number data from Home Assistant
state = await self.async_get_last_number_data()
if state and state.native_value is not None:
try:
# Convert minutes to seconds for storage
self._zone.quickrun_duration = int(state.native_value * 60)
return
except (ValueError, TypeError):
pass
# If no valid state exists, update from irrigation service
self._device = await self._irrigation_service.update(self._device)
for zone in self._device.zones:
if zone.zone_number == self._zone.zone_number:
self._zone = zone
break
async def async_added_to_hass(self) -> None:
"""Subscribe to updates."""
await self._async_load_value()
return await super().async_added_to_hass()
+626
View File
@@ -0,0 +1,626 @@
"""Platform for sensor integration."""
from collections.abc import Callable
import datetime
import json
import logging
from typing import Any
from wyzeapy import Wyzeapy
from wyzeapy.services.camera_service import Camera
from wyzeapy.services.irrigation_service import Irrigation, IrrigationService
from wyzeapy.services.lock_service import Lock
from wyzeapy.services.switch_service import Switch, SwitchUsageService
from homeassistant.components.sensor import (
RestoreSensor,
SensorDeviceClass,
SensorEntity,
SensorStateClass,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
ATTR_ATTRIBUTION,
PERCENTAGE,
EntityCategory,
UnitOfEnergy,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity import DeviceInfo
import homeassistant.helpers.entity_registry as er
from homeassistant.helpers.event import (
async_track_state_change_event,
async_track_time_change,
)
from .const import (
CAMERA_UPDATED,
CONF_CLIENT,
DOMAIN,
LOCK_UPDATED,
RESET_BUTTON_PRESSED,
)
from .token_manager import token_exception_handler
_LOGGER = logging.getLogger(__name__)
ATTRIBUTION = "Data provided by Wyze"
CAMERAS_WITH_BATTERIES = ["WVOD1", "HL_WCO2", "AN_RSCW", "GW_BE1"]
OUTDOOR_PLUGS = ["WLPPO"]
@token_exception_handler
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: Callable[[list[Any], bool], None],
) -> None:
"""This function sets up the config_entry.
:param hass: Home Assistant instance
:param config_entry: The current config_entry
:param async_add_entities: This function adds entities to the config_entry
:return:
"""
_LOGGER.debug("""Creating new WyzeApi sensor component""")
client: Wyzeapy = hass.data[DOMAIN][config_entry.entry_id][CONF_CLIENT]
# Get the list of locks so that we can create lock and keypad battery sensors
lock_service = await client.lock_service
camera_service = await client.camera_service
switch_usage_service = await client.switch_usage_service
irrigation_service = await client.irrigation_service
locks = await lock_service.get_locks()
sensors = []
for lock in locks:
sensors.append(WyzeLockBatterySensor(lock, WyzeLockBatterySensor.LOCK_BATTERY))
sensors.append(
WyzeLockBatterySensor(lock, WyzeLockBatterySensor.KEYPAD_BATTERY)
)
cameras = await camera_service.get_cameras()
sensors.extend(
[
WyzeCameraBatterySensor(camera)
for camera in cameras
if camera.product_model in CAMERAS_WITH_BATTERIES
]
)
plugs = await switch_usage_service.get_switches()
for plug in plugs:
if plug.product_model in OUTDOOR_PLUGS:
sensors.append(WyzePlugEnergySensor(plug, switch_usage_service))
sensors.append(WyzePlugDailyEnergySensor(plug))
# Get all irrigation devices
irrigation_devices = await irrigation_service.get_irrigations()
# Create sensor entities for each irrigation device
for device in irrigation_devices:
# Update the device to get its properties
device = await irrigation_service.update(device)
sensors.extend(
[
WyzeIrrigationRSSI(irrigation_service, device),
WyzeIrrigationIP(irrigation_service, device),
WyzeIrrigationSSID(irrigation_service, device),
]
)
async_add_entities(sensors, True)
class WyzeLockBatterySensor(SensorEntity):
"""Representation of a Wyze Lock or Lock Keypad Battery."""
@property
def enabled(self):
"""Return if the sensor is enabled."""
return self._enabled
LOCK_BATTERY = "lock_battery"
KEYPAD_BATTERY = "keypad_battery"
_attr_device_class = SensorDeviceClass.BATTERY
_attr_native_unit_of_measurement = PERCENTAGE
_attr_should_poll = False
def __init__(self, lock, battery_type) -> None:
"""Initialize the sensor."""
self._enabled = None
self._lock = lock
self._battery_type = battery_type
# make the battery unavailable by default, this will be toggled after the first update from the battery entity that
# has battery data.
self._available = False
@callback
def handle_lock_update(self, lock: Lock) -> None:
"""Helper function to Enable lock when Keypad has a battery.
Make it avaliable when either the lock battery or keypad battery exists.
"""
self._lock = lock
if self._lock.raw_dict.get("power") and self._battery_type == self.LOCK_BATTERY:
self._available = True
if (
self._lock.raw_dict.get("keypad", {}).get("power")
and self._battery_type == self.KEYPAD_BATTERY
):
if self.enabled is False:
self.enabled = True
self._available = True
self.async_write_ha_state()
async def async_added_to_hass(self) -> None:
"""Add listener on startup."""
self.async_on_remove(
async_dispatcher_connect(
self.hass,
f"{LOCK_UPDATED}-{self._lock.mac}",
self.handle_lock_update,
)
)
@property
def name(self) -> str:
"""Name of the Sensor."""
battery_type = self._battery_type.replace("_", " ").title()
return f"{self._lock.nickname} {battery_type}"
@property
def unique_id(self):
"""Unique ID of the sensor."""
return f"{self._lock.nickname}.{self._battery_type}"
@property
def available(self) -> bool:
"""Return if the sensor is available."""
return self._available
@property
def entity_registry_enabled_default(self) -> bool:
"""Return if the entity should be enabled."""
if self._battery_type == self.KEYPAD_BATTERY:
# The keypad battery may not be available if the lock has no keypad
return False
# The battery voltage will always be available for the lock
return True
@property
def device_info(self):
"""Return the device info."""
return {
"identifiers": {(DOMAIN, self._lock.mac)},
"connections": {
(
dr.CONNECTION_NETWORK_MAC,
self._lock.mac,
)
},
"name": f"{self._lock.nickname}.{self._battery_type}",
}
@property
def extra_state_attributes(self):
"""Return device attributes of the entity."""
return {
ATTR_ATTRIBUTION: ATTRIBUTION,
"device model": f"{self._lock.product_model}.{self._battery_type}",
}
@property
def native_value(self):
"""Return the state of the device."""
if self._battery_type == self.LOCK_BATTERY:
return str(self._lock.raw_dict.get("power"))
if self._battery_type == self.KEYPAD_BATTERY:
return str(self._lock.raw_dict.get("keypad", {}).get("power"))
return 0
@enabled.setter
def enabled(self, value):
self._enabled = value
class WyzeCameraBatterySensor(SensorEntity):
"""Representation of a Wyze Camera Battery."""
_attr_device_class = SensorDeviceClass.BATTERY
_attr_native_unit_of_measurement = PERCENTAGE
_attr_should_poll = False
def __init__(self, camera) -> None:
"""Initialize the sensor."""
self._camera = camera
@callback
def handle_camera_update(self, camera: Camera) -> None:
"""Handle camera updates."""
self._camera = camera
self.async_write_ha_state()
async def async_added_to_hass(self) -> None:
"""Add listener on startup."""
self.async_on_remove(
async_dispatcher_connect(
self.hass,
f"{CAMERA_UPDATED}-{self._camera.mac}",
self.handle_camera_update,
)
)
@property
def name(self) -> str:
"""Return the entity name."""
return f"{self._camera.nickname} Battery"
@property
def unique_id(self):
"""Unique ID of the sensor."""
return f"{self._camera.nickname}.battery"
@property
def device_info(self):
"""Return the device info."""
return {
"identifiers": {(DOMAIN, self._camera.mac)},
"connections": {
(
dr.CONNECTION_NETWORK_MAC,
self._camera.mac,
)
},
"name": self._camera.nickname,
"model": self._camera.product_model,
"manufacturer": "WyzeLabs",
}
@property
def extra_state_attributes(self):
"""Return device attributes of the entity."""
return {
ATTR_ATTRIBUTION: ATTRIBUTION,
"device model": f"{self._camera.product_model}.battery",
}
@property
def native_value(self):
"""Return the value of the sensor."""
return self._camera.device_params.get("electricity")
class WyzePlugEnergySensor(RestoreSensor):
"""Respresents an Outdoor Plug Total Energy Sensor."""
_attr_device_class = SensorDeviceClass.ENERGY
_attr_native_unit_of_measurement = UnitOfEnergy.KILO_WATT_HOUR
_attr_state_class = SensorStateClass.TOTAL_INCREASING
_attr_suggested_display_precision = 3
_attr_should_poll = False
_attr_name = "Total Energy Usage"
_previous_hour = None
_previous_value = None
_past_hours_previous_value = None
_current_value = 0
_past_hours_value = 0
_hourly_energy_usage_added = 0
def __init__(
self, switch: Switch, switch_usage_service: SwitchUsageService
) -> None:
"""Initialize an energy sensor."""
self._switch = switch
self._switch_usage_service = switch_usage_service
self._switch.usage_history = None # type: ignore[attr-defined]
@property
def unique_id(self):
"""Get the unique ID of the sensor."""
return f"{self._switch.nickname}.energy-{self._switch.mac}"
@property
def device_info(self):
"""Return the device info."""
return {
"identifiers": {(DOMAIN, self._switch.mac)},
"name": self._switch.nickname,
}
def update_energy(self):
"""Update the energy sensor."""
_now = int(datetime.datetime.now(datetime.UTC).hour)
self._hourly_energy_usage_added = 0
if (
self._switch.usage_history and len(self._switch.usage_history) > 0
): # Confirm there is data
_raw_data = self._switch.usage_history
_LOGGER.debug(_raw_data)
_current_day_list = json.loads(_raw_data[0]["data"])
if _now == 0: # Handle rolling to the next UTC day
self._past_hours_value = _current_day_list[23] / 1000
if len(_raw_data) > 1: # New Day's value
_next_day_list = json.loads(_raw_data[1]["data"])
self._current_value = _next_day_list[_now] / 1000
else:
self._current_value = 0
else:
self._past_hours_value = _current_day_list[_now - 1] / 1000
self._current_value = _current_day_list[_now] / 1000
# Set inital values to current values on startup.
# Has to be done after we check for current or next UTC day
if self._previous_hour is None:
self._previous_hour = _now
if self._past_hours_previous_value is None:
self._past_hours_previous_value = self._past_hours_value
if self._previous_value is None:
self._previous_value = self._current_value
if _now != self._previous_hour: # New Hour
if self._past_hours_value > self._previous_value:
self._hourly_energy_usage_added = (
self._past_hours_value - self._previous_value
)
self._hourly_energy_usage_added += self._current_value
self._previous_value = self._current_value
self._previous_hour = _now
self._past_hours_previous_value = self._past_hours_value
else: # Current Hour
if self._current_value > self._previous_value:
self._hourly_energy_usage_added += round(
self._current_value - self._previous_value, 3
)
self._previous_value = self._current_value
if self._past_hours_value > self._past_hours_previous_value:
self._hourly_energy_usage_added += round(
self._past_hours_value - self._past_hours_previous_value, 3
)
self._past_hours_previous_value = self._past_hours_value
_LOGGER.debug(
"Total Value Added to device %s is %s",
self._switch.mac,
self._hourly_energy_usage_added,
)
return self._hourly_energy_usage_added
@callback
def async_update_callback(self, switch: Switch):
"""Update the sensor's state."""
self._switch = switch
self.update_energy()
self._attr_native_value += self._hourly_energy_usage_added
self.async_write_ha_state()
@callback
def reset_energy_use(self, switch: Switch):
"""Reset the Energy Usage."""
_LOGGER.debug("Resetting Usage of %s to 0", self._switch.nickname)
self._switch = switch
self._attr_native_value = 0
self.async_write_ha_state()
async def async_added_to_hass(self) -> None:
"""Register Updater for the sensor and get previous data."""
state = await self.async_get_last_sensor_data()
if state:
self._attr_native_value = state.native_value
else:
self._attr_native_value = 0
self._switch.callback_function = self.async_update_callback
self._switch_usage_service.register_updater(
self._switch, 120
) # Every 2 minutes seems to work fine, probably could be longer
await self._switch_usage_service.start_update_manager()
self.async_on_remove(
async_dispatcher_connect(
self.hass,
f"{RESET_BUTTON_PRESSED}-{self._switch.mac}",
self.reset_energy_use,
)
)
async def async_will_remove_from_hass(self) -> None:
"""Remove updater."""
self._switch_usage_service.unregister_updater(self._switch)
class WyzePlugDailyEnergySensor(RestoreSensor):
"""Respresents an Outdoor Plug Daily Energy Sensor."""
_attr_device_class = SensorDeviceClass.ENERGY
_attr_native_unit_of_measurement = UnitOfEnergy.KILO_WATT_HOUR
_attr_state_class = SensorStateClass.TOTAL_INCREASING
_attr_suggested_display_precision = 3
_attr_name = "Daily Energy Usage"
def __init__(self, switch: Switch) -> None:
"""Initialize a daily energy sensor."""
self._switch = switch
@property
def unique_id(self):
"""Get the unique ID of the sensor."""
return f"{self._switch.nickname}.daily_energy-{self._switch.mac}"
@property
def should_poll(self) -> bool:
"""No polling needed."""
return False
@property
def device_info(self):
"""Return the device info."""
return {
"identifiers": {(DOMAIN, self._switch.mac)},
"name": self._switch.nickname,
}
@callback
def _update_daily_sensor(self, event):
"""Update the sensor when the total sensor updates."""
event_data = event.data
new_state = event_data["new_state"]
old_state = event_data["old_state"]
if not old_state or not new_state:
return
updated_energy = float(new_state.state) - float(old_state.state)
self._attr_native_value += updated_energy
self.async_write_ha_state()
async def _async_reset_at_midnight(self, now: datetime) -> None:
"""Reset the daily sensor."""
self._attr_native_value = 0
_LOGGER.debug("Resetting daily energy sensor %s to 0", self._switch.mac)
self.async_write_ha_state()
async def async_added_to_hass(self) -> None:
"""Get previous data and add listeners."""
state = await self.async_get_last_sensor_data()
if state:
self._attr_native_value = state.native_value
else:
self._attr_native_value = 0
registry = er.async_get(self.hass)
entity_id_total_sensor = registry.async_get_entity_id(
"sensor", DOMAIN, f"{self._switch.nickname}.energy-{self._switch.mac}"
)
self.async_on_remove(
async_track_state_change_event(
self.hass, [entity_id_total_sensor], self._update_daily_sensor
)
)
self.async_on_remove(
async_track_time_change(
self.hass, self._async_reset_at_midnight, hour=0, minute=0, second=0
)
)
class WyzeIrrigationBaseSensor(SensorEntity):
"""Base class for Wyze Irrigation sensors."""
_attr_has_entity_name = True
_attr_should_poll = False
def __init__(
self, irrigation_service: IrrigationService, irrigation: Irrigation
) -> None:
"""Initialize the irrigation base sensor."""
self._irrigation_service = irrigation_service
self._device = irrigation
@property
def device_info(self) -> DeviceInfo:
"""Return device information about this entity."""
return DeviceInfo(
identifiers={(DOMAIN, self._device.mac)},
name=self._device.nickname,
manufacturer="WyzeLabs",
model=self._device.product_model,
serial_number=self._device.sn,
connections={(dr.CONNECTION_NETWORK_MAC, self._device.mac)},
)
@callback
def async_update_callback(self, irrigation: Irrigation) -> None:
"""Update the irrigation's state."""
self._device = self._irrigation_service.update_device_props(irrigation)
self.async_schedule_update_ha_state()
async def async_added_to_hass(self) -> None:
"""Subscribe to updates."""
self._device.callback_function = self.async_update_callback
self._irrigation_service.register_updater(self._device, 30)
await self._irrigation_service.start_update_manager()
return await super().async_added_to_hass()
async def async_will_remove_from_hass(self) -> None:
"""Clean up when removed."""
self._irrigation_service.unregister_updater(self._device)
class WyzeIrrigationRSSI(WyzeIrrigationBaseSensor):
"""Representation of a Wyze Irrigation RSSI sensor."""
_attr_entity_category = EntityCategory.DIAGNOSTIC
_attr_entity_registry_enabled_default = False
@property
def name(self) -> str:
"""Return the name of the sensor."""
return "RSSI"
@property
def unique_id(self) -> str:
"""Return a unique ID for the sensor."""
return f"{self._device.mac}-rssi"
@property
def native_value(self) -> int:
"""Return the RSSI value."""
return self._device.RSSI
@property
def native_unit_of_measurement(self) -> str:
"""Return the unit of measurement."""
return "dBm"
class WyzeIrrigationIP(WyzeIrrigationBaseSensor):
"""Representation of a Wyze Irrigation IP sensor."""
_attr_entity_category = EntityCategory.DIAGNOSTIC
_attr_entity_registry_enabled_default = False
@property
def name(self) -> str:
"""Return the name of the sensor."""
return "IP Address"
@property
def unique_id(self) -> str:
"""Return a unique ID for the sensor."""
return f"{self._device.mac}-ip"
@property
def native_value(self) -> str:
"""Return the IP address."""
return self._device.IP
class WyzeIrrigationSSID(WyzeIrrigationBaseSensor):
"""Representation of a Wyze Irrigation SSID sensor."""
_attr_entity_category = EntityCategory.DIAGNOSTIC
_attr_entity_registry_enabled_default = False
@property
def name(self) -> str:
"""Return the name of the sensor."""
return "SSID"
@property
def unique_id(self) -> str:
"""Return a unique ID for the sensor."""
return f"{self._device.mac}-ssid"
@property
def native_value(self) -> str:
"""Return the SSID."""
return self._device.ssid
+142
View File
@@ -0,0 +1,142 @@
#!/usr/bin/python3
"""Platform for siren integration."""
import logging
from typing import Any, Callable
from aiohttp.client_exceptions import ClientConnectionError
from wyzeapy import CameraService, Wyzeapy
from wyzeapy.services.camera_service import Camera
from wyzeapy.exceptions import AccessTokenError, ParameterError, UnknownApiError
from homeassistant.components.siren import (
SirenEntity,
SirenEntityFeature,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from .const import CAMERA_UPDATED, CONF_CLIENT, DOMAIN
from .token_manager import token_exception_handler
_LOGGER = logging.getLogger(__name__)
ATTRIBUTION = "Data provided by Wyze"
@token_exception_handler
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: Callable[[list[Any], bool], None],
) -> None:
"""
This function sets up the config entry
:param hass: The Home Assistant Instance
:param config_entry: The current config entry
:param async_add_entities: This function adds entities to the config entry
:return:
"""
_LOGGER.debug("""Creating new WyzeApi siren component""")
client: Wyzeapy = hass.data[DOMAIN][config_entry.entry_id][CONF_CLIENT]
camera_service = await client.camera_service
sirens = []
for camera in await camera_service.get_cameras():
# The campan v1, v2 camera, and video doorbell pro don't have sirens
if camera.product_model not in ["WYZECP1_JEF", "WYZEC1-JZ", "GW_BE1"]:
sirens.append(WyzeCameraSiren(camera, camera_service))
async_add_entities(sirens, True)
class WyzeCameraSiren(SirenEntity):
"""Representation of a Wyze Camera Siren."""
_available: bool
_just_updated = False
def __init__(self, camera: Camera, camera_service: CameraService) -> None:
self._device = camera
self._service = camera_service
self._attr_supported_features = (
SirenEntityFeature.TURN_OFF | SirenEntityFeature.TURN_ON
)
@token_exception_handler
async def async_turn_on(self, **kwargs) -> None:
"""Turn the siren on."""
try:
await self._service.siren_on(self._device)
except (AccessTokenError, ParameterError, UnknownApiError) as err:
raise HomeAssistantError(f"Wyze returned an error: {err.args}") from err
except ClientConnectionError as err:
raise HomeAssistantError(err) from err
else:
self._device.siren = True
self._just_updated = True
self.async_schedule_update_ha_state()
@token_exception_handler
async def async_turn_off(self, **kwargs):
"""Turn the siren off."""
try:
await self._service.siren_off(self._device)
except (AccessTokenError, ParameterError, UnknownApiError) as err:
raise HomeAssistantError(f"Wyze returned an error: {err.args}") from err
except ClientConnectionError as err:
raise HomeAssistantError(err) from err
else:
self._device.siren = False
self._just_updated = True
self.async_schedule_update_ha_state()
@property
def should_poll(self) -> bool:
return False
@property
def is_on(self):
"""Return true if siren is on."""
return self._device.siren
@property
def available(self):
"""Return the connection status of this switch"""
return self._device.available
@property
def name(self) -> str:
return f"{self._device.nickname} Siren"
@property
def unique_id(self):
return f"{self._device.mac}-siren"
@property
def device_info(self):
return {
"identifiers": {(DOMAIN, self._device.mac)},
"name": self._device.nickname,
"manufacturer": "WyzeLabs",
"model": self._device.product_model,
}
@callback
def handle_camera_update(self, camera: Camera) -> None:
"""Update the camera object whenever there is an update"""
self._device = camera
self.async_write_ha_state()
async def async_added_to_hass(self) -> None:
self.async_on_remove(
async_dispatcher_connect(
self.hass,
f"{CAMERA_UPDATED}-{self._device.mac}",
self.handle_camera_update,
)
)
+77
View File
@@ -0,0 +1,77 @@
{
"title": "[%key:common::config_flow::data::title%]",
"config": {
"step": {
"user": {
"title": "[%key:common::config_flow::data::title%]",
"data": {
"username": "[%key:common::config_flow::data::username%]",
"password": "[%key:common::config_flow::data::password%]",
"keyid": "[%key:common::config_flow::data::key_id%]",
"apikey": "[%key:common::config_flow::data::api_key%]"
},
"data_description": {
"username": "Wyze or 3rd party OAuth email address",
"password": "Wyze or 3rd party OAuth password"
}
},
"2fa": {
"title": "[%key:common::config_flow::data::title%]",
"data": {
"verification_code": "[%key:common::config_flow::data::verification_code%]"
}
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
}
},
"options": {
"step": {
"init": {
"data": {
"bulb_local_control": "Use Local Control for Color Bulbs and Light Strips"
}
},
"user": {
"title": "[%key:common::config_flow::data::title%]",
"data": {
"username": "[%key:common::config_flow::data::username%]",
"password": "[%key:common::config_flow::data::password%]",
"keyid": "[%key:common::config_flow::data::key_id%]",
"apikey": "[%key:common::config_flow::data::api_key%]"
}
},
"2fa": {
"title": "[%key:common::config_flow::data::title%]",
"data": {
"verification_code": "[%key:common::config_flow::data::verification_code%]"
}
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
}
},
"issues": {
"entity_changed": {
"title": "Wyze Entity Error",
"description": "{automation} may need attention. {entity} may have changed or been removed. This is due to a change in outdoor plug devices. Once fixed you can ignore this issue."
},
"device_changed": {
"title": "Wyze Device Error",
"description": "{automation} may need attention. A device has changed or been removed. This is due to a change in outdoor plug devices. Once fixed you can ignore this issue."
}
}
}
+730
View File
@@ -0,0 +1,730 @@
#!/usr/bin/python3
"""Platform for switch integration."""
from collections.abc import Callable
from datetime import timedelta
import logging
from typing import Any
from aiohttp.client_exceptions import ClientConnectionError
from wyzeapy import BulbService, CameraService, SwitchService, Wyzeapy
from wyzeapy.exceptions import AccessTokenError, ParameterError, UnknownApiError
from wyzeapy.services.bulb_service import Bulb
from wyzeapy.services.camera_service import Camera
from wyzeapy.services.switch_service import Switch
from wyzeapy.types import Device, DeviceTypes, Event
from homeassistant.components.automation import (
automations_with_device,
automations_with_entity,
)
from homeassistant.components.script import scripts_with_device, scripts_with_entity
from homeassistant.components.switch import SwitchEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import (
device_registry as dr,
entity_registry as er,
issue_registry as ir,
)
from homeassistant.helpers.dispatcher import (
async_dispatcher_connect,
async_dispatcher_send,
)
from homeassistant.helpers.issue_registry import IssueSeverity
from .const import (
CAMERA_UPDATED,
CONF_CLIENT,
DOMAIN,
LIGHT_UPDATED,
WYZE_CAMERA_EVENT,
WYZE_NOTIFICATION_TOGGLE,
)
from .token_manager import token_exception_handler
_LOGGER = logging.getLogger(__name__)
ATTRIBUTION = "Data provided by Wyze"
SCAN_INTERVAL = timedelta(seconds=30)
OUTDOOR_PLUGS = "WLPPO"
OUTDOOR_PLUG_INDIVUAL_OUTLETS = "WLPPO-SUB"
MOTION_SWITCH_UNSUPPORTED = [
"GW_BE1",
"GW_GC1",
"GW_GC2",
] # Video doorbell pro, OG, OG 3x Telephoto
POWER_SWITCH_UNSUPPORTED = ["GW_BE1"] # Video doorbell pro (device has no off function)
NOTIFICATION_SWITCH_UNSUPPORTED = {
"GW_GC1",
"GW_GC2",
} # OG and OG 3x Telephoto models currently unsupported due to InvalidSignature2 error
# noinspection DuplicatedCode
@token_exception_handler
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: Callable[[list[Any], bool], None],
) -> None:
"""This function sets up the config entry.
:param hass: The Home Assistant Instance
:param config_entry: The current config entry
:param async_add_entities: This function adds entities to the config entry
:return:
"""
_LOGGER.debug("""Creating new WyzeApi light component""")
client: Wyzeapy = hass.data[DOMAIN][config_entry.entry_id][CONF_CLIENT]
switch_service = await client.switch_service
wall_switch_service = await client.wall_switch_service
camera_service = await client.camera_service
bulb_service = await client.bulb_service
switches: list[SwitchEntity] = []
has_outdoor_plug: bool = False
devices_to_migrate: list[str] = []
devices = []
device_registry = dr.async_get(hass)
base_switches = await switch_service.get_switches()
# The outdoor plug has a dummy switch that doesn't control anything
# on the device. So we add non-outdoor plug switches and then
# the switches for each individual outlet on the outdoor plug.
switches.extend(
WyzeSwitch(switch_service, switch)
for switch in base_switches
if switch.product_model not in [OUTDOOR_PLUGS, OUTDOOR_PLUG_INDIVUAL_OUTLETS]
)
for switch in base_switches:
if switch.product_model in [OUTDOOR_PLUG_INDIVUAL_OUTLETS]:
has_outdoor_plug = True
switches.append(WyzeSwitch(switch_service, switch))
switches.extend(
WyzeSwitch(wall_switch_service, switch)
for switch in await wall_switch_service.get_switches()
)
camera_switches = await camera_service.get_cameras()
for switch in camera_switches:
# Notification toggle switch
if switch.product_model not in NOTIFICATION_SWITCH_UNSUPPORTED:
switches.append(WyzeCameraNotificationSwitch(camera_service, switch))
# IoT Power switch
if switch.product_model not in POWER_SWITCH_UNSUPPORTED:
switches.append(WyzeSwitch(camera_service, switch))
# Motion toggle switch
if switch.product_model not in MOTION_SWITCH_UNSUPPORTED:
switches.append(WyzeCameraMotionSwitch(camera_service, switch))
switches.append(WyzeNotifications(client))
bulb_switches = await bulb_service.get_bulbs()
switches.extend(
WzyeLightstripSwitch(bulb_service, bulb)
for bulb in bulb_switches
if bulb.type is DeviceTypes.LIGHTSTRIP
)
# Catch old outdoor plug devices and entities and remove.
# This can be removed at a later date.
if has_outdoor_plug:
devices = dr.async_entries_for_config_entry(
device_registry, config_entry.entry_id
)
for device in devices:
for identifier in device.identifiers:
mac = identifier[1]
if (
"-" in mac and device.model == OUTDOOR_PLUG_INDIVUAL_OUTLETS
): # The old devices have a '-' in the mac
devices_to_migrate.append(device.id)
# Also catch the old dummy switch to remove below.
# Should only happen once while the old devices are still around.
if devices_to_migrate:
devices_to_migrate.extend(
device.id for device in devices if device.model == OUTDOOR_PLUGS
)
await async_migrate_switch_data(
hass, config_entry, devices_to_migrate, device_registry
)
async_add_entities(switches, True)
async def async_migrate_switch_data(
hass: HomeAssistant,
config_entry: ConfigEntry,
device_list,
device_registry,
):
"""Remove redundant switch devices and entities and flag for repair.
This can be removed in the future.
"""
entity_registry = er.async_get(hass)
entities = er.async_entries_for_config_entry(entity_registry, config_entry.entry_id)
for entity in entities:
if entity.device_id in device_list and entity.domain == "switch":
entity_automations = automations_with_entity(hass, entity.entity_id)
entity_automations.extend(scripts_with_entity(hass, entity.entity_id))
if entity_automations:
for issue in entity_automations:
ir.async_create_issue(
hass,
DOMAIN,
issue_id=f"{issue}-entity-issue",
is_fixable=False,
is_persistent=True,
severity=IssueSeverity.ERROR,
learn_more_url="https://github.com/SecKatie/ha-wyzeapi/issues/789",
translation_key="entity_changed",
translation_placeholders={
"entity": entity.entity_id,
"automation": issue,
},
)
entity_registry.async_remove(entity.id)
for device in device_list:
device_automations = automations_with_device(hass, device)
device_automations.extend(scripts_with_device(hass, device))
if device_automations:
for issue in device_automations:
ir.async_create_issue(
hass,
DOMAIN,
issue_id=f"{issue}-device-issue",
is_fixable=False,
is_persistent=True,
severity=IssueSeverity.ERROR,
learn_more_url="https://github.com/SecKatie/ha-wyzeapi/issues/789",
translation_key="device_changed",
translation_placeholders={
"automation": issue,
},
)
device_registry.async_remove_device(device)
class WyzeNotifications(SwitchEntity):
"""Class for notification switch."""
_attr_should_poll = False
_attr_name = "Wyze Notifications"
def __init__(self, client: Wyzeapy) -> None:
"""Initialize the switch."""
self._client = client
self._is_on = False
self._uid = WYZE_NOTIFICATION_TOGGLE
self._just_updated = False
@property
def is_on(self) -> bool:
"""Return if the switch is on."""
return self._is_on
@property
def device_info(self):
"""Return the device info."""
return {
"identifiers": {(DOMAIN, self._uid)},
"name": "Wyze Notifications",
"manufacturer": "WyzeLabs",
"model": "WyzeNotificationToggle",
}
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on the switch."""
try:
await self._client.enable_notifications()
except (AccessTokenError, ParameterError, UnknownApiError) as err:
raise HomeAssistantError(f"Wyze returned an error: {err.args}") from err
except ClientConnectionError as err:
raise HomeAssistantError(err) from err
else:
self._is_on = True
self._just_updated = True
self.async_schedule_update_ha_state()
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off the switch."""
try:
await self._client.disable_notifications()
except (AccessTokenError, ParameterError, UnknownApiError) as err:
raise HomeAssistantError(f"Wyze returned an error: {err.args}") from err
except ClientConnectionError as err:
raise HomeAssistantError(err) from err
else:
self._is_on = False
self._just_updated = True
self.async_schedule_update_ha_state()
@property
def available(self):
"""Return the connection status of this switch."""
return True
@property
def unique_id(self):
"""Return the unique ID."""
return self._uid
async def async_update(self):
"""Update the switch."""
if not self._just_updated:
self._is_on = await self._client.notifications_are_on
else:
self._just_updated = False
class WyzeSwitch(SwitchEntity):
"""Representation of a Wyze Switch."""
_on: bool
_available: bool
_just_updated = False
_old_event_ts: int = 0 # preload with 0 so that we know when it's been updated
_attr_should_poll = False
def __init__(self, service: CameraService | SwitchService, device: Device) -> None:
"""Initialize a Wyze Bulb."""
self._device = device
self._service = service
if type(self._device) is Camera:
self._device = Camera(self._device.raw_dict)
elif type(self._device) is Switch:
self._device = Switch(self._device.raw_dict)
@property
def device_info(self):
"""Return the device info.
Outdoor plug needs its own setup based on how the MAC's are
displayed and to keep the plugs organized by device.
"""
if self._device.product_model == OUTDOOR_PLUG_INDIVUAL_OUTLETS:
mac = self._device.mac.split("-")[0]
return {
"identifiers": {(DOMAIN, mac)},
"connections": {
(
dr.CONNECTION_NETWORK_MAC,
mac,
)
},
"name": f"Outdoor Plug {mac}",
"manufacturer": "WyzeLabs",
"model": self._device.product_model,
}
return {
"identifiers": {(DOMAIN, self._device.mac)},
"connections": {
(
dr.CONNECTION_NETWORK_MAC,
self._device.mac,
)
},
"name": self._device.nickname,
"manufacturer": "WyzeLabs",
"model": self._device.product_model,
}
@token_exception_handler
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on the switch."""
try:
await self._service.turn_on(self._device)
except (AccessTokenError, ParameterError, UnknownApiError) as err:
raise HomeAssistantError(f"Wyze returned an error: {err.args}") from err
except ClientConnectionError as err:
raise HomeAssistantError(err) from err
else:
self._device.on = True
self._just_updated = True
self.async_schedule_update_ha_state()
@token_exception_handler
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off the switch."""
try:
await self._service.turn_off(self._device)
except (AccessTokenError, ParameterError, UnknownApiError) as err:
raise HomeAssistantError(f"Wyze returned an error: {err.args}") from err
except ClientConnectionError as err:
raise HomeAssistantError(err) from err
else:
self._device.on = False
self._just_updated = True
self.async_schedule_update_ha_state()
@property
def name(self):
"""Return the display name of this switch."""
if type(self._device) is Camera:
return f"{self._device.nickname} Power"
return self._device.nickname
@property
def available(self):
"""Return the connection status of this switch."""
return self._device.available
@property
def is_on(self):
"""Return true if switch is on."""
return self._device.on
@property
def unique_id(self):
"""Return the unique ID."""
return f"{self._device.mac}-switch"
@property
def extra_state_attributes(self):
"""Return device attributes of the entity."""
dev_info = {}
if self._device.device_params.get("electricity"):
dev_info["Battery"] = str(
self._device.device_params.get("electricity") + "%"
)
# noinspection DuplicatedCode
if self._device.device_params.get("ip"):
dev_info["IP"] = str(self._device.device_params.get("ip"))
if self._device.device_params.get("rssi"):
dev_info["RSSI"] = str(self._device.device_params.get("rssi"))
if self._device.device_params.get("ssid"):
dev_info["SSID"] = str(self._device.device_params.get("ssid"))
return dev_info
@token_exception_handler
async def async_update(self):
"""Update the entity."""
if not self._just_updated:
self._device = await self._service.update(self._device)
else:
self._just_updated = False
@callback
def async_update_callback(self, switch: Switch):
"""Update the switch's state."""
self._device = switch
async_dispatcher_send(
self.hass,
f"{CAMERA_UPDATED}-{switch.mac}",
switch,
)
self.async_schedule_update_ha_state()
# if the switch is from a camera, lets check for new events
if isinstance(switch, Camera):
if (
self._old_event_ts > 0
and self._old_event_ts != switch.last_event_ts
and switch.last_event is not None
):
event: Event = switch.last_event
# The screenshot/video urls are not always in the same positions in the lists, so we have to loop
# through them
_screenshot_url = None
_video_url = None
_ai_tag_list = []
for resource in event.file_list:
_ai_tag_list = _ai_tag_list + resource["ai_tag_list"]
if resource["type"] == 1:
_screenshot_url = resource["url"]
elif resource["type"] == 2:
_video_url = resource["url"]
_LOGGER.debug("Camera: %s has a new event", switch.nickname)
self.hass.bus.fire(
WYZE_CAMERA_EVENT,
{
"device_name": switch.nickname,
"device_mac": switch.mac,
"ai_tag_list": _ai_tag_list,
"tag_list": event.tag_list,
"event_screenshot": _screenshot_url,
"event_video": _video_url,
},
)
self._old_event_ts = switch.last_event_ts
async def async_added_to_hass(self) -> None:
"""Subscribe to update events."""
self._device.callback_function = self.async_update_callback
self._service.register_updater(self._device, 30)
await self._service.start_update_manager()
return await super().async_added_to_hass()
async def async_will_remove_from_hass(self) -> None:
"""Unregister updated on removal."""
self._service.unregister_updater(self._device)
class WyzeCameraNotificationSwitch(SwitchEntity):
"""Representation of a Wyze Camera Notification Switch."""
_available: bool
def __init__(self, service: CameraService, device: Camera) -> None:
"""Initialize a Wyze Notification Switch."""
self._service = service
self._device = device
@property
def device_info(self):
"""Return the device info."""
return {
"identifiers": {(DOMAIN, self._device.mac)},
"name": self._device.nickname,
"manufacturer": "WyzeLabs",
"model": self._device.product_model,
}
@property
def should_poll(self) -> bool:
"""No polling needed."""
return False
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on the switch."""
try:
await self._service.turn_on_notifications(self._device)
except (AccessTokenError, ParameterError, UnknownApiError) as err:
raise HomeAssistantError(f"Wyze returned an error: {err.args}") from err
except ClientConnectionError as err:
raise HomeAssistantError(err) from err
else:
self._device.notify = True
self.async_write_ha_state()
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off the switch."""
try:
await self._service.turn_off_notifications(self._device)
except (AccessTokenError, ParameterError, UnknownApiError) as err:
raise HomeAssistantError(f"Wyze returned an error: {err.args}") from err
except ClientConnectionError as err:
raise HomeAssistantError(err) from err
else:
self._device.notify = False
self.async_write_ha_state()
@property
def name(self):
"""Return the display name of this switch."""
return f"{self._device.nickname} Notifications"
@property
def available(self):
"""Return the connection status of this switch."""
return self._device.available
@property
def is_on(self):
"""Return true if switch is on."""
return self._device.notify
@property
def unique_id(self):
"""Add a unique ID to the switch."""
return f"{self._device.mac}-notification_switch"
@callback
def handle_camera_update(self, camera: Camera) -> None:
"""Update the switch whenever there is an update."""
self._device = camera
self.async_write_ha_state()
async def async_added_to_hass(self) -> None:
"""Listen for camera updates."""
self.async_on_remove(
async_dispatcher_connect(
self.hass,
f"{CAMERA_UPDATED}-{self._device.mac}",
self.handle_camera_update,
)
)
class WyzeCameraMotionSwitch(SwitchEntity):
"""Representation of a Wyze Camera Motion Detection Switch."""
_available: bool
def __init__(self, service: CameraService, device: Camera) -> None:
"""Initialize a Wyze Notification Switch."""
self._service = service
self._device = device
@property
def device_info(self):
"""Return the device info."""
return {
"identifiers": {(DOMAIN, self._device.mac)},
"name": self._device.nickname,
"manufacturer": "WyzeLabs",
"model": self._device.product_model,
}
@property
def should_poll(self) -> bool:
"""No polling needed."""
return False
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on the switch."""
try:
await self._service.turn_on_motion_detection(self._device)
except (AccessTokenError, ParameterError, UnknownApiError) as err:
raise HomeAssistantError(f"Wyze returned an error: {err.args}") from err
except ClientConnectionError as err:
raise HomeAssistantError(err) from err
else:
self._device.motion = True
self.async_write_ha_state()
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off the switch."""
try:
await self._service.turn_off_motion_detection(self._device)
except (AccessTokenError, ParameterError, UnknownApiError) as err:
raise HomeAssistantError(f"Wyze returned an error: {err.args}") from err
except ClientConnectionError as err:
raise HomeAssistantError(err) from err
else:
self._device.motion = False
self.async_write_ha_state()
@property
def name(self):
"""Return the display name of this switch."""
return f"{self._device.nickname} Motion Detection"
@property
def available(self):
"""Return the connection status of this switch."""
return self._device.available
@property
def is_on(self):
"""Return true if switch is on."""
return self._device.motion
@property
def unique_id(self):
"""Add a unique ID to the switch."""
return f"{self._device.mac}-motion_switch"
@callback
def handle_camera_update(self, camera: Camera) -> None:
"""Update the switch whenever there is an update."""
self._device = camera
self.async_write_ha_state()
async def async_added_to_hass(self) -> None:
"""Listen for camera updates."""
self.async_on_remove(
async_dispatcher_connect(
self.hass,
f"{CAMERA_UPDATED}-{self._device.mac}",
self.handle_camera_update,
)
)
class WzyeLightstripSwitch(SwitchEntity):
"""Music Mode Switch for Wyze Light Strip."""
def __init__(self, service: BulbService, device: Device) -> None:
"""Initialize a Wyze Music Mode Switch."""
self._service = service
self._device = Bulb(device.raw_dict)
@property
def device_info(self):
"""Return the device info."""
return {
"identifiers": {(DOMAIN, self._device.mac)},
"name": self._device.nickname,
"manufacturer": "WyzeLabs",
"model": self._device.product_model,
}
@property
def should_poll(self) -> bool:
"""No polling needed."""
return False
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on the switch."""
try:
await self._service.music_mode_on(self._device)
except (AccessTokenError, ParameterError, UnknownApiError) as err:
raise HomeAssistantError(f"Wyze returned an error: {err.args}") from err
except ClientConnectionError as err:
raise HomeAssistantError(err) from err
else:
self._device.music_mode = True
self.async_schedule_update_ha_state()
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off the switch."""
try:
await self._service.music_mode_off(self._device)
except (AccessTokenError, ParameterError, UnknownApiError) as err:
raise HomeAssistantError(f"Wyze returned an error: {err.args}") from err
except ClientConnectionError as err:
raise HomeAssistantError(err) from err
else:
self._device.music_mode = False
self.async_schedule_update_ha_state()
@property
def name(self):
"""Return the display name of this switch."""
return f"{self._device.nickname} Music Mode for Effects"
@property
def available(self):
"""Return the connection status of this switch."""
return self._device.available
@property
def is_on(self):
"""Return true if switch is on."""
return self._device.music_mode
@property
def unique_id(self):
"""Add a unique ID to the switch."""
return f"{self._device.mac}-music_mode"
@callback
def handle_light_update(self, bulb: Bulb) -> None:
"""Update the switch whenever there is an update."""
self._device = bulb
self.async_write_ha_state()
async def async_added_to_hass(self) -> None:
"""Listen for light updates."""
self.async_on_remove(
async_dispatcher_connect(
self.hass,
f"{LIGHT_UPDATED}-{self._device.mac}",
self.handle_light_update,
)
)
@@ -0,0 +1,52 @@
import logging
from inspect import iscoroutinefunction
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_USERNAME, CONF_PASSWORD
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from wyzeapy.exceptions import AccessTokenError, LoginError
from wyzeapy.wyze_auth_lib import Token
from .const import DOMAIN, ACCESS_TOKEN, REFRESH_TOKEN, REFRESH_TIME
_LOGGER = logging.getLogger(__name__)
class TokenManager:
hass: HomeAssistant = None
config_entry: ConfigEntry = None
def __init__(self, hass: HomeAssistant, config_entry: ConfigEntry):
TokenManager.hass = hass
TokenManager.config_entry = config_entry
@staticmethod
async def token_callback(token: Token = None):
_LOGGER.debug("TokenManager: Received new token, updating config entry.")
if TokenManager.hass.config_entries.async_entries(DOMAIN):
for entry in TokenManager.hass.config_entries.async_entries(DOMAIN):
TokenManager.hass.config_entries.async_update_entry(
entry,
data={
CONF_USERNAME: entry.data.get(CONF_USERNAME),
CONF_PASSWORD: entry.data.get(CONF_PASSWORD),
ACCESS_TOKEN: token.access_token,
REFRESH_TOKEN: token.refresh_token,
REFRESH_TIME: str(token.refresh_time),
},
)
def token_exception_handler(func):
async def inner_function(*args, **kwargs):
try:
if iscoroutinefunction(func):
await func(*args, **kwargs)
else:
func(*args, **kwargs)
except (AccessTokenError, LoginError) as err:
_LOGGER.error("TokenManager detected a login issue please re-login.")
raise ConfigEntryAuthFailed("Unable to login, please re-login.") from err
return inner_function
@@ -0,0 +1,78 @@
{
"title": "Wyze Home Assistant Integration",
"config": {
"abort": {
"already_configured": "Device is already configured",
"reauth_successful": "Reauthentication Successful"
},
"error": {
"cannot_connect": "Failed to connect",
"invalid_auth": "Invalid authentication",
"unknown": "Unexpected error"
},
"step": {
"user": {
"title": "Enter Wyze Login Credentials",
"data": {
"username": "Email",
"password": "Password",
"key_id": "Keyid",
"api_key": "Apikey"
},
"data_description": {
"username": "Wyze or 3rd party OAuth email address",
"password": "Wyze or 3rd party OAuth password"
}
},
"2fa": {
"title": "Enter Wyze 2FA Verification Code",
"data": {
"verification_code": "2FA Verification Code"
}
}
}
},
"options": {
"step": {
"init": {
"data": {
"bulb_local_control": "Use Local Control for Color Bulbs and Light Strips"
}
},
"user": {
"title": "Enter Wyze Login Credentials",
"data": {
"username": "Email",
"password": "Password",
"key_id": "Keyid",
"api_key": "Apikey"
}
},
"2fa": {
"title": "Enter Wyze 2FA Verification Code",
"data": {
"verification_code": "2FA Verification Code"
}
}
},
"error": {
"cannot_connect": "Failed to connect",
"invalid_auth": "Invalid authentication",
"unknown": "Unexpected error"
},
"abort": {
"already_configured": "Device is already configured",
"reauth_successful": "Reauthentication Successful"
}
},
"issues": {
"entity_changed": {
"title": "Wyze Entity Error",
"description": "{automation} may need attention. {entity} may have changed or been removed. This is due to a change in outdoor plug devices. Once fixed you can ignore this issue."
},
"device_changed": {
"title": "Wyze Device Error",
"description": "{automation} may need attention. A device has changed or been removed. This is due to a change in outdoor plug devices. Once fixed you can ignore this issue."
}
}
}
@@ -0,0 +1,54 @@
{
"title": "Integração Wyze para Home Assistant",
"config": {
"abort": {
"already_configured": "O dispositivo já está configurado",
"reauth_successful": "Reautenticação bem-sucedida"
},
"error": {
"cannot_connect": "Falhou ao conectar",
"invalid_auth": "Autenticação inválida",
"unknown": "Erro inesperado"
},
"step": {
"user": {
"title": "Coloque as credenciais de login do Wyze",
"data": {
"password": "Senha",
"username": "Nome de usuário"
}
},
"2fa": {
"title": "Insira o código de verificação Wyze 2FA",
"data": {
"verification_code": "Código de verificação 2FA"
}
}
}
},
"options": {
"abort": {
"already_configured": "O dispositivo já está configurado"
},
"error": {
"cannot_connect": "Falhou ao conectar",
"invalid_auth": "Autenticação inválida",
"unknown": "Erro inesperado"
},
"step": {
"user": {
"title": "Coloque as credenciais de login do Wyze",
"data": {
"password": "Senha",
"username": "Nome de usuário"
}
},
"2fa": {
"title": "Insira o código de verificação Wyze 2FA",
"data": {
"verification_code": "Código de verificação 2FA"
}
}
}
}
}
@@ -0,0 +1,54 @@
{
"title": "Integração Wyze para Home Assistant",
"config": {
"abort": {
"already_configured": "O dispositivo já está configurado",
"reauth_successful": "Reautenticação bem-sucedida"
},
"error": {
"cannot_connect": "Falhou ao conectar",
"invalid_auth": "Autenticação inválida",
"unknown": "Erro inesperado"
},
"step": {
"user": {
"title": "Coloque as credenciais de login do Wyze",
"data": {
"password": "Senha",
"username": "Nome de usuário"
}
},
"2fa": {
"title": "Insira o código de verificação Wyze 2FA",
"data": {
"verification_code": "Código de verificação 2FA"
}
}
}
},
"options": {
"abort": {
"already_configured": "O dispositivo já está configurado"
},
"error": {
"cannot_connect": "Falhou ao conectar",
"invalid_auth": "Autenticação inválida",
"unknown": "Erro inesperado"
},
"step": {
"user": {
"title": "Coloque as credenciais de login do Wyze",
"data": {
"password": "Senha",
"username": "Nome de usuário"
}
},
"2fa": {
"title": "Insira o código de verificação Wyze 2FA",
"data": {
"verification_code": "Código de verificação 2FA"
}
}
}
}
}
@@ -0,0 +1,22 @@
{
"config": {
"abort": {
"already_configured": "Устройство уже сконфигурировано"
},
"error": {
"cannot_connect": "Не могу подключиться",
"invalid_auth": "Ошибка автризации",
"unknown": "Неизвестная ошибка"
},
"step": {
"user": {
"data": {
"host": "Хост",
"password": "Пароль",
"username": "Логин"
}
}
}
},
"title": "Интеграция Wyze Home Assistant"
}
+116
View File
@@ -0,0 +1,116 @@
import binascii
from typing import Dict
from Crypto.Cipher import AES
def decrypt_ecb(key: str, data: bytes) -> bytes:
key_bytes = key.encode()
cipher = AES.new(key_bytes, AES.MODE_ECB)
decrypted_data = cipher.decrypt(data)
return decrypted_data
def encrypt_ecb(key: str, data: bytes) -> bytes:
key_bytes = key.encode()
cipher = AES.new(key_bytes, AES.MODE_ECB)
encrypted_data = cipher.encrypt(data)
return encrypted_data
def pack_l1(flags: int, seq_no: int, data: bytes):
data_crc = crc(data)
result = b"\xab"
result += flags.to_bytes(1)
result += len(data).to_bytes(2)
result += data_crc.to_bytes(2)
result += seq_no.to_bytes(2)
result += data
return result
def parse_l1(data: bytes):
if data[0] != 0xAB:
raise ValueError("Unexpected data")
flags = data[1]
length = int.from_bytes(data[2:4])
data_crc = int.from_bytes(data[4:6])
seq_no = int.from_bytes(data[6:8])
l2_content = data[8:]
if len(l2_content) > length:
l2_content = l2_content[:length]
if len(l2_content) == length and crc(l2_content) != data_crc:
raise ValueError(f"CRC Checksum failed! {data_crc} != {crc(l2_content)}")
return l2_content, flags, seq_no, length - len(l2_content)
def pack_l2_dict(cmd: int, flags: int, content: Dict[int, bytes]):
result = cmd.to_bytes(1)
result += flags.to_bytes(1)
for k, v in content.items():
result += k.to_bytes(1)
result += len(v).to_bytes(2)
result += v
return result
def parse_l2_dict(data: bytes):
result_dict: Dict[int, bytes] = {}
cmd = data[0]
flags = data[1]
cur = 2
while cur < len(data):
key = data[cur]
length = int.from_bytes(data[cur + 1 : cur + 3])
value = data[cur + 3 : cur + 3 + length]
result_dict[key] = value
cur += 3 + length
return cmd, flags, result_dict
def pack_l2_lock_unlock(ble_id: int, ble_token: str, challenge: bytes, command):
if command == "unlock":
magic_bytes = binascii.unhexlify("01000000000000000000006C6F6F636B")
elif command == "lock":
magic_bytes = binascii.unhexlify("02000000000000000000006C6F6F636B")
else:
raise ValueError(f"Only accept `lock` or `unlock`, but got `{command}`")
encrypted_challenge = encrypt_ecb(ble_token[16:], challenge)
encrypted_challenge = b"".join(
(x ^ y).to_bytes(1) for x, y in zip(encrypted_challenge, magic_bytes)
)
result = (0x0400050002).to_bytes(5)
result += ble_id.to_bytes(2)
result += (0x040010).to_bytes(3)
result += encrypted_challenge
result += (0xAD000100F4000101F7000101).to_bytes(12)
return result
def crc(data):
magic = (
"0000c0c1c1810140c30103c00280c241c60106c00780c7410500c5c1c4810440"
"cc010cc00d80cd410f00cfc1ce810e400a00cac1cb810b40c90109c00880c841"
"d80118c01980d9411b00dbc1da811a401e00dec1df811f40dd011dc01c80dc41"
"1400d4c1d5811540d70117c01680d641d20112c01380d3411100d1c1d0811040"
"f00130c03180f1413300f3c1f28132403600f6c1f7813740f50135c03480f441"
"3c00fcc1fd813d40ff013fc03e80fe41fa013ac03b80fb413900f9c1f8813840"
"2800e8c1e9812940eb012bc02a80ea41ee012ec02f80ef412d00edc1ec812c40"
"e40124c02580e5412700e7c1e68126402200e2c1e3812340e10121c02080e041"
"a00160c06180a1416300a3c1a28162406600a6c1a7816740a50165c06480a441"
"6c00acc1ad816d40af016fc06e80ae41aa016ac06b80ab416900a9c1a8816840"
"7800b8c1b9817940bb017bc07a80ba41be017ec07f80bf417d00bdc1bc817c40"
"b40174c07580b5417700b7c1b68176407200b2c1b3817340b10171c07080b041"
"500090c191815140930153c052809241960156c057809741550095c194815440"
"9c015cc05d809d415f009fc19e815e405a009ac19b815b40990159c058809841"
"880148c0498089414b008bc18a814a404e008ec18f814f408d014dc04c808c41"
"440084c185814540870147c046808641820142c043808341410081c180814040"
)
magic = [
int.from_bytes(binascii.unhexlify(magic[x : x + 4]))
for x in range(0, len(magic), 4)
]
result = 0
for b in data:
result = magic[(result ^ (b & 255)) & 255] ^ (result >> 8)
return 65535 & result