Improve type hints in demo [2/3] (#77185)

pull/77357/head
epenet 2022-08-26 11:34:38 +02:00 committed by GitHub
parent 94cd8e801b
commit 452ee0284a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 81 additions and 34 deletions

View File

@ -41,7 +41,7 @@ class DemoCamera(Camera):
_attr_motion_detection_enabled = False _attr_motion_detection_enabled = False
_attr_supported_features = CameraEntityFeature.ON_OFF _attr_supported_features = CameraEntityFeature.ON_OFF
def __init__(self, name, content_type): def __init__(self, name: str, content_type: str) -> None:
"""Initialize demo camera component.""" """Initialize demo camera component."""
super().__init__() super().__init__()
self._attr_name = name self._attr_name = name

View File

@ -1,6 +1,8 @@
"""Config flow to configure demo component.""" """Config flow to configure demo component."""
from __future__ import annotations from __future__ import annotations
from typing import Any
import voluptuous as vol import voluptuous as vol
from homeassistant import config_entries from homeassistant import config_entries
@ -30,9 +32,9 @@ class DemoConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Get the options flow for this handler.""" """Get the options flow for this handler."""
return OptionsFlowHandler(config_entry) return OptionsFlowHandler(config_entry)
async def async_step_import(self, import_info) -> FlowResult: async def async_step_import(self, import_info: dict[str, Any]) -> FlowResult:
"""Set the config entry up from yaml.""" """Set the config entry up from yaml."""
return self.async_create_entry(title="Demo", data={}) return self.async_create_entry(title="Demo", data=import_info)
class OptionsFlowHandler(config_entries.OptionsFlow): class OptionsFlowHandler(config_entries.OptionsFlow):
@ -43,11 +45,15 @@ class OptionsFlowHandler(config_entries.OptionsFlow):
self.config_entry = config_entry self.config_entry = config_entry
self.options = dict(config_entry.options) self.options = dict(config_entry.options)
async def async_step_init(self, user_input=None) -> FlowResult: async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Manage the options.""" """Manage the options."""
return await self.async_step_options_1() return await self.async_step_options_1()
async def async_step_options_1(self, user_input=None) -> FlowResult: async def async_step_options_1(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Manage the options.""" """Manage the options."""
if user_input is not None: if user_input is not None:
self.options.update(user_input) self.options.update(user_input)
@ -70,7 +76,9 @@ class OptionsFlowHandler(config_entries.OptionsFlow):
), ),
) )
async def async_step_options_2(self, user_input=None) -> FlowResult: async def async_step_options_2(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Manage the options 2.""" """Manage the options 2."""
if user_input is not None: if user_input is not None:
self.options.update(user_input) self.options.update(user_input)
@ -101,6 +109,6 @@ class OptionsFlowHandler(config_entries.OptionsFlow):
), ),
) )
async def _update_options(self): async def _update_options(self) -> FlowResult:
"""Update config entry options.""" """Update config entry options."""
return self.async_create_entry(title="", data=self.options) return self.async_create_entry(title="", data=self.options)

View File

