Skip to content

Commit

Permalink
Merge pull request #2797 from alandtse/dev
Browse files Browse the repository at this point in the history
  • Loading branch information
@alandtse
alandtse authored Jan 14, 2025
2 parents 7c450e0 + 6152017 commit ab6c9f2
Show file tree
Hide file tree
Showing 5 changed files with 230 additions and 172 deletions.
202 changes: 118 additions & 84 deletions custom_components/alexa_media/alexa_entity.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -57,23 +57,30 @@ def is_skill(appliance: dict[str, Any]) -> bool:
return namespace and namespace == "SKILL"


def is_known_ha_bridge(appliance: Optional[dict[str, Any]]) -> bool:
"""Test whether a bridge appliance is a known HA bridge to avoid creating loops."""

if appliance is None:
return False

if appliance.get("manufacturerName") in ("t0bst4r", "Matterbridge"):
return True

# If we want to exclude all Matter devices (these can always be added
# directly to HA instead of going through AMP), we could test for a
# networkInterfaceIdentifier of type "MATTER" or capabilities on the
# "Alexa.Matter.NodeOperationalCredentials.FabricManagement" interface.

return False


def is_local(appliance: dict[str, Any]) -> bool:
"""Test whether locally connected.
This is mainly present to prevent loops with the official Alexa integration.
There is probably a better way to prevent that, but this works.
"""

if appliance.get("manufacturerName") == "t0bst4r":
# Home-Assistant-Matter-Hub is a new add-on (2024-10-27) which exposes selected
# HA entities to Alexa as Matter devices connected locally via Amazon Echo.
# "connectedVia" is not None so they need to be ignored to prevent duplicating them back into HA.
_LOGGER.debug(
'alexa_entity is_local: Return False for Home-Assistant-Matter-Hub manufacturer: "%s"',
appliance.get("manufacturerName"),
)
return False

if appliance.get("connectedVia"):
# connectedVia is a flag that determines which Echo devices holds the connection. Its blank for
# skill derived devices and includes an Echo name for zigbee and local devices.
Expand Down Expand Up @@ -181,6 +188,27 @@ def get_device_serial(appliance: dict[str, Any]) -> Optional[str]:
return None


def get_device_bridge(
appliance: dict[str, Any], appliances: dict[str, dict[str, Any]]
) -> Optional[dict[str, Any]]:
"""Find the bridge device for an appliance connected through e.g. a Matter bridge"""
if not appliance.get("connectedVia"):
# The appliance cannot be Matter if it does not connect to an Echo device
return None

# We expect the bridged devices to look like "AAA_SonarCloudService_UUID#DEVICENUM"
bridged_device_pattern = re.compile(
"(AAA_SonarCloudService_[a-f0-9\\-]+)#[0-9]+", flags=re.I
)

match = bridged_device_pattern.fullmatch(appliance.get("applianceId", ""))
if match is None:
return None

# We expect the bridge to share the prefix without the device num
return appliances[match.group(1)]


class AlexaEntity(TypedDict):
"""Class for Alexaentity."""

Expand Down Expand Up @@ -236,85 +264,91 @@ def parse_alexa_entities(network_details: Optional[dict[str, Any]]) -> AlexaEnti
contact_sensors = []
switches = []
location_details = network_details["locationDetails"]["locationDetails"]
# pylint: disable=too-many-nested-blocks

appliances = {}
for location in location_details.values():
amazon_bridge_details = location["amazonBridgeDetails"]["amazonBridgeDetails"]
for bridge in amazon_bridge_details.values():
appliance_details = bridge["applianceDetails"]["applianceDetails"]
for appliance in appliance_details.values():
processed_appliance = {
"id": appliance["entityId"],
"appliance_id": appliance["applianceId"],
"name": get_friendliest_name(appliance),
"is_hue_v1": is_hue_v1(appliance),
}
if is_alexa_guard(appliance):
guards.append(processed_appliance)
elif is_temperature_sensor(appliance):
serial = get_device_serial(appliance)
processed_appliance["device_serial"] = (
serial if serial else appliance["entityId"]
)
temperature_sensors.append(processed_appliance)
# Code for Amazon Smart Air Quality Monitor
elif is_air_quality_sensor(appliance):
serial = get_device_serial(appliance)
processed_appliance["device_serial"] = (
serial if serial else appliance["entityId"]
)
# create array of air quality sensors. We must store the instance id against
# the assetId so we know which sensors are which.
sensors = []
if (
appliance["friendlyDescription"]
== "Amazon Indoor Air Quality Monitor"
):
for cap in appliance["capabilities"]:
instance = cap.get("instance")
if instance:
friendlyName = cap["resources"].get("friendlyNames")
for entry in friendlyName:
assetId = entry["value"].get("assetId")
if assetId and assetId.startswith(
"Alexa.AirQuality"
):
unit = cap["configuration"]["unitOfMeasure"]
sensor = {
"sensorType": assetId,
"instance": instance,
"unit": unit,
}
sensors.append(sensor)
_LOGGER.debug(
"AIAQM sensor detected %s", sensor
)
processed_appliance["sensors"] = sensors

