Skip to content

Commit

Permalink
Merge pull request #649 from SecKatie/fix-to-be-removed-2025.11-12
Browse files Browse the repository at this point in the history
Fix-to-be-removed-2025.11-12
  • Loading branch information
SecKatie authored Dec 8, 2024
2 parents 350a39d + 634b41a commit 629b808
Show file tree
Hide file tree
Showing 2 changed files with 45 additions and 50 deletions.
82 changes: 41 additions & 41 deletions custom_components/wyzeapi/alarm_control_panel.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,10 +4,11 @@

import logging
from datetime import timedelta
from typing import Optional, Callable, List, Any
from typing import Optional, Callable, List, Any, cast
from aiohttp.client_exceptions import ClientConnectionError

from homeassistant.components.alarm_control_panel import (
AlarmControlPanelState,
AlarmControlPanelEntity,
AlarmControlPanelEntityFeature,
)
Expand All @@ -19,6 +20,7 @@
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

Expand Down Expand Up @@ -51,23 +53,39 @@ 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
_state = "disarmed"
_server_out_of_sync = False

_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

@property
def state(self):
return self._state
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=None) -> None:
async def async_alarm_disarm(self, code: Optional[str] = None) -> None:
"""Send disarm command."""
try:
await self._hms_service.set_mode(HMSMode.DISARMED)
Expand All @@ -80,7 +98,7 @@ async def async_alarm_disarm(self, code=None) -> None:
self._server_out_of_sync = True

@token_exception_handler
async def async_alarm_arm_home(self, code=None):
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:
Expand All @@ -92,7 +110,7 @@ async def async_alarm_arm_home(self, code=None):
self._server_out_of_sync = True

@token_exception_handler
async def async_alarm_arm_away(self, code=None):
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:
Expand All @@ -103,61 +121,43 @@ async def async_alarm_arm_away(self, code=None):
self._state = "armed_away"
self._server_out_of_sync = True

def alarm_arm_night(self, code=None):
raise NotImplementedError

def alarm_trigger(self, code=None):
raise NotImplementedError

def alarm_arm_custom_bypass(self, code=None):
raise NotImplementedError

@property
def supported_features(self) -> int:
return AlarmControlPanelEntityFeature.ARM_HOME | AlarmControlPanelEntityFeature.ARM_AWAY

@property
def device_info(self):
return {
"identifiers": {
def device_info(self) -> DeviceInfo:
return DeviceInfo(
identifiers={
(DOMAIN, self.unique_id)
},
"name": self.NAME,
"manufacturer": "WyzeLabs",
"model": self.DEVICE_MODEL
}

@property
def name(self) -> str:
return self.NAME

@property
def unique_id(self):
return self._hms_service.hms_id
name=self.NAME,
manufacturer=self.MANUFACTURER,
model=self.DEVICE_MODEL,
)

@property
def extra_state_attributes(self):
def extra_state_attributes(self) -> dict | None:
"""Return device attributes of the entity."""
return {
ATTR_ATTRIBUTION: ATTRIBUTION,
"device model": self.DEVICE_MODEL,
"mac": self.unique_id
}

@token_exception_handler
async def async_update(self):
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 = "disarmed"
self._state = AlarmControlPanelState.DISARMED
elif state is HMSMode.AWAY:
self._state = "armed_away"
self._state = AlarmControlPanelState.ARMED_AWAY
elif state is HMSMode.HOME:
self._state = "armed_home"
self._state = AlarmControlPanelState.ARMED_HOME
elif state is HMSMode.CHANGING:
self._state = "disarmed"
self._state = AlarmControlPanelState.DISARMED
else:
_LOGGER.warning(f"Received {state} from server")

Expand Down
13 changes: 4 additions & 9 deletions custom_components/wyzeapi/config_flow.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
from __future__ import annotations

import logging
from typing import Any, Dict
from typing import Any, Optional

import voluptuous as vol
from homeassistant import config_entries
Expand Down Expand Up @@ -45,8 +45,8 @@ async def get_client(self):
self.client = await Wyzeapy.create()

async def async_step_user(
self, user_input: Dict[str, Any] = None
) -> Dict[str, Any]:
self, user_input: Optional[dict[str, any]] = None
) -> dict[str, Any]:
"""Handle the initial step."""
await self.get_client()

Expand Down Expand Up @@ -88,7 +88,7 @@ async def async_step_user(
step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
)

async def async_step_2fa(self, user_input: Dict[str, Any] = None) -> Dict[str, Any]:
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)

Expand Down Expand Up @@ -138,11 +138,6 @@ def async_get_options_flow(config_entry):

class OptionsFlowHandler(config_entries.OptionsFlow):
"""Handle an option flow for Wyze."""

def __init__(self, config_entry: config_entries.ConfigEntry) -> None:
"""Initialize options flow."""
self.config_entry = config_entry

async def async_step_init(self, user_input=None):
"""Handle options flow."""
if user_input is not None:
Expand Down

0 comments on commit 629b808

Please sign in to comment.