evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add strict typing to Tractive integration (#56948)

Browse source 
* Strict typing

* Add few missing types

* Run hassfest

* Fix mypy errors

* Use List instead of list
This commit is contained in:
Maciej Bieniek 2021-10-03 09:13:12 +02:00 • committed by GitHub
parent 1aeab65f56
commit f3c76fb859
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
10 changed files with 166 additions and 134 deletions
Download patch file Download diff file Expand all files Collapse all files

1
.strict-typing
View file

@ -112,6 +112,7 @@ homeassistant.components.tautulli.*
homeassistant.components.tcp.*
homeassistant.components.tile.*
homeassistant.components.tplink.*
homeassistant.components.tractive.*
homeassistant.components.tradfri.*
homeassistant.components.tts.*
homeassistant.components.upcloud.*

51
homeassistant/components/tractive/__init__.py
View file

@ -4,6 +4,7 @@ from __future__ import annotations
import asyncio
from dataclasses import dataclass
import logging
from typing import Any, Final, List, cast
import aiotractive
@ -15,7 +16,7 @@ from homeassistant.const import (
CONF_PASSWORD,
EVENT_HOMEASSISTANT_STOP,
)
from homeassistant.core import HomeAssistant
from homeassistant.core import Event, HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.dispatcher import async_dispatcher_send
@ -36,10 +37,10 @@ from .const import (
TRACKER_POSITION_UPDATED,
)
PLATFORMS = ["binary_sensor", "device_tracker", "sensor", "switch"]
PLATFORMS: Final = ["binary_sensor", "device_tracker", "sensor", "switch"]
_LOGGER = logging.getLogger(__name__)
_LOGGER: Final = logging.getLogger(__name__)
@dataclass
@ -92,7 +93,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
async def cancel_listen_task(_):
async def cancel_listen_task(_: Event) -> None:
await tractive.unsubscribe()
entry.async_on_unload(
@ -102,13 +103,16 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
return True
async def _generate_trackables(client, trackable):
async def _generate_trackables(
client: aiotractive.Tractive,
trackable: aiotractive.trackable_object.TrackableObject,
) -> Trackables | None:
"""Generate trackables."""
trackable = await trackable.details()
# Check that the pet has tracker linked.
if not trackable["device_id"]:
return
return None
tracker = client.tracker(trackable["device_id"])
@ -132,37 +136,44 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
class TractiveClient:
"""A Tractive client."""
def __init__(self, hass, client, user_id):
def __init__(
self, hass: HomeAssistant, client: aiotractive.Tractive, user_id: str
) -> None:
"""Initialize the client."""
self._hass = hass
self._client = client
self._user_id = user_id
self._listen_task = None
self._listen_task: asyncio.Task | None = None
@property
def user_id(self):
def user_id(self) -> str:
"""Return user id."""
return self._user_id
async def trackable_objects(self):
async def trackable_objects(
self,
) -> list[aiotractive.trackable_object.TrackableObject]:
"""Get list of trackable objects."""
return await self._client.trackable_objects()
return cast(
List[aiotractive.trackable_object.TrackableObject],
await self._client.trackable_objects(),
)
def tracker(self, tracker_id):
def tracker(self, tracker_id: str) -> aiotractive.tracker.Tracker:
"""Get tracker by id."""
return self._client.tracker(tracker_id)
def subscribe(self):
def subscribe(self) -> None:
"""Start event listener coroutine."""
self._listen_task = asyncio.create_task(self._listen())
async def unsubscribe(self):
async def unsubscribe(self) -> None:
"""Stop event listener coroutine."""
if self._listen_task:
self._listen_task.cancel()
await self._client.close()
async def _listen(self):
async def _listen(self) -> None:
server_was_unavailable = False
while True:
try:
@ -191,7 +202,7 @@ class TractiveClient:
server_was_unavailable = True
continue
def _send_hardware_update(self, event):
def _send_hardware_update(self, event: dict[str, Any]) -> None:
# Sometimes hardware event doesn't contain complete data.
payload = {
ATTR_BATTERY_LEVEL: event["hardware"]["battery_level"],
@ -204,7 +215,7 @@ class TractiveClient:
TRACKER_HARDWARE_STATUS_UPDATED, event["tracker_id"], payload
)
def _send_activity_update(self, event):
def _send_activity_update(self, event: dict[str, Any]) -> None:
payload = {
ATTR_MINUTES_ACTIVE: event["progress"]["achieved_minutes"],
ATTR_DAILY_GOAL: event["progress"]["goal_minutes"],
@ -213,7 +224,7 @@ class TractiveClient:
TRACKER_ACTIVITY_STATUS_UPDATED, event["pet_id"], payload
)
def _send_position_update(self, event):
def _send_position_update(self, event: dict[str, Any]) -> None:
payload = {
"latitude": event["position"]["latlong"][0],
"longitude": event["position"]["latlong"][1],
@ -223,7 +234,9 @@ class TractiveClient:
TRACKER_POSITION_UPDATED, event["tracker_id"], payload
)
def _dispatch_tracker_event(self, event_name, tracker_id, payload):
def _dispatch_tracker_event(
self, event_name: str, tracker_id: str, payload: dict[str, Any]
) -> None:
async_dispatcher_send(
self._hass,
f"{event_name}-{tracker_id}",

50
homeassistant/components/tractive/binary_sensor.py
View file

@ -1,15 +1,20 @@
"""Support for Tractive binary sensors."""
from __future__ import annotations
from typing import Any, Final
from homeassistant.components.binary_sensor import (
DEVICE_CLASS_BATTERY_CHARGING,
BinarySensorEntity,
BinarySensorEntityDescription,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_BATTERY_CHARGING
from homeassistant.core import callback
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import Trackables
from .const import (
CLIENT,
DOMAIN,
@ -19,34 +24,36 @@ from .const import (
)
from .entity import TractiveEntity
TRACKERS_WITH_BUILTIN_BATTERY = ("TRNJA4", "TRAXL1")
TRACKERS_WITH_BUILTIN_BATTERY: Final = ("TRNJA4", "TRAXL1")
class TractiveBinarySensor(TractiveEntity, BinarySensorEntity):
"""Tractive sensor."""
def __init__(self, user_id, trackable, tracker_details, unique_id, description):
def __init__(
self, user_id: str, item: Trackables, description: BinarySensorEntityDescription
) -> None:
"""Initialize sensor entity."""
super().__init__(user_id, trackable, tracker_details)
super().__init__(user_id, item.trackable, item.tracker_details)
self._attr_name = f"{trackable['details']['name']} {description.name}"
self._attr_unique_id = unique_id
self._attr_name = f"{item.trackable['details']['name']} {description.name}"
self._attr_unique_id = f"{item.trackable['_id']}_{description.key}"
self.entity_description = description
@callback
def handle_server_unavailable(self):
def handle_server_unavailable(self) -> None:
"""Handle server unavailable."""
self._attr_available = False
self.async_write_ha_state()
@callback
def handle_hardware_status_update(self, event):
def handle_hardware_status_update(self, event: dict[str, Any]) -> None:
"""Handle hardware status update."""
self._attr_is_on = event[self.entity_description.key]
self._attr_available = True
self.async_write_ha_state()
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Handle entity which will be added."""
self.async_on_remove(
@ -66,31 +73,24 @@ class TractiveBinarySensor(TractiveEntity, BinarySensorEntity):
)
SENSOR_TYPE = BinarySensorEntityDescription(
SENSOR_TYPE: Final = BinarySensorEntityDescription(
key=ATTR_BATTERY_CHARGING,
name="Battery Charging",
device_class=DEVICE_CLASS_BATTERY_CHARGING,
)
async def async_setup_entry(hass, entry, async_add_entities):
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up Tractive device trackers."""
client = hass.data[DOMAIN][entry.entry_id][CLIENT]
trackables = hass.data[DOMAIN][entry.entry_id][TRACKABLES]
entities = []
for item in trackables:
if item.tracker_details["model_number"] not in TRACKERS_WITH_BUILTIN_BATTERY:
continue
entities.append(
TractiveBinarySensor(
client.user_id,
item.trackable,
item.tracker_details,
f"{item.trackable['_id']}_{SENSOR_TYPE.key}",
SENSOR_TYPE,
)
)
entities = [
TractiveBinarySensor(client.user_id, item, SENSOR_TYPE)
for item in trackables
if item.tracker_details["model_number"] in TRACKERS_WITH_BUILTIN_BATTERY
]
async_add_entities(entities)

8
homeassistant/components/tractive/config_flow.py
View file

@ -2,7 +2,7 @@
from __future__ import annotations
import logging
from typing import Any
from typing import Any, Final
import aiotractive
import voluptuous as vol
@ -15,9 +15,9 @@ from homeassistant.exceptions import HomeAssistantError
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
_LOGGER: Final = logging.getLogger(__name__)
USER_DATA_SCHEMA = vol.Schema(
USER_DATA_SCHEMA: Final = vol.Schema(
{vol.Required(CONF_EMAIL): str, vol.Required(CONF_PASSWORD): str}
)
@ -74,7 +74,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return await self.async_step_reauth_confirm()
async def async_step_reauth_confirm(
self, user_input: dict[str, Any] = None
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Dialog that informs the user that reauth is required."""

27
homeassistant/components/tractive/const.py
View file

@ -1,22 +1,23 @@
"""Constants for the tractive integration."""
from datetime import timedelta
from typing import Final
DOMAIN = "tractive"
DOMAIN: Final = "tractive"
RECONNECT_INTERVAL = timedelta(seconds=10)
RECONNECT_INTERVAL: Final = timedelta(seconds=10)
ATTR_DAILY_GOAL = "daily_goal"
ATTR_BUZZER = "buzzer"
ATTR_LED = "led"
ATTR_LIVE_TRACKING = "live_tracking"
ATTR_MINUTES_ACTIVE = "minutes_active"
ATTR_DAILY_GOAL: Final = "daily_goal"
ATTR_BUZZER: Final = "buzzer"
ATTR_LED: Final = "led"
ATTR_LIVE_TRACKING: Final = "live_tracking"
ATTR_MINUTES_ACTIVE: Final = "minutes_active"
CLIENT = "client"
TRACKABLES = "trackables"
CLIENT: Final = "client"
TRACKABLES: Final = "trackables"
TRACKER_HARDWARE_STATUS_UPDATED = f"{DOMAIN}_tracker_hardware_status_updated"
TRACKER_POSITION_UPDATED = f"{DOMAIN}_tracker_position_updated"
TRACKER_ACTIVITY_STATUS_UPDATED = f"{DOMAIN}_tracker_activity_updated"
TRACKER_HARDWARE_STATUS_UPDATED: Final = f"{DOMAIN}_tracker_hardware_status_updated"
TRACKER_POSITION_UPDATED: Final = f"{DOMAIN}_tracker_position_updated"
TRACKER_ACTIVITY_STATUS_UPDATED: Final = f"{DOMAIN}_tracker_activity_updated"
SERVER_UNAVAILABLE = f"{DOMAIN}_server_unavailable"
SERVER_UNAVAILABLE: Final = f"{DOMAIN}_server_unavailable"

65
homeassistant/components/tractive/device_tracker.py
View file

@ -1,12 +1,16 @@
"""Support for Tractive device trackers."""
from __future__ import annotations
import logging
from typing import Any
from homeassistant.components.device_tracker import SOURCE_TYPE_GPS
from homeassistant.components.device_tracker.config_entry import TrackerEntity
from homeassistant.core import callback
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import Trackables
from .const import (
CLIENT,
DOMAIN,
@ -17,26 +21,15 @@ from .const import (
)
from .entity import TractiveEntity
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass, entry, async_add_entities):
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up Tractive device trackers."""
client = hass.data[DOMAIN][entry.entry_id][CLIENT]
trackables = hass.data[DOMAIN][entry.entry_id][TRACKABLES]
entities = []
for item in trackables:
entities.append(
TractiveDeviceTracker(
client.user_id,
item.trackable,
item.tracker_details,
item.hw_info,
item.pos_report,
)
)
entities = [TractiveDeviceTracker(client.user_id, item) for item in trackables]
async_add_entities(entities)
@ -46,51 +39,51 @@ class TractiveDeviceTracker(TractiveEntity, TrackerEntity):
_attr_icon = "mdi:paw"
def __init__(self, user_id, trackable, tracker_details, hw_info, pos_report):
def __init__(self, user_id: str, item: Trackables) -> None:
"""Initialize tracker entity."""
super().__init__(user_id, trackable, tracker_details)
super().__init__(user_id, item.trackable, item.tracker_details)
self._battery_level = hw_info["battery_level"]
self._latitude = pos_report["latlong"][0]
self._longitude = pos_report["latlong"][1]
self._accuracy = pos_report["pos_uncertainty"]
self._battery_level: int = item.hw_info["battery_level"]
self._latitude: float = item.pos_report["latlong"][0]
self._longitude: float = item.pos_report["latlong"][1]
self._accuracy: int = item.pos_report["pos_uncertainty"]
self._attr_name = f"{self._tracker_id} {trackable['details']['name']}"
self._attr_unique_id = trackable["_id"]
self._attr_name = f"{self._tracker_id} {item.trackable['details']['name']}"
self._attr_unique_id = item.trackable["_id"]
@property
def source_type(self):
def source_type(self) -> str:
"""Return the source type, eg gps or router, of the device."""
return SOURCE_TYPE_GPS
@property
def latitude(self):
def latitude(self) -> float:
"""Return latitude value of the device."""
return self._latitude
@property
def longitude(self):
def longitude(self) -> float:
"""Return longitude value of the device."""
return self._longitude
@property
def location_accuracy(self):
def location_accuracy(self) -> int:
"""Return the gps accuracy of the device."""
return self._accuracy
@property
def battery_level(self):
def battery_level(self) -> int:
"""Return the battery level of the device."""
return self._battery_level
@callback
def _handle_hardware_status_update(self, event):
def _handle_hardware_status_update(self, event: dict[str, Any]) -> None:
self._battery_level = event["battery_level"]
self._attr_available = True
self.async_write_ha_state()
@callback
def _handle_position_update(self, event):
def _handle_position_update(self, event: dict[str, Any]) -> None:
self._latitude = event["latitude"]
self._longitude = event["longitude"]
self._accuracy = event["accuracy"]
@ -98,15 +91,11 @@ class TractiveDeviceTracker(TractiveEntity, TrackerEntity):
self.async_write_ha_state()
@callback
def _handle_server_unavailable(self):
self._latitude = None
self._longitude = None
self._accuracy = None
self._battery_level = None
def _handle_server_unavailable(self) -> None:
self._attr_available = False
self.async_write_ha_state()
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Handle entity which will be added."""
self.async_on_remove(

7
homeassistant/components/tractive/entity.py
View file

@ -1,4 +1,7 @@
"""A entity class for Tractive integration."""
from __future__ import annotations
from typing import Any
from homeassistant.helpers.entity import Entity
@ -8,7 +11,9 @@ from .const import DOMAIN
class TractiveEntity(Entity):
"""Tractive entity class."""
def __init__(self, user_id, trackable, tracker_details):
def __init__(
self, user_id: str, trackable: dict[str, Any], tracker_details: dict[str, Any]
) -> None:
"""Initialize tracker entity."""
self._attr_device_info = {
"identifiers": {(DOMAIN, tracker_details["_id"])},

66
homeassistant/components/tractive/sensor.py
View file

@ -2,17 +2,21 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Final
from homeassistant.components.sensor import SensorEntity, SensorEntityDescription
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
ATTR_BATTERY_LEVEL,
DEVICE_CLASS_BATTERY,
PERCENTAGE,
TIME_MINUTES,
)
from homeassistant.core import callback
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import Trackables
from .const import (
ATTR_DAILY_GOAL,
ATTR_MINUTES_ACTIVE,
@ -27,25 +31,37 @@ from .entity import TractiveEntity
@dataclass
class TractiveSensorEntityDescription(SensorEntityDescription):
"""Class describing Tractive sensor entities."""
class TractiveRequiredKeysMixin:
"""Mixin for required keys."""
entity_class: type[TractiveSensor] | None = None
entity_class: type[TractiveSensor]
@dataclass
class TractiveSensorEntityDescription(
SensorEntityDescription, TractiveRequiredKeysMixin
):
"""Class describing Tractive sensor entities."""
class TractiveSensor(TractiveEntity, SensorEntity):
"""Tractive sensor."""
def __init__(self, user_id, trackable, tracker_details, unique_id, description):
def __init__(
self,
user_id: str,
item: Trackables,
description: TractiveSensorEntityDescription,
) -> None:
"""Initialize sensor entity."""
super().__init__(user_id, trackable, tracker_details)
super().__init__(user_id, item.trackable, item.tracker_details)
self._attr_name = f"{trackable['details']['name']} {description.name}"
self._attr_unique_id = unique_id
self._attr_name = f"{item.trackable['details']['name']} {description.name}"
self._attr_unique_id = f"{item.trackable['_id']}_{description.key}"
self.entity_description = description
@callback
def handle_server_unavailable(self):
def handle_server_unavailable(self) -> None:
"""Handle server unavailable."""
self._attr_available = False
self.async_write_ha_state()
@ -55,13 +71,13 @@ class TractiveHardwareSensor(TractiveSensor):
"""Tractive hardware sensor."""
@callback
def handle_hardware_status_update(self, event):
def handle_hardware_status_update(self, event: dict[str, Any]) -> None:
"""Handle hardware status update."""
self._attr_native_value = event[self.entity_description.key]
self._attr_available = True
self.async_write_ha_state()
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Handle entity which will be added."""
self.async_on_remove(
@ -85,13 +101,13 @@ class TractiveActivitySensor(TractiveSensor):
"""Tractive active sensor."""
@callback
def handle_activity_status_update(self, event):
def handle_activity_status_update(self, event: dict[str, Any]) -> None:
"""Handle activity status update."""
self._attr_native_value = event[self.entity_description.key]
self._attr_available = True
self.async_write_ha_state()
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Handle entity which will be added."""
self.async_on_remove(
@ -111,7 +127,7 @@ class TractiveActivitySensor(TractiveSensor):
)
SENSOR_TYPES = (
SENSOR_TYPES: Final[tuple[TractiveSensorEntityDescription, ...]] = (
TractiveSensorEntityDescription(
key=ATTR_BATTERY_LEVEL,
name="Battery Level",
@ -136,23 +152,17 @@ SENSOR_TYPES = (
)
async def async_setup_entry(hass, entry, async_add_entities):
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up Tractive device trackers."""
client = hass.data[DOMAIN][entry.entry_id][CLIENT]
trackables = hass.data[DOMAIN][entry.entry_id][TRACKABLES]
entities = []
for item in trackables:
for description in SENSOR_TYPES:
entities.append(
description.entity_class(
client.user_id,
item.trackable,
item.tracker_details,
f"{item.trackable['_id']}_{description.key}",
description,
)
)
entities = [
description.entity_class(client.user_id, item, description)
for description in SENSOR_TYPES
for item in trackables
]
async_add_entities(entities)

14
homeassistant/components/tractive/switch.py
View file

@ -3,7 +3,7 @@ from __future__ import annotations
from dataclasses import dataclass
import logging
from typing import Any, Literal
from typing import Any, Final, Literal, cast
from aiotractive.exceptions import TractiveError
@ -26,7 +26,7 @@ from .const import (
)
from .entity import TractiveEntity
_LOGGER = logging.getLogger(__name__)
_LOGGER: Final = logging.getLogger(__name__)
@dataclass
@ -43,7 +43,7 @@ class TractiveSwitchEntityDescription(
"""Class describing Tractive switch entities."""
SWITCH_TYPES: tuple[TractiveSwitchEntityDescription, ...] = (
SWITCH_TYPES: Final[tuple[TractiveSwitchEntityDescription, ...]] = (
TractiveSwitchEntityDescription(
key=ATTR_BUZZER,
name="Tracker Buzzer",
@ -162,12 +162,14 @@ class TractiveSwitch(TractiveEntity, SwitchEntity):
async def async_set_buzzer(self, active: bool) -> dict[str, Any]:
"""Set the buzzer on/off."""
return await self._tracker.set_buzzer_active(active)
return cast(dict[str, Any], await self._tracker.set_buzzer_active(active))
async def async_set_led(self, active: bool) -> dict[str, Any]:
"""Set the LED on/off."""
return await self._tracker.set_led_active(active)
return cast(dict[str, Any], await self._tracker.set_led_active(active))
async def async_set_live_tracking(self, active: bool) -> dict[str, Any]:
"""Set the live tracking on/off."""
return await self._tracker.set_live_tracking_active(active)
return cast(
dict[str, Any], await self._tracker.set_live_tracking_active(active)
)

11
mypy.ini
View file

@ -1243,6 +1243,17 @@ no_implicit_optional = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.tractive.*]
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.tradfri.*]
check_untyped_defs = true
disallow_incomplete_defs = true