mirror of
https://github.com/home-assistant/core.git
synced 2025-07-23 13:17:32 +00:00
Add more type hints to helpers (#18196)
* Test typing for helpers.__init__ and temperature * Add type hints to helpers.sun * Add type hints to helpers.signal * Add type hints to helpers.entity_values * Add type hints to helpers.dispatcher
This commit is contained in:
parent
959fa81ea6
commit
922f34f72d
@ -1,9 +1,11 @@
|
|||||||
"""Helpers for Home Assistant dispatcher & internal component/platform."""
|
"""Helpers for Home Assistant dispatcher & internal component/platform."""
|
||||||
import logging
|
import logging
|
||||||
|
from typing import Any, Callable
|
||||||
|
|
||||||
from homeassistant.core import callback
|
from homeassistant.core import callback
|
||||||
from homeassistant.loader import bind_hass
|
from homeassistant.loader import bind_hass
|
||||||
from homeassistant.util.async_ import run_callback_threadsafe
|
from homeassistant.util.async_ import run_callback_threadsafe
|
||||||
|
from .typing import HomeAssistantType
|
||||||
|
|
||||||
|
|
||||||
_LOGGER = logging.getLogger(__name__)
|
_LOGGER = logging.getLogger(__name__)
|
||||||
@ -11,12 +13,13 @@ DATA_DISPATCHER = 'dispatcher'
|
|||||||
|
|
||||||
|
|
||||||
@bind_hass
|
@bind_hass
|
||||||
def dispatcher_connect(hass, signal, target):
|
def dispatcher_connect(hass: HomeAssistantType, signal: str,
|
||||||
|
target: Callable[..., None]) -> Callable[[], None]:
|
||||||
"""Connect a callable function to a signal."""
|
"""Connect a callable function to a signal."""
|
||||||
async_unsub = run_callback_threadsafe(
|
async_unsub = run_callback_threadsafe(
|
||||||
hass.loop, async_dispatcher_connect, hass, signal, target).result()
|
hass.loop, async_dispatcher_connect, hass, signal, target).result()
|
||||||
|
|
||||||
def remove_dispatcher():
|
def remove_dispatcher() -> None:
|
||||||
"""Remove signal listener."""
|
"""Remove signal listener."""
|
||||||
run_callback_threadsafe(hass.loop, async_unsub).result()
|
run_callback_threadsafe(hass.loop, async_unsub).result()
|
||||||
|
|
||||||
@ -25,7 +28,8 @@ def dispatcher_connect(hass, signal, target):
|
|||||||
|
|
||||||
@callback
|
@callback
|
||||||
@bind_hass
|
@bind_hass
|
||||||
def async_dispatcher_connect(hass, signal, target):
|
def async_dispatcher_connect(hass: HomeAssistantType, signal: str,
|
||||||
|
target: Callable[..., Any]) -> Callable[[], None]:
|
||||||
"""Connect a callable function to a signal.
|
"""Connect a callable function to a signal.
|
||||||
|
|
||||||
This method must be run in the event loop.
|
This method must be run in the event loop.
|
||||||
@ -39,7 +43,7 @@ def async_dispatcher_connect(hass, signal, target):
|
|||||||
hass.data[DATA_DISPATCHER][signal].append(target)
|
hass.data[DATA_DISPATCHER][signal].append(target)
|
||||||
|
|
||||||
@callback
|
@callback
|
||||||
def async_remove_dispatcher():
|
def async_remove_dispatcher() -> None:
|
||||||
"""Remove signal listener."""
|
"""Remove signal listener."""
|
||||||
try:
|
try:
|
||||||
hass.data[DATA_DISPATCHER][signal].remove(target)
|
hass.data[DATA_DISPATCHER][signal].remove(target)
|
||||||
@ -53,14 +57,15 @@ def async_dispatcher_connect(hass, signal, target):
|
|||||||
|
|
||||||
|
|
||||||
@bind_hass
|
@bind_hass
|
||||||
def dispatcher_send(hass, signal, *args):
|
def dispatcher_send(hass: HomeAssistantType, signal: str, *args: Any) -> None:
|
||||||
"""Send signal and data."""
|
"""Send signal and data."""
|
||||||
hass.loop.call_soon_threadsafe(async_dispatcher_send, hass, signal, *args)
|
hass.loop.call_soon_threadsafe(async_dispatcher_send, hass, signal, *args)
|
||||||
|
|
||||||
|
|
||||||
@callback
|
@callback
|
||||||
@bind_hass
|
@bind_hass
|
||||||
def async_dispatcher_send(hass, signal, *args):
|
def async_dispatcher_send(
|
||||||
|
hass: HomeAssistantType, signal: str, *args: Any) -> None:
|
||||||
"""Send signal and data.
|
"""Send signal and data.
|
||||||
|
|
||||||
This method must be run in the event loop.
|
This method must be run in the event loop.
|
||||||
|
@ -2,7 +2,7 @@
|
|||||||
from collections import OrderedDict
|
from collections import OrderedDict
|
||||||
import fnmatch
|
import fnmatch
|
||||||
import re
|
import re
|
||||||
from typing import Dict
|
from typing import Any, Dict, Optional, Pattern # noqa: F401
|
||||||
|
|
||||||
from homeassistant.core import split_entity_id
|
from homeassistant.core import split_entity_id
|
||||||
|
|
||||||
@ -10,15 +10,16 @@ from homeassistant.core import split_entity_id
|
|||||||
class EntityValues:
|
class EntityValues:
|
||||||
"""Class to store entity id based values."""
|
"""Class to store entity id based values."""
|
||||||
|
|
||||||
def __init__(self, exact: Dict = None, domain: Dict = None,
|
def __init__(self, exact: Optional[Dict] = None,
|
||||||
glob: Dict = None) -> None:
|
domain: Optional[Dict] = None,
|
||||||
|
glob: Optional[Dict] = None) -> None:
|
||||||
"""Initialize an EntityConfigDict."""
|
"""Initialize an EntityConfigDict."""
|
||||||
self._cache = {}
|
self._cache = {} # type: Dict[str, Dict]
|
||||||
self._exact = exact
|
self._exact = exact
|
||||||
self._domain = domain
|
self._domain = domain
|
||||||
|
|
||||||
if glob is None:
|
if glob is None:
|
||||||
compiled = None
|
compiled = None # type: Optional[Dict[Pattern[str], Any]]
|
||||||
else:
|
else:
|
||||||
compiled = OrderedDict()
|
compiled = OrderedDict()
|
||||||
for key, value in glob.items():
|
for key, value in glob.items():
|
||||||
@ -26,7 +27,7 @@ class EntityValues:
|
|||||||
|
|
||||||
self._glob = compiled
|
self._glob = compiled
|
||||||
|
|
||||||
def get(self, entity_id):
|
def get(self, entity_id: str) -> Dict:
|
||||||
"""Get config for an entity id."""
|
"""Get config for an entity id."""
|
||||||
if entity_id in self._cache:
|
if entity_id in self._cache:
|
||||||
return self._cache[entity_id]
|
return self._cache[entity_id]
|
||||||
|
@ -2,6 +2,7 @@
|
|||||||
import logging
|
import logging
|
||||||
import signal
|
import signal
|
||||||
import sys
|
import sys
|
||||||
|
from types import FrameType
|
||||||
|
|
||||||
from homeassistant.core import callback, HomeAssistant
|
from homeassistant.core import callback, HomeAssistant
|
||||||
from homeassistant.const import RESTART_EXIT_CODE
|
from homeassistant.const import RESTART_EXIT_CODE
|
||||||
@ -16,7 +17,7 @@ def async_register_signal_handling(hass: HomeAssistant) -> None:
|
|||||||
"""Register system signal handler for core."""
|
"""Register system signal handler for core."""
|
||||||
if sys.platform != 'win32':
|
if sys.platform != 'win32':
|
||||||
@callback
|
@callback
|
||||||
def async_signal_handle(exit_code):
|
def async_signal_handle(exit_code: int) -> None:
|
||||||
"""Wrap signal handling.
|
"""Wrap signal handling.
|
||||||
|
|
||||||
* queue call to shutdown task
|
* queue call to shutdown task
|
||||||
@ -49,7 +50,7 @@ def async_register_signal_handling(hass: HomeAssistant) -> None:
|
|||||||
old_sigint = None
|
old_sigint = None
|
||||||
|
|
||||||
@callback
|
@callback
|
||||||
def async_signal_handle(exit_code, frame):
|
def async_signal_handle(exit_code: int, frame: FrameType) -> None:
|
||||||
"""Wrap signal handling.
|
"""Wrap signal handling.
|
||||||
|
|
||||||
* queue call to shutdown task
|
* queue call to shutdown task
|
||||||
|
@ -1,23 +1,28 @@
|
|||||||
"""Helpers for sun events."""
|
"""Helpers for sun events."""
|
||||||
import datetime
|
import datetime
|
||||||
|
from typing import Optional, Union, TYPE_CHECKING
|
||||||
|
|
||||||
from homeassistant.const import SUN_EVENT_SUNRISE, SUN_EVENT_SUNSET
|
from homeassistant.const import SUN_EVENT_SUNRISE, SUN_EVENT_SUNSET
|
||||||
from homeassistant.core import callback
|
from homeassistant.core import callback
|
||||||
from homeassistant.util import dt as dt_util
|
from homeassistant.util import dt as dt_util
|
||||||
from homeassistant.loader import bind_hass
|
from homeassistant.loader import bind_hass
|
||||||
|
from .typing import HomeAssistantType
|
||||||
|
|
||||||
|
if TYPE_CHECKING:
|
||||||
|
import astral # pylint: disable=unused-import
|
||||||
|
|
||||||
DATA_LOCATION_CACHE = 'astral_location_cache'
|
DATA_LOCATION_CACHE = 'astral_location_cache'
|
||||||
|
|
||||||
|
|
||||||
@callback
|
@callback
|
||||||
@bind_hass
|
@bind_hass
|
||||||
def get_astral_location(hass):
|
def get_astral_location(hass: HomeAssistantType) -> 'astral.Location':
|
||||||
"""Get an astral location for the current Home Assistant configuration."""
|
"""Get an astral location for the current Home Assistant configuration."""
|
||||||
from astral import Location
|
from astral import Location
|
||||||
|
|
||||||
latitude = hass.config.latitude
|
latitude = hass.config.latitude
|
||||||
longitude = hass.config.longitude
|
longitude = hass.config.longitude
|
||||||
timezone = hass.config.time_zone.zone
|
timezone = str(hass.config.time_zone)
|
||||||
elevation = hass.config.elevation
|
elevation = hass.config.elevation
|
||||||
info = ('', '', latitude, longitude, timezone, elevation)
|
info = ('', '', latitude, longitude, timezone, elevation)
|
||||||
|
|
||||||
@ -33,9 +38,12 @@ def get_astral_location(hass):
|
|||||||
|
|
||||||
@callback
|
@callback
|
||||||
@bind_hass
|
@bind_hass
|
||||||
def get_astral_event_next(hass, event, utc_point_in_time=None, offset=None):
|
def get_astral_event_next(
|
||||||
|
hass: HomeAssistantType, event: str,
|
||||||
|
utc_point_in_time: Optional[datetime.datetime] = None,
|
||||||
|
offset: Optional[datetime.timedelta] = None) -> datetime.datetime:
|
||||||
"""Calculate the next specified solar event."""
|
"""Calculate the next specified solar event."""
|
||||||
import astral
|
from astral import AstralError
|
||||||
|
|
||||||
location = get_astral_location(hass)
|
location = get_astral_location(hass)
|
||||||
|
|
||||||
@ -51,19 +59,22 @@ def get_astral_event_next(hass, event, utc_point_in_time=None, offset=None):
|
|||||||
next_dt = getattr(location, event)(
|
next_dt = getattr(location, event)(
|
||||||
dt_util.as_local(utc_point_in_time).date() +
|
dt_util.as_local(utc_point_in_time).date() +
|
||||||
datetime.timedelta(days=mod),
|
datetime.timedelta(days=mod),
|
||||||
local=False) + offset
|
local=False) + offset # type: datetime.datetime
|
||||||
if next_dt > utc_point_in_time:
|
if next_dt > utc_point_in_time:
|
||||||
return next_dt
|
return next_dt
|
||||||
except astral.AstralError:
|
except AstralError:
|
||||||
pass
|
pass
|
||||||
mod += 1
|
mod += 1
|
||||||
|
|
||||||
|
|
||||||
@callback
|
@callback
|
||||||
@bind_hass
|
@bind_hass
|
||||||
def get_astral_event_date(hass, event, date=None):
|
def get_astral_event_date(
|
||||||
|
hass: HomeAssistantType, event: str,
|
||||||
|
date: Union[datetime.date, datetime.datetime, None] = None) \
|
||||||
|
-> Optional[datetime.datetime]:
|
||||||
"""Calculate the astral event time for the specified date."""
|
"""Calculate the astral event time for the specified date."""
|
||||||
import astral
|
from astral import AstralError
|
||||||
|
|
||||||
location = get_astral_location(hass)
|
location = get_astral_location(hass)
|
||||||
|
|
||||||
@ -74,15 +85,16 @@ def get_astral_event_date(hass, event, date=None):
|
|||||||
date = dt_util.as_local(date).date()
|
date = dt_util.as_local(date).date()
|
||||||
|
|
||||||
try:
|
try:
|
||||||
return getattr(location, event)(date, local=False)
|
return getattr(location, event)(date, local=False) # type: ignore
|
||||||
except astral.AstralError:
|
except AstralError:
|
||||||
# Event never occurs for specified date.
|
# Event never occurs for specified date.
|
||||||
return None
|
return None
|
||||||
|
|
||||||
|
|
||||||
@callback
|
@callback
|
||||||
@bind_hass
|
@bind_hass
|
||||||
def is_up(hass, utc_point_in_time=None):
|
def is_up(hass: HomeAssistantType,
|
||||||
|
utc_point_in_time: Optional[datetime.datetime] = None) -> bool:
|
||||||
"""Calculate if the sun is currently up."""
|
"""Calculate if the sun is currently up."""
|
||||||
if utc_point_in_time is None:
|
if utc_point_in_time is None:
|
||||||
utc_point_in_time = dt_util.utcnow()
|
utc_point_in_time = dt_util.utcnow()
|
||||||
|
2
tox.ini
2
tox.ini
@ -60,4 +60,4 @@ whitelist_externals=/bin/bash
|
|||||||
deps =
|
deps =
|
||||||
-r{toxinidir}/requirements_test.txt
|
-r{toxinidir}/requirements_test.txt
|
||||||
commands =
|
commands =
|
||||||
/bin/bash -c 'mypy homeassistant/*.py homeassistant/{auth,util}/ homeassistant/helpers/{icon,intent,json,location,state,translation,typing}.py'
|
/bin/bash -c 'mypy homeassistant/*.py homeassistant/{auth,util}/ homeassistant/helpers/{__init__,dispatcher,entity_values,icon,intent,json,location,signal,state,sun,temperature,translation,typing}.py'
|
||||||
|
Loading…
x
Reference in New Issue
Block a user