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

Add some more typing to HomeKit (#101959)

Browse Source
This commit is contained in:
J. Nick Koston 2023-10-13 14:03:01 -10:00 committed by GitHub
parent 2e3013f2f8
commit 76e2afbce9
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 77 additions and 63 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

58
homeassistant/components/homekit/type_media_players.py
View File

@ -2,6 +2,7 @@
import logging
from typing import Any
from pyhap.characteristic import Characteristic
from pyhap.const import CATEGORY_SWITCH
from homeassistant.components.media_player import (
@ -32,7 +33,7 @@ from homeassistant.const import (
STATE_STANDBY,
STATE_UNKNOWN,
)
from homeassistant.core import callback
from homeassistant.core import State, callback
from .accessories import TYPES, HomeAccessory
from .const import (
@ -82,11 +83,12 @@ MEDIA_PLAYER_OFF_STATES = (
class MediaPlayer(HomeAccessory):
"""Generate a Media Player accessory."""
def __init__(self, *args):
def __init__(self, *args: Any) -> None:
"""Initialize a Switch accessory object."""
super().__init__(*args, category=CATEGORY_SWITCH)
state = self.hass.states.get(self.entity_id)
self.chars = {
assert state
self.chars: dict[str, Characteristic | None] = {
FEATURE_ON_OFF: None,
FEATURE_PLAY_PAUSE: None,
FEATURE_PLAY_STOP: None,
@ -137,20 +139,20 @@ class MediaPlayer(HomeAccessory):
)
self.async_update_state(state)
def generate_service_name(self, mode):
def generate_service_name(self, mode: str) -> str:
"""Generate name for individual service."""
return cleanup_name_for_homekit(
f"{self.display_name} {MODE_FRIENDLY_NAME[mode]}"
)
def set_on_off(self, value):
def set_on_off(self, value: bool) -> None:
"""Move switch state to value if call came from HomeKit."""
_LOGGER.debug('%s: Set switch state for "on_off" to %s', self.entity_id, value)
service = SERVICE_TURN_ON if value else SERVICE_TURN_OFF
params = {ATTR_ENTITY_ID: self.entity_id}
self.async_call_service(DOMAIN, service, params)
def set_play_pause(self, value):
def set_play_pause(self, value: bool) -> None:
"""Move switch state to value if call came from HomeKit."""
_LOGGER.debug(
'%s: Set switch state for "play_pause" to %s', self.entity_id, value
@ -159,7 +161,7 @@ class MediaPlayer(HomeAccessory):
params = {ATTR_ENTITY_ID: self.entity_id}
self.async_call_service(DOMAIN, service, params)
def set_play_stop(self, value):
def set_play_stop(self, value: bool) -> None:
"""Move switch state to value if call came from HomeKit."""
_LOGGER.debug(
'%s: Set switch state for "play_stop" to %s', self.entity_id, value
@ -168,7 +170,7 @@ class MediaPlayer(HomeAccessory):
params = {ATTR_ENTITY_ID: self.entity_id}
self.async_call_service(DOMAIN, service, params)
def set_toggle_mute(self, value):
def set_toggle_mute(self, value: bool) -> None:
"""Move switch state to value if call came from HomeKit."""
_LOGGER.debug(
'%s: Set switch state for "toggle_mute" to %s', self.entity_id, value
@ -177,43 +179,43 @@ class MediaPlayer(HomeAccessory):
self.async_call_service(DOMAIN, SERVICE_VOLUME_MUTE, 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
if self.chars[FEATURE_ON_OFF]:
if on_off_char := self.chars[FEATURE_ON_OFF]:
hk_state = current_state not in MEDIA_PLAYER_OFF_STATES
_LOGGER.debug(
'%s: Set current state for "on_off" to %s', self.entity_id, hk_state
)
self.chars[FEATURE_ON_OFF].set_value(hk_state)
on_off_char.set_value(hk_state)
if self.chars[FEATURE_PLAY_PAUSE]:
if play_pause_char := self.chars[FEATURE_PLAY_PAUSE]:
hk_state = current_state == STATE_PLAYING
_LOGGER.debug(
'%s: Set current state for "play_pause" to %s',
self.entity_id,
hk_state,
)
self.chars[FEATURE_PLAY_PAUSE].set_value(hk_state)
play_pause_char.set_value(hk_state)
if self.chars[FEATURE_PLAY_STOP]:
if play_stop_char := self.chars[FEATURE_PLAY_STOP]:
hk_state = current_state == STATE_PLAYING
_LOGGER.debug(
'%s: Set current state for "play_stop" to %s',
self.entity_id,
hk_state,
)
self.chars[FEATURE_PLAY_STOP].set_value(hk_state)
play_stop_char.set_value(hk_state)
if self.chars[FEATURE_TOGGLE_MUTE]:
current_state = bool(new_state.attributes.get(ATTR_MEDIA_VOLUME_MUTED))
if toggle_mute_char := self.chars[FEATURE_TOGGLE_MUTE]:
mute_state = bool(new_state.attributes.get(ATTR_MEDIA_VOLUME_MUTED))
_LOGGER.debug(
'%s: Set current state for "toggle_mute" to %s',
self.entity_id,
current_state,
mute_state,
)
self.chars[FEATURE_TOGGLE_MUTE].set_value(current_state)
toggle_mute_char.set_value(mute_state)
@TYPES.register("TelevisionMediaPlayer")
@ -278,14 +280,14 @@ class TelevisionMediaPlayer(RemoteInputSelectAccessory):
self.async_update_state(state)
def set_on_off(self, value):
def set_on_off(self, value: bool) -> None:
"""Move switch state to value if call came from HomeKit."""
_LOGGER.debug('%s: Set switch state for "on_off" to %s', self.entity_id, value)
service = SERVICE_TURN_ON if value else SERVICE_TURN_OFF
params = {ATTR_ENTITY_ID: self.entity_id}
self.async_call_service(DOMAIN, service, params)
def set_mute(self, value):
def set_mute(self, value: bool) -> None:
"""Move switch state to value if call came from HomeKit."""
_LOGGER.debug(
'%s: Set switch state for "toggle_mute" to %s', self.entity_id, value
@ -293,27 +295,27 @@ class TelevisionMediaPlayer(RemoteInputSelectAccessory):
params = {ATTR_ENTITY_ID: self.entity_id, ATTR_MEDIA_VOLUME_MUTED: value}
self.async_call_service(DOMAIN, SERVICE_VOLUME_MUTE, params)
def set_volume(self, value):
def set_volume(self, value: bool) -> None:
"""Send volume step value if call came from HomeKit."""
_LOGGER.debug("%s: Set volume to %s", self.entity_id, value)
params = {ATTR_ENTITY_ID: self.entity_id, ATTR_MEDIA_VOLUME_LEVEL: value}
self.async_call_service(DOMAIN, SERVICE_VOLUME_SET, params)
def set_volume_step(self, value):
def set_volume_step(self, value: bool) -> None:
"""Send volume step value if call came from HomeKit."""
_LOGGER.debug("%s: Step volume by %s", self.entity_id, value)
service = SERVICE_VOLUME_DOWN if value else SERVICE_VOLUME_UP
params = {ATTR_ENTITY_ID: self.entity_id}
self.async_call_service(DOMAIN, service, params)
def set_input_source(self, value):
def set_input_source(self, value: int) -> None:
"""Send input set value if call came from HomeKit."""
_LOGGER.debug("%s: Set current input to %s", self.entity_id, value)
source_name = self._mapped_sources[self.sources[value]]
params = {ATTR_ENTITY_ID: self.entity_id, ATTR_INPUT_SOURCE: source_name}
self.async_call_service(DOMAIN, SERVICE_SELECT_SOURCE, params)
def set_remote_key(self, value):
def set_remote_key(self, value: int) -> None:
"""Send remote key value if call came from HomeKit."""
_LOGGER.debug("%s: Set remote key to %s", self.entity_id, value)
if (key_name := REMOTE_KEYS.get(value)) is None:
@ -322,7 +324,9 @@ class TelevisionMediaPlayer(RemoteInputSelectAccessory):
if key_name == KEY_PLAY_PAUSE and self._supports_play_pause:
# Handle Play Pause by directly updating the media player entity.
state = self.hass.states.get(self.entity_id).state
state_obj = self.hass.states.get(self.entity_id)
assert state_obj
state = state_obj.state
if state in (STATE_PLAYING, STATE_PAUSED):
service = (
SERVICE_MEDIA_PLAY if state == STATE_PAUSED else SERVICE_MEDIA_PAUSE
@ -340,7 +344,7 @@ class TelevisionMediaPlayer(RemoteInputSelectAccessory):
)
@callback
def async_update_state(self, new_state):
def async_update_state(self, new_state: State) -> None:
"""Update Television state after state changed."""
current_state = new_state.state

14
homeassistant/components/homekit/type_remotes.py
View File

@ -219,7 +219,7 @@ class RemoteInputSelectAccessory(HomeAccessory, ABC):
class ActivityRemote(RemoteInputSelectAccessory):
"""Generate a Activity Remote accessory."""
def __init__(self, *args):
def __init__(self, *args: Any) -> None:
"""Initialize a Activity Remote accessory object."""
super().__init__(
RemoteEntityFeature.ACTIVITY,
@ -227,23 +227,25 @@ class ActivityRemote(RemoteInputSelectAccessory):
ATTR_ACTIVITY_LIST,
*args,
)
self.async_update_state(self.hass.states.get(self.entity_id))
state = self.hass.states.get(self.entity_id)
assert state
self.async_update_state(state)
def set_on_off(self, value):
def set_on_off(self, value: bool) -> None:
"""Move switch state to value if call came from HomeKit."""
_LOGGER.debug('%s: Set switch state for "on_off" to %s', self.entity_id, value)
service = SERVICE_TURN_ON if value else SERVICE_TURN_OFF
params = {ATTR_ENTITY_ID: self.entity_id}
self.async_call_service(REMOTE_DOMAIN, service, params)
def set_input_source(self, value):
def set_input_source(self, value: int) -> None:
"""Send input set value if call came from HomeKit."""
_LOGGER.debug("%s: Set current input to %s", self.entity_id, value)
source = self._mapped_sources[self.sources[value]]
params = {ATTR_ENTITY_ID: self.entity_id, ATTR_ACTIVITY: source}
self.async_call_service(REMOTE_DOMAIN, SERVICE_TURN_ON, params)
def set_remote_key(self, value):
def set_remote_key(self, value: int) -> None:
"""Send remote key value if call came from HomeKit."""
_LOGGER.debug("%s: Set remote key to %s", self.entity_id, value)
if (key_name := REMOTE_KEYS.get(value)) is None:
@ -255,7 +257,7 @@ class ActivityRemote(RemoteInputSelectAccessory):
)
@callback
def async_update_state(self, new_state):
def async_update_state(self, new_state: State) -> None:
"""Update Television remote state after state changed."""
current_state = new_state.state
# Power state remote

10
homeassistant/components/homekit/type_security_systems.py
View File

@ -1,5 +1,6 @@
"""Class to hold all alarm control panel accessories."""
import logging
from typing import Any
from pyhap.const import CATEGORY_ALARM_SYSTEM
@ -23,7 +24,7 @@ from homeassistant.const import (
STATE_ALARM_DISARMED,
STATE_ALARM_TRIGGERED,
)
from homeassistant.core import callback
from homeassistant.core import State, callback
from .accessories import TYPES, HomeAccessory
from .const import (
@ -78,10 +79,11 @@ HK_TO_SERVICE = {
class SecuritySystem(HomeAccessory):
"""Generate an SecuritySystem accessory for an alarm control panel."""
def __init__(self, *args):
def __init__(self, *args: Any) -> None:
"""Initialize a SecuritySystem accessory object."""
super().__init__(*args, category=CATEGORY_ALARM_SYSTEM)
state = self.hass.states.get(self.entity_id)
assert state
self._alarm_code = self.config.get(ATTR_CODE)
supported_states = state.attributes.get(
@ -143,7 +145,7 @@ class SecuritySystem(HomeAccessory):
# GET to avoid an event storm after homekit startup
self.async_update_state(state)
def set_security_state(self, value):
def set_security_state(self, value: int) -> None:
"""Move security state to value if call came from HomeKit."""
_LOGGER.debug("%s: Set security state to %d", self.entity_id, value)
service = HK_TO_SERVICE[value]
@ -153,7 +155,7 @@ class SecuritySystem(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 security state after state changed."""
hass_state = new_state.state
if (current_state := HASS_TO_HOMEKIT_CURRENT.get(hass_state)) is not None:

58
homeassistant/components/homekit/type_sensors.py
View File

@ -3,7 +3,7 @@ from __future__ import annotations
from collections.abc import Callable
import logging
from typing import NamedTuple
from typing import Any, NamedTuple
from pyhap.const import CATEGORY_SENSOR
from pyhap.service import Service
@ -16,7 +16,7 @@ from homeassistant.const import (
STATE_ON,
UnitOfTemperature,
)
from homeassistant.core import callback
from homeassistant.core import State, callback
from .accessories import TYPES, HomeAccessory
from .const import (
@ -112,10 +112,11 @@ class TemperatureSensor(HomeAccessory):
Sensor entity must return temperature in °C, °F.
"""
def __init__(self, *args):
def __init__(self, *args: Any) -> None:
"""Initialize a TemperatureSensor accessory object."""
super().__init__(*args, category=CATEGORY_SENSOR)
state = self.hass.states.get(self.entity_id)
assert state
serv_temp = self.add_preload_service(SERV_TEMPERATURE_SENSOR)
self.char_temp = serv_temp.configure_char(
CHAR_CURRENT_TEMPERATURE, value=0, properties=PROP_CELSIUS
@ -125,7 +126,7 @@ class TemperatureSensor(HomeAccessory):
self.async_update_state(state)
@callback
def async_update_state(self, new_state):
def async_update_state(self, new_state: State) -> None:
"""Update temperature after state changed."""
unit = new_state.attributes.get(
ATTR_UNIT_OF_MEASUREMENT, UnitOfTemperature.CELSIUS
@ -142,10 +143,11 @@ class TemperatureSensor(HomeAccessory):
class HumiditySensor(HomeAccessory):
"""Generate a HumiditySensor accessory as humidity sensor."""
def __init__(self, *args):
def __init__(self, *args: Any) -> None:
"""Initialize a HumiditySensor accessory object."""
super().__init__(*args, category=CATEGORY_SENSOR)
state = self.hass.states.get(self.entity_id)
assert state
serv_humidity = self.add_preload_service(SERV_HUMIDITY_SENSOR)
self.char_humidity = serv_humidity.configure_char(
CHAR_CURRENT_HUMIDITY, value=0
@ -155,7 +157,7 @@ class HumiditySensor(HomeAccessory):
self.async_update_state(state)
@callback
def async_update_state(self, new_state):
def async_update_state(self, new_state: State) -> None:
"""Update accessory after state change."""
if (humidity := convert_to_float(new_state.state)) is not None:
self.char_humidity.set_value(humidity)
@ -166,18 +168,18 @@ class HumiditySensor(HomeAccessory):
class AirQualitySensor(HomeAccessory):
"""Generate a AirQualitySensor accessory as air quality sensor."""
def __init__(self, *args):
def __init__(self, *args: Any) -> None:
"""Initialize a AirQualitySensor accessory object."""
super().__init__(*args, category=CATEGORY_SENSOR)
state = self.hass.states.get(self.entity_id)
assert state
self.create_services()
# Set the state so it is in sync on initial
# GET to avoid an event storm after homekit startup
self.async_update_state(state)
def create_services(self):
def create_services(self) -> None:
"""Initialize a AirQualitySensor accessory object."""
serv_air_quality = self.add_preload_service(
SERV_AIR_QUALITY_SENSOR, [CHAR_AIR_PARTICULATE_DENSITY]
@ -188,7 +190,7 @@ class AirQualitySensor(HomeAccessory):
)
@callback
def async_update_state(self, new_state):
def async_update_state(self, new_state: State) -> None:
"""Update accessory after state change."""
if (density := convert_to_float(new_state.state)) is not None:
if self.char_density.value != density:
@ -203,7 +205,7 @@ class AirQualitySensor(HomeAccessory):
class PM10Sensor(AirQualitySensor):
"""Generate a PM10Sensor accessory as PM 10 sensor."""
def create_services(self):
def create_services(self) -> None:
"""Override the init function for PM 10 Sensor."""
serv_air_quality = self.add_preload_service(
SERV_AIR_QUALITY_SENSOR, [CHAR_PM10_DENSITY]
@ -212,7 +214,7 @@ class PM10Sensor(AirQualitySensor):
self.char_density = serv_air_quality.configure_char(CHAR_PM10_DENSITY, value=0)
@callback
def async_update_state(self, new_state):
def async_update_state(self, new_state: State) -> None:
"""Update accessory after state change."""
density = convert_to_float(new_state.state)
if density is None:
@ -230,7 +232,7 @@ class PM10Sensor(AirQualitySensor):
class PM25Sensor(AirQualitySensor):
"""Generate a PM25Sensor accessory as PM 2.5 sensor."""
def create_services(self):
def create_services(self) -> None:
"""Override the init function for PM 2.5 Sensor."""
serv_air_quality = self.add_preload_service(
SERV_AIR_QUALITY_SENSOR, [CHAR_PM25_DENSITY]
@ -239,7 +241,7 @@ class PM25Sensor(AirQualitySensor):
self.char_density = serv_air_quality.configure_char(CHAR_PM25_DENSITY, value=0)
@callback
def async_update_state(self, new_state):
def async_update_state(self, new_state: State) -> None:
"""Update accessory after state change."""
density = convert_to_float(new_state.state)
if density is None:
@ -257,7 +259,7 @@ class PM25Sensor(AirQualitySensor):
class NitrogenDioxideSensor(AirQualitySensor):
"""Generate a NitrogenDioxideSensor accessory as NO2 sensor."""
def create_services(self):
def create_services(self) -> None:
"""Override the init function for PM 2.5 Sensor."""
serv_air_quality = self.add_preload_service(
SERV_AIR_QUALITY_SENSOR, [CHAR_NITROGEN_DIOXIDE_DENSITY]
@ -268,7 +270,7 @@ class NitrogenDioxideSensor(AirQualitySensor):
)
@callback
def async_update_state(self, new_state):
def async_update_state(self, new_state: State) -> None:
"""Update accessory after state change."""
density = convert_to_float(new_state.state)
if density is None:
@ -289,7 +291,7 @@ class VolatileOrganicCompoundsSensor(AirQualitySensor):
Sensor entity must return VOC in µg/m3.
"""
def create_services(self):
def create_services(self) -> None:
"""Override the init function for VOC Sensor."""
serv_air_quality: Service = self.add_preload_service(
SERV_AIR_QUALITY_SENSOR, [CHAR_VOC_DENSITY]
@ -305,7 +307,7 @@ class VolatileOrganicCompoundsSensor(AirQualitySensor):
)
@callback
def async_update_state(self, new_state):
def async_update_state(self, new_state: State) -> None:
"""Update accessory after state change."""
density = convert_to_float(new_state.state)
if density is None:
@ -323,10 +325,11 @@ class VolatileOrganicCompoundsSensor(AirQualitySensor):
class CarbonMonoxideSensor(HomeAccessory):
"""Generate a CarbonMonoxidSensor accessory as CO sensor."""
def __init__(self, *args):
def __init__(self, *args: Any) -> None:
"""Initialize a CarbonMonoxideSensor accessory object."""
super().__init__(*args, category=CATEGORY_SENSOR)
state = self.hass.states.get(self.entity_id)
assert state
serv_co = self.add_preload_service(
SERV_CARBON_MONOXIDE_SENSOR,
[CHAR_CARBON_MONOXIDE_LEVEL, CHAR_CARBON_MONOXIDE_PEAK_LEVEL],
@ -343,7 +346,7 @@ class CarbonMonoxideSensor(HomeAccessory):
self.async_update_state(state)
@callback
def async_update_state(self, new_state):
def async_update_state(self, new_state: State) -> None:
"""Update accessory after state change."""
if (value := convert_to_float(new_state.state)) is not None:
self.char_level.set_value(value)
@ -358,10 +361,11 @@ class CarbonMonoxideSensor(HomeAccessory):
class CarbonDioxideSensor(HomeAccessory):
"""Generate a CarbonDioxideSensor accessory as CO2 sensor."""
def __init__(self, *args):
def __init__(self, *args: Any) -> None:
"""Initialize a CarbonDioxideSensor accessory object."""
super().__init__(*args, category=CATEGORY_SENSOR)
state = self.hass.states.get(self.entity_id)
assert state
serv_co2 = self.add_preload_service(
SERV_CARBON_DIOXIDE_SENSOR,
[CHAR_CARBON_DIOXIDE_LEVEL, CHAR_CARBON_DIOXIDE_PEAK_LEVEL],
@ -378,7 +382,7 @@ class CarbonDioxideSensor(HomeAccessory):
self.async_update_state(state)
@callback
def async_update_state(self, new_state):
def async_update_state(self, new_state: State) -> None:
"""Update accessory after state change."""
if (value := convert_to_float(new_state.state)) is not None:
self.char_level.set_value(value)
@ -393,10 +397,11 @@ class CarbonDioxideSensor(HomeAccessory):
class LightSensor(HomeAccessory):
"""Generate a LightSensor accessory as light sensor."""
def __init__(self, *args):
def __init__(self, *args: Any) -> None:
"""Initialize a LightSensor accessory object."""
super().__init__(*args, category=CATEGORY_SENSOR)
state = self.hass.states.get(self.entity_id)
assert state
serv_light = self.add_preload_service(SERV_LIGHT_SENSOR)
self.char_light = serv_light.configure_char(
CHAR_CURRENT_AMBIENT_LIGHT_LEVEL, value=0
@ -406,7 +411,7 @@ class LightSensor(HomeAccessory):
self.async_update_state(state)
@callback
def async_update_state(self, new_state):
def async_update_state(self, new_state: State) -> None:
"""Update accessory after state change."""
if (luminance := convert_to_float(new_state.state)) is not None:
self.char_light.set_value(luminance)
@ -417,10 +422,11 @@ class LightSensor(HomeAccessory):
class BinarySensor(HomeAccessory):
"""Generate a BinarySensor accessory as binary sensor."""
def __init__(self, *args):
def __init__(self, *args: Any) -> None:
"""Initialize a BinarySensor accessory object."""
super().__init__(*args, category=CATEGORY_SENSOR)
state = self.hass.states.get(self.entity_id)
assert state
device_class = state.attributes.get(ATTR_DEVICE_CLASS)
service_char = (
BINARY_SENSOR_SERVICE_MAP[device_class]
@ -439,7 +445,7 @@ class BinarySensor(HomeAccessory):
self.async_update_state(state)
@callback
def async_update_state(self, new_state):
def async_update_state(self, new_state: State) -> None:
"""Update accessory after state change."""
state = new_state.state
detected = self.format(state in (STATE_ON, STATE_HOME))