-
-
Notifications
You must be signed in to change notification settings - Fork 37.8k
Fix typing for wemo #62157
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Fix typing for wemo #62157
Changes from 1 commit
519ca6c
0866a33
2c6a51c
df897a0
e92b075
2a5c9c2
cca69dd
96d9573
21f3ea8
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,7 +1,9 @@ | ||
| """Support for WeMo device discovery.""" | ||
| from __future__ import annotations | ||
|
|
||
| from collections.abc import MutableSet, Sequence | ||
| import logging | ||
| from typing import Any, Optional, Tuple | ||
|
|
||
| import pywemo | ||
| import voluptuous as vol | ||
|
|
@@ -14,10 +16,11 @@ | |
| from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN | ||
| from homeassistant.config_entries import ConfigEntry | ||
| from homeassistant.const import CONF_DISCOVERY, EVENT_HOMEASSISTANT_STOP | ||
| from homeassistant.core import HomeAssistant, callback | ||
| from homeassistant.core import CALLBACK_TYPE, Event, HomeAssistant, callback | ||
| from homeassistant.helpers import config_validation as cv | ||
| from homeassistant.helpers.dispatcher import async_dispatcher_send | ||
| from homeassistant.helpers.event import async_call_later | ||
| from homeassistant.helpers.typing import ConfigType | ||
| from homeassistant.util.async_ import gather_with_concurrency | ||
|
|
||
| from .const import DOMAIN | ||
|
|
@@ -44,12 +47,15 @@ | |
|
|
||
| _LOGGER = logging.getLogger(__name__) | ||
|
|
||
| HostPortTuple = Tuple[str, Optional[str]] | ||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Is the port really a string?
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. No it is not. Nice catch, thank you! I missed the |
||
|
|
||
| def coerce_host_port(value): | ||
|
|
||
| def coerce_host_port(value: str) -> HostPortTuple: | ||
| """Validate that provided value is either just host or host:port. | ||
|
|
||
| Returns (host, None) or (host, port) respectively. | ||
| """ | ||
| port: str | None = None | ||
| host, _, port = value.partition(":") | ||
|
|
||
| if not host: | ||
|
|
@@ -82,7 +88,7 @@ def coerce_host_port(value): | |
| ) | ||
|
|
||
|
|
||
| async def async_setup(hass, config): | ||
| async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: | ||
| """Set up for WeMo devices.""" | ||
| hass.data[DOMAIN] = { | ||
| "config": config.get(DOMAIN, {}), | ||
|
|
@@ -112,11 +118,11 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: | |
| discovery_responder = pywemo.ssdp.DiscoveryResponder(registry.port) | ||
| await hass.async_add_executor_job(discovery_responder.start) | ||
|
|
||
| static_conf = config.get(CONF_STATIC, []) | ||
| static_conf: Sequence[HostPortTuple] = config.get(CONF_STATIC, []) | ||
| wemo_dispatcher = WemoDispatcher(entry) | ||
| wemo_discovery = WemoDiscovery(hass, wemo_dispatcher, static_conf) | ||
|
|
||
| async def async_stop_wemo(event): | ||
| async def async_stop_wemo(event: Event) -> None: | ||
| """Shutdown Wemo subscriptions and subscription thread on exit.""" | ||
| _LOGGER.debug("Shutting down WeMo event subscriptions") | ||
| await hass.async_add_executor_job(registry.stop) | ||
|
|
@@ -142,8 +148,8 @@ class WemoDispatcher: | |
| def __init__(self, config_entry: ConfigEntry) -> None: | ||
| """Initialize the WemoDispatcher.""" | ||
| self._config_entry = config_entry | ||
| self._added_serial_numbers = set() | ||
| self._loaded_components = set() | ||
| self._added_serial_numbers: MutableSet[str] = set() | ||
| self._loaded_components: MutableSet[str] = set() | ||
|
esev marked this conversation as resolved.
Outdated
|
||
|
|
||
| async def async_add_unique_device( | ||
| self, hass: HomeAssistant, wemo: pywemo.WeMoDevice | ||
|
|
@@ -191,16 +197,16 @@ def __init__( | |
| self, | ||
| hass: HomeAssistant, | ||
| wemo_dispatcher: WemoDispatcher, | ||
| static_config: list[tuple[[str, str | None]]], | ||
| static_config: Sequence[HostPortTuple], | ||
| ) -> None: | ||
| """Initialize the WemoDiscovery.""" | ||
| self._hass = hass | ||
| self._wemo_dispatcher = wemo_dispatcher | ||
| self._stop = None | ||
| self._stop: CALLBACK_TYPE | None = None | ||
| self._scan_delay = 0 | ||
| self._static_config = static_config | ||
|
|
||
| async def async_discover_and_schedule(self, *_) -> None: | ||
| async def async_discover_and_schedule(self, *_: tuple[Any]) -> None: | ||
| """Periodically scan the network looking for WeMo devices.""" | ||
| _LOGGER.debug("Scanning network for WeMo devices") | ||
| try: | ||
|
|
@@ -229,7 +235,7 @@ def async_stop_discovery(self) -> None: | |
| self._stop() | ||
| self._stop = None | ||
|
|
||
| async def discover_statics(self): | ||
| async def discover_statics(self) -> None: | ||
| """Initialize or Re-Initialize connections to statically configured devices.""" | ||
| if self._static_config: | ||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. If you reverse this condition and return, you can outdent below
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Done. 96d9573 |
||
| _LOGGER.debug("Adding statically configured WeMo devices") | ||
|
|
@@ -248,7 +254,7 @@ async def discover_statics(self): | |
| ) | ||
|
|
||
|
|
||
| def validate_static_config(host, port): | ||
| def validate_static_config(host: str, port: str | None) -> pywemo.WeMoDevice | None: | ||
| """Handle a static config.""" | ||
| url = pywemo.setup_url_for_address(host, port) | ||
|
|
||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
MutableSet unused now?
Uh oh!
There was an error while loading. Please reload this page.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yes. You're quick! I didn't get e92b075 pushed fast enough. :)