|
| 1 | +"""Diagnostics support for Luxtronik.""" |
| 2 | +from __future__ import annotations |
| 3 | + |
| 4 | +from functools import partial |
| 5 | +from ipaddress import IPv6Address, ip_address |
| 6 | + |
| 7 | +from async_timeout import timeout |
| 8 | +from getmac import get_mac_address |
| 9 | +from homeassistant.components.diagnostics import async_redact_data |
| 10 | +from homeassistant.config_entries import ConfigEntry |
| 11 | +from homeassistant.const import (CONF_HOST, CONF_PASSWORD, CONF_PORT, |
| 12 | + CONF_USERNAME) |
| 13 | +from homeassistant.core import HomeAssistant |
| 14 | +from homeassistant.helpers import device_registry |
| 15 | +from luxtronik import Luxtronik as Lux |
| 16 | + |
| 17 | +from .const import CONF_COORDINATOR, DOMAIN |
| 18 | + |
| 19 | +TO_REDACT = {CONF_USERNAME, CONF_PASSWORD} |
| 20 | + |
| 21 | + |
| 22 | +async def async_get_config_entry_diagnostics( |
| 23 | + hass: HomeAssistant, entry: ConfigEntry |
| 24 | +) -> dict: |
| 25 | + """Return diagnostics for a config entry.""" |
| 26 | + data: dict = entry.data |
| 27 | + client = Lux(data[CONF_HOST], data[CONF_PORT], True) |
| 28 | + client.read() |
| 29 | + |
| 30 | + mac = "" |
| 31 | + async with timeout(10): |
| 32 | + mac = await _async_get_mac_address(hass, data[CONF_HOST]) |
| 33 | + mac = mac[:9] + '*' |
| 34 | + |
| 35 | + entry_data = async_redact_data(entry.as_dict(), TO_REDACT) |
| 36 | + if "data" not in entry_data: |
| 37 | + entry_data["data"] = {} |
| 38 | + entry_data["data"]["mac"] = mac |
| 39 | + diag_data = { |
| 40 | + "entry": entry_data, |
| 41 | + "parameters": _dump_items(client.parameters.parameters), |
| 42 | + "calculations": _dump_items(client.calculations.calculations), |
| 43 | + "visibilities": _dump_items(client.visibilities.visibilities), |
| 44 | + } |
| 45 | + return diag_data |
| 46 | + |
| 47 | + |
| 48 | +def _dump_items(items: dict) -> dict: |
| 49 | + dump = dict() |
| 50 | + for index, item in items.items(): |
| 51 | + dump[f"{index:<4d} {item.name:<60}"] = f"{items.get(index)}" |
| 52 | + return dump |
| 53 | + |
| 54 | + |
| 55 | +async def _async_get_mac_address(hass: HomeAssistant, host: str) -> str | None: |
| 56 | + """Get mac address from host name, IPv4 address, or IPv6 address.""" |
| 57 | + # Help mypy, which has trouble with the async_add_executor_job + partial call |
| 58 | + mac_address: str | None |
| 59 | + # getmac has trouble using IPv6 addresses as the "hostname" parameter so |
| 60 | + # assume host is an IP address, then handle the case it's not. |
| 61 | + try: |
| 62 | + ip_addr = ip_address(host) |
| 63 | + except ValueError: |
| 64 | + mac_address = await hass.async_add_executor_job( |
| 65 | + partial(get_mac_address, hostname=host) |
| 66 | + ) |
| 67 | + else: |
| 68 | + if ip_addr.version == 4: |
| 69 | + mac_address = await hass.async_add_executor_job( |
| 70 | + partial(get_mac_address, ip=host) |
| 71 | + ) |
| 72 | + else: |
| 73 | + # Drop scope_id from IPv6 address by converting via int |
| 74 | + ip_addr = IPv6Address(int(ip_addr)) |
| 75 | + mac_address = await hass.async_add_executor_job( |
| 76 | + partial(get_mac_address, ip6=str(ip_addr)) |
| 77 | + ) |
| 78 | + |
| 79 | + if not mac_address: |
| 80 | + return None |
| 81 | + |
| 82 | + return device_registry.format_mac(mac_address) |
| 83 | + |
| 84 | + |
0 commit comments