From 237ef6419b5d77adf7dc09d7dd554965406103bb Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Sun, 29 May 2022 06:27:32 -1000 Subject: [PATCH] Add basic typing to emulated_hue (#72663) * Add basic typing to emulated_hue * type a few more places * fixes * numbers are always stringified * numbers are always stringified * coverage * drop assert --- .../components/emulated_hue/__init__.py | 233 +++--------------- .../components/emulated_hue/config.py | 213 ++++++++++++++++ .../components/emulated_hue/const.py | 2 + .../components/emulated_hue/hue_api.py | 96 +++++--- tests/components/emulated_hue/test_init.py | 16 +- tests/components/emulated_hue/test_upnp.py | 13 +- 6 files changed, 324 insertions(+), 249 deletions(-) create mode 100644 homeassistant/components/emulated_hue/config.py diff --git a/homeassistant/components/emulated_hue/__init__.py b/homeassistant/components/emulated_hue/__init__.py index d3586ab5fcf..71f98abed80 100644 --- a/homeassistant/components/emulated_hue/__init__.py +++ b/homeassistant/components/emulated_hue/__init__.py @@ -1,4 +1,6 @@ """Support for local control of entities by emulating a Philips Hue bridge.""" +from __future__ import annotations + import logging from aiohttp import web @@ -12,10 +14,29 @@ from homeassistant.const import ( EVENT_HOMEASSISTANT_STOP, ) from homeassistant.core import HomeAssistant -from homeassistant.helpers import storage import homeassistant.helpers.config_validation as cv from homeassistant.helpers.typing import ConfigType +from .config import ( + CONF_ADVERTISE_IP, + CONF_ADVERTISE_PORT, + CONF_ENTITY_HIDDEN, + CONF_ENTITY_NAME, + CONF_EXPOSE_BY_DEFAULT, + CONF_EXPOSED_DOMAINS, + CONF_HOST_IP, + CONF_LIGHTS_ALL_DIMMABLE, + CONF_LISTEN_PORT, + CONF_OFF_MAPS_TO_ON_DOMAINS, + CONF_UPNP_BIND_MULTICAST, + DEFAULT_LIGHTS_ALL_DIMMABLE, + DEFAULT_LISTEN_PORT, + DEFAULT_TYPE, + TYPE_ALEXA, + TYPE_GOOGLE, + Config, +) +from .const import DOMAIN from .hue_api import ( HueAllGroupsStateView, HueAllLightsStateView, @@ -27,46 +48,14 @@ from .hue_api import ( HueUnauthorizedUser, HueUsernameView, ) -from .upnp import DescriptionXmlView, create_upnp_datagram_endpoint - -DOMAIN = "emulated_hue" +from .upnp import ( + DescriptionXmlView, + UPNPResponderProtocol, + create_upnp_datagram_endpoint, +) _LOGGER = logging.getLogger(__name__) -NUMBERS_FILE = "emulated_hue_ids.json" -DATA_KEY = "emulated_hue.ids" -DATA_VERSION = "1" -SAVE_DELAY = 60 - -CONF_ADVERTISE_IP = "advertise_ip" -CONF_ADVERTISE_PORT = "advertise_port" -CONF_ENTITY_HIDDEN = "hidden" -CONF_ENTITY_NAME = "name" -CONF_EXPOSE_BY_DEFAULT = "expose_by_default" -CONF_EXPOSED_DOMAINS = "exposed_domains" -CONF_HOST_IP = "host_ip" -CONF_LIGHTS_ALL_DIMMABLE = "lights_all_dimmable" -CONF_LISTEN_PORT = "listen_port" -CONF_OFF_MAPS_TO_ON_DOMAINS = "off_maps_to_on_domains" -CONF_UPNP_BIND_MULTICAST = "upnp_bind_multicast" - -TYPE_ALEXA = "alexa" -TYPE_GOOGLE = "google_home" - -DEFAULT_LIGHTS_ALL_DIMMABLE = False -DEFAULT_LISTEN_PORT = 8300 -DEFAULT_UPNP_BIND_MULTICAST = True -DEFAULT_OFF_MAPS_TO_ON_DOMAINS = ["script", "scene"] -DEFAULT_EXPOSE_BY_DEFAULT = True -DEFAULT_EXPOSED_DOMAINS = [ - "switch", - "light", - "group", - "input_boolean", - "media_player", - "fan", -] -DEFAULT_TYPE = TYPE_GOOGLE CONFIG_ENTITY_SCHEMA = vol.Schema( { @@ -75,6 +64,7 @@ CONFIG_ENTITY_SCHEMA = vol.Schema( } ) + CONFIG_SCHEMA = vol.Schema( { DOMAIN: vol.Schema( @@ -102,8 +92,6 @@ CONFIG_SCHEMA = vol.Schema( extra=vol.ALLOW_EXTRA, ) -ATTR_EMULATED_HUE_NAME = "emulated_hue_name" - async def async_setup(hass: HomeAssistant, yaml_config: ConfigType) -> bool: """Activate the emulated_hue component.""" @@ -140,7 +128,7 @@ async def async_setup(hass: HomeAssistant, yaml_config: ConfigType) -> bool: config.advertise_ip, config.advertise_port or config.listen_port, ) - protocol = None + protocol: UPNPResponderProtocol | None = None async def stop_emulated_hue_bridge(event): """Stop the emulated hue bridge.""" @@ -161,7 +149,8 @@ async def async_setup(hass: HomeAssistant, yaml_config: ConfigType) -> bool: nonlocal site nonlocal runner - _, protocol = await listen + transport_protocol = await listen + protocol = transport_protocol[1] runner = web.AppRunner(app) await runner.setup() @@ -184,163 +173,3 @@ async def async_setup(hass: HomeAssistant, yaml_config: ConfigType) -> bool: hass.bus.async_listen_once(EVENT_HOMEASSISTANT_START, start_emulated_hue_bridge) return True - - -class Config: - """Hold configuration variables for the emulated hue bridge.""" - - def __init__(self, hass, conf, local_ip): - """Initialize the instance.""" - self.hass = hass - self.type = conf.get(CONF_TYPE) - self.numbers = None - self.store = None - self.cached_states = {} - self._exposed_cache = {} - - if self.type == TYPE_ALEXA: - _LOGGER.warning( - "Emulated Hue running in legacy mode because type has been " - "specified. More info at https://goo.gl/M6tgz8" - ) - - # Get the IP address that will be passed to the Echo during discovery - self.host_ip_addr = conf.get(CONF_HOST_IP) - if self.host_ip_addr is None: - self.host_ip_addr = local_ip - - # Get the port that the Hue bridge will listen on - self.listen_port = conf.get(CONF_LISTEN_PORT) - if not isinstance(self.listen_port, int): - self.listen_port = DEFAULT_LISTEN_PORT - _LOGGER.info( - "Listen port not specified, defaulting to %s", self.listen_port - ) - - # Get whether or not UPNP binds to multicast address (239.255.255.250) - # or to the unicast address (host_ip_addr) - self.upnp_bind_multicast = conf.get( - CONF_UPNP_BIND_MULTICAST, DEFAULT_UPNP_BIND_MULTICAST - ) - - # Get domains that cause both "on" and "off" commands to map to "on" - # This is primarily useful for things like scenes or scripts, which - # don't really have a concept of being off - self.off_maps_to_on_domains = conf.get(CONF_OFF_MAPS_TO_ON_DOMAINS) - if not isinstance(self.off_maps_to_on_domains, list): - self.off_maps_to_on_domains = DEFAULT_OFF_MAPS_TO_ON_DOMAINS - - # Get whether or not entities should be exposed by default, or if only - # explicitly marked ones will be exposed - self.expose_by_default = conf.get( - CONF_EXPOSE_BY_DEFAULT, DEFAULT_EXPOSE_BY_DEFAULT - ) - - # Get domains that are exposed by default when expose_by_default is - # True - self.exposed_domains = set( - conf.get(CONF_EXPOSED_DOMAINS, DEFAULT_EXPOSED_DOMAINS) - ) - - # Calculated effective advertised IP and port for network isolation - self.advertise_ip = conf.get(CONF_ADVERTISE_IP) or self.host_ip_addr - - self.advertise_port = conf.get(CONF_ADVERTISE_PORT) or self.listen_port - - self.entities = conf.get(CONF_ENTITIES, {}) - - self._entities_with_hidden_attr_in_config = {} - for entity_id in self.entities: - hidden_value = self.entities[entity_id].get(CONF_ENTITY_HIDDEN) - if hidden_value is not None: - self._entities_with_hidden_attr_in_config[entity_id] = hidden_value - - # Get whether all non-dimmable lights should be reported as dimmable - # for compatibility with older installations. - self.lights_all_dimmable = conf.get(CONF_LIGHTS_ALL_DIMMABLE) - - async def async_setup(self): - """Set up and migrate to storage.""" - self.store = storage.Store(self.hass, DATA_VERSION, DATA_KEY) - self.numbers = ( - await storage.async_migrator( - self.hass, self.hass.config.path(NUMBERS_FILE), self.store - ) - or {} - ) - - def entity_id_to_number(self, entity_id): - """Get a unique number for the entity id.""" - if self.type == TYPE_ALEXA: - return entity_id - - # Google Home - for number, ent_id in self.numbers.items(): - if entity_id == ent_id: - return number - - number = "1" - if self.numbers: - number = str(max(int(k) for k in self.numbers) + 1) - self.numbers[number] = entity_id - self.store.async_delay_save(lambda: self.numbers, SAVE_DELAY) - return number - - def number_to_entity_id(self, number): - """Convert unique number to entity id.""" - if self.type == TYPE_ALEXA: - return number - - # Google Home - assert isinstance(number, str) - return self.numbers.get(number) - - def get_entity_name(self, entity): - """Get the name of an entity.""" - if ( - entity.entity_id in self.entities - and CONF_ENTITY_NAME in self.entities[entity.entity_id] - ): - return self.entities[entity.entity_id][CONF_ENTITY_NAME] - - return entity.attributes.get(ATTR_EMULATED_HUE_NAME, entity.name) - - def is_entity_exposed(self, entity): - """Cache determine if an entity should be exposed on the emulated bridge.""" - entity_id = entity.entity_id - if entity_id not in self._exposed_cache: - self._exposed_cache[entity_id] = self._is_entity_exposed(entity) - return self._exposed_cache[entity_id] - - def filter_exposed_entities(self, states): - """Filter a list of all states down to exposed entities.""" - exposed = [] - for entity in states: - entity_id = entity.entity_id - if entity_id not in self._exposed_cache: - self._exposed_cache[entity_id] = self._is_entity_exposed(entity) - if self._exposed_cache[entity_id]: - exposed.append(entity) - return exposed - - def _is_entity_exposed(self, entity): - """Determine if an entity should be exposed on the emulated bridge. - - Async friendly. - """ - if entity.attributes.get("view") is not None: - # Ignore entities that are views - return False - - if entity.entity_id in self._entities_with_hidden_attr_in_config: - return not self._entities_with_hidden_attr_in_config[entity.entity_id] - - if not self.expose_by_default: - return False - # Expose an entity if the entity's domain is exposed by default and - # the configuration doesn't explicitly exclude it from being - # exposed, or if the entity is explicitly exposed - if entity.domain in self.exposed_domains: - return True - - return False diff --git a/homeassistant/components/emulated_hue/config.py b/homeassistant/components/emulated_hue/config.py new file mode 100644 index 00000000000..e39ec9839c8 --- /dev/null +++ b/homeassistant/components/emulated_hue/config.py @@ -0,0 +1,213 @@ +"""Support for local control of entities by emulating a Philips Hue bridge.""" +from __future__ import annotations + +from collections.abc import Iterable +import logging + +from homeassistant.const import CONF_ENTITIES, CONF_TYPE +from homeassistant.core import HomeAssistant, State +from homeassistant.helpers import storage +from homeassistant.helpers.typing import ConfigType + +TYPE_ALEXA = "alexa" +TYPE_GOOGLE = "google_home" + + +NUMBERS_FILE = "emulated_hue_ids.json" +DATA_KEY = "emulated_hue.ids" +DATA_VERSION = "1" +SAVE_DELAY = 60 + +CONF_ADVERTISE_IP = "advertise_ip" +CONF_ADVERTISE_PORT = "advertise_port" +CONF_ENTITY_HIDDEN = "hidden" +CONF_ENTITY_NAME = "name" +CONF_EXPOSE_BY_DEFAULT = "expose_by_default" +CONF_EXPOSED_DOMAINS = "exposed_domains" +CONF_HOST_IP = "host_ip" +CONF_LIGHTS_ALL_DIMMABLE = "lights_all_dimmable" +CONF_LISTEN_PORT = "listen_port" +CONF_OFF_MAPS_TO_ON_DOMAINS = "off_maps_to_on_domains" +CONF_UPNP_BIND_MULTICAST = "upnp_bind_multicast" + + +DEFAULT_LIGHTS_ALL_DIMMABLE = False +DEFAULT_LISTEN_PORT = 8300 +DEFAULT_UPNP_BIND_MULTICAST = True +DEFAULT_OFF_MAPS_TO_ON_DOMAINS = {"script", "scene"} +DEFAULT_EXPOSE_BY_DEFAULT = True +DEFAULT_EXPOSED_DOMAINS = [ + "switch", + "light", + "group", + "input_boolean", + "media_player", + "fan", +] +DEFAULT_TYPE = TYPE_GOOGLE + +ATTR_EMULATED_HUE_NAME = "emulated_hue_name" + + +_LOGGER = logging.getLogger(__name__) + + +class Config: + """Hold configuration variables for the emulated hue bridge.""" + + def __init__( + self, hass: HomeAssistant, conf: ConfigType, local_ip: str | None + ) -> None: + """Initialize the instance.""" + self.hass = hass + self.type = conf.get(CONF_TYPE) + self.numbers: dict[str, str] = {} + self.store: storage.Store | None = None + self.cached_states: dict[str, list] = {} + self._exposed_cache: dict[str, bool] = {} + + if self.type == TYPE_ALEXA: + _LOGGER.warning( + "Emulated Hue running in legacy mode because type has been " + "specified. More info at https://goo.gl/M6tgz8" + ) + + # Get the IP address that will be passed to the Echo during discovery + self.host_ip_addr = conf.get(CONF_HOST_IP) + if self.host_ip_addr is None: + self.host_ip_addr = local_ip + + # Get the port that the Hue bridge will listen on + self.listen_port = conf.get(CONF_LISTEN_PORT) + if not isinstance(self.listen_port, int): + self.listen_port = DEFAULT_LISTEN_PORT + _LOGGER.info( + "Listen port not specified, defaulting to %s", self.listen_port + ) + + # Get whether or not UPNP binds to multicast address (239.255.255.250) + # or to the unicast address (host_ip_addr) + self.upnp_bind_multicast = conf.get( + CONF_UPNP_BIND_MULTICAST, DEFAULT_UPNP_BIND_MULTICAST + ) + + # Get domains that cause both "on" and "off" commands to map to "on" + # This is primarily useful for things like scenes or scripts, which + # don't really have a concept of being off + off_maps_to_on_domains = conf.get(CONF_OFF_MAPS_TO_ON_DOMAINS) + if isinstance(off_maps_to_on_domains, list): + self.off_maps_to_on_domains = set(off_maps_to_on_domains) + else: + self.off_maps_to_on_domains = DEFAULT_OFF_MAPS_TO_ON_DOMAINS + + # Get whether or not entities should be exposed by default, or if only + # explicitly marked ones will be exposed + self.expose_by_default = conf.get( + CONF_EXPOSE_BY_DEFAULT, DEFAULT_EXPOSE_BY_DEFAULT + ) + + # Get domains that are exposed by default when expose_by_default is + # True + self.exposed_domains = set( + conf.get(CONF_EXPOSED_DOMAINS, DEFAULT_EXPOSED_DOMAINS) + ) + + # Calculated effective advertised IP and port for network isolation + self.advertise_ip = conf.get(CONF_ADVERTISE_IP) or self.host_ip_addr + + self.advertise_port = conf.get(CONF_ADVERTISE_PORT) or self.listen_port + + self.entities = conf.get(CONF_ENTITIES, {}) + + self._entities_with_hidden_attr_in_config = {} + for entity_id in self.entities: + hidden_value = self.entities[entity_id].get(CONF_ENTITY_HIDDEN) + if hidden_value is not None: + self._entities_with_hidden_attr_in_config[entity_id] = hidden_value + + # Get whether all non-dimmable lights should be reported as dimmable + # for compatibility with older installations. + self.lights_all_dimmable = conf.get(CONF_LIGHTS_ALL_DIMMABLE) + + async def async_setup(self) -> None: + """Set up and migrate to storage.""" + self.store = storage.Store(self.hass, DATA_VERSION, DATA_KEY) # type: ignore[arg-type] + self.numbers = ( + await storage.async_migrator( + self.hass, self.hass.config.path(NUMBERS_FILE), self.store + ) + or {} + ) + + def entity_id_to_number(self, entity_id: str) -> str: + """Get a unique number for the entity id.""" + if self.type == TYPE_ALEXA: + return entity_id + + # Google Home + for number, ent_id in self.numbers.items(): + if entity_id == ent_id: + return number + + number = "1" + if self.numbers: + number = str(max(int(k) for k in self.numbers) + 1) + self.numbers[number] = entity_id + assert self.store is not None + self.store.async_delay_save(lambda: self.numbers, SAVE_DELAY) + return number + + def number_to_entity_id(self, number: str) -> str | None: + """Convert unique number to entity id.""" + if self.type == TYPE_ALEXA: + return number + + # Google Home + return self.numbers.get(number) + + def get_entity_name(self, entity: State) -> str: + """Get the name of an entity.""" + if ( + entity.entity_id in self.entities + and CONF_ENTITY_NAME in self.entities[entity.entity_id] + ): + return self.entities[entity.entity_id][CONF_ENTITY_NAME] + + return entity.attributes.get(ATTR_EMULATED_HUE_NAME, entity.name) + + def is_entity_exposed(self, entity: State) -> bool: + """Cache determine if an entity should be exposed on the emulated bridge.""" + if (exposed := self._exposed_cache.get(entity.entity_id)) is not None: + return exposed + exposed = self._is_entity_exposed(entity) + self._exposed_cache[entity.entity_id] = exposed + return exposed + + def filter_exposed_entities(self, states: Iterable[State]) -> list[State]: + """Filter a list of all states down to exposed entities.""" + exposed: list[State] = [ + state for state in states if self.is_entity_exposed(state) + ] + return exposed + + def _is_entity_exposed(self, entity: State) -> bool: + """Determine if an entity should be exposed on the emulated bridge. + + Async friendly. + """ + if entity.attributes.get("view") is not None: + # Ignore entities that are views + return False + + if entity.entity_id in self._entities_with_hidden_attr_in_config: + return not self._entities_with_hidden_attr_in_config[entity.entity_id] + + if not self.expose_by_default: + return False + # Expose an entity if the entity's domain is exposed by default and + # the configuration doesn't explicitly exclude it from being + # exposed, or if the entity is explicitly exposed + if entity.domain in self.exposed_domains: + return True + + return False diff --git a/homeassistant/components/emulated_hue/const.py b/homeassistant/components/emulated_hue/const.py index bfd58c5a0e1..2bcd8cbac19 100644 --- a/homeassistant/components/emulated_hue/const.py +++ b/homeassistant/components/emulated_hue/const.py @@ -2,3 +2,5 @@ HUE_SERIAL_NUMBER = "001788FFFE23BFC2" HUE_UUID = "2f402f80-da50-11e1-9b23-001788255acc" + +DOMAIN = "emulated_hue" diff --git a/homeassistant/components/emulated_hue/hue_api.py b/homeassistant/components/emulated_hue/hue_api.py index b4f926afd31..e7a4876730c 100644 --- a/homeassistant/components/emulated_hue/hue_api.py +++ b/homeassistant/components/emulated_hue/hue_api.py @@ -1,10 +1,15 @@ """Support for a Hue API to control Home Assistant.""" +from __future__ import annotations + import asyncio import hashlib from http import HTTPStatus from ipaddress import ip_address import logging import time +from typing import Any + +from aiohttp import web from homeassistant import core from homeassistant.components import ( @@ -58,9 +63,12 @@ from homeassistant.const import ( STATE_ON, STATE_UNAVAILABLE, ) +from homeassistant.core import State from homeassistant.helpers.event import async_track_state_change_event from homeassistant.util.network import is_local +from .config import Config + _LOGGER = logging.getLogger(__name__) # How long to wait for a state change to happen @@ -111,7 +119,7 @@ class HueUnauthorizedUser(HomeAssistantView): extra_urls = ["/api/"] requires_auth = False - async def get(self, request): + async def get(self, request: web.Request) -> web.Response: """Handle a GET request.""" return self.json(UNAUTHORIZED_USER) @@ -124,8 +132,9 @@ class HueUsernameView(HomeAssistantView): extra_urls = ["/api/"] requires_auth = False - async def post(self, request): + async def post(self, request: web.Request) -> web.Response: """Handle a POST request.""" + assert request.remote is not None if not is_local(ip_address(request.remote)): return self.json_message("Only local IPs allowed", HTTPStatus.UNAUTHORIZED) @@ -147,13 +156,14 @@ class HueAllGroupsStateView(HomeAssistantView): name = "emulated_hue:all_groups:state" requires_auth = False - def __init__(self, config): + def __init__(self, config: Config) -> None: """Initialize the instance of the view.""" self.config = config @core.callback - def get(self, request, username): + def get(self, request: web.Request, username: str) -> web.Response: """Process a request to make the Brilliant Lightpad work.""" + assert request.remote is not None if not is_local(ip_address(request.remote)): return self.json_message("Only local IPs allowed", HTTPStatus.UNAUTHORIZED) @@ -167,13 +177,14 @@ class HueGroupView(HomeAssistantView): name = "emulated_hue:groups:state" requires_auth = False - def __init__(self, config): + def __init__(self, config: Config) -> None: """Initialize the instance of the view.""" self.config = config @core.callback - def put(self, request, username): + def put(self, request: web.Request, username: str) -> web.Response: """Process a request to make the Logitech Pop working.""" + assert request.remote is not None if not is_local(ip_address(request.remote)): return self.json_message("Only local IPs allowed", HTTPStatus.UNAUTHORIZED) @@ -197,13 +208,14 @@ class HueAllLightsStateView(HomeAssistantView): name = "emulated_hue:lights:state" requires_auth = False - def __init__(self, config): + def __init__(self, config: Config) -> None: """Initialize the instance of the view.""" self.config = config @core.callback - def get(self, request, username): + def get(self, request: web.Request, username: str) -> web.Response: """Process a request to get the list of available lights.""" + assert request.remote is not None if not is_local(ip_address(request.remote)): return self.json_message("Only local IPs allowed", HTTPStatus.UNAUTHORIZED) @@ -217,13 +229,14 @@ class HueFullStateView(HomeAssistantView): name = "emulated_hue:username:state" requires_auth = False - def __init__(self, config): + def __init__(self, config: Config) -> None: """Initialize the instance of the view.""" self.config = config @core.callback - def get(self, request, username): + def get(self, request: web.Request, username: str) -> web.Response: """Process a request to get the list of available lights.""" + assert request.remote is not None if not is_local(ip_address(request.remote)): return self.json_message("only local IPs allowed", HTTPStatus.UNAUTHORIZED) if username != HUE_API_USERNAME: @@ -245,13 +258,14 @@ class HueConfigView(HomeAssistantView): name = "emulated_hue:username:config" requires_auth = False - def __init__(self, config): + def __init__(self, config: Config) -> None: """Initialize the instance of the view.""" self.config = config @core.callback - def get(self, request, username=""): + def get(self, request: web.Request, username: str = "") -> web.Response: """Process a request to get the configuration.""" + assert request.remote is not None if not is_local(ip_address(request.remote)): return self.json_message("only local IPs allowed", HTTPStatus.UNAUTHORIZED) @@ -267,17 +281,18 @@ class HueOneLightStateView(HomeAssistantView): name = "emulated_hue:light:state" requires_auth = False - def __init__(self, config): + def __init__(self, config: Config) -> None: """Initialize the instance of the view.""" self.config = config @core.callback - def get(self, request, username, entity_id): + def get(self, request: web.Request, username: str, entity_id: str) -> web.Response: """Process a request to get the state of an individual light.""" + assert request.remote is not None if not is_local(ip_address(request.remote)): return self.json_message("Only local IPs allowed", HTTPStatus.UNAUTHORIZED) - hass = request.app["hass"] + hass: core.HomeAssistant = request.app["hass"] hass_entity_id = self.config.number_to_entity_id(entity_id) if hass_entity_id is None: @@ -307,17 +322,20 @@ class HueOneLightChangeView(HomeAssistantView): name = "emulated_hue:light:state" requires_auth = False - def __init__(self, config): + def __init__(self, config: Config) -> None: """Initialize the instance of the view.""" self.config = config - async def put(self, request, username, entity_number): # noqa: C901 + async def put( # noqa: C901 + self, request: web.Request, username: str, entity_number: str + ) -> web.Response: """Process a request to set the state of an individual light.""" + assert request.remote is not None if not is_local(ip_address(request.remote)): return self.json_message("Only local IPs allowed", HTTPStatus.UNAUTHORIZED) config = self.config - hass = request.app["hass"] + hass: core.HomeAssistant = request.app["hass"] entity_id = config.number_to_entity_id(entity_number) if entity_id is None: @@ -344,7 +362,7 @@ class HueOneLightChangeView(HomeAssistantView): color_modes = entity.attributes.get(light.ATTR_SUPPORTED_COLOR_MODES, []) # Parse the request - parsed = { + parsed: dict[str, Any] = { STATE_ON: False, STATE_BRIGHTNESS: None, STATE_HUE: None, @@ -416,10 +434,10 @@ class HueOneLightChangeView(HomeAssistantView): turn_on_needed = False # Convert the resulting "on" status into the service we need to call - service = SERVICE_TURN_ON if parsed[STATE_ON] else SERVICE_TURN_OFF + service: str | None = SERVICE_TURN_ON if parsed[STATE_ON] else SERVICE_TURN_OFF # Construct what we need to send to the service - data = {ATTR_ENTITY_ID: entity_id} + data: dict[str, Any] = {ATTR_ENTITY_ID: entity_id} # If the requested entity is a light, set the brightness, hue, # saturation and color temp @@ -596,7 +614,7 @@ class HueOneLightChangeView(HomeAssistantView): return self.json(json_response) -def get_entity_state(config, entity): +def get_entity_state(config: Config, entity: State) -> dict[str, Any]: """Retrieve and convert state and brightness values for an entity.""" cached_state_entry = config.cached_states.get(entity.entity_id, None) cached_state = None @@ -617,7 +635,7 @@ def get_entity_state(config, entity): # Remove the now stale cached entry. config.cached_states.pop(entity.entity_id) - data = { + data: dict[str, Any] = { STATE_ON: False, STATE_BRIGHTNESS: None, STATE_HUE: None, @@ -700,7 +718,7 @@ def get_entity_state(config, entity): return data -def entity_to_json(config, entity): +def entity_to_json(config: Config, entity: State) -> dict[str, Any]: """Convert an entity to its Hue bridge JSON representation.""" entity_features = entity.attributes.get(ATTR_SUPPORTED_FEATURES, 0) color_modes = entity.attributes.get(light.ATTR_SUPPORTED_COLOR_MODES, []) @@ -709,7 +727,7 @@ def entity_to_json(config, entity): state = get_entity_state(config, entity) - retval = { + retval: dict[str, Any] = { "state": { HUE_API_STATE_ON: state[STATE_ON], "reachable": entity.state != STATE_UNAVAILABLE, @@ -793,13 +811,15 @@ def entity_to_json(config, entity): return retval -def create_hue_success_response(entity_number, attr, value): +def create_hue_success_response( + entity_number: str, attr: str, value: str +) -> dict[str, Any]: """Create a success response for an attribute set on a light.""" success_key = f"/lights/{entity_number}/state/{attr}" return {"success": {success_key: value}} -def create_config_model(config, request): +def create_config_model(config: Config, request: web.Request) -> dict[str, Any]: """Create a config resource.""" return { "mac": "00:00:00:00:00:00", @@ -811,29 +831,29 @@ def create_config_model(config, request): } -def create_list_of_entities(config, request): +def create_list_of_entities(config: Config, request: web.Request) -> dict[str, Any]: """Create a list of all entities.""" - hass = request.app["hass"] - json_response = {} - - for entity in config.filter_exposed_entities(hass.states.async_all()): - number = config.entity_id_to_number(entity.entity_id) - json_response[number] = entity_to_json(config, entity) - + hass: core.HomeAssistant = request.app["hass"] + json_response: dict[str, Any] = { + config.entity_id_to_number(entity.entity_id): entity_to_json(config, entity) + for entity in config.filter_exposed_entities(hass.states.async_all()) + } return json_response -def hue_brightness_to_hass(value): +def hue_brightness_to_hass(value: int) -> int: """Convert hue brightness 1..254 to hass format 0..255.""" return min(255, round((value / HUE_API_STATE_BRI_MAX) * 255)) -def hass_to_hue_brightness(value): +def hass_to_hue_brightness(value: int) -> int: """Convert hass brightness 0..255 to hue 1..254 scale.""" return max(1, round((value / 255) * HUE_API_STATE_BRI_MAX)) -async def wait_for_state_change_or_timeout(hass, entity_id, timeout): +async def wait_for_state_change_or_timeout( + hass: core.HomeAssistant, entity_id: str, timeout: float +) -> None: """Wait for an entity to change state.""" ev = asyncio.Event() diff --git a/tests/components/emulated_hue/test_init.py b/tests/components/emulated_hue/test_init.py index 93bf8c0631f..024b0f3ddf7 100644 --- a/tests/components/emulated_hue/test_init.py +++ b/tests/components/emulated_hue/test_init.py @@ -1,13 +1,14 @@ """Test the Emulated Hue component.""" from datetime import timedelta -from unittest.mock import patch +from unittest.mock import AsyncMock, patch -from homeassistant.components.emulated_hue import ( +from homeassistant.components.emulated_hue.config import ( DATA_KEY, DATA_VERSION, SAVE_DELAY, Config, ) +from homeassistant.const import EVENT_HOMEASSISTANT_START from homeassistant.setup import async_setup_component from homeassistant.util import utcnow @@ -121,6 +122,13 @@ async def test_setup_works(hass): """Test setup works.""" hass.config.components.add("network") with patch( - "homeassistant.components.emulated_hue.create_upnp_datagram_endpoint" - ), patch("homeassistant.components.emulated_hue.async_get_source_ip"): + "homeassistant.components.emulated_hue.create_upnp_datagram_endpoint", + AsyncMock(), + ) as mock_create_upnp_datagram_endpoint, patch( + "homeassistant.components.emulated_hue.async_get_source_ip" + ): assert await async_setup_component(hass, "emulated_hue", {}) + + hass.bus.async_fire(EVENT_HOMEASSISTANT_START) + await hass.async_block_till_done() + assert len(mock_create_upnp_datagram_endpoint.mock_calls) == 2 diff --git a/tests/components/emulated_hue/test_upnp.py b/tests/components/emulated_hue/test_upnp.py index ec04ee7e19c..79daaadbbc9 100644 --- a/tests/components/emulated_hue/test_upnp.py +++ b/tests/components/emulated_hue/test_upnp.py @@ -2,6 +2,7 @@ from http import HTTPStatus import json import unittest +from unittest.mock import patch from aiohttp import web import defusedxml.ElementTree as ET @@ -52,11 +53,13 @@ def hue_client(aiohttp_client): async def setup_hue(hass): """Set up the emulated_hue integration.""" - assert await setup.async_setup_component( - hass, - emulated_hue.DOMAIN, - {emulated_hue.DOMAIN: {emulated_hue.CONF_LISTEN_PORT: BRIDGE_SERVER_PORT}}, - ) + with patch("homeassistant.components.emulated_hue.create_upnp_datagram_endpoint"): + assert await setup.async_setup_component( + hass, + emulated_hue.DOMAIN, + {emulated_hue.DOMAIN: {emulated_hue.CONF_LISTEN_PORT: BRIDGE_SERVER_PORT}}, + ) + await hass.async_block_till_done() def test_upnp_discovery_basic():