jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-23 13:17:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add more typing to HomeKit (#101896)

Browse Source
This commit is contained in:
J. Nick Koston 2023-10-12 08:43:53 -10:00 committed by GitHub
parent c4ce900567
commit cc3d1a11bd
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 90 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
homeassistant/components/homekit/iidmanager.py
View File

@ -36,7 +36,7 @@ class IIDStorage(Store):
old_major_version: int,
old_minor_version: int,
old_data: dict,
):
) -> dict:
"""Migrate to the new version."""
if old_major_version == 1:
# Convert v1 to v2 format which uses a unique iid set per accessory

51
homeassistant/components/homekit/type_cameras.py
View File

@ -2,6 +2,7 @@
import asyncio
from datetime import timedelta
import logging
from typing import Any
from haffmpeg.core import FFMPEG_STDERR, HAFFmpeg
from pyhap.camera import (
@ -14,7 +15,7 @@ from pyhap.const import CATEGORY_CAMERA
from homeassistant.components import camera
from homeassistant.components.ffmpeg import get_ffmpeg_manager
from homeassistant.const import STATE_ON
from homeassistant.core import State, callback
from homeassistant.core import HomeAssistant, State, callback
from homeassistant.helpers.event import (
EventStateChangedData,
async_track_state_change_event,
@ -22,7 +23,7 @@ from homeassistant.helpers.event import (
)
from homeassistant.helpers.typing import EventType
from .accessories import TYPES, HomeAccessory
from .accessories import TYPES, HomeAccessory, HomeDriver
from .const import (
CHAR_MOTION_DETECTED,
CHAR_MUTE,
@ -141,7 +142,15 @@ CONFIG_DEFAULTS = {
class Camera(HomeAccessory, PyhapCamera):
"""Generate a Camera accessory."""
def __init__(self, hass, driver, name, entity_id, aid, config):
def __init__(
self,
hass: HomeAssistant,
driver: HomeDriver,
name: str,
entity_id: str,
aid: int,
config: dict[str, Any],
) -> None:
"""Initialize a Camera accessory object."""
self._ffmpeg = get_ffmpeg_manager(hass)
for config_key, conf in CONFIG_DEFAULTS.items():
@ -242,12 +251,13 @@ class Camera(HomeAccessory, PyhapCamera):
self._async_update_doorbell_state(state)
async def run(self):
async def run(self) -> None:
"""Handle accessory driver started event.
Run inside the Home Assistant event loop.
"""
if self._char_motion_detected:
assert self.linked_motion_sensor
self._subscriptions.append(
async_track_state_change_event(
self.hass,
@ -257,6 +267,7 @@ class Camera(HomeAccessory, PyhapCamera):
)
if self._char_doorbell_detected:
assert self.linked_doorbell_sensor
self._subscriptions.append(
async_track_state_change_event(
self.hass,
@ -282,6 +293,7 @@ class Camera(HomeAccessory, PyhapCamera):
return
detected = new_state.state == STATE_ON
assert self._char_motion_detected
if self._char_motion_detected.value == detected:
return
@ -307,6 +319,8 @@ class Camera(HomeAccessory, PyhapCamera):
if not new_state:
return
assert self._char_doorbell_detected
assert self._char_doorbell_detected_switch
if new_state.state == STATE_ON:
self._char_doorbell_detected.set_value(DOORBELL_SINGLE_PRESS)
self._char_doorbell_detected_switch.set_value(DOORBELL_SINGLE_PRESS)
@ -318,11 +332,10 @@ class Camera(HomeAccessory, PyhapCamera):
)
@callback
def async_update_state(self, new_state):
def async_update_state(self, new_state: State | None) -> None:
"""Handle state change to update HomeKit value."""
pass # pylint: disable=unnecessary-pass
async def _async_get_stream_source(self):
async def _async_get_stream_source(self) -> str | None:
"""Find the camera stream source url."""
if stream_source := self.config.get(CONF_STREAM_SOURCE):
return stream_source
@ -337,7 +350,9 @@ class Camera(HomeAccessory, PyhapCamera):
)
return stream_source
async def start_stream(self, session_info, stream_config):
async def start_stream(
self, session_info: dict[str, Any], stream_config: dict[str, Any]
) -> bool:
"""Start a new stream with the given configuration."""
_LOGGER.debug(
"[%s] Starting stream with the following parameters: %s",
@ -418,7 +433,9 @@ class Camera(HomeAccessory, PyhapCamera):
return await self._async_ffmpeg_watch(session_info["id"])
async def _async_log_stderr_stream(self, stderr_reader):
async def _async_log_stderr_stream(
self, stderr_reader: asyncio.StreamReader
) -> None:
"""Log output from ffmpeg."""
_LOGGER.debug("%s: ffmpeg: started", self.display_name)
while True:
@ -428,7 +445,7 @@ class Camera(HomeAccessory, PyhapCamera):
_LOGGER.debug("%s: ffmpeg: %s", self.display_name, line.rstrip())
async def _async_ffmpeg_watch(self, session_id):
async def _async_ffmpeg_watch(self, session_id: str) -> bool:
"""Check to make sure ffmpeg is still running and cleanup if not."""
ffmpeg_pid = self.sessions[session_id][FFMPEG_PID]
if pid_is_alive(ffmpeg_pid):
@ -440,7 +457,7 @@ class Camera(HomeAccessory, PyhapCamera):
return False
@callback
def _async_stop_ffmpeg_watch(self, session_id):
def _async_stop_ffmpeg_watch(self, session_id: str) -> None:
"""Cleanup a streaming session after stopping."""
if FFMPEG_WATCHER not in self.sessions[session_id]:
return
@ -448,7 +465,7 @@ class Camera(HomeAccessory, PyhapCamera):
self.sessions[session_id].pop(FFMPEG_LOGGER).cancel()
@callback
def async_stop(self):
def async_stop(self) -> None:
"""Stop any streams when the accessory is stopped."""
for session_info in self.sessions.values():
self.hass.async_create_background_task(
@ -456,7 +473,7 @@ class Camera(HomeAccessory, PyhapCamera):
)
super().async_stop()
async def stop_stream(self, session_info):
async def stop_stream(self, session_info: dict[str, Any]) -> None:
"""Stop the stream for the given ``session_id``."""
session_id = session_info["id"]
if not (stream := session_info.get("stream")):
@ -467,7 +484,7 @@ class Camera(HomeAccessory, PyhapCamera):
if not pid_is_alive(stream.process.pid):
_LOGGER.info("[%s] Stream already stopped", session_id)
return True
return
for shutdown_method in ("close", "kill"):
_LOGGER.info("[%s] %s stream", session_id, shutdown_method)
@ -479,11 +496,13 @@ class Camera(HomeAccessory, PyhapCamera):
"[%s] Failed to %s stream", session_id, shutdown_method
)
async def reconfigure_stream(self, session_info, stream_config):
async def reconfigure_stream(
self, session_info: dict[str, Any], stream_config: dict[str, Any]
) -> bool:
"""Reconfigure the stream so that it uses the given ``stream_config``."""
return True
async def async_get_snapshot(self, image_size):
async def async_get_snapshot(self, image_size: dict[str, int]) -> bytes:
"""Return a jpeg of a snapshot from the camera."""
image = await camera.async_get_image(
self.hass,

19
homeassistant/components/homekit/type_humidifiers.py
View File

@ -1,5 +1,6 @@
"""Class to hold all thermostat accessories."""
import logging
from typing import Any
from pyhap.const import CATEGORY_HUMIDIFIER
@ -73,7 +74,7 @@ HC_STATE_DEHUMIDIFYING = 3
class HumidifierDehumidifier(HomeAccessory):
"""Generate a HumidifierDehumidifier accessory for a humidifier."""
def __init__(self, *args):
def __init__(self, *args: Any) -> None:
"""Initialize a HumidifierDehumidifier accessory object."""
super().__init__(*args, category=CATEGORY_HUMIDIFIER)
self._reload_on_change_attrs.extend(
@ -83,8 +84,9 @@ class HumidifierDehumidifier(HomeAccessory):
)
)
self.chars = []
self.chars: list[str] = []
state = self.hass.states.get(self.entity_id)
assert state
device_class = state.attributes.get(
ATTR_DEVICE_CLASS, HumidifierDeviceClass.HUMIDIFIER
)
@ -151,7 +153,7 @@ class HumidifierDehumidifier(HomeAccessory):
if humidity_state:
self._async_update_current_humidity(humidity_state)
async def run(self):
async def run(self) -> None:
"""Handle accessory driver started event.
Run inside the Home Assistant event loop.
@ -205,7 +207,8 @@ class HumidifierDehumidifier(HomeAccessory):
ex,
)
def _set_chars(self, char_values):
def _set_chars(self, char_values: dict[str, Any]) -> None:
"""Set characteristics based on the data coming from HomeKit."""
_LOGGER.debug("HumidifierDehumidifier _set_chars: %s", char_values)
if CHAR_TARGET_HUMIDIFIER_DEHUMIDIFIER in char_values:
@ -225,6 +228,7 @@ class HumidifierDehumidifier(HomeAccessory):
if self._target_humidity_char_name in char_values:
state = self.hass.states.get(self.entity_id)
assert state
max_humidity = state.attributes.get(ATTR_MAX_HUMIDITY, DEFAULT_MAX_HUMIDITY)
max_humidity = round(max_humidity)
max_humidity = min(max_humidity, 100)
@ -232,6 +236,11 @@ class HumidifierDehumidifier(HomeAccessory):
min_humidity = state.attributes.get(ATTR_MIN_HUMIDITY, DEFAULT_MIN_HUMIDITY)
min_humidity = round(min_humidity)
min_humidity = max(min_humidity, 0)
# The min/max humidity values here should be clamped to the HomeKit
# min/max that was set when the accessory was added to HomeKit so
# that the user cannot set a value outside of the range that was
# originally set as it could cause HomeKit to report the accessory
# as not responding.
humidity = round(char_values[self._target_humidity_char_name])
@ -252,7 +261,7 @@ class HumidifierDehumidifier(HomeAccessory):
)
@callback
def async_update_state(self, new_state):
def async_update_state(self, new_state: State) -> None:
"""Update state without rechecking the device features."""
is_active = new_state.state == STATE_ON

19
homeassistant/components/homekit/type_lights.py
View File

@ -1,7 +1,9 @@
"""Class to hold all light accessories."""
from __future__ import annotations
from datetime import datetime
import logging
from typing import Any
from pyhap.const import CATEGORY_LIGHTBULB
@ -29,7 +31,7 @@ from homeassistant.const import (
SERVICE_TURN_ON,
STATE_ON,
)
from homeassistant.core import callback
from homeassistant.core import CALLBACK_TYPE, State, callback
from homeassistant.helpers.event import async_call_later
from homeassistant.util.color import (
color_temperature_kelvin_to_mired,
@ -68,7 +70,7 @@ class Light(HomeAccessory):
Currently supports: state, brightness, color temperature, rgb_color.
"""
def __init__(self, *args):
def __init__(self, *args: Any) -> None:
"""Initialize a new Light accessory object."""
super().__init__(*args, category=CATEGORY_LIGHTBULB)
self._reload_on_change_attrs.extend(
@ -79,10 +81,11 @@ class Light(HomeAccessory):
)
)
self.chars = []
self._event_timer = None
self._pending_events = {}
self._event_timer: CALLBACK_TYPE | None = None
self._pending_events: dict[str, Any] = {}
state = self.hass.states.get(self.entity_id)
assert state
attributes = state.attributes
self.color_modes = color_modes = (
attributes.get(ATTR_SUPPORTED_COLOR_MODES) or []
@ -140,7 +143,7 @@ class Light(HomeAccessory):
self.async_update_state(state)
serv_light.setter_callback = self._set_chars
def _set_chars(self, char_values):
def _set_chars(self, char_values: dict[str, Any]) -> None:
_LOGGER.debug("Light _set_chars: %s", char_values)
# Newest change always wins
if CHAR_COLOR_TEMPERATURE in self._pending_events and (
@ -159,14 +162,14 @@ class Light(HomeAccessory):
)
@callback
def _async_send_events(self, *_):
def _async_send_events(self, _now: datetime) -> None:
"""Process all changes at once."""
_LOGGER.debug("Coalesced _set_chars: %s", self._pending_events)
char_values = self._pending_events
self._pending_events = {}
events = []
service = SERVICE_TURN_ON
params = {ATTR_ENTITY_ID: self.entity_id}
params: dict[str, Any] = {ATTR_ENTITY_ID: self.entity_id}
if CHAR_ON in char_values:
if not char_values[CHAR_ON]:
@ -231,7 +234,7 @@ class Light(HomeAccessory):
self.async_call_service(DOMAIN, service, params, ", ".join(events))
@callback
def async_update_state(self, new_state):
def async_update_state(self, new_state: State) -> None:
"""Update light after state change."""
# Handle State
state = new_state.state

49
homeassistant/components/homekit/type_switches.py
View File

@ -2,8 +2,9 @@
from __future__ import annotations
import logging
from typing import NamedTuple
from typing import Any, NamedTuple
from pyhap.characteristic import Characteristic
from pyhap.const import (
CATEGORY_FAUCET,
CATEGORY_OUTLET,
@ -30,7 +31,7 @@ from homeassistant.const import (
SERVICE_TURN_ON,
STATE_ON,
)
from homeassistant.core import callback, split_entity_id
from homeassistant.core import State, callback, split_entity_id
from homeassistant.helpers.event import async_call_later
from .accessories import TYPES, HomeAccessory
@ -78,10 +79,11 @@ ACTIVATE_ONLY_RESET_SECONDS = 10
class Outlet(HomeAccessory):
"""Generate an Outlet accessory."""
def __init__(self, *args):
def __init__(self, *args: Any) -> None:
"""Initialize an Outlet accessory object."""
super().__init__(*args, category=CATEGORY_OUTLET)
state = self.hass.states.get(self.entity_id)
assert state
serv_outlet = self.add_preload_service(SERV_OUTLET)
self.char_on = serv_outlet.configure_char(
@ -94,7 +96,7 @@ class Outlet(HomeAccessory):
# GET to avoid an event storm after homekit startup
self.async_update_state(state)
def set_state(self, value):
def set_state(self, value: bool) -> None:
"""Move switch state to value if call came from HomeKit."""
_LOGGER.debug("%s: Set switch state to %s", self.entity_id, value)
params = {ATTR_ENTITY_ID: self.entity_id}
@ -102,7 +104,7 @@ class Outlet(HomeAccessory):
self.async_call_service(DOMAIN, service, params)
@callback
def async_update_state(self, new_state):
def async_update_state(self, new_state: State) -> None:
"""Update switch state after state changed."""
current_state = new_state.state == STATE_ON
_LOGGER.debug("%s: Set current state to %s", self.entity_id, current_state)
@ -113,13 +115,14 @@ class Outlet(HomeAccessory):
class Switch(HomeAccessory):
"""Generate a Switch accessory."""
def __init__(self, *args):
def __init__(self, *args: Any) -> None:
"""Initialize a Switch accessory object."""
super().__init__(*args, category=CATEGORY_SWITCH)
self._domain, self._object_id = split_entity_id(self.entity_id)
state = self.hass.states.get(self.entity_id)
assert state
self.activate_only = self.is_activate(self.hass.states.get(self.entity_id))
self.activate_only = self.is_activate(state)
serv_switch = self.add_preload_service(SERV_SWITCH)
self.char_on = serv_switch.configure_char(
@ -129,16 +132,16 @@ class Switch(HomeAccessory):
# GET to avoid an event storm after homekit startup
self.async_update_state(state)
def is_activate(self, state):
def is_activate(self, state: State) -> bool:
"""Check if entity is activate only."""
return self._domain in ACTIVATE_ONLY_SWITCH_DOMAINS
def reset_switch(self, *args):
def reset_switch(self, *args: Any) -> None:
"""Reset switch to emulate activate click."""
_LOGGER.debug("%s: Reset switch to off", self.entity_id)
self.char_on.set_value(False)
def set_state(self, value):
def set_state(self, value: bool) -> None:
"""Move switch state to value if call came from HomeKit."""
_LOGGER.debug("%s: Set switch state to %s", self.entity_id, value)
if self.activate_only and not value:
@ -162,7 +165,7 @@ class Switch(HomeAccessory):
async_call_later(self.hass, ACTIVATE_ONLY_RESET_SECONDS, self.reset_switch)
@callback
def async_update_state(self, new_state):
def async_update_state(self, new_state: State) -> None:
"""Update switch state after state changed."""
self.activate_only = self.is_activate(new_state)
if self.activate_only:
@ -180,10 +183,12 @@ class Switch(HomeAccessory):
class Vacuum(Switch):
"""Generate a Switch accessory."""
def set_state(self, value):
def set_state(self, value: bool) -> None:
"""Move switch state to value if call came from HomeKit."""
_LOGGER.debug("%s: Set switch state to %s", self.entity_id, value)
state = self.hass.states.get(self.entity_id)
assert state
features = state.attributes.get(ATTR_SUPPORTED_FEATURES, 0)
if value:
@ -198,7 +203,7 @@ class Vacuum(Switch):
)
@callback
def async_update_state(self, new_state):
def async_update_state(self, new_state: State) -> None:
"""Update switch state after state changed."""
current_state = new_state.state in (STATE_CLEANING, STATE_ON)
_LOGGER.debug("%s: Set current state to %s", self.entity_id, current_state)
@ -209,10 +214,12 @@ class Vacuum(Switch):
class Valve(HomeAccessory):
"""Generate a Valve accessory."""
def __init__(self, *args):
def __init__(self, *args: Any) -> None:
"""Initialize a Valve accessory object."""
super().__init__(*args)
state = self.hass.states.get(self.entity_id)
assert state
valve_type = self.config[CONF_TYPE]
self.category = VALVE_TYPE[valve_type].category
@ -228,7 +235,7 @@ class Valve(HomeAccessory):
# GET to avoid an event storm after homekit startup
self.async_update_state(state)
def set_state(self, value):
def set_state(self, value: bool) -> None:
"""Move value state to value if call came from HomeKit."""
_LOGGER.debug("%s: Set switch state to %s", self.entity_id, value)
self.char_in_use.set_value(value)
@ -237,7 +244,7 @@ class Valve(HomeAccessory):
self.async_call_service(DOMAIN, service, params)
@callback
def async_update_state(self, new_state):
def async_update_state(self, new_state: State) -> None:
"""Update switch state after state changed."""
current_state = 1 if new_state.state == STATE_ON else 0
_LOGGER.debug("%s: Set active state to %s", self.entity_id, current_state)
@ -250,12 +257,14 @@ class Valve(HomeAccessory):
class SelectSwitch(HomeAccessory):
"""Generate a Switch accessory that contains multiple switches."""
def __init__(self, *args):
def __init__(self, *args: Any) -> None:
"""Initialize a Switch accessory object."""
super().__init__(*args, category=CATEGORY_SWITCH)
self.domain = split_entity_id(self.entity_id)[0]
state = self.hass.states.get(self.entity_id)
self.select_chars = {}
assert state
self.select_chars: dict[str, Characteristic] = {}
options = state.attributes[ATTR_OPTIONS]
for option in options:
serv_option = self.add_preload_service(
@ -275,14 +284,14 @@ class SelectSwitch(HomeAccessory):
# GET to avoid an event storm after homekit startup
self.async_update_state(state)
def select_option(self, option):
def select_option(self, option: str) -> None:
"""Set option from HomeKit."""
_LOGGER.debug("%s: Set option to %s", self.entity_id, option)
params = {ATTR_ENTITY_ID: self.entity_id, "option": option}
self.async_call_service(self.domain, SERVICE_SELECT_OPTION, params)
@callback
def async_update_state(self, new_state):
def async_update_state(self, new_state: State) -> None:
"""Update switch state after state changed."""
current_option = cleanup_name_for_homekit(new_state.state)
for option, char in self.select_chars.items():