Skip to content

Commit

Permalink
🏷️ Add strict(er) typing
Browse files Browse the repository at this point in the history
The code now passes 'mypy tech pylint` as if it were a core component
  • Loading branch information
anarion80 committed Mar 16, 2024
1 parent 1702233 commit c6f8a21
Show file tree
Hide file tree
Showing 9 changed files with 959 additions and 769 deletions.
86 changes: 53 additions & 33 deletions custom_components/tech/__init__.py
Original file line number Diff line number Diff line change
@@ -1,8 +1,10 @@
"""The Tech Controllers integration."""
import asyncio
import logging
from typing import Any, Final

from aiohttp import ClientSession
from typing_extensions import TypeVar

from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_DESCRIPTION, CONF_NAME, CONF_TOKEN
Expand All @@ -14,6 +16,9 @@
entity_registry as er,
)
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.device_registry import DeviceEntry, DeviceRegistry
from homeassistant.helpers.entity_registry import EntityRegistry
from homeassistant.helpers.typing import ConfigType
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed

from . import assets
Expand All @@ -30,19 +35,21 @@
)
from .tech import Tech, TechError, TechLoginError

_LOGGER = logging.getLogger(__name__)
_LOGGER: Final = logging.getLogger(__name__)

CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)

_DataT = TypeVar("_DataT", default=dict[str, Any])

async def async_setup(hass: HomeAssistant, config: dict): # pylint: disable=unused-argument

async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: # pylint: disable=unused-argument
"""Set up the Tech Controllers component."""
return True


