Home Assistant Unofficial Reference 2024.12.1
hue_api.py
Go to the documentation of this file.
1 """Support for a Hue API to control Home Assistant."""
2 
3 from __future__ import annotations
4 
5 import asyncio
6 from collections.abc import Iterable
7 from functools import lru_cache
8 import hashlib
9 from http import HTTPStatus
10 from ipaddress import ip_address
11 import logging
12 import time
13 from typing import Any
14 
15 from aiohttp import web
16 
17 from homeassistant import core
18 from homeassistant.components import (
19  climate,
20  cover,
21  fan,
22  humidifier,
23  light,
24  media_player,
25  scene,
26  script,
27 )
29  SERVICE_SET_TEMPERATURE,
30  ClimateEntityFeature,
31 )
33  ATTR_CURRENT_POSITION,
34  ATTR_POSITION,
35  CoverEntityFeature,
36 )
37 from homeassistant.components.fan import ATTR_PERCENTAGE, FanEntityFeature
38 from homeassistant.components.http import KEY_HASS, HomeAssistantView
39 from homeassistant.components.humidifier import ATTR_HUMIDITY, SERVICE_SET_HUMIDITY
41  ATTR_BRIGHTNESS,
42  ATTR_COLOR_TEMP,
43  ATTR_HS_COLOR,
44  ATTR_TRANSITION,
45  ATTR_XY_COLOR,
46  ColorMode,
47  LightEntityFeature,
48 )
50  ATTR_MEDIA_VOLUME_LEVEL,
51  MediaPlayerEntityFeature,
52 )
53 from homeassistant.const import (
54  ATTR_ENTITY_ID,
55  ATTR_SUPPORTED_FEATURES,
56  ATTR_TEMPERATURE,
57  SERVICE_CLOSE_COVER,
58  SERVICE_OPEN_COVER,
59  SERVICE_SET_COVER_POSITION,
60  SERVICE_TURN_OFF,
61  SERVICE_TURN_ON,
62  SERVICE_VOLUME_SET,
63  STATE_CLOSED,
64  STATE_OFF,
65  STATE_ON,
66  STATE_UNAVAILABLE,
67 )
68 from homeassistant.core import Event, EventStateChangedData, State
69 from homeassistant.helpers.event import async_track_state_change_event
70 from homeassistant.util.json import json_loads
71 from homeassistant.util.network import is_local
72 
73 from .config import Config
74 
75 _LOGGER = logging.getLogger(__name__)
76 _OFF_STATES: dict[str, str] = {cover.DOMAIN: STATE_CLOSED}
77 
78 # How long to wait for a state change to happen
79 STATE_CHANGE_WAIT_TIMEOUT = 5.0
80 # How long an entry state's cache will be valid for in seconds.
81 STATE_CACHED_TIMEOUT = 2.0
82 
83 STATE_BRIGHTNESS = "bri"
84 STATE_COLORMODE = "colormode"
85 STATE_HUE = "hue"
86 STATE_SATURATION = "sat"
87 STATE_COLOR_TEMP = "ct"
88 STATE_TRANSITION = "tt"
89 STATE_XY = "xy"
90 
91 # Hue API states, defined separately in case they change
92 HUE_API_STATE_ON = "on"
93 HUE_API_STATE_BRI = "bri"
94 HUE_API_STATE_COLORMODE = "colormode"
95 HUE_API_STATE_HUE = "hue"
96 HUE_API_STATE_SAT = "sat"
97 HUE_API_STATE_CT = "ct"
98 HUE_API_STATE_XY = "xy"
99 HUE_API_STATE_EFFECT = "effect"
100 HUE_API_STATE_TRANSITION = "transitiontime"
101 
102 # Hue API min/max values - https://developers.meethue.com/develop/hue-api/lights-api/
103 HUE_API_STATE_BRI_MIN = 1 # Brightness
104 HUE_API_STATE_BRI_MAX = 254
105 HUE_API_STATE_HUE_MIN = 0 # Hue
106 HUE_API_STATE_HUE_MAX = 65535
107 HUE_API_STATE_SAT_MIN = 0 # Saturation
108 HUE_API_STATE_SAT_MAX = 254
109 HUE_API_STATE_CT_MIN = 153 # Color temp
110 HUE_API_STATE_CT_MAX = 500
111 
112 HUE_API_USERNAME = "nouser"
113 UNAUTHORIZED_USER = [
114  {"error": {"address": "/", "description": "unauthorized user", "type": "1"}}
115 ]
116 
117 DIMMABLE_SUPPORTED_FEATURES_BY_DOMAIN = {
118  cover.DOMAIN: CoverEntityFeature.SET_POSITION,
119  fan.DOMAIN: FanEntityFeature.SET_SPEED,
120  media_player.DOMAIN: MediaPlayerEntityFeature.VOLUME_SET,
121  climate.DOMAIN: ClimateEntityFeature.TARGET_TEMPERATURE,
122 }
123 
124 ENTITY_FEATURES_BY_DOMAIN = {
125  cover.DOMAIN: CoverEntityFeature,
126  fan.DOMAIN: FanEntityFeature,
127  media_player.DOMAIN: MediaPlayerEntityFeature,
128  climate.DOMAIN: ClimateEntityFeature,
129 }
130 
131 
132 @lru_cache(maxsize=32)
133 def _remote_is_allowed(address: str) -> bool:
134  """Check if remote address is allowed."""
135  return is_local(ip_address(address))
136 
137 
138 class HueUnauthorizedUser(HomeAssistantView):
139  """Handle requests to find the emulated hue bridge."""
140 
141  url = "/api"
142  name = "emulated_hue:api:unauthorized_user"
143  extra_urls = ["/api/"]
144  requires_auth = False
145 
146  async def get(self, request: web.Request) -> web.Response:
147  """Handle a GET request."""
148  return self.json(UNAUTHORIZED_USER)
149 
150 
151 class HueUsernameView(HomeAssistantView):
152  """Handle requests to create a username for the emulated hue bridge."""
153 
154  url = "/api"
155  name = "emulated_hue:api:create_username"
156  extra_urls = ["/api/"]
157  requires_auth = False
158 
159  async def post(self, request: web.Request) -> web.Response:
160  """Handle a POST request."""
161  assert request.remote is not None
162  if not _remote_is_allowed(request.remote):
163  return self.json_message("Only local IPs allowed", HTTPStatus.UNAUTHORIZED)
164 
165  try:
166  data = await request.json(loads=json_loads)
167  except ValueError:
168  return self.json_message("Invalid JSON", HTTPStatus.BAD_REQUEST)
169 
170  if "devicetype" not in data:
171  return self.json_message("devicetype not specified", HTTPStatus.BAD_REQUEST)
172 
173  return self.json([{"success": {"username": HUE_API_USERNAME}}])
174 
175 
176 class HueAllGroupsStateView(HomeAssistantView):
177  """Handle requests for getting info about entity groups."""
178 
179  url = "/api/{username}/groups"
180  name = "emulated_hue:all_groups:state"
181  requires_auth = False
182 
183  def __init__(self, config: Config) -> None:
184  """Initialize the instance of the view."""
185  self.configconfig = config
186 
187  @core.callback
188  def get(self, request: web.Request, username: str) -> web.Response:
189  """Process a request to make the Brilliant Lightpad work."""
190  assert request.remote is not None
191  if not _remote_is_allowed(request.remote):
192  return self.json_message("Only local IPs allowed", HTTPStatus.UNAUTHORIZED)
193 
194  return self.json({})
195 
196 
197 class HueGroupView(HomeAssistantView):
198  """Group handler to get Logitech Pop working."""
199 
200  url = "/api/{username}/groups/0/action"
201  name = "emulated_hue:groups:state"
202  requires_auth = False
203 
204  def __init__(self, config: Config) -> None:
205  """Initialize the instance of the view."""
206  self.configconfig = config
207 
208  @core.callback
209  def put(self, request: web.Request, username: str) -> web.Response:
210  """Process a request to make the Logitech Pop working."""
211  assert request.remote is not None
212  if not _remote_is_allowed(request.remote):
213  return self.json_message("Only local IPs allowed", HTTPStatus.UNAUTHORIZED)
214 
215  return self.json(
216  [
217  {
218  "error": {
219  "address": "/groups/0/action/scene",
220  "type": 7,
221  "description": "invalid value, dummy for parameter, scene",
222  }
223  }
224  ]
225  )
226 
227 
228 class HueAllLightsStateView(HomeAssistantView):
229  """Handle requests for getting info about all entities."""
230 
231  url = "/api/{username}/lights"
232  name = "emulated_hue:lights:state"
233  requires_auth = False
234 
235  def __init__(self, config: Config) -> None:
236  """Initialize the instance of the view."""
237  self.configconfig = config
238 
239  @core.callback
240  def get(self, request: web.Request, username: str) -> web.Response:
241  """Process a request to get the list of available lights."""
242  assert request.remote is not None
243  if not _remote_is_allowed(request.remote):
244  return self.json_message("Only local IPs allowed", HTTPStatus.UNAUTHORIZED)
245 
246  return self.json(create_list_of_entities(self.configconfig, request))
247 
248 
249 class HueFullStateView(HomeAssistantView):
250  """Return full state view of emulated hue."""
251 
252  url = "/api/{username}"
253  name = "emulated_hue:username:state"
254  requires_auth = False
255 
256  def __init__(self, config: Config) -> None:
257  """Initialize the instance of the view."""
258  self.configconfig = config
259 
260  @core.callback
261  def get(self, request: web.Request, username: str) -> web.Response:
262  """Process a request to get the list of available lights."""
263  assert request.remote is not None
264  if not _remote_is_allowed(request.remote):
265  return self.json_message("only local IPs allowed", HTTPStatus.UNAUTHORIZED)
266  if username != HUE_API_USERNAME:
267  return self.json(UNAUTHORIZED_USER)
268 
269  json_response = {
270  "lights": create_list_of_entities(self.configconfig, request),
271  "config": create_config_model(self.configconfig, request),
272  }
273 
274  return self.json(json_response)
275 
276 
277 class HueConfigView(HomeAssistantView):
278  """Return config view of emulated hue."""
279 
280  url = "/api/{username}/config"
281  extra_urls = ["/api/config"]
282  name = "emulated_hue:username:config"
283  requires_auth = False
284 
285  def __init__(self, config: Config) -> None:
286  """Initialize the instance of the view."""
287  self.configconfig = config
288 
289  @core.callback
290  def get(self, request: web.Request, username: str = "") -> web.Response:
291  """Process a request to get the configuration."""
292  assert request.remote is not None
293  if not _remote_is_allowed(request.remote):
294  return self.json_message("only local IPs allowed", HTTPStatus.UNAUTHORIZED)
295 
296  json_response = create_config_model(self.configconfig, request)
297 
298  return self.json(json_response)
299 
300 
301 class HueOneLightStateView(HomeAssistantView):
302  """Handle requests for getting info about a single entity."""
303 
304  url = "/api/{username}/lights/{entity_id}"
305  name = "emulated_hue:light:state"
306  requires_auth = False
307 
308  def __init__(self, config: Config) -> None:
309  """Initialize the instance of the view."""
310  self.configconfig = config
311 
312  @core.callback
313  def get(self, request: web.Request, username: str, entity_id: str) -> web.Response:
314  """Process a request to get the state of an individual light."""
315  assert request.remote is not None
316  if not _remote_is_allowed(request.remote):
317  return self.json_message("Only local IPs allowed", HTTPStatus.UNAUTHORIZED)
318 
319  hass = request.app[KEY_HASS]
320  hass_entity_id = self.configconfig.number_to_entity_id(entity_id)
321 
322  if hass_entity_id is None:
323  _LOGGER.error(
324  "Unknown entity number: %s not found in emulated_hue_ids.json",
325  entity_id,
326  )
327  return self.json_message("Entity not found", HTTPStatus.NOT_FOUND)
328 
329  if (state := hass.states.get(hass_entity_id)) is None:
330  _LOGGER.error("Entity not found: %s", hass_entity_id)
331  return self.json_message("Entity not found", HTTPStatus.NOT_FOUND)
332 
333  if not self.configconfig.is_state_exposed(state):
334  _LOGGER.error("Entity not exposed: %s", entity_id)
335  return self.json_message("Entity not exposed", HTTPStatus.UNAUTHORIZED)
336 
337  json_response = state_to_json(self.configconfig, state)
338 
339  return self.json(json_response)
340 
341 
342 class HueOneLightChangeView(HomeAssistantView):
343  """Handle requests for setting info about entities."""
344 
345  url = "/api/{username}/lights/{entity_number}/state"
346  name = "emulated_hue:light:state"
347  requires_auth = False
348 
349  def __init__(self, config: Config) -> None:
350  """Initialize the instance of the view."""
351  self.configconfig = config
352 
353  async def put( # noqa: C901
354  self, request: web.Request, username: str, entity_number: str
355  ) -> web.Response:
356  """Process a request to set the state of an individual light."""
357  assert request.remote is not None
358  if not _remote_is_allowed(request.remote):
359  return self.json_message("Only local IPs allowed", HTTPStatus.UNAUTHORIZED)
360 
361  config = self.configconfig
362  hass = request.app[KEY_HASS]
363  entity_id = config.number_to_entity_id(entity_number)
364 
365  if entity_id is None:
366  _LOGGER.error("Unknown entity number: %s", entity_number)
367  return self.json_message("Entity not found", HTTPStatus.NOT_FOUND)
368 
369  if (entity := hass.states.get(entity_id)) is None:
370  _LOGGER.error("Entity not found: %s", entity_id)
371  return self.json_message("Entity not found", HTTPStatus.NOT_FOUND)
372 
373  if not config.is_state_exposed(entity):
374  _LOGGER.error("Entity not exposed: %s", entity_id)
375  return self.json_message("Entity not exposed", HTTPStatus.UNAUTHORIZED)
376 
377  try:
378  request_json = await request.json()
379  except ValueError:
380  _LOGGER.error("Received invalid json")
381  return self.json_message("Invalid JSON", HTTPStatus.BAD_REQUEST)
382 
383  # Get the entity's supported features
384  entity_features = entity.attributes.get(ATTR_SUPPORTED_FEATURES, 0)
385  if entity.domain == light.DOMAIN:
386  color_modes = entity.attributes.get(light.ATTR_SUPPORTED_COLOR_MODES) or []
387 
388  # Parse the request
389  parsed: dict[str, Any] = {
390  STATE_ON: False,
391  STATE_BRIGHTNESS: None,
392  STATE_HUE: None,
393  STATE_SATURATION: None,
394  STATE_COLOR_TEMP: None,
395  STATE_XY: None,
396  STATE_TRANSITION: None,
397  }
398 
399  if HUE_API_STATE_ON in request_json:
400  if not isinstance(request_json[HUE_API_STATE_ON], bool):
401  _LOGGER.error("Unable to parse data: %s", request_json)
402  return self.json_message("Bad request", HTTPStatus.BAD_REQUEST)
403  parsed[STATE_ON] = request_json[HUE_API_STATE_ON]
404  else:
405  parsed[STATE_ON] = _hass_to_hue_state(entity)
406 
407  for key, attr in (
408  (HUE_API_STATE_BRI, STATE_BRIGHTNESS),
409  (HUE_API_STATE_HUE, STATE_HUE),
410  (HUE_API_STATE_SAT, STATE_SATURATION),
411  (HUE_API_STATE_CT, STATE_COLOR_TEMP),
412  (HUE_API_STATE_TRANSITION, STATE_TRANSITION),
413  ):
414  if key in request_json:
415  try:
416  parsed[attr] = int(request_json[key])
417  except ValueError:
418  _LOGGER.error("Unable to parse data (2): %s", request_json)
419  return self.json_message("Bad request", HTTPStatus.BAD_REQUEST)
420  if HUE_API_STATE_XY in request_json:
421  try:
422  parsed[STATE_XY] = (
423  float(request_json[HUE_API_STATE_XY][0]),
424  float(request_json[HUE_API_STATE_XY][1]),
425  )
426  except ValueError:
427  _LOGGER.error("Unable to parse data (2): %s", request_json)
428  return self.json_message("Bad request", HTTPStatus.BAD_REQUEST)
429 
430  if HUE_API_STATE_BRI in request_json:
431  if entity.domain == light.DOMAIN:
432  if light.brightness_supported(color_modes):
433  parsed[STATE_ON] = parsed[STATE_BRIGHTNESS] > 0
434  else:
435  parsed[STATE_BRIGHTNESS] = None
436 
437  elif entity.domain == scene.DOMAIN:
438  parsed[STATE_BRIGHTNESS] = None
439  parsed[STATE_ON] = True
440 
441  elif entity.domain in [
442  script.DOMAIN,
443  media_player.DOMAIN,
444  fan.DOMAIN,
445  cover.DOMAIN,
446  climate.DOMAIN,
447  humidifier.DOMAIN,
448  ]:
449  # Convert 0-254 to 0-100
450  level = (parsed[STATE_BRIGHTNESS] / HUE_API_STATE_BRI_MAX) * 100
451  parsed[STATE_BRIGHTNESS] = round(level)
452  parsed[STATE_ON] = True
453 
454  # Choose general HA domain
455  domain = core.DOMAIN
456 
457  # Entity needs separate call to turn on
458  turn_on_needed = False
459 
460  # Convert the resulting "on" status into the service we need to call
461  service: str | None = SERVICE_TURN_ON if parsed[STATE_ON] else SERVICE_TURN_OFF
462 
463  # Construct what we need to send to the service
464  data: dict[str, Any] = {ATTR_ENTITY_ID: entity_id}
465 
466  # If the requested entity is a light, set the brightness, hue,
467  # saturation and color temp
468  if entity.domain == light.DOMAIN:
469  if parsed[STATE_ON]:
470  if (
471  light.brightness_supported(color_modes)
472  and parsed[STATE_BRIGHTNESS] is not None
473  ):
474  data[ATTR_BRIGHTNESS] = hue_brightness_to_hass(
475  parsed[STATE_BRIGHTNESS]
476  )
477 
478  if light.color_supported(color_modes):
479  if any((parsed[STATE_HUE], parsed[STATE_SATURATION])):
480  if parsed[STATE_HUE] is not None:
481  hue = parsed[STATE_HUE]
482  else:
483  hue = 0
484 
485  if parsed[STATE_SATURATION] is not None:
486  sat = parsed[STATE_SATURATION]
487  else:
488  sat = 0
489 
490  # Convert hs values to hass hs values
491  hue = int((hue / HUE_API_STATE_HUE_MAX) * 360)
492  sat = int((sat / HUE_API_STATE_SAT_MAX) * 100)
493 
494  data[ATTR_HS_COLOR] = (hue, sat)
495 
496  if parsed[STATE_XY] is not None:
497  data[ATTR_XY_COLOR] = parsed[STATE_XY]
498 
499  if (
500  light.color_temp_supported(color_modes)
501  and parsed[STATE_COLOR_TEMP] is not None
502  ):
503  data[ATTR_COLOR_TEMP] = parsed[STATE_COLOR_TEMP]
504 
505  if (
506  entity_features & LightEntityFeature.TRANSITION
507  and parsed[STATE_TRANSITION] is not None
508  ):
509  data[ATTR_TRANSITION] = parsed[STATE_TRANSITION] / 10
510 
511  # If the requested entity is a script, add some variables
512  elif entity.domain == script.DOMAIN:
513  data["variables"] = {
514  "requested_state": STATE_ON if parsed[STATE_ON] else STATE_OFF
515  }
516 
517  if parsed[STATE_BRIGHTNESS] is not None:
518  data["variables"]["requested_level"] = parsed[STATE_BRIGHTNESS]
519 
520  # If the requested entity is a climate, set the temperature
521  elif entity.domain == climate.DOMAIN:
522  # We don't support turning climate devices on or off,
523  # only setting the temperature
524  service = None
525 
526  if (
527  entity_features & ClimateEntityFeature.TARGET_TEMPERATURE
528  and parsed[STATE_BRIGHTNESS] is not None
529  ):
530  domain = entity.domain
531  service = SERVICE_SET_TEMPERATURE
532  data[ATTR_TEMPERATURE] = parsed[STATE_BRIGHTNESS]
533 
534  # If the requested entity is a humidifier, set the humidity
535  elif entity.domain == humidifier.DOMAIN:
536  if parsed[STATE_BRIGHTNESS] is not None:
537  turn_on_needed = True
538  domain = entity.domain
539  service = SERVICE_SET_HUMIDITY
540  data[ATTR_HUMIDITY] = parsed[STATE_BRIGHTNESS]
541 
542  # If the requested entity is a media player, convert to volume
543  elif entity.domain == media_player.DOMAIN:
544  if (
545  entity_features & MediaPlayerEntityFeature.VOLUME_SET
546  and parsed[STATE_BRIGHTNESS] is not None
547  ):
548  turn_on_needed = True
549  domain = entity.domain
550  service = SERVICE_VOLUME_SET
551  # Convert 0-100 to 0.0-1.0
552  data[ATTR_MEDIA_VOLUME_LEVEL] = parsed[STATE_BRIGHTNESS] / 100.0
553 
554  # If the requested entity is a cover, convert to open_cover/close_cover
555  elif entity.domain == cover.DOMAIN:
556  domain = entity.domain
557  if service == SERVICE_TURN_ON:
558  service = SERVICE_OPEN_COVER
559  else:
560  service = SERVICE_CLOSE_COVER
561 
562  if (
563  entity_features & CoverEntityFeature.SET_POSITION
564  and parsed[STATE_BRIGHTNESS] is not None
565  ):
566  domain = entity.domain
567  service = SERVICE_SET_COVER_POSITION
568  data[ATTR_POSITION] = parsed[STATE_BRIGHTNESS]
569 
570  # If the requested entity is a fan, convert to speed
571  elif (
572  entity.domain == fan.DOMAIN
573  and entity_features & FanEntityFeature.SET_SPEED
574  and parsed[STATE_BRIGHTNESS] is not None
575  ):
576  domain = entity.domain
577  # Convert 0-100 to a fan speed
578  data[ATTR_PERCENTAGE] = parsed[STATE_BRIGHTNESS]
579 
580  # Map the off command to on
581  if entity.domain in config.off_maps_to_on_domains:
582  service = SERVICE_TURN_ON
583 
584  # Separate call to turn on needed
585  if turn_on_needed:
586  await hass.services.async_call(
587  core.DOMAIN,
588  SERVICE_TURN_ON,
589  {ATTR_ENTITY_ID: entity_id},
590  blocking=False,
591  )
592 
593  if service is not None:
594  state_will_change = parsed[STATE_ON] != _hass_to_hue_state(entity)
595 
596  await hass.services.async_call(domain, service, data, blocking=False)
597 
598  if state_will_change:
599  # Wait for the state to change.
601  hass, entity_id, STATE_CACHED_TIMEOUT
602  )
603 
604  # Create success responses for all received keys
605  json_response = [
607  entity_number, HUE_API_STATE_ON, parsed[STATE_ON]
608  )
609  ]
610 
611  for key, val in (
612  (STATE_BRIGHTNESS, HUE_API_STATE_BRI),
613  (STATE_HUE, HUE_API_STATE_HUE),
614  (STATE_SATURATION, HUE_API_STATE_SAT),
615  (STATE_COLOR_TEMP, HUE_API_STATE_CT),
616  (STATE_XY, HUE_API_STATE_XY),
617  (STATE_TRANSITION, HUE_API_STATE_TRANSITION),
618  ):
619  if parsed[key] is not None:
620  json_response.append(
621  create_hue_success_response(entity_number, val, parsed[key])
622  )
623 
624  if entity.domain in config.off_maps_to_on_domains:
625  # Caching is required because things like scripts and scenes won't
626  # report as "off" to Alexa if an "off" command is received, because
627  # they'll map to "on". Thus, instead of reporting its actual
628  # status, we report what Alexa will want to see, which is the same
629  # as the actual requested command.
630  config.cached_states[entity_id] = [parsed, None]
631  else:
632  config.cached_states[entity_id] = [parsed, time.time()]
633 
634  return self.json(json_response)
635 
636 
637 def get_entity_state_dict(config: Config, entity: State) -> dict[str, Any]:
638  """Retrieve and convert state and brightness values for an entity."""
639  cached_state_entry = config.cached_states.get(entity.entity_id, None)
640  cached_state = None
641 
642  # Check if we have a cached entry, and if so if it hasn't expired.
643  if cached_state_entry is not None:
644  entry_state, entry_time = cached_state_entry
645  if entry_time is None:
646  # Handle the case where the entity is listed in config.off_maps_to_on_domains.
647  cached_state = entry_state
648  elif time.time() - entry_time < STATE_CACHED_TIMEOUT and entry_state[
649  STATE_ON
650  ] == _hass_to_hue_state(entity):
651  # We only want to use the cache if the actual state of the entity
652  # is in sync so that it can be detected as an error by Alexa.
653  cached_state = entry_state
654  else:
655  # Remove the now stale cached entry.
656  config.cached_states.pop(entity.entity_id)
657 
658  if cached_state is None:
659  return _build_entity_state_dict(entity)
660 
661  data: dict[str, Any] = cached_state
662  # Make sure brightness is valid
663  if data[STATE_BRIGHTNESS] is None:
664  data[STATE_BRIGHTNESS] = HUE_API_STATE_BRI_MAX if data[STATE_ON] else 0
665 
666  # Make sure hue/saturation are valid
667  if (data[STATE_HUE] is None) or (data[STATE_SATURATION] is None):
668  data[STATE_HUE] = 0
669  data[STATE_SATURATION] = 0
670 
671  # If the light is off, set the color to off
672  if data[STATE_BRIGHTNESS] == 0:
673  data[STATE_HUE] = 0
674  data[STATE_SATURATION] = 0
675 
676  _clamp_values(data)
677  return data
678 
679 
680 @lru_cache(maxsize=512)
681 def _build_entity_state_dict(entity: State) -> dict[str, Any]:
682  """Build a state dict for an entity."""
683  is_on = _hass_to_hue_state(entity)
684  data: dict[str, Any] = {
685  STATE_ON: is_on,
686  STATE_BRIGHTNESS: None,
687  STATE_HUE: None,
688  STATE_SATURATION: None,
689  STATE_COLOR_TEMP: None,
690  }
691  attributes = entity.attributes
692  if is_on:
693  data[STATE_BRIGHTNESS] = hass_to_hue_brightness(
694  attributes.get(ATTR_BRIGHTNESS) or 0
695  )
696  if (hue_sat := attributes.get(ATTR_HS_COLOR)) is not None:
697  hue = hue_sat[0]
698  sat = hue_sat[1]
699  # Convert hass hs values back to hue hs values
700  data[STATE_HUE] = int((hue / 360.0) * HUE_API_STATE_HUE_MAX)
701  data[STATE_SATURATION] = int((sat / 100.0) * HUE_API_STATE_SAT_MAX)
702  else:
703  data[STATE_HUE] = HUE_API_STATE_HUE_MIN
704  data[STATE_SATURATION] = HUE_API_STATE_SAT_MIN
705  data[STATE_COLOR_TEMP] = attributes.get(ATTR_COLOR_TEMP) or 0
706 
707  else:
708  data[STATE_BRIGHTNESS] = 0
709  data[STATE_HUE] = 0
710  data[STATE_SATURATION] = 0
711  data[STATE_COLOR_TEMP] = 0
712 
713  if entity.domain == climate.DOMAIN:
714  temperature = attributes.get(ATTR_TEMPERATURE, 0)
715  # Convert 0-100 to 0-254
716  data[STATE_BRIGHTNESS] = round(temperature * HUE_API_STATE_BRI_MAX / 100)
717  elif entity.domain == humidifier.DOMAIN:
718  humidity = attributes.get(ATTR_HUMIDITY, 0)
719  # Convert 0-100 to 0-254
720  data[STATE_BRIGHTNESS] = round(humidity * HUE_API_STATE_BRI_MAX / 100)
721  elif entity.domain == media_player.DOMAIN:
722  level = attributes.get(ATTR_MEDIA_VOLUME_LEVEL, 1.0 if is_on else 0.0)
723  # Convert 0.0-1.0 to 0-254
724  data[STATE_BRIGHTNESS] = round(min(1.0, level) * HUE_API_STATE_BRI_MAX)
725  elif entity.domain == fan.DOMAIN:
726  percentage = attributes.get(ATTR_PERCENTAGE) or 0
727  # Convert 0-100 to 0-254
728  data[STATE_BRIGHTNESS] = round(percentage * HUE_API_STATE_BRI_MAX / 100)
729  elif entity.domain == cover.DOMAIN:
730  level = attributes.get(ATTR_CURRENT_POSITION, 0)
731  data[STATE_BRIGHTNESS] = round(level / 100 * HUE_API_STATE_BRI_MAX)
732  _clamp_values(data)
733  return data
734 
735 
736 def _clamp_values(data: dict[str, Any]) -> None:
737  """Clamp brightness, hue, saturation, and color temp to valid values."""
738  for key, v_min, v_max in (
739  (STATE_BRIGHTNESS, HUE_API_STATE_BRI_MIN, HUE_API_STATE_BRI_MAX),
740  (STATE_HUE, HUE_API_STATE_HUE_MIN, HUE_API_STATE_HUE_MAX),
741  (STATE_SATURATION, HUE_API_STATE_SAT_MIN, HUE_API_STATE_SAT_MAX),
742  (STATE_COLOR_TEMP, HUE_API_STATE_CT_MIN, HUE_API_STATE_CT_MAX),
743  ):
744  if data[key] is not None:
745  data[key] = max(v_min, min(data[key], v_max))
746 
747 
748 @lru_cache(maxsize=1024)
749 def _entity_unique_id(entity_id: str) -> str:
750  """Return the emulated_hue unique id for the entity_id."""
751  unique_id = hashlib.md5(entity_id.encode()).hexdigest()
752  return (
753  f"00:{unique_id[0:2]}:{unique_id[2:4]}:"
754  f"{unique_id[4:6]}:{unique_id[6:8]}:{unique_id[8:10]}:"
755  f"{unique_id[10:12]}:{unique_id[12:14]}-{unique_id[14:16]}"
756  )
757 
758 
759 def state_to_json(config: Config, state: State) -> dict[str, Any]:
760  """Convert an entity to its Hue bridge JSON representation."""
761  color_modes = state.attributes.get(light.ATTR_SUPPORTED_COLOR_MODES) or []
762  unique_id = _entity_unique_id(state.entity_id)
763  state_dict = get_entity_state_dict(config, state)
764 
765  json_state: dict[str, str | bool | int] = {
766  HUE_API_STATE_ON: state_dict[STATE_ON],
767  "reachable": state.state != STATE_UNAVAILABLE,
768  "mode": "homeautomation",
769  }
770  retval: dict[str, str | dict[str, str | bool | int]] = {
771  "state": json_state,
772  "name": config.get_entity_name(state),
773  "uniqueid": unique_id,
774  "manufacturername": "Home Assistant",
775  "swversion": "123",
776  }
777  is_light = state.domain == light.DOMAIN
778  color_supported = is_light and light.color_supported(color_modes)
779  color_temp_supported = is_light and light.color_temp_supported(color_modes)
780  if color_supported and color_temp_supported:
781  # Extended Color light (Zigbee Device ID: 0x0210)
782  # Same as Color light, but which supports additional setting of color temperature
783  retval["type"] = "Extended color light"
784  retval["modelid"] = "HASS231"
785  json_state.update(
786  {
787  HUE_API_STATE_BRI: state_dict[STATE_BRIGHTNESS],
788  HUE_API_STATE_HUE: state_dict[STATE_HUE],
789  HUE_API_STATE_SAT: state_dict[STATE_SATURATION],
790  HUE_API_STATE_CT: state_dict[STATE_COLOR_TEMP],
791  HUE_API_STATE_EFFECT: "none",
792  }
793  )
794  if state_dict[STATE_HUE] > 0 or state_dict[STATE_SATURATION] > 0:
795  json_state[HUE_API_STATE_COLORMODE] = "hs"
796  else:
797  json_state[HUE_API_STATE_COLORMODE] = "ct"
798  elif color_supported:
799  # Color light (Zigbee Device ID: 0x0200)
800  # Supports on/off, dimming and color control (hue/saturation, enhanced hue, color loop and XY)
801  retval["type"] = "Color light"
802  retval["modelid"] = "HASS213"
803  json_state.update(
804  {
805  HUE_API_STATE_BRI: state_dict[STATE_BRIGHTNESS],
806  HUE_API_STATE_COLORMODE: "hs",
807  HUE_API_STATE_HUE: state_dict[STATE_HUE],
808  HUE_API_STATE_SAT: state_dict[STATE_SATURATION],
809  HUE_API_STATE_EFFECT: "none",
810  }
811  )
812  elif color_temp_supported:
813  # Color temperature light (Zigbee Device ID: 0x0220)
814  # Supports groups, scenes, on/off, dimming, and setting of a color temperature
815  retval["type"] = "Color temperature light"
816  retval["modelid"] = "HASS312"
817  json_state.update(
818  {
819  HUE_API_STATE_COLORMODE: "ct",
820  HUE_API_STATE_CT: state_dict[STATE_COLOR_TEMP],
821  HUE_API_STATE_BRI: state_dict[STATE_BRIGHTNESS],
822  }
823  )
824  elif state_supports_hue_brightness(state, color_modes):
825  # Dimmable light (Zigbee Device ID: 0x0100)
826  # Supports groups, scenes, on/off and dimming
827  retval["type"] = "Dimmable light"
828  retval["modelid"] = "HASS123"
829  json_state.update({HUE_API_STATE_BRI: state_dict[STATE_BRIGHTNESS]})
830  elif not config.lights_all_dimmable:
831  # On/Off light (ZigBee Device ID: 0x0000)
832  # Supports groups, scenes and on/off control
833  retval["type"] = "On/Off light"
834  retval["productname"] = "On/Off light"
835  retval["modelid"] = "HASS321"
836  else:
837  # Dimmable light (Zigbee Device ID: 0x0100)
838  # Supports groups, scenes, on/off and dimming
839  # Reports fixed brightness for compatibility with Alexa.
840  retval["type"] = "Dimmable light"
841  retval["modelid"] = "HASS123"
842  json_state.update({HUE_API_STATE_BRI: HUE_API_STATE_BRI_MAX})
843 
844  return retval
845 
846 
848  state: State, color_modes: Iterable[ColorMode]
849 ) -> bool:
850  """Return True if the state supports brightness."""
851  domain = state.domain
852  if domain == light.DOMAIN:
853  return light.brightness_supported(color_modes)
854  if not (required_feature := DIMMABLE_SUPPORTED_FEATURES_BY_DOMAIN.get(domain)):
855  return False
856  features = state.attributes.get(ATTR_SUPPORTED_FEATURES, 0)
857  enum = ENTITY_FEATURES_BY_DOMAIN[domain]
858  features = enum(features) if type(features) is int else features # noqa: E721
859  return required_feature in features
860 
861 
863  entity_number: str, attr: str, value: str
864 ) -> dict[str, Any]:
865  """Create a success response for an attribute set on a light."""
866  success_key = f"/lights/{entity_number}/state/{attr}"
867  return {"success": {success_key: value}}
868 
869 
870 def create_config_model(config: Config, request: web.Request) -> dict[str, Any]:
871  """Create a config resource."""
872  return {
873  "name": "HASS BRIDGE",
874  "mac": "00:00:00:00:00:00",
875  "swversion": "01003542",
876  "apiversion": "1.17.0",
877  "whitelist": {HUE_API_USERNAME: {"name": "HASS BRIDGE"}},
878  "ipaddress": f"{config.advertise_ip}:{config.advertise_port}",
879  "linkbutton": True,
880  }
881 
882 
883 def create_list_of_entities(config: Config, request: web.Request) -> dict[str, Any]:
884  """Create a list of all entities."""
885  hass = request.app[KEY_HASS]
886  return {
887  config.entity_id_to_number(entity_id): state_to_json(config, state)
888  for entity_id in config.get_exposed_entity_ids()
889  if (state := hass.states.get(entity_id))
890  }
891 
892 
893 def hue_brightness_to_hass(value: int) -> int:
894  """Convert hue brightness 1..254 to hass format 0..255."""
895  return min(255, round((value / HUE_API_STATE_BRI_MAX) * 255))
896 
897 
898 def hass_to_hue_brightness(value: int) -> int:
899  """Convert hass brightness 0..255 to hue 1..254 scale."""
900  return max(1, round((value / 255) * HUE_API_STATE_BRI_MAX))
901 
902 
903 def _hass_to_hue_state(entity: State) -> bool:
904  """Convert hass entity states to simple True/False on/off state for Hue."""
905  return entity.state != _OFF_STATES.get(entity.domain, STATE_OFF)
906 
907 
909  hass: core.HomeAssistant, entity_id: str, timeout: float
910 ) -> None:
911  """Wait for an entity to change state."""
912  ev = asyncio.Event()
913 
914  @core.callback
915  def _async_event_changed(event: Event[EventStateChangedData]) -> None:
916  ev.set()
917 
918  unsub = async_track_state_change_event(hass, [entity_id], _async_event_changed)
919 
920  try:
921  async with asyncio.timeout(STATE_CHANGE_WAIT_TIMEOUT):
922  await ev.wait()
923  except TimeoutError:
924  pass
925  finally:
926  unsub()
web.Response get(self, web.Request request, str username)
Definition: hue_api.py:188
web.Response get(self, web.Request request, str username)
Definition: hue_api.py:240
web.Response get(self, web.Request request, str username="")
Definition: hue_api.py:290
web.Response get(self, web.Request request, str username)
Definition: hue_api.py:261
web.Response put(self, web.Request request, str username)
Definition: hue_api.py:209
web.Response put(self, web.Request request, str username, str entity_number)
Definition: hue_api.py:355
web.Response get(self, web.Request request, str username, str entity_id)
Definition: hue_api.py:313
web.Response post(self, web.Request request)
Definition: hue_api.py:159
dict[str, Any] state_to_json(Config config, State state)
Definition: hue_api.py:759
dict[str, Any] create_list_of_entities(Config config, web.Request request)
Definition: hue_api.py:883
dict[str, Any] get_entity_state_dict(Config config, State entity)
Definition: hue_api.py:637
bool state_supports_hue_brightness(State state, Iterable[ColorMode] color_modes)
Definition: hue_api.py:849
None wait_for_state_change_or_timeout(core.HomeAssistant hass, str entity_id, float timeout)
Definition: hue_api.py:910
None _clamp_values(dict[str, Any] data)
Definition: hue_api.py:736
dict[str, Any] _build_entity_state_dict(State entity)
Definition: hue_api.py:681
dict[str, Any] create_hue_success_response(str entity_number, str attr, str value)
Definition: hue_api.py:864
dict[str, Any] create_config_model(Config config, web.Request request)
Definition: hue_api.py:870
bool is_local(ConfigEntry entry)
Definition: __init__.py:58
CALLBACK_TYPE async_track_state_change_event(HomeAssistant hass, str|Iterable[str] entity_ids, Callable[[Event[EventStateChangedData]], Any] action, HassJobType|None job_type=None)
Definition: event.py:314