# Add as both temperature and air quality sensor
temperature_sensors.append(processed_appliance)
air_quality_sensors.append(processed_appliance)
elif is_switch(appliance):
switches.append(processed_appliance)
elif is_light(appliance):
processed_appliance["brightness"] = has_capability(
appliance, "Alexa.BrightnessController", "brightness"
)
processed_appliance["color"] = has_capability(
appliance, "Alexa.ColorController", "color"
)
processed_appliance["color_temperature"] = has_capability(
appliance,
"Alexa.ColorTemperatureController",
"colorTemperatureInKelvin",
)
lights.append(processed_appliance)
elif is_contact_sensor(appliance):
processed_appliance["battery_level"] = has_capability(
appliance, "Alexa.BatteryLevelSensor", "batteryLevel"
)
contact_sensors.append(processed_appliance)
else:
_LOGGER.debug("Found unsupported device %s", appliance)
appliances[appliance["applianceId"]] = appliance

for appliance in appliances.values():
device_bridge = get_device_bridge(appliance, appliances)
if is_known_ha_bridge(device_bridge):
_LOGGER.debug("Found Home Assistant bridge, skipping %s", appliance)
continue

processed_appliance = {
"id": appliance["entityId"],
"appliance_id": appliance["applianceId"],
"name": get_friendliest_name(appliance),
"is_hue_v1": is_hue_v1(appliance),
}
if is_alexa_guard(appliance):
guards.append(processed_appliance)
elif is_temperature_sensor(appliance):
serial = get_device_serial(appliance)
processed_appliance["device_serial"] = (
serial if serial else appliance["entityId"]
)
temperature_sensors.append(processed_appliance)
# Code for Amazon Smart Air Quality Monitor
elif is_air_quality_sensor(appliance):
serial = get_device_serial(appliance)
processed_appliance["device_serial"] = (
serial if serial else appliance["entityId"]
)
# create array of air quality sensors. We must store the instance id against
# the assetId so we know which sensors are which.
sensors = []
if appliance["friendlyDescription"] == "Amazon Indoor Air Quality Monitor":
for cap in appliance["capabilities"]:
instance = cap.get("instance")
if not instance:
continue

friendlyName = cap["resources"].get("friendlyNames")
for entry in friendlyName:
assetId = entry["value"].get("assetId")
if not assetId or not assetId.startswith("Alexa.AirQuality"):
continue

unit = cap["configuration"]["unitOfMeasure"]
sensor = {
"sensorType": assetId,
"instance": instance,
"unit": unit,
}
sensors.append(sensor)
_LOGGER.debug("AIAQM sensor detected %s", sensor)
processed_appliance["sensors"] = sensors

# Add as both temperature and air quality sensor
temperature_sensors.append(processed_appliance)
air_quality_sensors.append(processed_appliance)
elif is_switch(appliance):
switches.append(processed_appliance)
elif is_light(appliance):
processed_appliance["brightness"] = has_capability(
appliance, "Alexa.BrightnessController", "brightness"
)
processed_appliance["color"] = has_capability(
appliance, "Alexa.ColorController", "color"
)
processed_appliance["color_temperature"] = has_capability(
appliance,
"Alexa.ColorTemperatureController",
"colorTemperatureInKelvin",
)
lights.append(processed_appliance)
elif is_contact_sensor(appliance):
processed_appliance["battery_level"] = has_capability(
appliance, "Alexa.BatteryLevelSensor", "batteryLevel"
)
contact_sensors.append(processed_appliance)
else:
_LOGGER.debug("Found unsupported device %s", appliance)

return {
"light": lights,
Expand Down
58 changes: 33 additions & 25 deletions custom_components/alexa_media/config_flow.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -26,12 +26,19 @@
hide_email,
obfuscate,
)
from awesomeversion import AwesomeVersion
from homeassistant import config_entries
from homeassistant.components.http.view import HomeAssistantView
from homeassistant.components.persistent_notification import (
async_dismiss as async_dismiss_persistent_notification,
)
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD, CONF_SCAN_INTERVAL, CONF_URL
from homeassistant.const import (
CONF_EMAIL,
CONF_PASSWORD,
CONF_SCAN_INTERVAL,
CONF_URL,
__version__ as HAVERSION,
)
from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult, UnknownFlow
from homeassistant.exceptions import Unauthorized
Expand Down Expand Up @@ -871,7 +878,8 @@ class OptionsFlowHandler(config_entries.OptionsFlow):
def __init__(self, config_entry: config_entries.ConfigEntry) -> None:
"""Initialize options flow."""
self.config = OrderedDict()
self._config_entry = config_entry
if AwesomeVersion(HAVERSION) < "2024.12":
self.config_entry = config_entry