async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Tech Controllers from a config entry."""
_LOGGER.debug("Setting up component's entry.")
_LOGGER.debug("Setting up component's entry")
_LOGGER.debug("Entry id: %s", str(entry.entry_id))
_LOGGER.debug(
"Entry -> title: %s, data: %s, id: %s, domain: %s",
Expand All @@ -51,19 +58,20 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
entry.entry_id,
entry.domain,
)
language_code = hass.config.language
user_id = entry.data[USER_ID]
token = entry.data[CONF_TOKEN]
# Store an API object for your platforms to access
language_code: str = hass.config.language
user_id: str = entry.data[USER_ID]
token: str = entry.data[CONF_TOKEN]
hass.data.setdefault(DOMAIN, {})
websession = async_get_clientsession(hass)
websession: ClientSession = async_get_clientsession(hass)

coordinator = TechCoordinator(hass, websession, user_id, token)
coordinator: DataUpdateCoordinator[dict[str, dict[str, Any]]] = TechCoordinator(
hass, websession, user_id, token
)
hass.data[DOMAIN][entry.entry_id] = coordinator

await coordinator.async_config_entry_first_refresh()

await assets.load_subtitles(language_code, Tech(websession, user_id, token))
await assets.load_subtitles(Tech(websession, user_id, token), language_code)

hass.async_create_task(
hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
Expand All @@ -72,27 +80,28 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
return True


async def async_migrate_entry(hass: HomeAssistant, config_entry: ConfigEntry):
async def async_migrate_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
"""Migrate old entry."""
_LOGGER.info("Migrating from version %s", config_entry.version)

device_registry = dr.async_get(hass)
entity_registry = er.async_get(hass)
udid = config_entry.data[UDID]
device_registry: DeviceRegistry = dr.async_get(hass)
entity_registry: EntityRegistry = er.async_get(hass)
udid: str = config_entry.data[UDID]

if config_entry.version == 1:
version = 2
version: int = 2

http_session = async_get_clientsession(hass)
api = Tech(
http_session: ClientSession = async_get_clientsession(hass)
api: Tech = Tech(
http_session, config_entry.data[USER_ID], config_entry.data[CONF_TOKEN]
)
controllers = await api.list_modules()
controller = next(obj for obj in controllers if obj.get(UDID) == udid)
controllers: list[dict[str, Any]] = await api.list_modules()
controller: dict[str, Any] = next(
obj for obj in controllers if obj.get(UDID) == udid
)
api.modules.setdefault(udid, {"last_update": None, "zones": {}, "tiles": {}})
zones = await api.get_module_zones(udid)
zones: dict[str, Any] = await api.get_module_zones(udid)

data = {
data: dict[str, Any] = {
USER_ID: api.user_id,
CONF_TOKEN: api.token,
CONTROLLER: controller,
Expand All @@ -118,7 +127,7 @@ async def async_migrate_entry(hass: HomeAssistant, config_entry: ConfigEntry):

# Create new devices as version 1 did not have any:
for z in zones:
zone = zones[z]
zone: dict[str, Any] = zones[z]
device_registry.async_get_or_create(
config_entry_id=config_entry.entry_id,
identifiers={(DOMAIN, zone[CONF_DESCRIPTION][CONF_NAME])},
Expand All @@ -139,16 +148,16 @@ async def async_migrate_entry(hass: HomeAssistant, config_entry: ConfigEntry):
# Update all entities and link them to appropriate devices
# plus update unique_id, everything else as it was
for unique_id, old_entity_entry in old_entity_entries.items():
if old_entity_entry.original_name != "":
device = device_registry.async_get_device(
if old_entity_entry.original_name is not None:
device: DeviceEntry | None = device_registry.async_get_device(
{(DOMAIN, old_entity_entry.original_name)},
set(),
)
if device and device.name == old_entity_entry.original_name:
# since thsi entity stays, remove it from collection to remove
del remaining_entities_to_remove[unique_id]
entity_registry.async_update_entity(
old_entity_entry.entity_id,
entity_id=old_entity_entry.entity_id,
area_id=old_entity_entry.area_id,
device_class=old_entity_entry.device_class,
device_id=device.id,
Expand All @@ -166,16 +175,21 @@ async def async_migrate_entry(hass: HomeAssistant, config_entry: ConfigEntry):

for entity_to_remove in remaining_entities_to_remove:
entity_registry.async_remove(
remaining_entities_to_remove[entity_to_remove].entity_id
entity_id=remaining_entities_to_remove[entity_to_remove].entity_id
)

_LOGGER.info("Migration to version %s successful", version)

return True


async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry):
"""Unload a config entry."""
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry.
Returns:
True if successful, False if cannot unload.
"""
unload_ok = all(
await asyncio.gather(
*[
Expand All @@ -190,13 +204,17 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry):
return unload_ok


class TechCoordinator(DataUpdateCoordinator):
class TechCoordinator(DataUpdateCoordinator[dict[str, dict[str, Any]]]):
"""TECH API data update coordinator."""

config_entry: ConfigEntry

def __init__(
self, hass: HomeAssistant, session: ClientSession, user_id: str, token: str
self,
hass: HomeAssistant,
session: ClientSession,
user_id: str,
token: str,
) -> None:
"""Initialize my coordinator."""
super().__init__(
Expand All @@ -209,7 +227,9 @@ def __init__(
)
self.api = Tech(session, user_id, token)

async def _async_update_data(self):
async def _async_update_data(
self,
):
"""Fetch data from TECH API endpoint(s)."""

_LOGGER.debug(
Expand Down
66 changes: 51 additions & 15 deletions custom_components/tech/assets.py
Original file line number Diff line number Diff line change
@@ -1,20 +1,24 @@
"""Assets for translations."""
import logging
from typing import Any

from .const import DEFAULT_ICON, ICON_BY_ID, ICON_BY_TYPE, TXT_ID_BY_TYPE
from .tech import Tech

logging.basicConfig(level=logging.DEBUG)
_LOGGER = logging.getLogger(__name__)
_LOGGER: logging.Logger = logging.getLogger(__name__)

TRANSLATIONS = None
TRANSLATIONS: dict[str, Any] = {}


async def load_subtitles(language, api):
async def load_subtitles(
api: Tech,
language: str = "pl",
) -> None:
"""Load subtitles for the specified language.
Args:
language (str): The language code for the subtitles. Defaults to "pl".
api: object to use Tech api
api (Tech, required): object to use Tech api
language (str, optional): The language code for the subtitles. Defaults to "pl".
Returns:
None
Expand All @@ -24,25 +28,57 @@ async def load_subtitles(language, api):
TRANSLATIONS = await api.get_translations(language)


def get_text(text_id) -> str:
"""Get text by id."""
def get_text(text_id: int) -> str:
"""Get text by id.
Args:
text_id (int): The ID of the text to retrieve.
Returns:
str: The text corresponding to the ID, or a default string if not found.
"""
if text_id != 0:
return TRANSLATIONS["data"].get(str(text_id), f"txtId {text_id}")
else:
return f"txtId {text_id}"


def get_text_by_type(text_type) -> str:
"""Get text by type."""
text_id = TXT_ID_BY_TYPE.get(text_type, f"type {text_type}")
def get_text_by_type(text_type: int) -> str:
"""Get text by type.
Args:
text_type (int): The type of the text to retrieve.
Returns:
str: The text corresponding to the type, or a default string if not found.
"""
text_id: int = TXT_ID_BY_TYPE.get(text_type, text_type)
return get_text(text_id)


def get_icon(icon_id) -> str:
"""Get icon by id."""
def get_icon(icon_id: int) -> str:
"""Get icon by id.
Args:
icon_id (int): The ID of the icon to retrieve.
Returns:
str: The icon corresponding to the ID, or a default string if not found.
"""
return ICON_BY_ID.get(icon_id, DEFAULT_ICON)


def get_icon_by_type(icon_type) -> str:
"""Get icon by type."""
def get_icon_by_type(icon_type: int) -> str:
"""Get icon by type.
Args:
icon_type (int): The type of the icon to retrieve.
Returns:
str: The icon corresponding to the type, or a default string if not found.
"""
return ICON_BY_TYPE.get(icon_type, DEFAULT_ICON)
Loading

0 comments on commit c6f8a21

Please sign in to comment.