Small typing cleanups for HomeKit (#101790)
parent
bd38fd9516
commit
39fd5897cb
|
@ -140,6 +140,6 @@ class AccessoryAidStorage:
|
||||||
return await self.store.async_save(self._data_to_save())
|
return await self.store.async_save(self._data_to_save())
|
||||||
|
|
||||||
@callback
|
@callback
|
||||||
def _data_to_save(self) -> dict:
|
def _data_to_save(self) -> dict[str, dict[str, int]]:
|
||||||
"""Return data of entity map to store in a file."""
|
"""Return data of entity map to store in a file."""
|
||||||
return {ALLOCATIONS_KEY: self.allocations}
|
return {ALLOCATIONS_KEY: self.allocations}
|
||||||
|
|
|
@ -257,7 +257,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
|
|
||||||
async def async_step_accessory(self, accessory_input: dict) -> FlowResult:
|
async def async_step_accessory(self, accessory_input: dict[str, Any]) -> FlowResult:
|
||||||
"""Handle creation a single accessory in accessory mode."""
|
"""Handle creation a single accessory in accessory mode."""
|
||||||
entity_id = accessory_input[CONF_ENTITY_ID]
|
entity_id = accessory_input[CONF_ENTITY_ID]
|
||||||
port = accessory_input[CONF_PORT]
|
port = accessory_input[CONF_PORT]
|
||||||
|
@ -283,7 +283,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||||
title=f"{name}:{entry_data[CONF_PORT]}", data=entry_data
|
title=f"{name}:{entry_data[CONF_PORT]}", data=entry_data
|
||||||
)
|
)
|
||||||
|
|
||||||
async def async_step_import(self, user_input: dict) -> FlowResult:
|
async def async_step_import(self, user_input: dict[str, Any]) -> FlowResult:
|
||||||
"""Handle import from yaml."""
|
"""Handle import from yaml."""
|
||||||
if not self._async_is_unique_name_port(user_input):
|
if not self._async_is_unique_name_port(user_input):
|
||||||
return self.async_abort(reason="port_name_in_use")
|
return self.async_abort(reason="port_name_in_use")
|
||||||
|
@ -318,7 +318,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||||
return suggested_name
|
return suggested_name
|
||||||
|
|
||||||
@callback
|
@callback
|
||||||
def _async_is_unique_name_port(self, user_input: dict[str, str]) -> bool:
|
def _async_is_unique_name_port(self, user_input: dict[str, Any]) -> bool:
|
||||||
"""Determine is a name or port is already used."""
|
"""Determine is a name or port is already used."""
|
||||||
name = user_input[CONF_NAME]
|
name = user_input[CONF_NAME]
|
||||||
port = user_input[CONF_PORT]
|
port = user_input[CONF_PORT]
|
||||||
|
|
|
@ -1,5 +1,6 @@
|
||||||
"""Class to hold all cover accessories."""
|
"""Class to hold all cover accessories."""
|
||||||
import logging
|
import logging
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
from pyhap.const import (
|
from pyhap.const import (
|
||||||
CATEGORY_DOOR,
|
CATEGORY_DOOR,
|
||||||
|
@ -7,6 +8,7 @@ from pyhap.const import (
|
||||||
CATEGORY_WINDOW,
|
CATEGORY_WINDOW,
|
||||||
CATEGORY_WINDOW_COVERING,
|
CATEGORY_WINDOW_COVERING,
|
||||||
)
|
)
|
||||||
|
from pyhap.service import Service
|
||||||
|
|
||||||
from homeassistant.components.cover import (
|
from homeassistant.components.cover import (
|
||||||
ATTR_CURRENT_POSITION,
|
ATTR_CURRENT_POSITION,
|
||||||
|
@ -98,7 +100,7 @@ class GarageDoorOpener(HomeAccessory):
|
||||||
and support no more than open, close, and stop.
|
and support no more than open, close, and stop.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(self, *args):
|
def __init__(self, *args: Any) -> None:
|
||||||
"""Initialize a GarageDoorOpener accessory object."""
|
"""Initialize a GarageDoorOpener accessory object."""
|
||||||
super().__init__(*args, category=CATEGORY_GARAGE_DOOR_OPENER)
|
super().__init__(*args, category=CATEGORY_GARAGE_DOOR_OPENER)
|
||||||
state = self.hass.states.get(self.entity_id)
|
state = self.hass.states.get(self.entity_id)
|
||||||
|
@ -203,12 +205,12 @@ class OpeningDeviceBase(HomeAccessory):
|
||||||
WindowCovering
|
WindowCovering
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(self, *args, category, service):
|
def __init__(self, *args: Any, category: int, service: Service) -> None:
|
||||||
"""Initialize a OpeningDeviceBase accessory object."""
|
"""Initialize a OpeningDeviceBase accessory object."""
|
||||||
super().__init__(*args, category=category)
|
super().__init__(*args, category=category)
|
||||||
state = self.hass.states.get(self.entity_id)
|
state = self.hass.states.get(self.entity_id)
|
||||||
|
assert state
|
||||||
self.features = state.attributes.get(ATTR_SUPPORTED_FEATURES, 0)
|
self.features: int = state.attributes.get(ATTR_SUPPORTED_FEATURES, 0)
|
||||||
self._supports_stop = self.features & CoverEntityFeature.STOP
|
self._supports_stop = self.features & CoverEntityFeature.STOP
|
||||||
self.chars = []
|
self.chars = []
|
||||||
if self._supports_stop:
|
if self._supports_stop:
|
||||||
|
@ -276,14 +278,15 @@ class OpeningDevice(OpeningDeviceBase, HomeAccessory):
|
||||||
The cover entity must support: set_cover_position.
|
The cover entity must support: set_cover_position.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(self, *args, category, service):
|
def __init__(self, *args: Any, category: int, service: Service) -> None:
|
||||||
"""Initialize a WindowCovering accessory object."""
|
"""Initialize a WindowCovering accessory object."""
|
||||||
super().__init__(*args, category=category, service=service)
|
super().__init__(*args, category=category, service=service)
|
||||||
state = self.hass.states.get(self.entity_id)
|
state = self.hass.states.get(self.entity_id)
|
||||||
|
assert state
|
||||||
self.char_current_position = self.serv_cover.configure_char(
|
self.char_current_position = self.serv_cover.configure_char(
|
||||||
CHAR_CURRENT_POSITION, value=0
|
CHAR_CURRENT_POSITION, value=0
|
||||||
)
|
)
|
||||||
target_args = {"value": 0}
|
target_args: dict[str, Any] = {"value": 0}
|
||||||
if self.features & CoverEntityFeature.SET_POSITION:
|
if self.features & CoverEntityFeature.SET_POSITION:
|
||||||
target_args["setter_callback"] = self.move_cover
|
target_args["setter_callback"] = self.move_cover
|
||||||
else:
|
else:
|
||||||
|
@ -307,7 +310,7 @@ class OpeningDevice(OpeningDeviceBase, HomeAccessory):
|
||||||
)
|
)
|
||||||
self.async_update_state(state)
|
self.async_update_state(state)
|
||||||
|
|
||||||
def move_cover(self, value):
|
def move_cover(self, value: int) -> None:
|
||||||
"""Move cover to value if call came from HomeKit."""
|
"""Move cover to value if call came from HomeKit."""
|
||||||
_LOGGER.debug("%s: Set position to %d", self.entity_id, value)
|
_LOGGER.debug("%s: Set position to %d", self.entity_id, value)
|
||||||
params = {ATTR_ENTITY_ID: self.entity_id, ATTR_POSITION: value}
|
params = {ATTR_ENTITY_ID: self.entity_id, ATTR_POSITION: value}
|
||||||
|
@ -338,7 +341,7 @@ class Door(OpeningDevice):
|
||||||
The entity must support: set_cover_position.
|
The entity must support: set_cover_position.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(self, *args):
|
def __init__(self, *args: Any) -> None:
|
||||||
"""Initialize a Door accessory object."""
|
"""Initialize a Door accessory object."""
|
||||||
super().__init__(*args, category=CATEGORY_DOOR, service=SERV_DOOR)
|
super().__init__(*args, category=CATEGORY_DOOR, service=SERV_DOOR)
|
||||||
|
|
||||||
|
@ -350,7 +353,7 @@ class Window(OpeningDevice):
|
||||||
The entity must support: set_cover_position.
|
The entity must support: set_cover_position.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(self, *args):
|
def __init__(self, *args: Any) -> None:
|
||||||
"""Initialize a Window accessory object."""
|
"""Initialize a Window accessory object."""
|
||||||
super().__init__(*args, category=CATEGORY_WINDOW, service=SERV_WINDOW)
|
super().__init__(*args, category=CATEGORY_WINDOW, service=SERV_WINDOW)
|
||||||
|
|
||||||
|
@ -362,7 +365,7 @@ class WindowCovering(OpeningDevice):
|
||||||
The entity must support: set_cover_position.
|
The entity must support: set_cover_position.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(self, *args):
|
def __init__(self, *args: Any) -> None:
|
||||||
"""Initialize a WindowCovering accessory object."""
|
"""Initialize a WindowCovering accessory object."""
|
||||||
super().__init__(
|
super().__init__(
|
||||||
*args, category=CATEGORY_WINDOW_COVERING, service=SERV_WINDOW_COVERING
|
*args, category=CATEGORY_WINDOW_COVERING, service=SERV_WINDOW_COVERING
|
||||||
|
@ -377,12 +380,13 @@ class WindowCoveringBasic(OpeningDeviceBase, HomeAccessory):
|
||||||
stop_cover (optional).
|
stop_cover (optional).
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(self, *args):
|
def __init__(self, *args: Any) -> None:
|
||||||
"""Initialize a WindowCoveringBasic accessory object."""
|
"""Initialize a WindowCoveringBasic accessory object."""
|
||||||
super().__init__(
|
super().__init__(
|
||||||
*args, category=CATEGORY_WINDOW_COVERING, service=SERV_WINDOW_COVERING
|
*args, category=CATEGORY_WINDOW_COVERING, service=SERV_WINDOW_COVERING
|
||||||
)
|
)
|
||||||
state = self.hass.states.get(self.entity_id)
|
state = self.hass.states.get(self.entity_id)
|
||||||
|
assert state
|
||||||
self.char_current_position = self.serv_cover.configure_char(
|
self.char_current_position = self.serv_cover.configure_char(
|
||||||
CHAR_CURRENT_POSITION, value=0
|
CHAR_CURRENT_POSITION, value=0
|
||||||
)
|
)
|
||||||
|
@ -394,7 +398,7 @@ class WindowCoveringBasic(OpeningDeviceBase, HomeAccessory):
|
||||||
)
|
)
|
||||||
self.async_update_state(state)
|
self.async_update_state(state)
|
||||||
|
|
||||||
def move_cover(self, value):
|
def move_cover(self, value: int) -> None:
|
||||||
"""Move cover to value if call came from HomeKit."""
|
"""Move cover to value if call came from HomeKit."""
|
||||||
_LOGGER.debug("%s: Set position to %d", self.entity_id, value)
|
_LOGGER.debug("%s: Set position to %d", self.entity_id, value)
|
||||||
|
|
||||||
|
@ -436,7 +440,7 @@ class WindowCoveringBasic(OpeningDeviceBase, HomeAccessory):
|
||||||
super().async_update_state(new_state)
|
super().async_update_state(new_state)
|
||||||
|
|
||||||
|
|
||||||
def _hass_state_to_position_start(state):
|
def _hass_state_to_position_start(state: str) -> int:
|
||||||
"""Convert hass state to homekit position state."""
|
"""Convert hass state to homekit position state."""
|
||||||
if state == STATE_OPENING:
|
if state == STATE_OPENING:
|
||||||
return HK_POSITION_GOING_TO_MAX
|
return HK_POSITION_GOING_TO_MAX
|
||||||
|
|
|
@ -1,5 +1,6 @@
|
||||||
"""Class to hold all light accessories."""
|
"""Class to hold all fan accessories."""
|
||||||
import logging
|
import logging
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
from pyhap.const import CATEGORY_FAN
|
from pyhap.const import CATEGORY_FAN
|
||||||
|
|
||||||
|
@ -27,7 +28,7 @@ from homeassistant.const import (
|
||||||
STATE_OFF,
|
STATE_OFF,
|
||||||
STATE_ON,
|
STATE_ON,
|
||||||
)
|
)
|
||||||
from homeassistant.core import callback
|
from homeassistant.core import State, callback
|
||||||
|
|
||||||
from .accessories import TYPES, HomeAccessory
|
from .accessories import TYPES, HomeAccessory
|
||||||
from .const import (
|
from .const import (
|
||||||
|
@ -54,12 +55,12 @@ class Fan(HomeAccessory):
|
||||||
Currently supports: state, speed, oscillate, direction.
|
Currently supports: state, speed, oscillate, direction.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(self, *args):
|
def __init__(self, *args: Any) -> None:
|
||||||
"""Initialize a new Fan accessory object."""
|
"""Initialize a new Fan accessory object."""
|
||||||
super().__init__(*args, category=CATEGORY_FAN)
|
super().__init__(*args, category=CATEGORY_FAN)
|
||||||
self.chars = []
|
self.chars: list[str] = []
|
||||||
state = self.hass.states.get(self.entity_id)
|
state = self.hass.states.get(self.entity_id)
|
||||||
|
assert state
|
||||||
self._reload_on_change_attrs.extend(
|
self._reload_on_change_attrs.extend(
|
||||||
(
|
(
|
||||||
ATTR_PERCENTAGE_STEP,
|
ATTR_PERCENTAGE_STEP,
|
||||||
|
@ -69,7 +70,7 @@ class Fan(HomeAccessory):
|
||||||
|
|
||||||
features = state.attributes.get(ATTR_SUPPORTED_FEATURES, 0)
|
features = state.attributes.get(ATTR_SUPPORTED_FEATURES, 0)
|
||||||
percentage_step = state.attributes.get(ATTR_PERCENTAGE_STEP, 1)
|
percentage_step = state.attributes.get(ATTR_PERCENTAGE_STEP, 1)
|
||||||
self.preset_modes = state.attributes.get(ATTR_PRESET_MODES)
|
self.preset_modes: list[str] | None = state.attributes.get(ATTR_PRESET_MODES)
|
||||||
|
|
||||||
if features & FanEntityFeature.DIRECTION:
|
if features & FanEntityFeature.DIRECTION:
|
||||||
self.chars.append(CHAR_ROTATION_DIRECTION)
|
self.chars.append(CHAR_ROTATION_DIRECTION)
|
||||||
|
@ -136,7 +137,7 @@ class Fan(HomeAccessory):
|
||||||
self.async_update_state(state)
|
self.async_update_state(state)
|
||||||
serv_fan.setter_callback = self._set_chars
|
serv_fan.setter_callback = self._set_chars
|
||||||
|
|
||||||
def _set_chars(self, char_values):
|
def _set_chars(self, char_values: dict[str, Any]) -> None:
|
||||||
_LOGGER.debug("Fan _set_chars: %s", char_values)
|
_LOGGER.debug("Fan _set_chars: %s", char_values)
|
||||||
if CHAR_ACTIVE in char_values:
|
if CHAR_ACTIVE in char_values:
|
||||||
if char_values[CHAR_ACTIVE]:
|
if char_values[CHAR_ACTIVE]:
|
||||||
|
@ -167,23 +168,23 @@ class Fan(HomeAccessory):
|
||||||
if CHAR_TARGET_FAN_STATE in char_values:
|
if CHAR_TARGET_FAN_STATE in char_values:
|
||||||
self.set_single_preset_mode(char_values[CHAR_TARGET_FAN_STATE])
|
self.set_single_preset_mode(char_values[CHAR_TARGET_FAN_STATE])
|
||||||
|
|
||||||
def set_single_preset_mode(self, value):
|
def set_single_preset_mode(self, value: int) -> None:
|
||||||
"""Set auto call came from HomeKit."""
|
"""Set auto call came from HomeKit."""
|
||||||
params = {ATTR_ENTITY_ID: self.entity_id}
|
params: dict[str, Any] = {ATTR_ENTITY_ID: self.entity_id}
|
||||||
if value:
|
if value:
|
||||||
|
assert self.preset_modes
|
||||||
_LOGGER.debug(
|
_LOGGER.debug(
|
||||||
"%s: Set auto to 1 (%s)", self.entity_id, self.preset_modes[0]
|
"%s: Set auto to 1 (%s)", self.entity_id, self.preset_modes[0]
|
||||||
)
|
)
|
||||||
params[ATTR_PRESET_MODE] = self.preset_modes[0]
|
params[ATTR_PRESET_MODE] = self.preset_modes[0]
|
||||||
self.async_call_service(DOMAIN, SERVICE_SET_PRESET_MODE, params)
|
self.async_call_service(DOMAIN, SERVICE_SET_PRESET_MODE, params)
|
||||||
else:
|
elif current_state := self.hass.states.get(self.entity_id):
|
||||||
current_state = self.hass.states.get(self.entity_id)
|
percentage: float = current_state.attributes.get(ATTR_PERCENTAGE) or 50.0
|
||||||
percentage = current_state.attributes.get(ATTR_PERCENTAGE) or 50
|
|
||||||
params[ATTR_PERCENTAGE] = percentage
|
params[ATTR_PERCENTAGE] = percentage
|
||||||
_LOGGER.debug("%s: Set auto to 0", self.entity_id)
|
_LOGGER.debug("%s: Set auto to 0", self.entity_id)
|
||||||
self.async_call_service(DOMAIN, SERVICE_TURN_ON, params)
|
self.async_call_service(DOMAIN, SERVICE_TURN_ON, params)
|
||||||
|
|
||||||
def set_preset_mode(self, value, preset_mode):
|
def set_preset_mode(self, value: int, preset_mode: str) -> None:
|
||||||
"""Set preset_mode if call came from HomeKit."""
|
"""Set preset_mode if call came from HomeKit."""
|
||||||
_LOGGER.debug(
|
_LOGGER.debug(
|
||||||
"%s: Set preset_mode %s to %d", self.entity_id, preset_mode, value
|
"%s: Set preset_mode %s to %d", self.entity_id, preset_mode, value
|
||||||
|
@ -195,35 +196,35 @@ class Fan(HomeAccessory):
|
||||||
else:
|
else:
|
||||||
self.async_call_service(DOMAIN, SERVICE_TURN_ON, params)
|
self.async_call_service(DOMAIN, SERVICE_TURN_ON, params)
|
||||||
|
|
||||||
def set_state(self, value):
|
def set_state(self, value: int) -> None:
|
||||||
"""Set state if call came from HomeKit."""
|
"""Set state if call came from HomeKit."""
|
||||||
_LOGGER.debug("%s: Set state to %d", self.entity_id, value)
|
_LOGGER.debug("%s: Set state to %d", self.entity_id, value)
|
||||||
service = SERVICE_TURN_ON if value == 1 else SERVICE_TURN_OFF
|
service = SERVICE_TURN_ON if value == 1 else SERVICE_TURN_OFF
|
||||||
params = {ATTR_ENTITY_ID: self.entity_id}
|
params = {ATTR_ENTITY_ID: self.entity_id}
|
||||||
self.async_call_service(DOMAIN, service, params)
|
self.async_call_service(DOMAIN, service, params)
|
||||||
|
|
||||||
def set_direction(self, value):
|
def set_direction(self, value: int) -> None:
|
||||||
"""Set state if call came from HomeKit."""
|
"""Set state if call came from HomeKit."""
|
||||||
_LOGGER.debug("%s: Set direction to %d", self.entity_id, value)
|
_LOGGER.debug("%s: Set direction to %d", self.entity_id, value)
|
||||||
direction = DIRECTION_REVERSE if value == 1 else DIRECTION_FORWARD
|
direction = DIRECTION_REVERSE if value == 1 else DIRECTION_FORWARD
|
||||||
params = {ATTR_ENTITY_ID: self.entity_id, ATTR_DIRECTION: direction}
|
params = {ATTR_ENTITY_ID: self.entity_id, ATTR_DIRECTION: direction}
|
||||||
self.async_call_service(DOMAIN, SERVICE_SET_DIRECTION, params, direction)
|
self.async_call_service(DOMAIN, SERVICE_SET_DIRECTION, params, direction)
|
||||||
|
|
||||||
def set_oscillating(self, value):
|
def set_oscillating(self, value: int) -> None:
|
||||||
"""Set state if call came from HomeKit."""
|
"""Set state if call came from HomeKit."""
|
||||||
_LOGGER.debug("%s: Set oscillating to %d", self.entity_id, value)
|
_LOGGER.debug("%s: Set oscillating to %d", self.entity_id, value)
|
||||||
oscillating = value == 1
|
oscillating = value == 1
|
||||||
params = {ATTR_ENTITY_ID: self.entity_id, ATTR_OSCILLATING: oscillating}
|
params = {ATTR_ENTITY_ID: self.entity_id, ATTR_OSCILLATING: oscillating}
|
||||||
self.async_call_service(DOMAIN, SERVICE_OSCILLATE, params, oscillating)
|
self.async_call_service(DOMAIN, SERVICE_OSCILLATE, params, oscillating)
|
||||||
|
|
||||||
def set_percentage(self, value):
|
def set_percentage(self, value: float) -> None:
|
||||||
"""Set state if call came from HomeKit."""
|
"""Set state if call came from HomeKit."""
|
||||||
_LOGGER.debug("%s: Set speed to %d", self.entity_id, value)
|
_LOGGER.debug("%s: Set speed to %d", self.entity_id, value)
|
||||||
params = {ATTR_ENTITY_ID: self.entity_id, ATTR_PERCENTAGE: value}
|
params = {ATTR_ENTITY_ID: self.entity_id, ATTR_PERCENTAGE: value}
|
||||||
self.async_call_service(DOMAIN, SERVICE_SET_PERCENTAGE, params, value)
|
self.async_call_service(DOMAIN, SERVICE_SET_PERCENTAGE, params, value)
|
||||||
|
|
||||||
@callback
|
@callback
|
||||||
def async_update_state(self, new_state):
|
def async_update_state(self, new_state: State) -> None:
|
||||||
"""Update fan after state change."""
|
"""Update fan after state change."""
|
||||||
# Handle State
|
# Handle State
|
||||||
state = new_state.state
|
state = new_state.state
|
||||||
|
|
Loading…
Reference in New Issue