Skip to content

Commit

Permalink
Add strict typing to Tractive integration (home-assistant#56948)
Browse files Browse the repository at this point in the history
* Strict typing

* Add few missing types

* Run hassfest

* Fix mypy errors

* Use List instead of list
  • Loading branch information
bieniu authored Oct 3, 2021
1 parent 1aeab65 commit f3c76fb
Show file tree
Hide file tree
Showing 10 changed files with 166 additions and 134 deletions.
1 change: 1 addition & 0 deletions .strict-typing
Original file line number Diff line number Diff line change
Expand Up @@ -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.*
Expand Down
51 changes: 32 additions & 19 deletions homeassistant/components/tractive/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
import asyncio
from dataclasses import dataclass
import logging
from typing import Any, Final, List, cast

import aiotractive

Expand All @@ -15,7 +16,7 @@
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
Expand All @@ -36,10 +37,10 @@
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
Expand Down Expand Up @@ -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(
Expand All @@ -102,13 +103,16 @@ async def cancel_listen_task(_):
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"])

Expand All @@ -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:
Expand Down Expand Up @@ -191,7 +202,7 @@ async def _listen(self):
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"],
Expand All @@ -204,7 +215,7 @@ def _send_hardware_update(self, event):
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"],
Expand All @@ -213,7 +224,7 @@ def _send_activity_update(self, event):
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],
Expand All @@ -223,7 +234,9 @@ def _send_position_update(self, event):
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}",
Expand Down
50 changes: 25 additions & 25 deletions homeassistant/components/tractive/binary_sensor.py
Original file line number Diff line number Diff line change
@@ -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,
Expand All @@ -19,34 +24,36 @@
)
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(
Expand All @@ -66,31 +73,24 @@ async def async_added_to_hass(self):
)


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 changes: 4 additions & 4 deletions homeassistant/components/tractive/config_flow.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -15,9 +15,9 @@

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}
)

Expand Down Expand Up @@ -74,7 +74,7 @@ async def async_step_reauth(self, _: dict[str, Any]) -> FlowResult:
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."""

Expand Down
27 changes: 14 additions & 13 deletions homeassistant/components/tractive/const.py
Original file line number Diff line number Diff line change
@@ -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"
Loading

0 comments on commit f3c76fb

Please sign in to comment.