async def async_step_init(
self, user_input: dict[str, Any] | None = None
Expand All @@ -883,7 +891,7 @@ async def async_step_init(
(
vol.Optional(
CONF_PUBLIC_URL,
default=self._config_entry.data.get(
default=self.config_entry.data.get(
CONF_PUBLIC_URL, DEFAULT_PUBLIC_URL
),
),
Expand All @@ -892,28 +900,28 @@ async def async_step_init(
(
vol.Optional(
CONF_INCLUDE_DEVICES,
default=self._config_entry.data.get(CONF_INCLUDE_DEVICES, ""),
default=self.config_entry.data.get(CONF_INCLUDE_DEVICES, ""),
),
str,
),
(
vol.Optional(
CONF_EXCLUDE_DEVICES,
default=self._config_entry.data.get(CONF_EXCLUDE_DEVICES, ""),
default=self.config_entry.data.get(CONF_EXCLUDE_DEVICES, ""),
),
str,
),
(
vol.Optional(
CONF_SCAN_INTERVAL,
default=self._config_entry.data.get(CONF_SCAN_INTERVAL, 120),
default=self.config_entry.data.get(CONF_SCAN_INTERVAL, 120),
),
int,
),
(
vol.Optional(
CONF_QUEUE_DELAY,
default=self._config_entry.data.get(
default=self.config_entry.data.get(
CONF_QUEUE_DELAY, DEFAULT_QUEUE_DELAY
),
),
Expand All @@ -922,7 +930,7 @@ async def async_step_init(
(
vol.Optional(
CONF_EXTENDED_ENTITY_DISCOVERY,
default=self._config_entry.data.get(
default=self.config_entry.data.get(
CONF_EXTENDED_ENTITY_DISCOVERY,
DEFAULT_EXTENDED_ENTITY_DISCOVERY,
),
Expand All @@ -932,7 +940,7 @@ async def async_step_init(
(
vol.Optional(
CONF_DEBUG,
default=self._config_entry.data.get(CONF_DEBUG, DEFAULT_DEBUG),
default=self.config_entry.data.get(CONF_DEBUG, DEFAULT_DEBUG),
),
bool,
),
Expand All @@ -941,36 +949,36 @@ async def async_step_init(

if user_input is not None:
"""Preserve these parameters"""
if CONF_URL in self._config_entry.data:
user_input[CONF_URL] = self._config_entry.data[CONF_URL]
if CONF_EMAIL in self._config_entry.data:
user_input[CONF_EMAIL] = self._config_entry.data[CONF_EMAIL]
if CONF_PASSWORD in self._config_entry.data:
user_input[CONF_PASSWORD] = self._config_entry.data[CONF_PASSWORD]
if CONF_SECURITYCODE in self._config_entry.data:
user_input[CONF_SECURITYCODE] = self._config_entry.data[
if CONF_URL in self.config_entry.data:
user_input[CONF_URL] = self.config_entry.data[CONF_URL]
if CONF_EMAIL in self.config_entry.data:
user_input[CONF_EMAIL] = self.config_entry.data[CONF_EMAIL]
if CONF_PASSWORD in self.config_entry.data:
user_input[CONF_PASSWORD] = self.config_entry.data[CONF_PASSWORD]
if CONF_SECURITYCODE in self.config_entry.data:
user_input[CONF_SECURITYCODE] = self.config_entry.data[
CONF_SECURITYCODE
]
if CONF_OTPSECRET in self._config_entry.data:
user_input[CONF_OTPSECRET] = self._config_entry.data[CONF_OTPSECRET]
if CONF_OAUTH in self._config_entry.data:
user_input[CONF_OAUTH] = self._config_entry.data[CONF_OAUTH]
if CONF_OTPSECRET in self.config_entry.data:
user_input[CONF_OTPSECRET] = self.config_entry.data[CONF_OTPSECRET]
if CONF_OAUTH in self.config_entry.data:
user_input[CONF_OAUTH] = self.config_entry.data[CONF_OAUTH]
"""Ensure public_url ends with trailing slash"""
if CONF_PUBLIC_URL in self._config_entry.data:
if CONF_PUBLIC_URL in self.config_entry.data:
if not user_input[CONF_PUBLIC_URL].endswith("/"):
user_input[CONF_PUBLIC_URL] = user_input[CONF_PUBLIC_URL] + "/"
"""Remove leading/trailing spaces in device strings"""
if CONF_INCLUDE_DEVICES in self._config_entry.data:
if CONF_INCLUDE_DEVICES in self.config_entry.data:
user_input[CONF_INCLUDE_DEVICES] = user_input[
CONF_INCLUDE_DEVICES
].strip()
if CONF_EXCLUDE_DEVICES in self._config_entry.data:
if CONF_EXCLUDE_DEVICES in self.config_entry.data:
user_input[CONF_EXCLUDE_DEVICES] = user_input[
CONF_EXCLUDE_DEVICES
].strip()

self.hass.config_entries.async_update_entry(
self._config_entry, data=user_input, options=self._config_entry.options
self.config_entry, data=user_input, options=self.config_entry.options
)
return self.async_create_entry(title="", data={})

Expand Down
Loading

0 comments on commit ab6c9f2

Please sign in to comment.