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

Tibber strict typing (#79407)

Browse source 
Co-authored-by: Franck Nijhof <git@frenck.dev>
This commit is contained in:
Daniel Hjelseth Høyer 2022-10-26 12:41:23 +02:00 • committed by GitHub
parent d65e639f00
commit e3233f72ce
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
9 changed files with 77 additions and 37 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.strict-typing
View file

@ -260,6 +260,7 @@ homeassistant.components.tag.*
homeassistant.components.tailscale.*
homeassistant.components.tautulli.*
homeassistant.components.tcp.*
homeassistant.components.tibber.*
homeassistant.components.tile.*
homeassistant.components.tilt_ble.*
homeassistant.components.tolo.*

4
homeassistant/components/tibber/__init__.py
View file

@ -12,7 +12,7 @@ from homeassistant.const import (
EVENT_HOMEASSISTANT_STOP,
Platform,
)
from homeassistant.core import HomeAssistant
from homeassistant.core import Event, HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import discovery
from homeassistant.helpers.aiohttp_client import async_get_clientsession
@ -46,7 +46,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
)
hass.data[DOMAIN] = tibber_connection
async def _close(event):
async def _close(event: Event) -> None:
await tibber_connection.rt_disconnect()
entry.async_on_unload(hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, _close))

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

@ -1,5 +1,8 @@
"""Adds config flow for Tibber integration."""
from __future__ import annotations
import asyncio
from typing import Any
import aiohttp
import tibber
@ -7,6 +10,7 @@ import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_ACCESS_TOKEN
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import DOMAIN
@ -19,7 +23,9 @@ class TibberConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
VERSION = 1
async def async_step_user(self, user_input=None):
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the initial step."""
self._async_abort_entries_match()

2
homeassistant/components/tibber/manifest.json
View file

@ -3,7 +3,7 @@
"domain": "tibber",
"name": "Tibber",
"documentation": "https://www.home-assistant.io/integrations/tibber",
"requirements": ["pyTibber==0.25.4"],
"requirements": ["pyTibber==0.25.6"],
"codeowners": ["@danielhiversen"],
"quality_scale": "silver",
"config_flow": true,

16
homeassistant/components/tibber/notify.py
View file

@ -1,19 +1,29 @@
"""Support for Tibber notifications."""
from __future__ import annotations
import asyncio
from collections.abc import Callable
import logging
from typing import Any
from homeassistant.components.notify import (
ATTR_TITLE,
ATTR_TITLE_DEFAULT,
BaseNotificationService,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from . import DOMAIN as TIBBER_DOMAIN
_LOGGER = logging.getLogger(__name__)
async def async_get_service(hass, config, discovery_info=None):
async def async_get_service(
hass: HomeAssistant,
config: ConfigType,
discovery_info: DiscoveryInfoType | None = None,
) -> TibberNotificationService:
"""Get the Tibber notification service."""
tibber_connection = hass.data[TIBBER_DOMAIN]
return TibberNotificationService(tibber_connection.send_notification)
@ -22,11 +32,11 @@ async def async_get_service(hass, config, discovery_info=None):
class TibberNotificationService(BaseNotificationService):
"""Implement the notification service for Tibber."""
def __init__(self, notify):
def __init__(self, notify: Callable) -> None:
"""Initialize the service."""
self._notify = notify
async def async_send_message(self, message=None, **kwargs):
async def async_send_message(self, message: str = "", **kwargs: Any) -> None:
"""Send a message to Tibber devices."""
title = kwargs.get(ATTR_TITLE, ATTR_TITLE_DEFAULT)
try:

69
homeassistant/components/tibber/sensor.py
View file

@ -2,11 +2,14 @@
from __future__ import annotations
import asyncio
import datetime
from datetime import timedelta
import logging
from random import randrange
from typing import Any
import aiohttp
import tibber
from homeassistant.components.recorder import get_instance
from homeassistant.components.recorder.models import StatisticData, StatisticMetaData
@ -31,7 +34,7 @@ from homeassistant.const import (
POWER_WATT,
SIGNAL_STRENGTH_DECIBELS,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.core import Event, HomeAssistant, callback
from homeassistant.exceptions import PlatformNotReady
from homeassistant.helpers.device_registry import async_get as async_get_dev_reg
from homeassistant.helpers.entity import DeviceInfo, EntityCategory
@ -296,7 +299,9 @@ async def async_setup_entry(
class TibberSensor(SensorEntity):
"""Representation of a generic Tibber sensor."""
def __init__(self, *args, tibber_home, **kwargs):
def __init__(
self, *args: Any, tibber_home: tibber.TibberHome, **kwargs: Any
) -> None:
"""Initialize the sensor."""
super().__init__(*args, **kwargs)
self._tibber_home = tibber_home
@ -305,11 +310,11 @@ class TibberSensor(SensorEntity):
self._home_name = tibber_home.info["viewer"]["home"]["address"].get(
"address1", ""
)
self._device_name = None
self._model = None
self._device_name: None | str = None
self._model: None | str = None
@property
def device_info(self):
def device_info(self) -> DeviceInfo:
"""Return the device_info of the device."""
device_info = DeviceInfo(
identifiers={(TIBBER_DOMAIN, self._tibber_home.home_id)},
@ -324,10 +329,10 @@ class TibberSensor(SensorEntity):
class TibberSensorElPrice(TibberSensor):
"""Representation of a Tibber sensor for el price."""
def __init__(self, tibber_home):
def __init__(self, tibber_home: tibber.TibberHome) -> None:
"""Initialize the sensor."""
super().__init__(tibber_home=tibber_home)
self._last_updated = None
self._last_updated: datetime.datetime | None = None
self._spread_load_constant = randrange(5000)
self._attr_available = False
@ -380,7 +385,7 @@ class TibberSensorElPrice(TibberSensor):
self._attr_native_unit_of_measurement = self._tibber_home.price_unit
@Throttle(MIN_TIME_BETWEEN_UPDATES)
async def _fetch_data(self):
async def _fetch_data(self) -> None:
_LOGGER.debug("Fetching data")
try:
await self._tibber_home.update_info_and_price_info()
@ -401,10 +406,10 @@ class TibberDataSensor(TibberSensor, CoordinatorEntity["TibberDataCoordinator"])
def __init__(
self,
tibber_home,
tibber_home: tibber.TibberHome,
coordinator: TibberDataCoordinator,
entity_description: SensorEntityDescription,
):
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator=coordinator, tibber_home=tibber_home)
self.entity_description = entity_description
@ -419,7 +424,7 @@ class TibberDataSensor(TibberSensor, CoordinatorEntity["TibberDataCoordinator"])
self._device_name = self._home_name
@property
def native_value(self):
def native_value(self) -> Any:
"""Return the value of the sensor."""
return getattr(self._tibber_home, self.entity_description.key)
@ -429,11 +434,11 @@ class TibberSensorRT(TibberSensor, CoordinatorEntity["TibberRtDataCoordinator"])
def __init__(
self,
tibber_home,
tibber_home: tibber.TibberHome,
description: SensorEntityDescription,
initial_state,
initial_state: float,
coordinator: TibberRtDataCoordinator,
):
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator=coordinator, tibber_home=tibber_home)
self.entity_description = description
@ -486,12 +491,17 @@ class TibberSensorRT(TibberSensor, CoordinatorEntity["TibberRtDataCoordinator"])
class TibberRtDataCoordinator(DataUpdateCoordinator):
"""Handle Tibber realtime data."""
def __init__(self, async_add_entities, tibber_home, hass):
def __init__(
self,
async_add_entities: AddEntitiesCallback,
tibber_home: tibber.TibberHome,
hass: HomeAssistant,
) -> None:
"""Initialize the data handler."""
self._async_add_entities = async_add_entities
self._tibber_home = tibber_home
self.hass = hass
self._added_sensors = set()
self._added_sensors: set[str] = set()
super().__init__(
hass,
_LOGGER,
@ -506,12 +516,12 @@ class TibberRtDataCoordinator(DataUpdateCoordinator):
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, self._handle_ha_stop)
@callback
def _handle_ha_stop(self, _event) -> None:
def _handle_ha_stop(self, _event: Event) -> None:
"""Handle Home Assistant stopping."""
self._async_remove_device_updates_handler()
@callback
def _add_sensors(self):
def _add_sensors(self) -> None:
"""Add sensor."""
if not (live_measurement := self.get_live_measurement()):
return
@ -534,7 +544,7 @@ class TibberRtDataCoordinator(DataUpdateCoordinator):
if new_entities:
self._async_add_entities(new_entities)
def get_live_measurement(self):
def get_live_measurement(self) -> Any:
"""Get live measurement data."""
if errors := self.data.get("errors"):
_LOGGER.error(errors[0])
@ -545,7 +555,7 @@ class TibberRtDataCoordinator(DataUpdateCoordinator):
class TibberDataCoordinator(DataUpdateCoordinator):
"""Handle Tibber data and insert statistics."""
def __init__(self, hass, tibber_connection):
def __init__(self, hass: HomeAssistant, tibber_connection: tibber.Tibber) -> None:
"""Initialize the data handler."""
super().__init__(
hass,
@ -555,13 +565,13 @@ class TibberDataCoordinator(DataUpdateCoordinator):
)
self._tibber_connection = tibber_connection
async def _async_update_data(self):
async def _async_update_data(self) -> None:
"""Update data via API."""
await self._tibber_connection.fetch_consumption_data_active_homes()
await self._tibber_connection.fetch_production_data_active_homes()
await self._insert_statistics()
async def _insert_statistics(self):
async def _insert_statistics(self) -> None:
"""Insert Tibber statistics."""
for home in self._tibber_connection.get_homes():
sensors = []
@ -602,9 +612,10 @@ class TibberDataCoordinator(DataUpdateCoordinator):
else home.hourly_consumption_data
)
start = dt_util.parse_datetime(hourly_data[0]["from"]) - timedelta(
hours=1
)
from_time = dt_util.parse_datetime(hourly_data[0]["from"])
if from_time is None:
continue
start = from_time - timedelta(hours=1)
stat = await get_instance(self.hass).async_add_executor_job(
statistics_during_period,
self.hass,
@ -623,15 +634,17 @@ class TibberDataCoordinator(DataUpdateCoordinator):
if data.get(sensor_type) is None:
continue
start = dt_util.parse_datetime(data["from"])
if last_stats_time is not None and start <= last_stats_time:
from_time = dt_util.parse_datetime(data["from"])
if from_time is None or (
last_stats_time is not None and from_time <= last_stats_time
):
continue
_sum += data[sensor_type]
statistics.append(
StatisticData(
start=start,
start=from_time,
state=data[sensor_type],
sum=_sum,
)

10
mypy.ini
View file

@ -2353,6 +2353,16 @@ disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.tibber.*]
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.tile.*]
check_untyped_defs = true
disallow_incomplete_defs = true

2
requirements_all.txt
View file

@ -1409,7 +1409,7 @@ pyRFXtrx==0.30.0
pySwitchmate==0.5.1
# homeassistant.components.tibber
pyTibber==0.25.4
pyTibber==0.25.6
# homeassistant.components.dlink
pyW215==0.7.0

2
requirements_test_all.txt
View file

@ -1009,7 +1009,7 @@ pyMetno==0.9.0
pyRFXtrx==0.30.0
# homeassistant.components.tibber
pyTibber==0.25.4
pyTibber==0.25.6
# homeassistant.components.nextbus
py_nextbusnext==0.1.5