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

Add type annotations for Netatmo (#52811)

Browse source
This commit is contained in:
Tobias Sauerwein 2021-07-21 23:36:57 +02:00 • committed by GitHub
parent 84c482441d
commit 583deada83
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
21 changed files with 288 additions and 177 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.strict-typing
View file

@ -60,6 +60,7 @@ homeassistant.components.mailbox.*
homeassistant.components.media_player.*
homeassistant.components.mysensors.*
homeassistant.components.nam.*
homeassistant.components.netatmo.*
homeassistant.components.network.*
homeassistant.components.no_ip.*
homeassistant.components.notify.*

14
homeassistant/components/netatmo/__init__.py
View file

@ -1,4 +1,6 @@
"""The Netatmo integration."""
from __future__ import annotations
import logging
import secrets
@ -67,7 +69,7 @@ CONFIG_SCHEMA = vol.Schema(
)
async def async_setup(hass: HomeAssistant, config: dict):
async def async_setup(hass: HomeAssistant, config: dict) -> bool:
"""Set up the Netatmo component."""
hass.data[DOMAIN] = {
DATA_PERSONS: {},
@ -121,7 +123,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
async def unregister_webhook(_):
async def unregister_webhook(_: None) -> None:
if CONF_WEBHOOK_ID not in entry.data:
return
_LOGGER.debug("Unregister Netatmo webhook (%s)", entry.data[CONF_WEBHOOK_ID])
@ -138,7 +140,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"No webhook to be dropped for %s", entry.data[CONF_WEBHOOK_ID]
)
async def register_webhook(event):
async def register_webhook(_: None) -> None:
if CONF_WEBHOOK_ID not in entry.data:
data = {**entry.data, CONF_WEBHOOK_ID: secrets.token_hex()}
hass.config_entries.async_update_entry(entry, data=data)
@ -175,7 +177,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async_handle_webhook,
)
async def handle_event(event):
async def handle_event(event: dict) -> None:
"""Handle webhook events."""
if event["data"][WEBHOOK_PUSH_TYPE] == WEBHOOK_ACTIVATION:
if activation_listener is not None:
@ -219,7 +221,7 @@ async def async_config_entry_updated(hass: HomeAssistant, entry: ConfigEntry) ->
async_dispatcher_send(hass, f"signal-{DOMAIN}-public-update-{entry.entry_id}")
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry):
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
if CONF_WEBHOOK_ID in entry.data:
webhook_unregister(hass, entry.data[CONF_WEBHOOK_ID])
@ -236,7 +238,7 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry):
return unload_ok
async def async_remove_entry(hass: HomeAssistant, entry: ConfigEntry):
async def async_remove_entry(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Cleanup when entry is removed."""
if (
CONF_WEBHOOK_ID in entry.data

6
homeassistant/components/netatmo/api.py
View file

@ -1,4 +1,6 @@
"""API for Netatmo bound to HASS OAuth."""
from typing import cast
from aiohttp import ClientSession
import pyatmo
@ -17,8 +19,8 @@ class AsyncConfigEntryNetatmoAuth(pyatmo.auth.AbstractAsyncAuth):
super().__init__(websession)
self._oauth_session = oauth_session
async def async_get_access_token(self):
async def async_get_access_token(self) -> str:
"""Return a valid access token for Netatmo API."""
if not self._oauth_session.valid_token:
await self._oauth_session.async_ensure_token_valid()
return self._oauth_session.token["access_token"]
return cast(str, self._oauth_session.token["access_token"])

88
homeassistant/components/netatmo/camera.py
View file

@ -1,15 +1,20 @@
"""Support for the Netatmo cameras."""
from __future__ import annotations
import logging
from typing import Any, cast
import aiohttp
import pyatmo
import voluptuous as vol
from homeassistant.components.camera import SUPPORT_STREAM, Camera
from homeassistant.core import callback
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import PlatformNotReady
from homeassistant.helpers import config_validation as cv, entity_platform
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import (
ATTR_CAMERA_LIGHT_MODE,
@ -31,11 +36,12 @@ from .const import (
SERVICE_SET_PERSON_AWAY,
SERVICE_SET_PERSONS_HOME,
SIGNAL_NAME,
UNKNOWN,
WEBHOOK_LIGHT_MODE,
WEBHOOK_NACAMERA_CONNECTION,
WEBHOOK_PUSH_TYPE,
)
from .data_handler import CAMERA_DATA_CLASS_NAME
from .data_handler import CAMERA_DATA_CLASS_NAME, NetatmoDataHandler
from .netatmo_entity_base import NetatmoBase
_LOGGER = logging.getLogger(__name__)
@ -43,7 +49,9 @@ _LOGGER = logging.getLogger(__name__)
DEFAULT_QUALITY = "high"
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 the Netatmo camera platform."""
if "access_camera" not in entry.data["token"]["scope"]:
_LOGGER.info(
@ -108,12 +116,12 @@ class NetatmoCamera(NetatmoBase, Camera):
def __init__(
self,
data_handler,
camera_id,
camera_type,
home_id,
quality,
):
data_handler: NetatmoDataHandler,
camera_id: str,
camera_type: str,
home_id: str,
quality: str,
) -> None:
"""Set up for access to the Netatmo camera images."""
Camera.__init__(self)
super().__init__(data_handler)
@ -124,17 +132,19 @@ class NetatmoCamera(NetatmoBase, Camera):
self._id = camera_id
self._home_id = home_id
self._device_name = self._data.get_camera(camera_id=camera_id).get("name")
self._device_name = self._data.get_camera(camera_id=camera_id).get(
"name", UNKNOWN
)
self._attr_name = f"{MANUFACTURER} {self._device_name}"
self._model = camera_type
self._attr_unique_id = f"{self._id}-{self._model}"
self._quality = quality
self._vpnurl = None
self._localurl = None
self._status = None
self._sd_status = None
self._alim_status = None
self._is_local = None
self._vpnurl: str | None = None
self._localurl: str | None = None
self._status: str | None = None
self._sd_status: str | None = None
self._alim_status: str | None = None
self._is_local: str | None = None
self._light_state = None
async def async_added_to_hass(self) -> None:
@ -153,7 +163,7 @@ class NetatmoCamera(NetatmoBase, Camera):
self.hass.data[DOMAIN][DATA_CAMERAS][self._id] = self._device_name
@callback
def handle_event(self, event):
def handle_event(self, event: dict) -> None:
"""Handle webhook events."""
data = event["data"]
@ -179,7 +189,15 @@ class NetatmoCamera(NetatmoBase, Camera):
self.async_write_ha_state()
return
async def async_camera_image(self):
@property
def _data(self) -> pyatmo.AsyncCameraData:
"""Return data for this entity."""
return cast(
pyatmo.AsyncCameraData,
self.data_handler.data[self._data_classes[0]["name"]],
)
async def async_camera_image(self) -> bytes | None:
"""Return a still image response from the camera."""
try:
return await self._data.async_get_live_snapshot(camera_id=self._id)
@ -194,43 +212,43 @@ class NetatmoCamera(NetatmoBase, Camera):
return None
@property
def available(self):
def available(self) -> bool:
"""Return True if entity is available."""
return bool(self._alim_status == "on" or self._status == "disconnected")
@property
def supported_features(self):
def supported_features(self) -> int:
"""Return supported features."""
return SUPPORT_STREAM
@property
def brand(self):
def brand(self) -> str:
"""Return the camera brand."""
return MANUFACTURER
@property
def motion_detection_enabled(self):
def motion_detection_enabled(self) -> bool:
"""Return the camera motion detection status."""
return bool(self._status == "on")
@property
def is_on(self):
def is_on(self) -> bool:
"""Return true if on."""
return self.is_streaming
async def async_turn_off(self):
async def async_turn_off(self) -> None:
"""Turn off camera."""
await self._data.async_set_state(
home_id=self._home_id, camera_id=self._id, monitoring="off"
)
async def async_turn_on(self):
async def async_turn_on(self) -> None:
"""Turn on camera."""
await self._data.async_set_state(
home_id=self._home_id, camera_id=self._id, monitoring="on"
)
async def stream_source(self):
async def stream_source(self) -> str:
"""Return the stream source."""
url = "{0}/live/files/{1}/index.m3u8"
if self._localurl:
@ -238,12 +256,12 @@ class NetatmoCamera(NetatmoBase, Camera):
return url.format(self._vpnurl, self._quality)
@property
def model(self):
def model(self) -> str:
"""Return the camera model."""
return MODELS[self._model]
@callback
def async_update_callback(self):
def async_update_callback(self) -> None:
"""Update the entity's state."""
camera = self._data.get_camera(self._id)
self._vpnurl, self._localurl = self._data.camera_urls(self._id)
@ -275,7 +293,7 @@ class NetatmoCamera(NetatmoBase, Camera):
}
)
def process_events(self, events):
def process_events(self, events: dict) -> dict:
"""Add meta data to events."""
for event in events.values():
if "video_id" not in event:
@ -290,9 +308,9 @@ class NetatmoCamera(NetatmoBase, Camera):
] = f"{self._vpnurl}/vod/{event['video_id']}/files/{self._quality}/index.m3u8"
return events
async def _service_set_persons_home(self, **kwargs):
async def _service_set_persons_home(self, **kwargs: Any) -> None:
"""Service to change current home schedule."""
persons = kwargs.get(ATTR_PERSONS)
persons = kwargs.get(ATTR_PERSONS, {})
person_ids = []
for person in persons:
for pid, data in self._data.persons.items():
@ -304,7 +322,7 @@ class NetatmoCamera(NetatmoBase, Camera):
)
_LOGGER.debug("Set %s as at home", persons)
async def _service_set_person_away(self, **kwargs):
async def _service_set_person_away(self, **kwargs: Any) -> None:
"""Service to mark a person as away or set the home as empty."""
person = kwargs.get(ATTR_PERSON)
person_id = None
@ -327,10 +345,10 @@ class NetatmoCamera(NetatmoBase, Camera):
)
_LOGGER.debug("Set home as empty")
async def _service_set_camera_light(self, **kwargs):
async def _service_set_camera_light(self, **kwargs: Any) -> None:
"""Service to set light mode."""
mode = kwargs.get(ATTR_CAMERA_LIGHT_MODE)
_LOGGER.debug("Turn %s camera light for '%s'", mode, self.name)
mode = str(kwargs.get(ATTR_CAMERA_LIGHT_MODE))
_LOGGER.debug("Turn %s camera light for '%s'", mode, self._attr_name)
await self._data.async_set_state(
home_id=self._home_id,
camera_id=self._id,

90
homeassistant/components/netatmo/climate.py
View file

@ -2,6 +2,7 @@
from __future__ import annotations
import logging
from typing import cast
import pyatmo
import voluptuous as vol
@ -19,6 +20,7 @@ from homeassistant.components.climate.const import (
SUPPORT_PRESET_MODE,
SUPPORT_TARGET_TEMPERATURE,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
ATTR_BATTERY_LEVEL,
ATTR_TEMPERATURE,
@ -26,11 +28,13 @@ from homeassistant.const import (
STATE_OFF,
TEMP_CELSIUS,
)
from homeassistant.core import callback
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import PlatformNotReady
from homeassistant.helpers import config_validation as cv, entity_platform
from homeassistant.helpers.device_registry import async_get_registry
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import (
ATTR_HEATING_POWER_REQUEST,
@ -49,7 +53,11 @@ from .const import (
SERVICE_SET_SCHEDULE,
SIGNAL_NAME,
)
from .data_handler import HOMEDATA_DATA_CLASS_NAME, HOMESTATUS_DATA_CLASS_NAME
from .data_handler import (
HOMEDATA_DATA_CLASS_NAME,
HOMESTATUS_DATA_CLASS_NAME,
NetatmoDataHandler,
)
from .netatmo_entity_base import NetatmoBase
_LOGGER = logging.getLogger(__name__)
@ -106,8 +114,12 @@ DEFAULT_MAX_TEMP = 30
NA_THERM = "NATherm1"
NA_VALVE = "NRV"
SUGGESTED_AREA = "suggested_area"
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 the Netatmo energy platform."""
data_handler = hass.data[DOMAIN][entry.entry_id][DATA_HANDLER]
@ -163,7 +175,9 @@ async def async_setup_entry(hass, entry, async_add_entities):
class NetatmoThermostat(NetatmoBase, ClimateEntity):
"""Representation a Netatmo thermostat."""
def __init__(self, data_handler, home_id, room_id):
def __init__(
self, data_handler: NetatmoDataHandler, home_id: str, room_id: str
) -> None:
"""Initialize the sensor."""
ClimateEntity.__init__(self)
super().__init__(data_handler)
@ -189,29 +203,29 @@ class NetatmoThermostat(NetatmoBase, ClimateEntity):
self._home_status = self.data_handler.data[self._home_status_class]
self._room_status = self._home_status.rooms[room_id]
self._room_data = self._data.rooms[home_id][room_id]
self._room_data: dict = self._data.rooms[home_id][room_id]
self._model = NA_VALVE
for module in self._room_data.get("module_ids"):
self._model: str = NA_VALVE
for module in self._room_data.get("module_ids", []):
if self._home_status.thermostats.get(module):
self._model = NA_THERM
break
self._device_name = self._data.rooms[home_id][room_id]["name"]
self._attr_name = f"{MANUFACTURER} {self._device_name}"
self._current_temperature = None
self._target_temperature = None
self._preset = None
self._away = None
self._current_temperature: float | None = None
self._target_temperature: float | None = None
self._preset: str | None = None
self._away: bool | None = None
self._operation_list = [HVAC_MODE_AUTO, HVAC_MODE_HEAT]
self._support_flags = SUPPORT_FLAGS
self._hvac_mode = None
self._hvac_mode: str = HVAC_MODE_AUTO
self._battery_level = None
self._connected = None
self._connected: bool | None = None
self._away_temperature = None
self._hg_temperature = None
self._boilerstatus = None
self._away_temperature: float | None = None
self._hg_temperature: float | None = None
self._boilerstatus: bool | None = None
self._setpoint_duration = None
self._selected_schedule = None
@ -240,9 +254,10 @@ class NetatmoThermostat(NetatmoBase, ClimateEntity):
registry = await async_get_registry(self.hass)
device = registry.async_get_device({(DOMAIN, self._id)}, set())
assert device
self.hass.data[DOMAIN][DATA_DEVICE_IDS][self._home_id] = device.id
async def handle_event(self, event):
async def handle_event(self, event: dict) -> None:
"""Handle webhook events."""
data = event["data"]
@ -307,22 +322,29 @@ class NetatmoThermostat(NetatmoBase, ClimateEntity):
return
@property
def supported_features(self):
def _data(self) -> pyatmo.AsyncHomeData:
"""Return data for this entity."""
return cast(
pyatmo.AsyncHomeData, self.data_handler.data[self._data_classes[0]["name"]]
)
@property
def supported_features(self) -> int:
"""Return the list of supported features."""
return self._support_flags
@property
def temperature_unit(self):
def temperature_unit(self) -> str:
"""Return the unit of measurement."""
return TEMP_CELSIUS
@property
def current_temperature(self):
def current_temperature(self) -> float | None:
"""Return the current temperature."""
return self._current_temperature
@property
def target_temperature(self):
def target_temperature(self) -> float | None:
"""Return the temperature we try to reach."""
return self._target_temperature
@ -332,12 +354,12 @@ class NetatmoThermostat(NetatmoBase, ClimateEntity):
return PRECISION_HALVES
@property
def hvac_mode(self):
def hvac_mode(self) -> str:
"""Return hvac operation ie. heat, cool mode."""
return self._hvac_mode
@property
def hvac_modes(self):
def hvac_modes(self) -> list[str]:
"""Return the list of available hvac operation modes."""
return self._operation_list
@ -418,7 +440,7 @@ class NetatmoThermostat(NetatmoBase, ClimateEntity):
"""Return a list of available preset modes."""
return SUPPORT_PRESET
async def async_set_temperature(self, **kwargs):
async def async_set_temperature(self, **kwargs: dict) -> None:
"""Set new target temperature for 2 hours."""
temp = kwargs.get(ATTR_TEMPERATURE)
if temp is None:
@ -429,7 +451,7 @@ class NetatmoThermostat(NetatmoBase, ClimateEntity):
self.async_write_ha_state()
async def async_turn_off(self):
async def async_turn_off(self) -> None:
"""Turn the entity off."""
if self._model == NA_VALVE:
await self._home_status.async_set_room_thermpoint(
@ -443,7 +465,7 @@ class NetatmoThermostat(NetatmoBase, ClimateEntity):
)
self.async_write_ha_state()
async def async_turn_on(self):
async def async_turn_on(self) -> None:
"""Turn the entity on."""
await self._home_status.async_set_room_thermpoint(self._id, STATE_NETATMO_HOME)
self.async_write_ha_state()
@ -454,7 +476,7 @@ class NetatmoThermostat(NetatmoBase, ClimateEntity):
return bool(self._connected)
@callback
def async_update_callback(self):
def async_update_callback(self) -> None:
"""Update the entity's state."""
self._home_status = self.data_handler.data[self._home_status_class]
if self._home_status is None:
@ -487,8 +509,6 @@ class NetatmoThermostat(NetatmoBase, ClimateEntity):
if "current_temperature" not in roomstatus:
return
if self._model is None:
self._model = roomstatus["module_type"]
self._current_temperature = roomstatus["current_temperature"]
self._target_temperature = roomstatus["target_temperature"]
self._preset = NETATMO_MAP_PRESET[roomstatus["setpoint_mode"]]
@ -511,7 +531,7 @@ class NetatmoThermostat(NetatmoBase, ClimateEntity):
ATTR_SELECTED_SCHEDULE
] = self._selected_schedule
def _build_room_status(self):
def _build_room_status(self) -> dict:
"""Construct room status."""
try:
roomstatus = {
@ -570,7 +590,7 @@ class NetatmoThermostat(NetatmoBase, ClimateEntity):
return {}
async def _async_service_set_schedule(self, **kwargs):
async def _async_service_set_schedule(self, **kwargs: dict) -> None:
schedule_name = kwargs.get(ATTR_SCHEDULE_NAME)
schedule_id = None
for sid, name in self.hass.data[DOMAIN][DATA_SCHEDULES][self._home_id].items():
@ -592,12 +612,14 @@ class NetatmoThermostat(NetatmoBase, ClimateEntity):
)
@property
def device_info(self):
def device_info(self) -> DeviceInfo:
"""Return the device info for the thermostat."""
return {**super().device_info, "suggested_area": self._room_data["name"]}
device_info: DeviceInfo = super().device_info
device_info["suggested_area"] = self._room_data["name"]
return device_info
def get_all_home_ids(home_data: pyatmo.HomeData) -> list[str]:
def get_all_home_ids(home_data: pyatmo.HomeData | None) -> list[str]:
"""Get all the home ids returned by NetAtmo API."""
if home_data is None:
return []

27
homeassistant/components/netatmo/config_flow.py
View file

@ -1,4 +1,6 @@
"""Config flow for Netatmo."""
from __future__ import annotations
import logging
import uuid
@ -7,6 +9,7 @@ import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_SHOW_ON_MAP
from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers import config_entry_oauth2_flow, config_validation as cv
from .const import (
@ -32,7 +35,9 @@ class NetatmoFlowHandler(
@staticmethod
@callback
def async_get_options_flow(config_entry):
def async_get_options_flow(
config_entry: config_entries.ConfigEntry,
) -> config_entries.OptionsFlow:
"""Get the options flow for this handler."""
return NetatmoOptionsFlowHandler(config_entry)
@ -62,7 +67,7 @@ class NetatmoFlowHandler(
return {"scope": " ".join(scopes)}
async def async_step_user(self, user_input=None):
async def async_step_user(self, user_input: dict | None = None) -> FlowResult:
"""Handle a flow start."""
await self.async_set_unique_id(DOMAIN)
@ -81,17 +86,19 @@ class NetatmoOptionsFlowHandler(config_entries.OptionsFlow):
self.options = dict(config_entry.options)
self.options.setdefault(CONF_WEATHER_AREAS, {})
async def async_step_init(self, user_input=None):
async def async_step_init(self, user_input: dict | None = None) -> FlowResult:
"""Manage the Netatmo options."""
return await self.async_step_public_weather_areas()
async def async_step_public_weather_areas(self, user_input=None):
async def async_step_public_weather_areas(
self, user_input: dict | None = None
) -> FlowResult:
"""Manage configuration of Netatmo public weather areas."""
errors = {}
errors: dict = {}
if user_input is not None:
new_client = user_input.pop(CONF_NEW_AREA, None)
areas = user_input.pop(CONF_WEATHER_AREAS, None)
areas = user_input.pop(CONF_WEATHER_AREAS, [])
user_input[CONF_WEATHER_AREAS] = {
area: self.options[CONF_WEATHER_AREAS][area] for area in areas
}
@ -110,7 +117,7 @@ class NetatmoOptionsFlowHandler(config_entries.OptionsFlow):
vol.Optional(
CONF_WEATHER_AREAS,
default=weather_areas,
): cv.multi_select(weather_areas),
): cv.multi_select({wa: None for wa in weather_areas}),
vol.Optional(CONF_NEW_AREA): str,
}
)
@ -120,7 +127,7 @@ class NetatmoOptionsFlowHandler(config_entries.OptionsFlow):
errors=errors,
)
async def async_step_public_weather(self, user_input=None):
async def async_step_public_weather(self, user_input: dict) -> FlowResult:
"""Manage configuration of Netatmo public weather sensors."""
if user_input is not None and CONF_NEW_AREA not in user_input:
self.options[CONF_WEATHER_AREAS][
@ -181,14 +188,14 @@ class NetatmoOptionsFlowHandler(config_entries.OptionsFlow):
return self.async_show_form(step_id="public_weather", data_schema=data_schema)
def _create_options_entry(self):
def _create_options_entry(self) -> FlowResult:
"""Update config entry options."""
return self.async_create_entry(
title="Netatmo Public Weather", data=self.options
)
def fix_coordinates(user_input):
def fix_coordinates(user_input: dict) -> dict:
"""Fix coordinates if they don't comply with the Netatmo API."""
# Ensure coordinates have acceptable length for the Netatmo API
for coordinate in (CONF_LAT_NE, CONF_LAT_SW, CONF_LON_NE, CONF_LON_SW):

3
homeassistant/components/netatmo/const.py
View file

@ -6,6 +6,7 @@ from homeassistant.components.select import DOMAIN as SELECT_DOMAIN
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
API = "api"
UNKNOWN = "unknown"
DOMAIN = "netatmo"
MANUFACTURER = "Netatmo"
@ -76,7 +77,7 @@ DATA_SCHEDULES = "netatmo_schedules"
NETATMO_WEBHOOK_URL = None
NETATMO_EVENT = "netatmo_event"
DEFAULT_PERSON = "Unknown"
DEFAULT_PERSON = UNKNOWN
DEFAULT_DISCOVERY = True
DEFAULT_WEBHOOKS = False

29
homeassistant/components/netatmo/data_handler.py
View file

@ -8,6 +8,7 @@ from datetime import timedelta
from itertools import islice
import logging
from time import time
from typing import Any
import pyatmo
@ -75,11 +76,11 @@ class NetatmoDataHandler:
self._auth = hass.data[DOMAIN][entry.entry_id][AUTH]
self.listeners: list[CALLBACK_TYPE] = []
self.data_classes: dict = {}
self.data = {}
self._queue = deque()
self.data: dict = {}
self._queue: deque = deque()
self._webhook: bool = False
async def async_setup(self):
async def async_setup(self) -> None:
"""Set up the Netatmo data handler."""
async_track_time_interval(
@ -94,7 +95,7 @@ class NetatmoDataHandler:
)
)
async def async_update(self, event_time):
async def async_update(self, event_time: timedelta) -> None:
"""
Update device.
@ -115,17 +116,17 @@ class NetatmoDataHandler:
self._queue.rotate(BATCH_SIZE)
@callback
def async_force_update(self, data_class_entry):
def async_force_update(self, data_class_entry: str) -> None:
"""Prioritize data retrieval for given data class entry."""
self.data_classes[data_class_entry].next_scan = time()
self._queue.rotate(-(self._queue.index(self.data_classes[data_class_entry])))
async def async_cleanup(self):
async def async_cleanup(self) -> None:
"""Clean up the Netatmo data handler."""
for listener in self.listeners:
listener()
async def handle_event(self, event):
async def handle_event(self, event: dict) -> None:
"""Handle webhook events."""
if event["data"][WEBHOOK_PUSH_TYPE] == WEBHOOK_ACTIVATION:
_LOGGER.info("%s webhook successfully registered", MANUFACTURER)
@ -139,7 +140,7 @@ class NetatmoDataHandler:
_LOGGER.debug("%s camera reconnected", MANUFACTURER)
self.async_force_update(CAMERA_DATA_CLASS_NAME)
async def async_fetch_data(self, data_class_entry):
async def async_fetch_data(self, data_class_entry: str) -> None:
"""Fetch data and notify."""
if self.data[data_class_entry] is None:
return
@ -163,8 +164,12 @@ class NetatmoDataHandler:
update_callback()
async def register_data_class(
self, data_class_name, data_class_entry, update_callback, **kwargs
):
self,
data_class_name: str,
data_class_entry: str,
update_callback: CALLBACK_TYPE,
**kwargs: Any,
) -> None:
"""Register data class."""
if data_class_entry in self.data_classes:
if update_callback not in self.data_classes[data_class_entry].subscriptions:
@ -189,7 +194,9 @@ class NetatmoDataHandler:
self._queue.append(self.data_classes[data_class_entry])
_LOGGER.debug("Data class %s added", data_class_entry)
async def unregister_data_class(self, data_class_entry, update_callback):
async def unregister_data_class(
self, data_class_entry: str, update_callback: CALLBACK_TYPE | None
) -> None:
"""Unregister data class."""
self.data_classes[data_class_entry].subscriptions.remove(update_callback)

18
homeassistant/components/netatmo/device_trigger.py
View file

@ -63,7 +63,9 @@ TRIGGER_SCHEMA = DEVICE_TRIGGER_BASE_SCHEMA.extend(
)
async def async_validate_trigger_config(hass, config):
async def async_validate_trigger_config(
hass: HomeAssistant, config: ConfigType
) -> ConfigType:
"""Validate config."""
config = TRIGGER_SCHEMA(config)
@ -129,10 +131,10 @@ async def async_attach_trigger(
device = device_registry.async_get(config[CONF_DEVICE_ID])
if not device:
return
return lambda: None
if device.model not in DEVICES:
return
return lambda: None
event_config = {
event_trigger.CONF_PLATFORM: "event",
@ -142,10 +144,14 @@ async def async_attach_trigger(
ATTR_DEVICE_ID: config[ATTR_DEVICE_ID],
},
}
# if config[CONF_TYPE] in SUBTYPES:
# event_config[event_trigger.CONF_EVENT_DATA]["data"] = {
# "mode": config[CONF_SUBTYPE]
# }
if config[CONF_TYPE] in SUBTYPES:
event_config[event_trigger.CONF_EVENT_DATA]["data"] = {
"mode": config[CONF_SUBTYPE]
}
event_config.update(
{event_trigger.CONF_EVENT_DATA: {"data": {"mode": config[CONF_SUBTYPE]}}}
)
event_config = event_trigger.TRIGGER_SCHEMA(event_config)
return await event_trigger.async_attach_trigger(

4
homeassistant/components/netatmo/helper.py
View file

@ -1,6 +1,6 @@
"""Helper for Netatmo integration."""
from dataclasses import dataclass
from uuid import uuid4
from uuid import UUID, uuid4
@dataclass
@ -14,4 +14,4 @@ class NetatmoArea:
lon_sw: float
mode: str
show_on_map: bool
uuid: str = uuid4()
uuid: UUID = uuid4()

34
homeassistant/components/netatmo/light.py
View file

@ -1,10 +1,17 @@
"""Support for the Netatmo camera lights."""
from __future__ import annotations
import logging
from typing import cast
import pyatmo
from homeassistant.components.light import LightEntity
from homeassistant.core import callback
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import PlatformNotReady
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import (
DATA_HANDLER,
@ -12,6 +19,7 @@ from .const import (
EVENT_TYPE_LIGHT_MODE,
MANUFACTURER,
SIGNAL_NAME,
UNKNOWN,
WEBHOOK_LIGHT_MODE,
WEBHOOK_PUSH_TYPE,
)
@ -21,7 +29,9 @@ from .netatmo_entity_base import NetatmoBase
_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 the Netatmo camera light platform."""
if "access_camera" not in entry.data["token"]["scope"]:
_LOGGER.info(
@ -79,7 +89,7 @@ class NetatmoLight(NetatmoBase, LightEntity):
self._id = camera_id
self._home_id = home_id
self._model = camera_type
self._device_name = self._data.get_camera(camera_id).get("name")
self._device_name: str = self._data.get_camera(camera_id).get("name", UNKNOWN)
self._attr_name = f"{MANUFACTURER} {self._device_name}"
self._is_on = False
self._attr_unique_id = f"{self._id}-light"
@ -97,7 +107,7 @@ class NetatmoLight(NetatmoBase, LightEntity):
)
@callback
def handle_event(self, event):
def handle_event(self, event: dict) -> None:
"""Handle webhook events."""
data = event["data"]
@ -114,17 +124,25 @@ class NetatmoLight(NetatmoBase, LightEntity):
self.async_write_ha_state()
return
@property
def _data(self) -> pyatmo.AsyncCameraData:
"""Return data for this entity."""
return cast(
pyatmo.AsyncCameraData,
self.data_handler.data[self._data_classes[0]["name"]],
)
@property
def available(self) -> bool:
"""If the webhook is not established, mark as unavailable."""
return bool(self.data_handler.webhook)
@property
def is_on(self):
def is_on(self) -> bool:
"""Return true if light is on."""
return self._is_on
async def async_turn_on(self, **kwargs):
async def async_turn_on(self, **kwargs: dict) -> None:
"""Turn camera floodlight on."""
_LOGGER.debug("Turn camera '%s' on", self.name)
await self._data.async_set_state(
@ -133,7 +151,7 @@ class NetatmoLight(NetatmoBase, LightEntity):
floodlight="on",
)
async def async_turn_off(self, **kwargs):
async def async_turn_off(self, **kwargs: dict) -> None:
"""Turn camera floodlight into auto mode."""
_LOGGER.debug("Turn camera '%s' to auto mode", self.name)
await self._data.async_set_state(
@ -143,6 +161,6 @@ class NetatmoLight(NetatmoBase, LightEntity):
)
@callback
def async_update_callback(self):
def async_update_callback(self) -> None:
"""Update the entity's state."""
self._is_on = bool(self._data.get_light_state(self._id) == "on")

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

@ -3,7 +3,7 @@
"name": "Netatmo",
"documentation": "https://www.home-assistant.io/integrations/netatmo",
"requirements": [
"pyatmo==5.2.0"
"pyatmo==5.2.1"
],
"after_dependencies": [
"cloud",

13
homeassistant/components/netatmo/media_source.py
View file

@ -11,7 +11,6 @@ from homeassistant.components.media_player.const import (
MEDIA_TYPE_VIDEO,
)
from homeassistant.components.media_player.errors import BrowseError
from homeassistant.components.media_source.const import MEDIA_MIME_TYPES
from homeassistant.components.media_source.error import MediaSourceError, Unresolvable
from homeassistant.components.media_source.models import (
BrowseMediaSource,
@ -31,7 +30,7 @@ class IncompatibleMediaSource(MediaSourceError):
"""Incompatible media source attributes."""
async def async_get_media_source(hass: HomeAssistant):
async def async_get_media_source(hass: HomeAssistant) -> NetatmoSource:
"""Set up Netatmo media source."""
return NetatmoSource(hass)
@ -54,7 +53,9 @@ class NetatmoSource(MediaSource):
return PlayMedia(url, MIME_TYPE)
async def async_browse_media(
self, item: MediaSourceItem, media_types: tuple[str] = MEDIA_MIME_TYPES
self,
item: MediaSourceItem,
media_types: tuple[str] = ("video",),
) -> BrowseMediaSource:
"""Return media."""
try:
@ -65,7 +66,7 @@ class NetatmoSource(MediaSource):
return self._browse_media(source, camera_id, event_id)
def _browse_media(
self, source: str, camera_id: str, event_id: int
self, source: str, camera_id: str, event_id: int | None
) -> BrowseMediaSource:
"""Browse media."""
if camera_id and camera_id not in self.events:
@ -77,7 +78,7 @@ class NetatmoSource(MediaSource):
return self._build_item_response(source, camera_id, event_id)
def _build_item_response(
self, source: str, camera_id: str, event_id: int = None
self, source: str, camera_id: str, event_id: int | None = None
) -> BrowseMediaSource:
if event_id and event_id in self.events[camera_id]:
created = dt.datetime.fromtimestamp(event_id)
@ -148,7 +149,7 @@ class NetatmoSource(MediaSource):
return media
def remove_html_tags(text):
def remove_html_tags(text: str) -> str:
"""Remove html tags from string."""
clean = re.compile("<.*?>")
return re.sub(clean, "", text)

23
homeassistant/components/netatmo/netatmo_entity_base.py
View file

@ -3,7 +3,7 @@ from __future__ import annotations
from homeassistant.const import ATTR_ATTRIBUTION
from homeassistant.core import CALLBACK_TYPE, callback
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.entity import DeviceInfo, Entity
from .const import (
DATA_DEVICE_IDS,
@ -25,12 +25,14 @@ class NetatmoBase(Entity):
self._data_classes: list[dict] = []
self._listeners: list[CALLBACK_TYPE] = []
self._device_name = None
self._id = None
self._model = None
self._device_name: str = ""
self._id: str = ""
self._model: str = ""
self._attr_name = None
self._attr_unique_id = None
self._attr_extra_state_attributes = {ATTR_ATTRIBUTION: DEFAULT_ATTRIBUTION}
self._attr_extra_state_attributes: dict = {
ATTR_ATTRIBUTION: DEFAULT_ATTRIBUTION
}
async def async_added_to_hass(self) -> None:
"""Entity created."""
@ -71,7 +73,7 @@ class NetatmoBase(Entity):
self.async_update_callback()
async def async_will_remove_from_hass(self):
async def async_will_remove_from_hass(self) -> None:
"""Run when entity will be removed from hass."""
await super().async_will_remove_from_hass()
@ -84,17 +86,12 @@ class NetatmoBase(Entity):
)
@callback
def async_update_callback(self):
def async_update_callback(self) -> None:
"""Update the entity's state."""
raise NotImplementedError
@property
def _data(self):
"""Return data for this entity."""
return self.data_handler.data[self._data_classes[0]["name"]]
@property
def device_info(self):
def device_info(self) -> DeviceInfo:
"""Return the device info for the sensor."""
return {
"identifiers": {(DOMAIN, self._id)},

67
homeassistant/components/netatmo/sensor.py
View file

@ -2,9 +2,12 @@
from __future__ import annotations
import logging
from typing import NamedTuple
from typing import NamedTuple, cast
import pyatmo
from homeassistant.components.sensor import SensorEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
ATTR_LATITUDE,
ATTR_LONGITUDE,
@ -24,19 +27,21 @@ from homeassistant.const import (
SPEED_KILOMETERS_PER_HOUR,
TEMP_CELSIUS,
)
from homeassistant.core import callback
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import PlatformNotReady
from homeassistant.helpers.device_registry import async_entries_for_config_entry
from homeassistant.helpers.dispatcher import (
async_dispatcher_connect,
async_dispatcher_send,
)
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import CONF_WEATHER_AREAS, DATA_HANDLER, DOMAIN, MANUFACTURER, SIGNAL_NAME
from .data_handler import (
HOMECOACH_DATA_CLASS_NAME,
PUBLICDATA_DATA_CLASS_NAME,
WEATHERSTATION_DATA_CLASS_NAME,
NetatmoDataHandler,
)
from .helper import NetatmoArea
from .netatmo_entity_base import NetatmoBase
@ -267,12 +272,14 @@ BATTERY_VALUES = {
PUBLIC = "public"
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 the Netatmo weather and homecoach platform."""
data_handler = hass.data[DOMAIN][entry.entry_id][DATA_HANDLER]
platform_not_ready = True
async def find_entities(data_class_name):
async def find_entities(data_class_name: str) -> list:
"""Find all entities."""
all_module_infos = {}
data = data_handler.data
@ -330,7 +337,7 @@ async def async_setup_entry(hass, entry, async_add_entities):
device_registry = await hass.helpers.device_registry.async_get_registry()
async def add_public_entities(update=True):
async def add_public_entities(update: bool = True) -> None:
"""Retrieve Netatmo public weather entities."""
entities = {
device.name: device.id
@ -396,7 +403,13 @@ async def async_setup_entry(hass, entry, async_add_entities):
class NetatmoSensor(NetatmoBase, SensorEntity):
"""Implementation of a Netatmo sensor."""
def __init__(self, data_handler, data_class_name, module_info, sensor_type):
def __init__(
self,
data_handler: NetatmoDataHandler,
data_class_name: str,
module_info: dict,
sensor_type: str,
) -> None:
"""Initialize the sensor."""
super().__init__(data_handler)
@ -434,20 +447,21 @@ class NetatmoSensor(NetatmoBase, SensorEntity):
self._attr_entity_registry_enabled_default = metadata.enable_default
@property
def available(self):
def _data(self) -> pyatmo.AsyncWeatherStationData:
"""Return data for this entity."""
return cast(
pyatmo.AsyncWeatherStationData,
self.data_handler.data[self._data_classes[0]["name"]],
)
@property
def available(self) -> bool:
"""Return entity availability."""
return self.state is not None
@callback
def async_update_callback(self):
def async_update_callback(self) -> None:
"""Update the entity's state."""
if self._data is None:
if self.state is None:
return
_LOGGER.warning("No data from update")
self._attr_state = None
return
data = self._data.get_last_data(station_id=self._station_id, exclude=3600).get(
self._id
)
@ -531,7 +545,7 @@ def process_battery(data: int, model: str) -> str:
return "Very Low"
def process_health(health):
def process_health(health: int) -> str:
"""Process health index and return string for display."""
if health == 0:
return "Healthy"
@ -541,11 +555,10 @@ def process_health(health):
return "Fair"
if health == 3:
return "Poor"
if health == 4:
return "Unhealthy"
return "Unhealthy"
def process_rf(strength):
def process_rf(strength: int) -> str:
"""Process wifi signal strength and return string for display."""
if strength >= 90:
return "Low"
@ -556,7 +569,7 @@ def process_rf(strength):
return "Full"
def process_wifi(strength):
def process_wifi(strength: int) -> str:
"""Process wifi signal strength and return string for display."""
if strength >= 86:
return "Low"
@ -570,7 +583,9 @@ def process_wifi(strength):
class NetatmoPublicSensor(NetatmoBase, SensorEntity):
"""Represent a single sensor in a Netatmo."""
def __init__(self, data_handler, area, sensor_type):
def __init__(
self, data_handler: NetatmoDataHandler, area: NetatmoArea, sensor_type: str
) -> None:
"""Initialize the sensor."""
super().__init__(data_handler)
@ -611,13 +626,15 @@ class NetatmoPublicSensor(NetatmoBase, SensorEntity):
)
@property
def _data(self):
return self.data_handler.data[self._signal_name]
def _data(self) -> pyatmo.AsyncPublicData:
"""Return data for this entity."""
return cast(pyatmo.AsyncPublicData, self.data_handler.data[self._signal_name])
async def async_added_to_hass(self) -> None:
"""Entity created."""
await super().async_added_to_hass()
assert self.device_info and "name" in self.device_info
self.data_handler.listeners.append(
async_dispatcher_connect(
self.hass,
@ -626,7 +643,7 @@ class NetatmoPublicSensor(NetatmoBase, SensorEntity):
)
)
async def async_config_update_callback(self, area):
async def async_config_update_callback(self, area: NetatmoArea) -> None:
"""Update the entity's config."""
if self.area == area:
return
@ -661,7 +678,7 @@ class NetatmoPublicSensor(NetatmoBase, SensorEntity):
)
@callback
def async_update_callback(self):
def async_update_callback(self) -> None:
"""Update the entity's state."""
data = None

15
homeassistant/components/netatmo/webhook.py
View file

@ -1,7 +1,10 @@
"""The Netatmo integration."""
import logging
from aiohttp.web import Request
from homeassistant.const import ATTR_DEVICE_ID, ATTR_ID, ATTR_NAME
from homeassistant.core import HomeAssistant
from homeassistant.helpers.dispatcher import async_dispatcher_send
from .const import (
@ -25,7 +28,9 @@ SUBEVENT_TYPE_MAP = {
}
async def async_handle_webhook(hass, webhook_id, request):
async def async_handle_webhook(
hass: HomeAssistant, webhook_id: str, request: Request
) -> None:
"""Handle webhook callback."""
try:
data = await request.json()
@ -47,12 +52,12 @@ async def async_handle_webhook(hass, webhook_id, request):
async_evaluate_event(hass, data)
def async_evaluate_event(hass, event_data):
def async_evaluate_event(hass: HomeAssistant, event_data: dict) -> None:
"""Evaluate events from webhook."""
event_type = event_data.get(ATTR_EVENT_TYPE)
event_type = event_data.get(ATTR_EVENT_TYPE, "None")
if event_type == "person":
for person in event_data.get(ATTR_PERSONS):
for person in event_data.get(ATTR_PERSONS, {}):
person_event_data = dict(event_data)
person_event_data[ATTR_ID] = person.get(ATTR_ID)
person_event_data[ATTR_NAME] = hass.data[DOMAIN][DATA_PERSONS].get(
@ -67,7 +72,7 @@ def async_evaluate_event(hass, event_data):
async_send_event(hass, event_type, event_data)
def async_send_event(hass, event_type, data):
def async_send_event(hass: HomeAssistant, event_type: str, data: dict) -> None:
"""Send events."""
_LOGGER.debug("%s: %s", event_type, data)
async_dispatcher_send(

14
mypy.ini
View file

@ -671,6 +671,17 @@ no_implicit_optional = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.netatmo.*]
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.network.*]
check_untyped_defs = true
disallow_incomplete_defs = true
@ -1388,9 +1399,6 @@ ignore_errors = true
[mypy-homeassistant.components.nest.legacy.*]
ignore_errors = true
[mypy-homeassistant.components.netatmo.*]
ignore_errors = true
[mypy-homeassistant.components.netio.*]
ignore_errors = true

2
requirements_all.txt
View file

@ -1316,7 +1316,7 @@ pyarlo==0.2.4
pyatag==0.3.5.3
# homeassistant.components.netatmo
pyatmo==5.2.0
pyatmo==5.2.1
# homeassistant.components.atome
pyatome==0.1.1

2
requirements_test_all.txt
View file

@ -747,7 +747,7 @@ pyarlo==0.2.4
pyatag==0.3.5.3
# homeassistant.components.netatmo
pyatmo==5.2.0
pyatmo==5.2.1
# homeassistant.components.apple_tv
pyatv==0.8.1

1
script/hassfest/mypy_config.py
View file

@ -110,7 +110,6 @@ IGNORED_MODULES: Final[list[str]] = [
"homeassistant.components.neato.*",
"homeassistant.components.ness_alarm.*",
"homeassistant.components.nest.legacy.*",
"homeassistant.components.netatmo.*",
"homeassistant.components.netio.*",
"homeassistant.components.nightscout.*",
"homeassistant.components.nilu.*",

12
tests/components/netatmo/test_select.py
View file

@ -16,10 +16,10 @@ async def test_select_schedule_thermostats(hass, config_entry, caplog, netatmo_a
await hass.async_block_till_done()
webhook_id = config_entry.data[CONF_WEBHOOK_ID]
select_entity_livingroom = "select.netatmo_myhome"
select_entity = "select.netatmo_myhome"
assert hass.states.get(select_entity_livingroom).state == "Default"
assert hass.states.get(select_entity_livingroom).attributes[ATTR_OPTIONS] == [
assert hass.states.get(select_entity).state == "Default"
assert hass.states.get(select_entity).attributes[ATTR_OPTIONS] == [
"Default",
"Winter",
]
@ -33,7 +33,7 @@ async def test_select_schedule_thermostats(hass, config_entry, caplog, netatmo_a
}
await simulate_webhook(hass, webhook_id, response)
assert hass.states.get(select_entity_livingroom).state == "Winter"
assert hass.states.get(select_entity).state == "Winter"
# Test setting a different schedule
with patch(
@ -43,7 +43,7 @@ async def test_select_schedule_thermostats(hass, config_entry, caplog, netatmo_a
SELECT_DOMAIN,
SERVICE_SELECT_OPTION,
{
ATTR_ENTITY_ID: select_entity_livingroom,
ATTR_ENTITY_ID: select_entity,
ATTR_OPTION: "Default",
},
blocking=True,
@ -62,4 +62,4 @@ async def test_select_schedule_thermostats(hass, config_entry, caplog, netatmo_a
}
await simulate_webhook(hass, webhook_id, response)
assert hass.states.get(select_entity_livingroom).state == "Default"
assert hass.states.get(select_entity).state == "Default"