Bump reolink-aio to 0.7.14 and improve typing of Reolink (#103129)

* Improve typing

* fix mypy

* Further improve typing

* Restore Literal typing

* Bump reolink_aio to 0.7.13

* Bump reolink-aio to 0.7.14
This commit is contained in:
starkillerOG 2023-11-03 17:05:27 +01:00 committed by GitHub
parent c1d979dc07
commit 1df69f52e5
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
10 changed files with 37 additions and 29 deletions

View file

@ -4,7 +4,7 @@ from __future__ import annotations
import asyncio
from collections.abc import Mapping
import logging
from typing import Any
from typing import Any, Literal
import aiohttp
from aiohttp.web import Request
@ -81,7 +81,7 @@ class ReolinkHost:
return self._unique_id
@property
def api(self):
def api(self) -> Host:
"""Return the API object."""
return self._api
@ -313,7 +313,7 @@ class ReolinkHost:
"""Call the API of the camera device to update the internal states."""
await self._api.get_states()
async def disconnect(self):
async def disconnect(self) -> None:
"""Disconnect from the API, so the connection will be released."""
try:
await self._api.unsubscribe()
@ -335,7 +335,7 @@ class ReolinkHost:
err,
)
async def _async_start_long_polling(self, initial=False):
async def _async_start_long_polling(self, initial=False) -> None:
"""Start ONVIF long polling task."""
if self._long_poll_task is None:
try:
@ -364,7 +364,7 @@ class ReolinkHost:
self._lost_subscription = False
self._long_poll_task = asyncio.create_task(self._async_long_polling())
async def _async_stop_long_polling(self):
async def _async_stop_long_polling(self) -> None:
"""Stop ONVIF long polling task."""
if self._long_poll_task is not None:
self._long_poll_task.cancel()
@ -372,7 +372,7 @@ class ReolinkHost:
await self._api.unsubscribe(sub_type=SubType.long_poll)
async def stop(self, event=None):
async def stop(self, event=None) -> None:
"""Disconnect the API."""
if self._cancel_poll is not None:
self._cancel_poll()
@ -433,7 +433,7 @@ class ReolinkHost:
else:
self._lost_subscription = False
async def _renew(self, sub_type: SubType) -> None:
async def _renew(self, sub_type: Literal[SubType.push, SubType.long_poll]) -> None:
"""Execute the renew of the subscription."""
if not self._api.subscribed(sub_type):
_LOGGER.debug(
@ -512,8 +512,10 @@ class ReolinkHost:
_LOGGER.debug("Registered webhook: %s", event_id)
def unregister_webhook(self):
def unregister_webhook(self) -> None:
"""Unregister the webhook for motion events."""
if self.webhook_id is None:
return
_LOGGER.debug("Unregistering webhook %s", self.webhook_id)
webhook.async_unregister(self._hass, self.webhook_id)
self.webhook_id = None