@ -18,11 +18,11 @@ def setup_scanner(
) -> bool: ) -> bool:
"""Set up the demo tracker.""" """Set up the demo tracker."""
def offset(): def offset() -> float:
"""Return random offset.""" """Return random offset."""
return (random.randrange(500, 2000)) / 2e5 * random.choice((-1, 1)) return (random.randrange(500, 2000)) / 2e5 * random.choice((-1, 1))
def random_see(dev_id, name): def random_see(dev_id: str, name: str) -> None:
"""Randomize a sighting.""" """Randomize a sighting."""
see( see(
dev_id=dev_id, dev_id=dev_id,

View File

@ -4,6 +4,7 @@ from __future__ import annotations
from hashlib import sha1 from hashlib import sha1
import logging import logging
import os import os
from typing import Any
from homeassistant.components.mailbox import CONTENT_TYPE_MPEG, Mailbox, StreamError from homeassistant.components.mailbox import CONTENT_TYPE_MPEG, Mailbox, StreamError
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
@ -27,10 +28,10 @@ async def async_get_handler(
class DemoMailbox(Mailbox): class DemoMailbox(Mailbox):
"""Demo Mailbox.""" """Demo Mailbox."""
def __init__(self, hass, name): def __init__(self, hass: HomeAssistant, name: str) -> None:
"""Initialize Demo mailbox.""" """Initialize Demo mailbox."""
super().__init__(hass, name) super().__init__(hass, name)
self._messages = {} self._messages: dict[str, dict[str, Any]] = {}
txt = "Lorem ipsum dolor sit amet, consectetur adipiscing elit. " txt = "Lorem ipsum dolor sit amet, consectetur adipiscing elit. "
for idx in range(0, 10): for idx in range(0, 10):
msgtime = int(dt.as_timestamp(dt.utcnow()) - 3600 * 24 * (10 - idx)) msgtime = int(dt.as_timestamp(dt.utcnow()) - 3600 * 24 * (10 - idx))
@ -48,21 +49,21 @@ class DemoMailbox(Mailbox):
self._messages[msgsha] = msg self._messages[msgsha] = msg
@property @property
def media_type(self): def media_type(self) -> str:
"""Return the supported media type.""" """Return the supported media type."""
return CONTENT_TYPE_MPEG return CONTENT_TYPE_MPEG
@property @property
def can_delete(self): def can_delete(self) -> bool:
"""Return if messages can be deleted.""" """Return if messages can be deleted."""
return True return True
@property @property
def has_media(self): def has_media(self) -> bool:
"""Return if messages have attached media files.""" """Return if messages have attached media files."""
return True return True
async def async_get_media(self, msgid): async def async_get_media(self, msgid: str) -> bytes:
"""Return the media blob for the msgid.""" """Return the media blob for the msgid."""
if msgid not in self._messages: if msgid not in self._messages:
raise StreamError("Message not found") raise StreamError("Message not found")
@ -71,7 +72,7 @@ class DemoMailbox(Mailbox):
with open(audio_path, "rb") as file: with open(audio_path, "rb") as file:
return file.read() return file.read()
async def async_get_messages(self): async def async_get_messages(self) -> list[dict[str, Any]]:
"""Return a list of the current messages.""" """Return a list of the current messages."""
return sorted( return sorted(
self._messages.values(), self._messages.values(),
@ -79,7 +80,7 @@ class DemoMailbox(Mailbox):
reverse=True, reverse=True,
) )
async def async_delete(self, msgid): async def async_delete(self, msgid: str) -> bool:
"""Delete the specified messages.""" """Delete the specified messages."""
if msgid in self._messages: if msgid in self._messages:
_LOGGER.info("Deleting: %s", msgid) _LOGGER.info("Deleting: %s", msgid)

View File

@ -1,10 +1,20 @@
"""Demo notification service.""" """Demo notification service."""
from __future__ import annotations
from typing import Any
from homeassistant.components.notify import BaseNotificationService from homeassistant.components.notify import BaseNotificationService
from homeassistant.core import HomeAssistant
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
EVENT_NOTIFY = "notify" EVENT_NOTIFY = "notify"
def get_service(hass, config, discovery_info=None): def get_service(
hass: HomeAssistant,
config: ConfigType,
discovery_info: DiscoveryInfoType | None = None,
) -> BaseNotificationService:
"""Get the demo notification service.""" """Get the demo notification service."""
return DemoNotificationService(hass) return DemoNotificationService(hass)
@ -12,16 +22,16 @@ def get_service(hass, config, discovery_info=None):
class DemoNotificationService(BaseNotificationService): class DemoNotificationService(BaseNotificationService):
"""Implement demo notification service.""" """Implement demo notification service."""
def __init__(self, hass): def __init__(self, hass: HomeAssistant) -> None:
"""Initialize the service.""" """Initialize the service."""
self.hass = hass self.hass = hass
@property @property
def targets(self): def targets(self) -> dict[str, str]:
"""Return a dictionary of registered targets.""" """Return a dictionary of registered targets."""
return {"test target name": "test target id"} return {"test target name": "test target id"}
def send_message(self, message="", **kwargs): def send_message(self, message: str = "", **kwargs: Any) -> None:
"""Send a message to a user.""" """Send a message to a user."""
kwargs["message"] = message kwargs["message"] = message
self.hass.bus.fire(EVENT_NOTIFY, kwargs) self.hass.bus.fire(EVENT_NOTIFY, kwargs)

View File

@ -12,11 +12,17 @@ from homeassistant.components.stt.const import (
AudioSampleRates, AudioSampleRates,
SpeechResultState, SpeechResultState,
) )
from homeassistant.core import HomeAssistant
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
SUPPORT_LANGUAGES = ["en", "de"] SUPPORT_LANGUAGES = ["en", "de"]
async def async_get_engine(hass, config, discovery_info=None): async def async_get_engine(
hass: HomeAssistant,
config: ConfigType,
discovery_info: DiscoveryInfoType | None = None,
) -> Provider:
"""Set up Demo speech component.""" """Set up Demo speech component."""
return DemoProvider() return DemoProvider()

View File

@ -1,9 +1,19 @@
"""Support for the demo for text to speech service.""" """Support for the demo for text to speech service."""
from __future__ import annotations
import os import os
from typing import Any
import voluptuous as vol import voluptuous as vol
from homeassistant.components.tts import CONF_LANG, PLATFORM_SCHEMA, Provider from homeassistant.components.tts import (
CONF_LANG,
PLATFORM_SCHEMA,
Provider,
TtsAudioType,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
SUPPORT_LANGUAGES = ["en", "de"] SUPPORT_LANGUAGES = ["en", "de"]
@ -14,7 +24,11 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
) )
def get_engine(hass, config, discovery_info=None): def get_engine(
hass: HomeAssistant,
config: ConfigType,
discovery_info: DiscoveryInfoType | None = None,
) -> Provider:
"""Set up Demo speech component.""" """Set up Demo speech component."""
return DemoProvider(config.get(CONF_LANG, DEFAULT_LANG)) return DemoProvider(config.get(CONF_LANG, DEFAULT_LANG))
@ -22,27 +36,29 @@ def get_engine(hass, config, discovery_info=None):
class DemoProvider(Provider): class DemoProvider(Provider):
"""Demo speech API provider.""" """Demo speech API provider."""
def __init__(self, lang): def __init__(self, lang: str) -> None:
"""Initialize demo provider.""" """Initialize demo provider."""
self._lang = lang self._lang = lang
self.name = "Demo" self.name = "Demo"
@property @property
def default_language(self): def default_language(self) -> str:
"""Return the default language.""" """Return the default language."""
return self._lang return self._lang
@property @property
def supported_languages(self): def supported_languages(self) -> list[str]:
"""Return list of supported languages.""" """Return list of supported languages."""
return SUPPORT_LANGUAGES return SUPPORT_LANGUAGES
@property @property
def supported_options(self): def supported_options(self) -> list[str]:
"""Return list of supported options like voice, emotions.""" """Return list of supported options like voice, emotions."""
return ["voice", "age"] return ["voice", "age"]
def get_tts_audio(self, message, language, options=None): def get_tts_audio(
self, message: str, language: str, options: dict[str, Any] | None = None
) -> TtsAudioType:
"""Load TTS from demo.""" """Load TTS from demo."""
filename = os.path.join(os.path.dirname(__file__), "tts.mp3") filename = os.path.join(os.path.dirname(__file__), "tts.mp3")
try: try:

View File

@ -51,21 +51,27 @@ class DemoWaterHeater(WaterHeaterEntity):
_attr_supported_features = SUPPORT_FLAGS_HEATER _attr_supported_features = SUPPORT_FLAGS_HEATER
def __init__( def __init__(
self, name, target_temperature, unit_of_measurement, away, current_operation self,
): name: str,
target_temperature: int,
unit_of_measurement: str,
away: bool,
current_operation: str,
) -> None:
"""Initialize the water_heater device.""" """Initialize the water_heater device."""
self._attr_name = name self._attr_name = name
if target_temperature is not None: if target_temperature is not None:
self._attr_supported_features = ( self._attr_supported_features = (
self.supported_features | WaterHeaterEntityFeature.TARGET_TEMPERATURE self._attr_supported_features
| WaterHeaterEntityFeature.TARGET_TEMPERATURE
) )
if away is not None: if away is not None:
self._attr_supported_features = ( self._attr_supported_features = (
self.supported_features | WaterHeaterEntityFeature.AWAY_MODE self._attr_supported_features | WaterHeaterEntityFeature.AWAY_MODE
) )
if current_operation is not None: if current_operation is not None:
self._attr_supported_features = ( self._attr_supported_features = (
self.supported_features | WaterHeaterEntityFeature.OPERATION_MODE self._attr_supported_features | WaterHeaterEntityFeature.OPERATION_MODE
) )
self._attr_target_temperature = target_temperature self._attr_target_temperature = target_temperature
self._attr_temperature_unit = unit_of_measurement self._attr_temperature_unit = unit_of_measurement