evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add basic typing to emulated_hue (#72663)

Browse source 
* Add basic typing to emulated_hue

* type a few more places

* fixes

* numbers are always stringified

* numbers are always stringified

* coverage

* drop assert
This commit is contained in:
J. Nick Koston 2022-05-29 06:27:32 -10:00 • committed by GitHub
parent d603952872
commit 237ef6419b
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 324 additions and 249 deletions
Download patch file Download diff file Expand all files Collapse all files

233
homeassistant/components/emulated_hue/__init__.py
View file

@ -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

213
homeassistant/components/emulated_hue/config.py Normal file
View file

@ -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

2
homeassistant/components/emulated_hue/const.py
View file

@ -2,3 +2,5 @@
HUE_SERIAL_NUMBER = "001788FFFE23BFC2"
HUE_UUID = "2f402f80-da50-11e1-9b23-001788255acc"
DOMAIN = "emulated_hue"

96
homeassistant/components/emulated_hue/hue_api.py
View file

@ -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()

16
tests/components/emulated_hue/test_init.py
View file

@ -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

3
tests/components/emulated_hue/test_upnp.py
View file

@ -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."""
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():