Add strict typing to device_tracker (#50930)
* Add strict typing to device_tracker * Update homeassistant/components/device_tracker/legacy.py Co-authored-by: Martin Hjelmare <marhje52@gmail.com> Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
parent
2e316f6fd5
commit
b704f0e729
7 changed files with 157 additions and 120 deletions
|
@ -21,6 +21,7 @@ homeassistant.components.camera.*
|
|||
homeassistant.components.canary.*
|
||||
homeassistant.components.cover.*
|
||||
homeassistant.components.device_automation.*
|
||||
homeassistant.components.device_tracker.*
|
||||
homeassistant.components.elgato.*
|
||||
homeassistant.components.fitbit.*
|
||||
homeassistant.components.fritzbox.*
|
||||
|
|
|
@ -37,12 +37,12 @@ from .legacy import ( # noqa: F401
|
|||
|
||||
|
||||
@bind_hass
|
||||
def is_on(hass: HomeAssistant, entity_id: str):
|
||||
def is_on(hass: HomeAssistant, entity_id: str) -> bool:
|
||||
"""Return the state if any or a specified device is home."""
|
||||
return hass.states.is_state(entity_id, STATE_HOME)
|
||||
|
||||
|
||||
async def async_setup(hass: HomeAssistant, config: ConfigType):
|
||||
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
|
||||
"""Set up the device tracker."""
|
||||
await async_setup_legacy_integration(hass, config)
|
||||
|
||||
|
|
|
@ -4,6 +4,7 @@ from __future__ import annotations
|
|||
from typing import final
|
||||
|
||||
from homeassistant.components import zone
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import (
|
||||
ATTR_BATTERY_LEVEL,
|
||||
ATTR_GPS_ACCURACY,
|
||||
|
@ -12,13 +13,15 @@ from homeassistant.const import (
|
|||
STATE_HOME,
|
||||
STATE_NOT_HOME,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.entity import Entity
|
||||
from homeassistant.helpers.entity_component import EntityComponent
|
||||
from homeassistant.helpers.typing import StateType
|
||||
|
||||
from .const import ATTR_HOST_NAME, ATTR_IP, ATTR_MAC, ATTR_SOURCE_TYPE, DOMAIN, LOGGER
|
||||
|
||||
|
||||
async def async_setup_entry(hass, entry):
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Set up an entry."""
|
||||
component: EntityComponent | None = hass.data.get(DOMAIN)
|
||||
|
||||
|
@ -28,16 +31,17 @@ async def async_setup_entry(hass, entry):
|
|||
return await component.async_setup_entry(entry)
|
||||
|
||||
|
||||
async def async_unload_entry(hass, entry):
|
||||
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Unload an entry."""
|
||||
return await hass.data[DOMAIN].async_unload_entry(entry)
|
||||
component: EntityComponent = hass.data[DOMAIN]
|
||||
return await component.async_unload_entry(entry)
|
||||
|
||||
|
||||
class BaseTrackerEntity(Entity):
|
||||
"""Represent a tracked device."""
|
||||
|
||||
@property
|
||||
def battery_level(self):
|
||||
def battery_level(self) -> int | None:
|
||||
"""Return the battery level of the device.
|
||||
|
||||
Percentage from 0-100.
|
||||
|
@ -45,16 +49,16 @@ class BaseTrackerEntity(Entity):
|
|||
return None
|
||||
|
||||
@property
|
||||
def source_type(self):
|
||||
def source_type(self) -> str:
|
||||
"""Return the source type, eg gps or router, of the device."""
|
||||
raise NotImplementedError
|
||||
|
||||
@property
|
||||
def state_attributes(self):
|
||||
def state_attributes(self) -> dict[str, StateType]:
|
||||
"""Return the device state attributes."""
|
||||
attr = {ATTR_SOURCE_TYPE: self.source_type}
|
||||
attr: dict[str, StateType] = {ATTR_SOURCE_TYPE: self.source_type}
|
||||
|
||||
if self.battery_level:
|
||||
if self.battery_level is not None:
|
||||
attr[ATTR_BATTERY_LEVEL] = self.battery_level
|
||||
|
||||
return attr
|
||||
|
@ -64,17 +68,17 @@ class TrackerEntity(BaseTrackerEntity):
|
|||
"""Base class for a tracked device."""
|
||||
|
||||
@property
|
||||
def should_poll(self):
|
||||
def should_poll(self) -> bool:
|
||||
"""No polling for entities that have location pushed."""
|
||||
return False
|
||||
|
||||
@property
|
||||
def force_update(self):
|
||||
def force_update(self) -> bool:
|
||||
"""All updates need to be written to the state machine if we're not polling."""
|
||||
return not self.should_poll
|
||||
|
||||
@property
|
||||
def location_accuracy(self):
|
||||
def location_accuracy(self) -> int:
|
||||
"""Return the location accuracy of the device.
|
||||
|
||||
Value in meters.
|
||||
|
@ -97,9 +101,9 @@ class TrackerEntity(BaseTrackerEntity):
|
|||
raise NotImplementedError
|
||||
|
||||
@property
|
||||
def state(self):
|
||||
def state(self) -> str | None:
|
||||
"""Return the state of the device."""
|
||||
if self.location_name:
|
||||
if self.location_name is not None:
|
||||
return self.location_name
|
||||
|
||||
if self.latitude is not None and self.longitude is not None:
|
||||
|
@ -118,11 +122,11 @@ class TrackerEntity(BaseTrackerEntity):
|
|||
|
||||
@final
|
||||
@property
|
||||
def state_attributes(self):
|
||||
def state_attributes(self) -> dict[str, StateType]:
|
||||
"""Return the device state attributes."""
|
||||
attr = {}
|
||||
attr: dict[str, StateType] = {}
|
||||
attr.update(super().state_attributes)
|
||||
if self.latitude is not None:
|
||||
if self.latitude is not None and self.longitude is not None:
|
||||
attr[ATTR_LATITUDE] = self.latitude
|
||||
attr[ATTR_LONGITUDE] = self.longitude
|
||||
attr[ATTR_GPS_ACCURACY] = self.location_accuracy
|
||||
|
@ -162,9 +166,9 @@ class ScannerEntity(BaseTrackerEntity):
|
|||
|
||||
@final
|
||||
@property
|
||||
def state_attributes(self):
|
||||
def state_attributes(self) -> dict[str, StateType]:
|
||||
"""Return the device state attributes."""
|
||||
attr = {}
|
||||
attr: dict[str, StateType] = {}
|
||||
attr.update(super().state_attributes)
|
||||
if self.ip_address is not None:
|
||||
attr[ATTR_IP] = self.ip_address
|
||||
|
|
|
@ -1,37 +1,38 @@
|
|||
"""Device tracker constants."""
|
||||
from datetime import timedelta
|
||||
import logging
|
||||
from typing import Final
|
||||
|
||||
LOGGER = logging.getLogger(__package__)
|
||||
LOGGER: Final = logging.getLogger(__package__)
|
||||
|
||||
DOMAIN = "device_tracker"
|
||||
DOMAIN: Final = "device_tracker"
|
||||
|
||||
PLATFORM_TYPE_LEGACY = "legacy"
|
||||
PLATFORM_TYPE_ENTITY = "entity_platform"
|
||||
PLATFORM_TYPE_LEGACY: Final = "legacy"
|
||||
PLATFORM_TYPE_ENTITY: Final = "entity_platform"
|
||||
|
||||
SOURCE_TYPE_GPS = "gps"
|
||||
SOURCE_TYPE_ROUTER = "router"
|
||||
SOURCE_TYPE_BLUETOOTH = "bluetooth"
|
||||
SOURCE_TYPE_BLUETOOTH_LE = "bluetooth_le"
|
||||
SOURCE_TYPE_GPS: Final = "gps"
|
||||
SOURCE_TYPE_ROUTER: Final = "router"
|
||||
SOURCE_TYPE_BLUETOOTH: Final = "bluetooth"
|
||||
SOURCE_TYPE_BLUETOOTH_LE: Final = "bluetooth_le"
|
||||
|
||||
CONF_SCAN_INTERVAL = "interval_seconds"
|
||||
SCAN_INTERVAL = timedelta(seconds=12)
|
||||
CONF_SCAN_INTERVAL: Final = "interval_seconds"
|
||||
SCAN_INTERVAL: Final = timedelta(seconds=12)
|
||||
|
||||
CONF_TRACK_NEW = "track_new_devices"
|
||||
DEFAULT_TRACK_NEW = True
|
||||
CONF_TRACK_NEW: Final = "track_new_devices"
|
||||
DEFAULT_TRACK_NEW: Final = True
|
||||
|
||||
CONF_CONSIDER_HOME = "consider_home"
|
||||
DEFAULT_CONSIDER_HOME = timedelta(seconds=180)
|
||||
CONF_CONSIDER_HOME: Final = "consider_home"
|
||||
DEFAULT_CONSIDER_HOME: Final = timedelta(seconds=180)
|
||||
|
||||
CONF_NEW_DEVICE_DEFAULTS = "new_device_defaults"
|
||||
CONF_NEW_DEVICE_DEFAULTS: Final = "new_device_defaults"
|
||||
|
||||
ATTR_ATTRIBUTES = "attributes"
|
||||
ATTR_BATTERY = "battery"
|
||||
ATTR_DEV_ID = "dev_id"
|
||||
ATTR_GPS = "gps"
|
||||
ATTR_HOST_NAME = "host_name"
|
||||
ATTR_LOCATION_NAME = "location_name"
|
||||
ATTR_MAC = "mac"
|
||||
ATTR_SOURCE_TYPE = "source_type"
|
||||
ATTR_CONSIDER_HOME = "consider_home"
|
||||
ATTR_IP = "ip"
|
||||
ATTR_ATTRIBUTES: Final = "attributes"
|
||||
ATTR_BATTERY: Final = "battery"
|
||||
ATTR_DEV_ID: Final = "dev_id"
|
||||
ATTR_GPS: Final = "gps"
|
||||
ATTR_HOST_NAME: Final = "host_name"
|
||||
ATTR_LOCATION_NAME: Final = "location_name"
|
||||
ATTR_MAC: Final = "mac"
|
||||
ATTR_SOURCE_TYPE: Final = "source_type"
|
||||
ATTR_CONSIDER_HOME: Final = "consider_home"
|
||||
ATTR_IP: Final = "ip"
|
||||
|
|
|
@ -1,6 +1,8 @@
|
|||
"""Provides device automations for Device Tracker."""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Final
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.automation import AutomationActionType
|
||||
|
@ -21,9 +23,9 @@ from homeassistant.helpers.typing import ConfigType
|
|||
|
||||
from . import DOMAIN
|
||||
|
||||
TRIGGER_TYPES = {"enters", "leaves"}
|
||||
TRIGGER_TYPES: Final[set[str]] = {"enters", "leaves"}
|
||||
|
||||
TRIGGER_SCHEMA = TRIGGER_BASE_SCHEMA.extend(
|
||||
TRIGGER_SCHEMA: Final = TRIGGER_BASE_SCHEMA.extend(
|
||||
{
|
||||
vol.Required(CONF_ENTITY_ID): cv.entity_id,
|
||||
vol.Required(CONF_TYPE): vol.In(TRIGGER_TYPES),
|
||||
|
@ -88,7 +90,9 @@ async def async_attach_trigger(
|
|||
)
|
||||
|
||||
|
||||
async def async_get_trigger_capabilities(hass: HomeAssistant, config: ConfigType):
|
||||
async def async_get_trigger_capabilities(
|
||||
hass: HomeAssistant, config: ConfigType
|
||||
) -> dict[str, vol.Schema]:
|
||||
"""List trigger capabilities."""
|
||||
zones = {
|
||||
ent.entity_id: ent.name
|
||||
|
|
|
@ -2,7 +2,7 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from collections.abc import Sequence
|
||||
from collections.abc import Coroutine, Sequence
|
||||
from datetime import timedelta
|
||||
import hashlib
|
||||
from types import ModuleType
|
||||
|
@ -28,7 +28,7 @@ from homeassistant.const import (
|
|||
STATE_HOME,
|
||||
STATE_NOT_HOME,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
from homeassistant.core import HomeAssistant, ServiceCall, callback
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
from homeassistant.helpers import config_per_platform, discovery
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
|
@ -38,7 +38,7 @@ from homeassistant.helpers.event import (
|
|||
async_track_utc_time_change,
|
||||
)
|
||||
from homeassistant.helpers.restore_state import RestoreEntity
|
||||
from homeassistant.helpers.typing import ConfigType, GPSType
|
||||
from homeassistant.helpers.typing import ConfigType, GPSType, StateType
|
||||
from homeassistant.setup import async_prepare_setup_platform, async_start_setup
|
||||
from homeassistant.util import dt as dt_util
|
||||
from homeassistant.util.yaml import dump
|
||||
|
@ -69,9 +69,9 @@ from .const import (
|
|||
SOURCE_TYPE_ROUTER,
|
||||
)
|
||||
|
||||
SERVICE_SEE = "see"
|
||||
SERVICE_SEE: Final = "see"
|
||||
|
||||
SOURCE_TYPES = (
|
||||
SOURCE_TYPES: Final[tuple[str, ...]] = (
|
||||
SOURCE_TYPE_GPS,
|
||||
SOURCE_TYPE_ROUTER,
|
||||
SOURCE_TYPE_BLUETOOTH,
|
||||
|
@ -92,9 +92,11 @@ PLATFORM_SCHEMA: Final = cv.PLATFORM_SCHEMA.extend(
|
|||
vol.Optional(CONF_NEW_DEVICE_DEFAULTS, default={}): NEW_DEVICE_DEFAULTS_SCHEMA,
|
||||
}
|
||||
)
|
||||
PLATFORM_SCHEMA_BASE = cv.PLATFORM_SCHEMA_BASE.extend(PLATFORM_SCHEMA.schema)
|
||||
PLATFORM_SCHEMA_BASE: Final[vol.Schema] = cv.PLATFORM_SCHEMA_BASE.extend(
|
||||
PLATFORM_SCHEMA.schema
|
||||
)
|
||||
|
||||
SERVICE_SEE_PAYLOAD_SCHEMA = vol.Schema(
|
||||
SERVICE_SEE_PAYLOAD_SCHEMA: Final[vol.Schema] = vol.Schema(
|
||||
vol.All(
|
||||
cv.has_at_least_one_key(ATTR_MAC, ATTR_DEV_ID),
|
||||
{
|
||||
|
@ -115,23 +117,23 @@ SERVICE_SEE_PAYLOAD_SCHEMA = vol.Schema(
|
|||
)
|
||||
)
|
||||
|
||||
YAML_DEVICES = "known_devices.yaml"
|
||||
EVENT_NEW_DEVICE = "device_tracker_new_device"
|
||||
YAML_DEVICES: Final = "known_devices.yaml"
|
||||
EVENT_NEW_DEVICE: Final = "device_tracker_new_device"
|
||||
|
||||
|
||||
def see(
|
||||
hass: HomeAssistant,
|
||||
mac: str = None,
|
||||
dev_id: str = None,
|
||||
host_name: str = None,
|
||||
location_name: str = None,
|
||||
gps: GPSType = None,
|
||||
gps_accuracy=None,
|
||||
battery: int = None,
|
||||
attributes: dict = None,
|
||||
):
|
||||
mac: str | None = None,
|
||||
dev_id: str | None = None,
|
||||
host_name: str | None = None,
|
||||
location_name: str | None = None,
|
||||
gps: GPSType | None = None,
|
||||
gps_accuracy: int | None = None,
|
||||
battery: int | None = None,
|
||||
attributes: dict | None = None,
|
||||
) -> None:
|
||||
"""Call service to notify you see device."""
|
||||
data = {
|
||||
data: dict[str, Any] = {
|
||||
key: value
|
||||
for key, value in (
|
||||
(ATTR_MAC, mac),
|
||||
|
@ -144,7 +146,7 @@ def see(
|
|||
)
|
||||
if value is not None
|
||||
}
|
||||
if attributes:
|
||||
if attributes is not None:
|
||||
data[ATTR_ATTRIBUTES] = attributes
|
||||
hass.services.call(DOMAIN, SERVICE_SEE, data)
|
||||
|
||||
|
@ -163,7 +165,9 @@ async def async_setup_integration(hass: HomeAssistant, config: ConfigType) -> No
|
|||
if setup_tasks:
|
||||
await asyncio.wait(setup_tasks)
|
||||
|
||||
async def async_platform_discovered(p_type, info):
|
||||
async def async_platform_discovered(
|
||||
p_type: str, info: dict[str, Any] | None
|
||||
) -> None:
|
||||
"""Load a platform."""
|
||||
platform = await async_create_platform_type(hass, config, p_type, {})
|
||||
|
||||
|
@ -179,7 +183,7 @@ async def async_setup_integration(hass: HomeAssistant, config: ConfigType) -> No
|
|||
hass, tracker.async_update_stale, second=range(0, 60, 5)
|
||||
)
|
||||
|
||||
async def async_see_service(call):
|
||||
async def async_see_service(call: ServiceCall) -> None:
|
||||
"""Service to see a device."""
|
||||
# Temp workaround for iOS, introduced in 0.65
|
||||
data = dict(call.data)
|
||||
|
@ -199,7 +203,7 @@ async def async_setup_integration(hass: HomeAssistant, config: ConfigType) -> No
|
|||
class DeviceTrackerPlatform:
|
||||
"""Class to hold platform information."""
|
||||
|
||||
LEGACY_SETUP = (
|
||||
LEGACY_SETUP: Final[tuple[str, ...]] = (
|
||||
"async_get_scanner",
|
||||
"get_scanner",
|
||||
"async_setup_scanner",
|
||||
|
@ -211,17 +215,22 @@ class DeviceTrackerPlatform:
|
|||
config: dict = attr.ib()
|
||||
|
||||
@property
|
||||
def type(self):
|
||||
def type(self) -> str | None:
|
||||
"""Return platform type."""
|
||||
for methods, platform_type in ((self.LEGACY_SETUP, PLATFORM_TYPE_LEGACY),):
|
||||
for meth in methods:
|
||||
if hasattr(self.platform, meth):
|
||||
return platform_type
|
||||
|
||||
methods, platform_type = self.LEGACY_SETUP, PLATFORM_TYPE_LEGACY
|
||||
for method in methods:
|
||||
if hasattr(self.platform, method):
|
||||
return platform_type
|
||||
return None
|
||||
|
||||
async def async_setup_legacy(self, hass, tracker, discovery_info=None):
|
||||
async def async_setup_legacy(
|
||||
self,
|
||||
hass: HomeAssistant,
|
||||
tracker: DeviceTracker,
|
||||
discovery_info: dict[str, Any] | None = None,
|
||||
) -> None:
|
||||
"""Set up a legacy platform."""
|
||||
assert self.type == PLATFORM_TYPE_LEGACY
|
||||
full_name = f"{DOMAIN}.{self.name}"
|
||||
LOGGER.info("Setting up %s", full_name)
|
||||
with async_start_setup(hass, [full_name]):
|
||||
|
@ -229,20 +238,22 @@ class DeviceTrackerPlatform:
|
|||
scanner = None
|
||||
setup = None
|
||||
if hasattr(self.platform, "async_get_scanner"):
|
||||
scanner = await self.platform.async_get_scanner(
|
||||
scanner = await self.platform.async_get_scanner( # type: ignore[attr-defined]
|
||||
hass, {DOMAIN: self.config}
|
||||
)
|
||||
elif hasattr(self.platform, "get_scanner"):
|
||||
scanner = await hass.async_add_executor_job(
|
||||
self.platform.get_scanner, hass, {DOMAIN: self.config}
|
||||
self.platform.get_scanner, # type: ignore[attr-defined]
|
||||
hass,
|
||||
{DOMAIN: self.config},
|
||||
)
|
||||
elif hasattr(self.platform, "async_setup_scanner"):
|
||||
setup = await self.platform.async_setup_scanner(
|
||||
setup = await self.platform.async_setup_scanner( # type: ignore[attr-defined]
|
||||
hass, self.config, tracker.async_see, discovery_info
|
||||
)
|
||||
elif hasattr(self.platform, "setup_scanner"):
|
||||
setup = await hass.async_add_executor_job(
|
||||
self.platform.setup_scanner,
|
||||
self.platform.setup_scanner, # type: ignore[attr-defined]
|
||||
hass,
|
||||
self.config,
|
||||
tracker.see,
|
||||
|
@ -251,12 +262,12 @@ class DeviceTrackerPlatform:
|
|||
else:
|
||||
raise HomeAssistantError("Invalid legacy device_tracker platform.")
|
||||
|
||||
if scanner:
|
||||
if scanner is not None:
|
||||
async_setup_scanner_platform(
|
||||
hass, self.config, scanner, tracker.async_see, self.type
|
||||
)
|
||||
|
||||
if not setup and not scanner:
|
||||
if setup is None and scanner is None:
|
||||
LOGGER.error(
|
||||
"Error setting up platform %s %s", self.type, self.name
|
||||
)
|
||||
|
@ -270,9 +281,11 @@ class DeviceTrackerPlatform:
|
|||
)
|
||||
|
||||
|
||||
async def async_extract_config(hass, config):
|
||||
async def async_extract_config(
|
||||
hass: HomeAssistant, config: ConfigType
|
||||
) -> list[DeviceTrackerPlatform]:
|
||||
"""Extract device tracker config and split between legacy and modern."""
|
||||
legacy = []
|
||||
legacy: list[DeviceTrackerPlatform] = []
|
||||
|
||||
for platform in await asyncio.gather(
|
||||
*(
|
||||
|
@ -294,7 +307,7 @@ async def async_extract_config(hass, config):
|
|||
|
||||
|
||||
async def async_create_platform_type(
|
||||
hass, config, p_type, p_config
|
||||
hass: HomeAssistant, config: ConfigType, p_type: str, p_config: dict
|
||||
) -> DeviceTrackerPlatform | None:
|
||||
"""Determine type of platform."""
|
||||
platform = await async_prepare_setup_platform(hass, config, DOMAIN, p_type)
|
||||
|
@ -310,9 +323,9 @@ def async_setup_scanner_platform(
|
|||
hass: HomeAssistant,
|
||||
config: ConfigType,
|
||||
scanner: DeviceScanner,
|
||||
async_see_device: Callable,
|
||||
async_see_device: Callable[..., Coroutine[None, None, None]],
|
||||
platform: str,
|
||||
):
|
||||
) -> None:
|
||||
"""Set up the connect scanner-based platform to device tracker.
|
||||
|
||||
This method must be run in the event loop.
|
||||
|
@ -324,7 +337,7 @@ def async_setup_scanner_platform(
|
|||
# Initial scan of each mac we also tell about host name for config
|
||||
seen: Any = set()
|
||||
|
||||
async def async_device_tracker_scan(now: dt_util.dt.datetime | None):
|
||||
async def async_device_tracker_scan(now: dt_util.dt.datetime | None) -> None:
|
||||
"""Handle interval matches."""
|
||||
if update_lock.locked():
|
||||
LOGGER.warning(
|
||||
|
@ -350,7 +363,7 @@ def async_setup_scanner_platform(
|
|||
except NotImplementedError:
|
||||
extra_attributes = {}
|
||||
|
||||
kwargs = {
|
||||
kwargs: dict[str, Any] = {
|
||||
"mac": mac,
|
||||
"host_name": host_name,
|
||||
"source_type": SOURCE_TYPE_ROUTER,
|
||||
|
@ -361,7 +374,7 @@ def async_setup_scanner_platform(
|
|||
}
|
||||
|
||||
zone_home = hass.states.get(hass.components.zone.ENTITY_ID_HOME)
|
||||
if zone_home:
|
||||
if zone_home is not None:
|
||||
kwargs["gps"] = [
|
||||
zone_home.attributes[ATTR_LATITUDE],
|
||||
zone_home.attributes[ATTR_LONGITUDE],
|
||||
|
@ -374,7 +387,7 @@ def async_setup_scanner_platform(
|
|||
hass.async_create_task(async_device_tracker_scan(None))
|
||||
|
||||
|
||||
async def get_tracker(hass, config):
|
||||
async def get_tracker(hass: HomeAssistant, config: ConfigType) -> DeviceTracker:
|
||||
"""Create a tracker."""
|
||||
yaml_path = hass.config.path(YAML_DEVICES)
|
||||
|
||||
|
@ -400,12 +413,12 @@ class DeviceTracker:
|
|||
hass: HomeAssistant,
|
||||
consider_home: timedelta,
|
||||
track_new: bool,
|
||||
defaults: dict,
|
||||
devices: Sequence,
|
||||
defaults: dict[str, Any],
|
||||
devices: Sequence[Device],
|
||||
) -> None:
|
||||
"""Initialize a device tracker."""
|
||||
self.hass = hass
|
||||
self.devices = {dev.dev_id: dev for dev in devices}
|
||||
self.devices: dict[str, Device] = {dev.dev_id: dev for dev in devices}
|
||||
self.mac_to_dev = {dev.mac: dev for dev in devices if dev.mac}
|
||||
self.consider_home = consider_home
|
||||
self.track_new = (
|
||||
|
@ -436,7 +449,7 @@ class DeviceTracker:
|
|||
picture: str | None = None,
|
||||
icon: str | None = None,
|
||||
consider_home: timedelta | None = None,
|
||||
):
|
||||
) -> None:
|
||||
"""Notify the device tracker that you see a device."""
|
||||
self.hass.create_task(
|
||||
self.async_see(
|
||||
|
@ -556,7 +569,7 @@ class DeviceTracker:
|
|||
)
|
||||
)
|
||||
|
||||
async def async_update_config(self, path, dev_id, device):
|
||||
async def async_update_config(self, path: str, dev_id: str, device: Device) -> None:
|
||||
"""Add device to YAML configuration file.
|
||||
|
||||
This method is a coroutine.
|
||||
|
@ -567,7 +580,7 @@ class DeviceTracker:
|
|||
)
|
||||
|
||||
@callback
|
||||
def async_update_stale(self, now: dt_util.dt.datetime):
|
||||
def async_update_stale(self, now: dt_util.dt.datetime) -> None:
|
||||
"""Update stale devices.
|
||||
|
||||
This method must be run in the event loop.
|
||||
|
@ -576,18 +589,18 @@ class DeviceTracker:
|
|||
if (device.track and device.last_update_home) and device.stale(now):
|
||||
self.hass.async_create_task(device.async_update_ha_state(True))
|
||||
|
||||
async def async_setup_tracked_device(self):
|
||||
async def async_setup_tracked_device(self) -> None:
|
||||
"""Set up all not exists tracked devices.
|
||||
|
||||
This method is a coroutine.
|
||||
"""
|
||||
|
||||
async def async_init_single_device(dev):
|
||||
async def async_init_single_device(dev: Device) -> None:
|
||||
"""Init a single device_tracker entity."""
|
||||
await dev.async_added_to_hass()
|
||||
dev.async_write_ha_state()
|
||||
|
||||
tasks = []
|
||||
tasks: list[asyncio.Task] = []
|
||||
for device in self.devices.values():
|
||||
if device.track and not device.last_seen:
|
||||
tasks.append(
|
||||
|
@ -610,8 +623,8 @@ class Device(RestoreEntity):
|
|||
attributes: dict | None = None
|
||||
|
||||
# Track if the last update of this device was HOME.
|
||||
last_update_home = False
|
||||
_state = STATE_NOT_HOME
|
||||
last_update_home: bool = False
|
||||
_state: str = STATE_NOT_HOME
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
|
@ -644,6 +657,7 @@ class Device(RestoreEntity):
|
|||
self.config_name = name
|
||||
|
||||
# Configured picture
|
||||
self.config_picture: str | None
|
||||
if gravatar is not None:
|
||||
self.config_picture = get_gravatar_for_email(gravatar)
|
||||
else:
|
||||
|
@ -656,32 +670,32 @@ class Device(RestoreEntity):
|
|||
self._attributes: dict[str, Any] = {}
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
def name(self) -> str:
|
||||
"""Return the name of the entity."""
|
||||
return self.config_name or self.host_name or self.dev_id or DEVICE_DEFAULT_NAME
|
||||
|
||||
@property
|
||||
def state(self):
|
||||
def state(self) -> str:
|
||||
"""Return the state of the device."""
|
||||
return self._state
|
||||
|
||||
@property
|
||||
def entity_picture(self):
|
||||
def entity_picture(self) -> str | None:
|
||||
"""Return the picture of the device."""
|
||||
return self.config_picture
|
||||
|
||||
@final
|
||||
@property
|
||||
def state_attributes(self):
|
||||
def state_attributes(self) -> dict[str, StateType]:
|
||||
"""Return the device state attributes."""
|
||||
attributes = {ATTR_SOURCE_TYPE: self.source_type}
|
||||
attributes: dict[str, StateType] = {ATTR_SOURCE_TYPE: self.source_type}
|
||||
|
||||
if self.gps:
|
||||
if self.gps is not None:
|
||||
attributes[ATTR_LATITUDE] = self.gps[0]
|
||||
attributes[ATTR_LONGITUDE] = self.gps[1]
|
||||
attributes[ATTR_GPS_ACCURACY] = self.gps_accuracy
|
||||
|
||||
if self.battery:
|
||||
if self.battery is not None:
|
||||
attributes[ATTR_BATTERY] = self.battery
|
||||
|
||||
return attributes
|
||||
|
@ -742,13 +756,13 @@ class Device(RestoreEntity):
|
|||
or (now or dt_util.utcnow()) - self.last_seen > self.consider_home
|
||||
)
|
||||
|
||||
def mark_stale(self):
|
||||
def mark_stale(self) -> None:
|
||||
"""Mark the device state as stale."""
|
||||
self._state = STATE_NOT_HOME
|
||||
self.gps = None
|
||||
self.last_update_home = False
|
||||
|
||||
async def async_update(self):
|
||||
async def async_update(self) -> None:
|
||||
"""Update state of entity.
|
||||
|
||||
This method is a coroutine.
|
||||
|
@ -773,7 +787,7 @@ class Device(RestoreEntity):
|
|||
self._state = STATE_HOME
|
||||
self.last_update_home = True
|
||||
|
||||
async def async_added_to_hass(self):
|
||||
async def async_added_to_hass(self) -> None:
|
||||
"""Add an entity."""
|
||||
await super().async_added_to_hass()
|
||||
state = await self.async_get_last_state()
|
||||
|
@ -807,7 +821,7 @@ class DeviceScanner:
|
|||
"""Scan for devices."""
|
||||
raise NotImplementedError()
|
||||
|
||||
async def async_scan_devices(self) -> Any:
|
||||
async def async_scan_devices(self) -> list[str]:
|
||||
"""Scan for devices."""
|
||||
assert (
|
||||
self.hass is not None
|
||||
|
@ -829,7 +843,7 @@ class DeviceScanner:
|
|||
"""Get the extra attributes of a device."""
|
||||
raise NotImplementedError()
|
||||
|
||||
async def async_get_extra_attributes(self, device: str) -> Any:
|
||||
async def async_get_extra_attributes(self, device: str) -> dict:
|
||||
"""Get the extra attributes of a device."""
|
||||
assert (
|
||||
self.hass is not None
|
||||
|
@ -837,7 +851,9 @@ class DeviceScanner:
|
|||
return await self.hass.async_add_executor_job(self.get_extra_attributes, device)
|
||||
|
||||
|
||||
async def async_load_config(path: str, hass: HomeAssistant, consider_home: timedelta):
|
||||
async def async_load_config(
|
||||
path: str, hass: HomeAssistant, consider_home: timedelta
|
||||
) -> list[Device]:
|
||||
"""Load devices from YAML configuration file.
|
||||
|
||||
This method is a coroutine.
|
||||
|
@ -857,7 +873,7 @@ async def async_load_config(path: str, hass: HomeAssistant, consider_home: timed
|
|||
),
|
||||
}
|
||||
)
|
||||
result = []
|
||||
result: list[Device] = []
|
||||
try:
|
||||
devices = await hass.async_add_executor_job(load_yaml_config_file, path)
|
||||
except HomeAssistantError as err:
|
||||
|
@ -880,7 +896,7 @@ async def async_load_config(path: str, hass: HomeAssistant, consider_home: timed
|
|||
return result
|
||||
|
||||
|
||||
def update_config(path: str, dev_id: str, device: Device):
|
||||
def update_config(path: str, dev_id: str, device: Device) -> None:
|
||||
"""Add device to YAML configuration file."""
|
||||
with open(path, "a") as out:
|
||||
device_config = {
|
||||
|
@ -896,7 +912,7 @@ def update_config(path: str, dev_id: str, device: Device):
|
|||
out.write(dump(device_config))
|
||||
|
||||
|
||||
def get_gravatar_for_email(email: str):
|
||||
def get_gravatar_for_email(email: str) -> str:
|
||||
"""Return an 80px Gravatar for the given email address.
|
||||
|
||||
Async friendly.
|
||||
|
|
11
mypy.ini
11
mypy.ini
|
@ -242,6 +242,17 @@ no_implicit_optional = true
|
|||
warn_return_any = true
|
||||
warn_unreachable = true
|
||||
|
||||
[mypy-homeassistant.components.device_tracker.*]
|
||||
check_untyped_defs = true
|
||||
disallow_incomplete_defs = true
|
||||
disallow_subclassing_any = true
|
||||
disallow_untyped_calls = true
|
||||
disallow_untyped_decorators = true
|
||||
disallow_untyped_defs = true
|
||||
no_implicit_optional = true
|
||||
warn_return_any = true
|
||||
warn_unreachable = true
|
||||
|
||||
[mypy-homeassistant.components.elgato.*]
|
||||
check_untyped_defs = true
|
||||
disallow_incomplete_defs = true
|
||||
|
|
Loading…
Add table
Reference in a new issue