Improve acmeda typing (#106812)

This commit is contained in:
Marc Mueller 2024-01-01 22:16:22 +01:00 committed by GitHub
parent 73ccd0d310
commit aec8dc13b2
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 39 additions and 23 deletions

View File

@ -42,6 +42,7 @@ homeassistant.components
homeassistant.components.abode.*
homeassistant.components.accuweather.*
homeassistant.components.acer_projector.*
homeassistant.components.acmeda.*
homeassistant.components.actiontec.*
homeassistant.components.adax.*
homeassistant.components.adguard.*

View File

@ -66,12 +66,12 @@ class AcmedaBase(entity.Entity):
@property
def unique_id(self) -> str:
"""Return the unique ID of this roller."""
return self.roller.id
return self.roller.id # type: ignore[no-any-return]
@property
def device_id(self) -> str:
"""Return the ID of this roller."""
return self.roller.id
return self.roller.id # type: ignore[no-any-return]
@property
def device_info(self) -> dr.DeviceInfo:

View File

@ -30,7 +30,7 @@ async def async_setup_entry(
current: set[int] = set()
@callback
def async_add_acmeda_covers():
def async_add_acmeda_covers() -> None:
async_add_acmeda_entities(
hass, AcmedaCover, config_entry, current, async_add_entities
)
@ -95,7 +95,7 @@ class AcmedaCover(AcmedaBase, CoverEntity):
@property
def is_closed(self) -> bool:
"""Return if the cover is closed."""
return self.roller.closed_percent == 100
return self.roller.closed_percent == 100 # type: ignore[no-any-return]
async def async_close_cover(self, **kwargs: Any) -> None:
"""Close the roller."""

View File

@ -1,6 +1,8 @@
"""Helper functions for Acmeda Pulse."""
from __future__ import annotations
from aiopulse import Roller
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import device_registry as dr
@ -16,7 +18,7 @@ def async_add_acmeda_entities(
config_entry: ConfigEntry,
current: set[int],
async_add_entities: AddEntitiesCallback,
):
) -> None:
"""Add any new entities."""
hub = hass.data[DOMAIN][config_entry.entry_id]
LOGGER.debug("Looking for new %s on: %s", entity_class.__name__, hub.host)
@ -34,7 +36,9 @@ def async_add_acmeda_entities(
async_add_entities(new_items)
async def update_devices(hass: HomeAssistant, config_entry: ConfigEntry, api):
async def update_devices(
hass: HomeAssistant, config_entry: ConfigEntry, api: dict[int, Roller]
) -> None:
"""Tell hass that device info has been updated."""
dev_registry = dr.async_get(hass)

View File

@ -2,9 +2,12 @@
from __future__ import annotations
import asyncio
from collections.abc import Callable
import aiopulse
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.dispatcher import async_dispatcher_send
from .const import ACMEDA_ENTITY_REMOVE, ACMEDA_HUB_UPDATE, LOGGER
@ -14,31 +17,29 @@ from .helpers import update_devices
class PulseHub:
"""Manages a single Pulse Hub."""
def __init__(self, hass, config_entry):
api: aiopulse.Hub
def __init__(self, hass: HomeAssistant, config_entry: ConfigEntry) -> None:
"""Initialize the system."""
self.config_entry = config_entry
self.hass = hass
self.api: aiopulse.Hub | None = None
self.tasks = []
self.current_rollers = {}
self.cleanup_callbacks = []
self.tasks: list[asyncio.Task[None]] = []
self.current_rollers: dict[int, aiopulse.Roller] = {}
self.cleanup_callbacks: list[Callable[[], None]] = []
@property
def title(self):
def title(self) -> str:
"""Return the title of the hub shown in the integrations list."""
return f"{self.api.id} ({self.api.host})"
@property
def host(self):
def host(self) -> str:
"""Return the host of this hub."""
return self.config_entry.data["host"]
return self.config_entry.data["host"] # type: ignore[no-any-return]
async def async_setup(self, tries=0):
async def async_setup(self, tries: int = 0) -> bool:
"""Set up a hub based on host parameter."""
host = self.host
hub = aiopulse.Hub(host)
self.api = hub
self.api = hub = aiopulse.Hub(self.host)
hub.callback_subscribe(self.async_notify_update)
self.tasks.append(asyncio.create_task(hub.run()))
@ -46,7 +47,7 @@ class PulseHub:
LOGGER.debug("Hub setup complete")
return True
async def async_reset(self):
async def async_reset(self) -> bool:
"""Reset this hub to default state."""
for cleanup_callback in self.cleanup_callbacks:
@ -66,7 +67,7 @@ class PulseHub:
return True
async def async_notify_update(self, update_type):
async def async_notify_update(self, update_type: aiopulse.UpdateType) -> None:
"""Evaluate entities when hub reports that update has occurred."""
LOGGER.debug("Hub {update_type.name} updated")

View File

@ -25,7 +25,7 @@ async def async_setup_entry(
current: set[int] = set()
@callback
def async_add_acmeda_sensors():
def async_add_acmeda_sensors() -> None:
async_add_acmeda_entities(
hass, AcmedaBattery, config_entry, current, async_add_entities
)
@ -48,4 +48,4 @@ class AcmedaBattery(AcmedaBase, SensorEntity):
@property
def native_value(self) -> float | int | None:
"""Return the state of the device."""
return self.roller.battery
return self.roller.battery # type: ignore[no-any-return]

View File

@ -180,6 +180,16 @@ disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.acmeda.*]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.actiontec.*]
check_untyped_defs = true
disallow_incomplete_defs = true