mirror of
https://github.com/home-assistant/core.git
synced 2025-07-19 11:17:21 +00:00
Adjust async_step_dhcp signature for strict typing (#59751)
Co-authored-by: epenet <epenet@users.noreply.github.com>
This commit is contained in:
parent
a78176e192
commit
476a59d248
@ -6,8 +6,7 @@ from urllib.parse import urlsplit
|
|||||||
import voluptuous as vol
|
import voluptuous as vol
|
||||||
|
|
||||||
from homeassistant import config_entries
|
from homeassistant import config_entries
|
||||||
from homeassistant.components import zeroconf
|
from homeassistant.components import dhcp, zeroconf
|
||||||
from homeassistant.components.dhcp import HOSTNAME, IP_ADDRESS, MAC_ADDRESS
|
|
||||||
from homeassistant.config_entries import SOURCE_IGNORE
|
from homeassistant.config_entries import SOURCE_IGNORE
|
||||||
from homeassistant.const import (
|
from homeassistant.const import (
|
||||||
CONF_HOST,
|
CONF_HOST,
|
||||||
@ -153,13 +152,13 @@ class AxisFlowHandler(config_entries.ConfigFlow, domain=AXIS_DOMAIN):
|
|||||||
|
|
||||||
return await self.async_step_user()
|
return await self.async_step_user()
|
||||||
|
|
||||||
async def async_step_dhcp(self, discovery_info: dict):
|
async def async_step_dhcp(self, discovery_info: dhcp.DhcpServiceInfo) -> FlowResult:
|
||||||
"""Prepare configuration for a DHCP discovered Axis device."""
|
"""Prepare configuration for a DHCP discovered Axis device."""
|
||||||
return await self._process_discovered_device(
|
return await self._process_discovered_device(
|
||||||
{
|
{
|
||||||
CONF_HOST: discovery_info[IP_ADDRESS],
|
CONF_HOST: discovery_info[dhcp.IP_ADDRESS],
|
||||||
CONF_MAC: format_mac(discovery_info.get(MAC_ADDRESS, "")),
|
CONF_MAC: format_mac(discovery_info[dhcp.MAC_ADDRESS]),
|
||||||
CONF_NAME: discovery_info.get(HOSTNAME),
|
CONF_NAME: discovery_info[dhcp.HOSTNAME],
|
||||||
CONF_PORT: DEFAULT_PORT,
|
CONF_PORT: DEFAULT_PORT,
|
||||||
}
|
}
|
||||||
)
|
)
|
||||||
|
@ -6,6 +6,7 @@ from ipaddress import ip_address as make_ip_address
|
|||||||
import logging
|
import logging
|
||||||
import os
|
import os
|
||||||
import threading
|
import threading
|
||||||
|
from typing import Final, TypedDict
|
||||||
|
|
||||||
from aiodiscover import DiscoverHosts
|
from aiodiscover import DiscoverHosts
|
||||||
from aiodiscover.discovery import (
|
from aiodiscover.discovery import (
|
||||||
@ -45,15 +46,23 @@ from homeassistant.util.network import is_invalid, is_link_local, is_loopback
|
|||||||
FILTER = "udp and (port 67 or 68)"
|
FILTER = "udp and (port 67 or 68)"
|
||||||
REQUESTED_ADDR = "requested_addr"
|
REQUESTED_ADDR = "requested_addr"
|
||||||
MESSAGE_TYPE = "message-type"
|
MESSAGE_TYPE = "message-type"
|
||||||
HOSTNAME = "hostname"
|
HOSTNAME: Final = "hostname"
|
||||||
MAC_ADDRESS = "macaddress"
|
MAC_ADDRESS: Final = "macaddress"
|
||||||
IP_ADDRESS = "ip"
|
IP_ADDRESS: Final = "ip"
|
||||||
DHCP_REQUEST = 3
|
DHCP_REQUEST = 3
|
||||||
SCAN_INTERVAL = timedelta(minutes=60)
|
SCAN_INTERVAL = timedelta(minutes=60)
|
||||||
|
|
||||||
_LOGGER = logging.getLogger(__name__)
|
_LOGGER = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
|
class DhcpServiceInfo(TypedDict):
|
||||||
|
"""Prepared info from dhcp entries."""
|
||||||
|
|
||||||
|
ip: str
|
||||||
|
hostname: str
|
||||||
|
macaddress: str
|
||||||
|
|
||||||
|
|
||||||
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
|
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
|
||||||
"""Set up the dhcp component."""
|
"""Set up the dhcp component."""
|
||||||
|
|
||||||
@ -150,11 +159,11 @@ class WatcherBase:
|
|||||||
self.hass,
|
self.hass,
|
||||||
entry["domain"],
|
entry["domain"],
|
||||||
{"source": config_entries.SOURCE_DHCP},
|
{"source": config_entries.SOURCE_DHCP},
|
||||||
{
|
DhcpServiceInfo(
|
||||||
IP_ADDRESS: ip_address,
|
ip=ip_address,
|
||||||
HOSTNAME: lowercase_hostname,
|
hostname=lowercase_hostname,
|
||||||
MAC_ADDRESS: data[MAC_ADDRESS],
|
macaddress=data[MAC_ADDRESS],
|
||||||
},
|
),
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@ -7,7 +7,7 @@ from typing import Any, Final
|
|||||||
import voluptuous as vol
|
import voluptuous as vol
|
||||||
|
|
||||||
from homeassistant import config_entries
|
from homeassistant import config_entries
|
||||||
from homeassistant.components.dhcp import HOSTNAME, IP_ADDRESS, MAC_ADDRESS
|
from homeassistant.components import dhcp
|
||||||
from homeassistant.const import CONF_HOST, CONF_MAC, CONF_MODE, CONF_NAME, CONF_PROTOCOL
|
from homeassistant.const import CONF_HOST, CONF_MAC, CONF_MODE, CONF_NAME, CONF_PROTOCOL
|
||||||
from homeassistant.core import callback
|
from homeassistant.core import callback
|
||||||
from homeassistant.data_entry_flow import FlowResult
|
from homeassistant.data_entry_flow import FlowResult
|
||||||
@ -82,12 +82,12 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
|||||||
},
|
},
|
||||||
)
|
)
|
||||||
|
|
||||||
async def async_step_dhcp(self, discovery_info: DiscoveryInfoType) -> FlowResult:
|
async def async_step_dhcp(self, discovery_info: dhcp.DhcpServiceInfo) -> FlowResult:
|
||||||
"""Handle discovery via dhcp."""
|
"""Handle discovery via dhcp."""
|
||||||
self._discovered_device = {
|
self._discovered_device = {
|
||||||
FLUX_HOST: discovery_info[IP_ADDRESS],
|
FLUX_HOST: discovery_info[dhcp.IP_ADDRESS],
|
||||||
FLUX_MODEL: discovery_info[HOSTNAME],
|
FLUX_MODEL: discovery_info[dhcp.HOSTNAME],
|
||||||
FLUX_MAC: discovery_info[MAC_ADDRESS].replace(":", ""),
|
FLUX_MAC: discovery_info[dhcp.MAC_ADDRESS].replace(":", ""),
|
||||||
}
|
}
|
||||||
return await self._async_handle_discovery()
|
return await self._async_handle_discovery()
|
||||||
|
|
||||||
|
@ -8,12 +8,11 @@ from goalzero import Yeti, exceptions
|
|||||||
import voluptuous as vol
|
import voluptuous as vol
|
||||||
|
|
||||||
from homeassistant import config_entries
|
from homeassistant import config_entries
|
||||||
from homeassistant.components.dhcp import IP_ADDRESS, MAC_ADDRESS
|
from homeassistant.components import dhcp
|
||||||
from homeassistant.const import CONF_HOST, CONF_NAME
|
from homeassistant.const import CONF_HOST, CONF_NAME
|
||||||
from homeassistant.data_entry_flow import FlowResult
|
from homeassistant.data_entry_flow import FlowResult
|
||||||
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
||||||
from homeassistant.helpers.device_registry import format_mac
|
from homeassistant.helpers.device_registry import format_mac
|
||||||
from homeassistant.helpers.typing import DiscoveryInfoType
|
|
||||||
|
|
||||||
from .const import DEFAULT_NAME, DOMAIN
|
from .const import DEFAULT_NAME, DOMAIN
|
||||||
|
|
||||||
@ -27,13 +26,13 @@ class GoalZeroFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
|
|||||||
|
|
||||||
def __init__(self) -> None:
|
def __init__(self) -> None:
|
||||||
"""Initialize a Goal Zero Yeti flow."""
|
"""Initialize a Goal Zero Yeti flow."""
|
||||||
self.ip_address = None
|
self.ip_address: str | None = None
|
||||||
|
|
||||||
async def async_step_dhcp(self, discovery_info: DiscoveryInfoType) -> FlowResult:
|
async def async_step_dhcp(self, discovery_info: dhcp.DhcpServiceInfo) -> FlowResult:
|
||||||
"""Handle dhcp discovery."""
|
"""Handle dhcp discovery."""
|
||||||
self.ip_address = discovery_info[IP_ADDRESS]
|
self.ip_address = discovery_info[dhcp.IP_ADDRESS]
|
||||||
|
|
||||||
await self.async_set_unique_id(discovery_info[MAC_ADDRESS])
|
await self.async_set_unique_id(discovery_info[dhcp.MAC_ADDRESS])
|
||||||
self._abort_if_unique_id_configured(updates={CONF_HOST: self.ip_address})
|
self._abort_if_unique_id_configured(updates={CONF_HOST: self.ip_address})
|
||||||
self._async_abort_entries_match({CONF_HOST: self.ip_address})
|
self._async_abort_entries_match({CONF_HOST: self.ip_address})
|
||||||
|
|
||||||
|
@ -8,12 +8,10 @@ from aioguardian.errors import GuardianError
|
|||||||
import voluptuous as vol
|
import voluptuous as vol
|
||||||
|
|
||||||
from homeassistant import config_entries
|
from homeassistant import config_entries
|
||||||
from homeassistant.components import zeroconf
|
from homeassistant.components import dhcp, zeroconf
|
||||||
from homeassistant.components.dhcp import IP_ADDRESS
|
|
||||||
from homeassistant.const import CONF_IP_ADDRESS, CONF_PORT
|
from homeassistant.const import CONF_IP_ADDRESS, CONF_PORT
|
||||||
from homeassistant.core import HomeAssistant, callback
|
from homeassistant.core import HomeAssistant, callback
|
||||||
from homeassistant.data_entry_flow import FlowResult
|
from homeassistant.data_entry_flow import FlowResult
|
||||||
from homeassistant.helpers.typing import DiscoveryInfoType
|
|
||||||
|
|
||||||
from .const import CONF_UID, DOMAIN, LOGGER
|
from .const import CONF_UID, DOMAIN, LOGGER
|
||||||
|
|
||||||
@ -99,10 +97,10 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
|||||||
title=info[CONF_UID], data={CONF_UID: info["uid"], **user_input}
|
title=info[CONF_UID], data={CONF_UID: info["uid"], **user_input}
|
||||||
)
|
)
|
||||||
|
|
||||||
async def async_step_dhcp(self, discovery_info: DiscoveryInfoType) -> FlowResult:
|
async def async_step_dhcp(self, discovery_info: dhcp.DhcpServiceInfo) -> FlowResult:
|
||||||
"""Handle the configuration via dhcp."""
|
"""Handle the configuration via dhcp."""
|
||||||
self.discovery_info = {
|
self.discovery_info = {
|
||||||
CONF_IP_ADDRESS: discovery_info[IP_ADDRESS],
|
CONF_IP_ADDRESS: discovery_info[dhcp.IP_ADDRESS],
|
||||||
CONF_PORT: DEFAULT_PORT,
|
CONF_PORT: DEFAULT_PORT,
|
||||||
}
|
}
|
||||||
return await self._async_handle_discovery()
|
return await self._async_handle_discovery()
|
||||||
|
@ -10,8 +10,7 @@ import getmac
|
|||||||
import voluptuous as vol
|
import voluptuous as vol
|
||||||
|
|
||||||
from homeassistant import config_entries, data_entry_flow
|
from homeassistant import config_entries, data_entry_flow
|
||||||
from homeassistant.components import zeroconf
|
from homeassistant.components import dhcp, zeroconf
|
||||||
from homeassistant.components.dhcp import IP_ADDRESS, MAC_ADDRESS
|
|
||||||
from homeassistant.components.ssdp import (
|
from homeassistant.components.ssdp import (
|
||||||
ATTR_SSDP_LOCATION,
|
ATTR_SSDP_LOCATION,
|
||||||
ATTR_UPNP_MANUFACTURER,
|
ATTR_UPNP_MANUFACTURER,
|
||||||
@ -285,12 +284,12 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
|||||||
return await self.async_step_confirm()
|
return await self.async_step_confirm()
|
||||||
|
|
||||||
async def async_step_dhcp(
|
async def async_step_dhcp(
|
||||||
self, discovery_info: DiscoveryInfoType
|
self, discovery_info: dhcp.DhcpServiceInfo
|
||||||
) -> data_entry_flow.FlowResult:
|
) -> data_entry_flow.FlowResult:
|
||||||
"""Handle a flow initialized by dhcp discovery."""
|
"""Handle a flow initialized by dhcp discovery."""
|
||||||
LOGGER.debug("Samsung device found via DHCP: %s", discovery_info)
|
LOGGER.debug("Samsung device found via DHCP: %s", discovery_info)
|
||||||
self._mac = discovery_info[MAC_ADDRESS]
|
self._mac = discovery_info[dhcp.MAC_ADDRESS]
|
||||||
self._host = discovery_info[IP_ADDRESS]
|
self._host = discovery_info[dhcp.IP_ADDRESS]
|
||||||
await self._async_start_discovery_with_mac_address()
|
await self._async_start_discovery_with_mac_address()
|
||||||
await self._async_set_device_unique_id()
|
await self._async_set_device_unique_id()
|
||||||
self.context["title_placeholders"] = {"device": self._title}
|
self.context["title_placeholders"] = {"device": self._title}
|
||||||
|
@ -9,7 +9,7 @@ from kasa.discover import Discover
|
|||||||
import voluptuous as vol
|
import voluptuous as vol
|
||||||
|
|
||||||
from homeassistant import config_entries
|
from homeassistant import config_entries
|
||||||
from homeassistant.components.dhcp import IP_ADDRESS, MAC_ADDRESS
|
from homeassistant.components import dhcp
|
||||||
from homeassistant.const import CONF_DEVICE, CONF_HOST, CONF_MAC, CONF_NAME
|
from homeassistant.const import CONF_DEVICE, CONF_HOST, CONF_MAC, CONF_NAME
|
||||||
from homeassistant.core import callback
|
from homeassistant.core import callback
|
||||||
from homeassistant.data_entry_flow import FlowResult
|
from homeassistant.data_entry_flow import FlowResult
|
||||||
@ -32,10 +32,10 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
|||||||
self._discovered_devices: dict[str, SmartDevice] = {}
|
self._discovered_devices: dict[str, SmartDevice] = {}
|
||||||
self._discovered_device: SmartDevice | None = None
|
self._discovered_device: SmartDevice | None = None
|
||||||
|
|
||||||
async def async_step_dhcp(self, discovery_info: DiscoveryInfoType) -> FlowResult:
|
async def async_step_dhcp(self, discovery_info: dhcp.DhcpServiceInfo) -> FlowResult:
|
||||||
"""Handle discovery via dhcp."""
|
"""Handle discovery via dhcp."""
|
||||||
return await self._async_handle_discovery(
|
return await self._async_handle_discovery(
|
||||||
discovery_info[IP_ADDRESS], discovery_info[MAC_ADDRESS]
|
discovery_info[dhcp.IP_ADDRESS], discovery_info[dhcp.MAC_ADDRESS]
|
||||||
)
|
)
|
||||||
|
|
||||||
async def async_step_discovery(
|
async def async_step_discovery(
|
||||||
|
@ -32,6 +32,7 @@ from homeassistant.util.decorator import Registry
|
|||||||
import homeassistant.util.uuid as uuid_util
|
import homeassistant.util.uuid as uuid_util
|
||||||
|
|
||||||
if TYPE_CHECKING:
|
if TYPE_CHECKING:
|
||||||
|
from homeassistant.components.dhcp import DhcpServiceInfo
|
||||||
from homeassistant.components.zeroconf import ZeroconfServiceInfo
|
from homeassistant.components.zeroconf import ZeroconfServiceInfo
|
||||||
|
|
||||||
_LOGGER = logging.getLogger(__name__)
|
_LOGGER = logging.getLogger(__name__)
|
||||||
@ -1378,10 +1379,10 @@ class ConfigFlow(data_entry_flow.FlowHandler):
|
|||||||
return await self.async_step_discovery(cast(dict, discovery_info))
|
return await self.async_step_discovery(cast(dict, discovery_info))
|
||||||
|
|
||||||
async def async_step_dhcp(
|
async def async_step_dhcp(
|
||||||
self, discovery_info: DiscoveryInfoType
|
self, discovery_info: DhcpServiceInfo
|
||||||
) -> data_entry_flow.FlowResult:
|
) -> data_entry_flow.FlowResult:
|
||||||
"""Handle a flow initialized by DHCP discovery."""
|
"""Handle a flow initialized by DHCP discovery."""
|
||||||
return await self.async_step_discovery(discovery_info)
|
return await self.async_step_discovery(cast(dict, discovery_info))
|
||||||
|
|
||||||
async def async_step_usb(
|
async def async_step_usb(
|
||||||
self, discovery_info: DiscoveryInfoType
|
self, discovery_info: DiscoveryInfoType
|
||||||
|
@ -5,7 +5,7 @@ import logging
|
|||||||
from typing import Any, Awaitable, Callable, Union
|
from typing import Any, Awaitable, Callable, Union
|
||||||
|
|
||||||
from homeassistant import config_entries
|
from homeassistant import config_entries
|
||||||
from homeassistant.components import zeroconf
|
from homeassistant.components import dhcp, zeroconf
|
||||||
from homeassistant.core import HomeAssistant
|
from homeassistant.core import HomeAssistant
|
||||||
from homeassistant.data_entry_flow import FlowResult
|
from homeassistant.data_entry_flow import FlowResult
|
||||||
from homeassistant.helpers.typing import UNDEFINED, DiscoveryInfoType, UndefinedType
|
from homeassistant.helpers.typing import UNDEFINED, DiscoveryInfoType, UndefinedType
|
||||||
@ -82,6 +82,15 @@ class DiscoveryFlowHandler(config_entries.ConfigFlow):
|
|||||||
|
|
||||||
return await self.async_step_confirm()
|
return await self.async_step_confirm()
|
||||||
|
|
||||||
|
async def async_step_dhcp(self, discovery_info: dhcp.DhcpServiceInfo) -> FlowResult:
|
||||||
|
"""Handle a flow initialized by dhcp discovery."""
|
||||||
|
if self._async_in_progress() or self._async_current_entries():
|
||||||
|
return self.async_abort(reason="single_instance_allowed")
|
||||||
|
|
||||||
|
await self.async_set_unique_id(self._domain)
|
||||||
|
|
||||||
|
return await self.async_step_confirm()
|
||||||
|
|
||||||
async def async_step_homekit(
|
async def async_step_homekit(
|
||||||
self, discovery_info: zeroconf.ZeroconfServiceInfo
|
self, discovery_info: zeroconf.ZeroconfServiceInfo
|
||||||
) -> FlowResult:
|
) -> FlowResult:
|
||||||
@ -106,7 +115,6 @@ class DiscoveryFlowHandler(config_entries.ConfigFlow):
|
|||||||
|
|
||||||
async_step_ssdp = async_step_discovery
|
async_step_ssdp = async_step_discovery
|
||||||
async_step_mqtt = async_step_discovery
|
async_step_mqtt = async_step_discovery
|
||||||
async_step_dhcp = async_step_discovery
|
|
||||||
|
|
||||||
async def async_step_import(self, _: dict[str, Any] | None) -> FlowResult:
|
async def async_step_import(self, _: dict[str, Any] | None) -> FlowResult:
|
||||||
"""Handle a flow initialized by import."""
|
"""Handle a flow initialized by import."""
|
||||||
|
Loading…
x
Reference in New Issue
Block a user