Code maintenance for HomematicIP Cloud (#28980)

* Add more type hints to Homematic IP Cloud

* Rename device to entity in async_setup_entry
pull/29058/head
SukramJ 2019-11-25 14:17:14 +01:00 committed by Pascal Vizeli
parent b927f40f00
commit 1043712c54
12 changed files with 190 additions and 163 deletions

View File

@ -1,8 +1,10 @@
"""Support for HomematicIP Cloud devices."""
import logging
from pathlib import Path
from typing import Optional
from homematicip.aio.group import AsyncHeatingGroup
from homematicip.aio.home import AsyncHome
from homematicip.base.helpers import handle_config
import voluptuous as vol
@ -135,7 +137,7 @@ async def async_setup(hass: HomeAssistantType, config: ConfigType) -> bool:
)
)
async def _async_activate_eco_mode_with_duration(service):
async def _async_activate_eco_mode_with_duration(service) -> None:
"""Service to activate eco mode with duration."""
duration = service.data[ATTR_DURATION]
hapid = service.data.get(ATTR_ACCESSPOINT_ID)
@ -155,7 +157,7 @@ async def async_setup(hass: HomeAssistantType, config: ConfigType) -> bool:
schema=SCHEMA_ACTIVATE_ECO_MODE_WITH_DURATION,
)
async def _async_activate_eco_mode_with_period(service):
async def _async_activate_eco_mode_with_period(service) -> None:
"""Service to activate eco mode with period."""
endtime = service.data[ATTR_ENDTIME]
hapid = service.data.get(ATTR_ACCESSPOINT_ID)
@ -175,7 +177,7 @@ async def async_setup(hass: HomeAssistantType, config: ConfigType) -> bool:
schema=SCHEMA_ACTIVATE_ECO_MODE_WITH_PERIOD,
)
async def _async_activate_vacation(service):
async def _async_activate_vacation(service) -> None:
"""Service to activate vacation."""
endtime = service.data[ATTR_ENDTIME]
temperature = service.data[ATTR_TEMPERATURE]
@ -196,7 +198,7 @@ async def async_setup(hass: HomeAssistantType, config: ConfigType) -> bool:
schema=SCHEMA_ACTIVATE_VACATION,
)
async def _async_deactivate_eco_mode(service):
async def _async_deactivate_eco_mode(service) -> None:
"""Service to deactivate eco mode."""
hapid = service.data.get(ATTR_ACCESSPOINT_ID)
@ -215,7 +217,7 @@ async def async_setup(hass: HomeAssistantType, config: ConfigType) -> bool:
schema=SCHEMA_DEACTIVATE_ECO_MODE,
)
async def _async_deactivate_vacation(service):
async def _async_deactivate_vacation(service) -> None:
"""Service to deactivate vacation."""
hapid = service.data.get(ATTR_ACCESSPOINT_ID)
@ -234,7 +236,7 @@ async def async_setup(hass: HomeAssistantType, config: ConfigType) -> bool:
schema=SCHEMA_DEACTIVATE_VACATION,
)
async def _set_active_climate_profile(service):
async def _set_active_climate_profile(service) -> None:
"""Service to set the active climate profile."""
entity_id_list = service.data[ATTR_ENTITY_ID]
climate_profile_index = service.data[ATTR_CLIMATE_PROFILE_INDEX] - 1
@ -257,7 +259,7 @@ async def async_setup(hass: HomeAssistantType, config: ConfigType) -> bool:
schema=SCHEMA_SET_ACTIVE_CLIMATE_PROFILE,
)
async def _async_dump_hap_config(service):
async def _async_dump_hap_config(service) -> None:
"""Service to dump the configuration of a Homematic IP Access Point."""
config_path = (
service.data.get(ATTR_CONFIG_OUTPUT_PATH) or hass.config.config_dir
@ -287,7 +289,7 @@ async def async_setup(hass: HomeAssistantType, config: ConfigType) -> bool:
schema=SCHEMA_DUMP_HAP_CONFIG,
)
def _get_home(hapid: str):
def _get_home(hapid: str) -> Optional[AsyncHome]:
"""Return a HmIP home."""
hap = hass.data[DOMAIN].get(hapid)
if hap:
@ -324,7 +326,7 @@ async def async_setup_entry(hass: HomeAssistantType, entry: ConfigEntry) -> bool
return True
async def async_unload_entry(hass, entry):
async def async_unload_entry(hass: HomeAssistantType, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
hap = hass.data[DOMAIN].pop(entry.data[HMIPC_HAPID])
return await hap.async_reset()

View File

@ -1,5 +1,6 @@
"""Support for HomematicIP Cloud alarm control panel."""
import logging
from typing import Any, Dict
from homematicip.functionalHomes import SecurityAndAlarmHome
@ -21,7 +22,9 @@ _LOGGER = logging.getLogger(__name__)
CONST_ALARM_CONTROL_PANEL_NAME = "HmIP Alarm Control Panel"
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
async def async_setup_platform(
hass, config, async_add_entities, discovery_info=None
) -> None:
"""Set up the HomematicIP Cloud alarm control devices."""
pass
@ -40,9 +43,10 @@ class HomematicipAlarmControlPanel(AlarmControlPanel):
def __init__(self, hap: HomematicipHAP) -> None:
"""Initialize the alarm control panel."""
self._home = hap.home
_LOGGER.info("Setting up %s", self.name)
@property
def device_info(self):
def device_info(self) -> Dict[str, Any]:
"""Return device specific attributes."""
return {
"identifiers": {(HMIPC_DOMAIN, f"ACP {self._home.id}")},
@ -70,26 +74,26 @@ class HomematicipAlarmControlPanel(AlarmControlPanel):
return STATE_ALARM_DISARMED
@property
def _security_and_alarm(self):
def _security_and_alarm(self) -> SecurityAndAlarmHome:
return self._home.get_functionalHome(SecurityAndAlarmHome)
async def async_alarm_disarm(self, code=None):
async def async_alarm_disarm(self, code=None) -> None:
"""Send disarm command."""
await self._home.set_security_zones_activation(False, False)
async def async_alarm_arm_home(self, code=None):
async def async_alarm_arm_home(self, code=None) -> None:
"""Send arm home command."""
await self._home.set_security_zones_activation(False, True)
async def async_alarm_arm_away(self, code=None):
async def async_alarm_arm_away(self, code=None) -> None:
"""Send arm away command."""
await self._home.set_security_zones_activation(True, True)
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Register callbacks."""
self._home.on_update(self._async_device_changed)
def _async_device_changed(self, *args, **kwargs):
def _async_device_changed(self, *args, **kwargs) -> None:
"""Handle device state changes."""
_LOGGER.debug("Event %s (%s)", self.name, CONST_ALARM_CONTROL_PANEL_NAME)
self.async_schedule_update_ha_state()

View File

@ -1,5 +1,6 @@
"""Support for HomematicIP Cloud binary sensor."""
import logging
from typing import Any, Dict
from homematicip.aio.device import (
AsyncAccelerationSensor,
@ -72,7 +73,9 @@ SAM_DEVICE_ATTRIBUTES = {
}
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
async def async_setup_platform(
hass, config, async_add_entities, discovery_info=None
) -> None:
"""Set up the HomematicIP Cloud binary sensor devices."""
pass
@ -82,17 +85,17 @@ async def async_setup_entry(
) -> None:
"""Set up the HomematicIP Cloud binary sensor from a config entry."""
hap = hass.data[HMIPC_DOMAIN][config_entry.data[HMIPC_HAPID]]
devices = []
entities = []
for device in hap.home.devices:
if isinstance(device, AsyncAccelerationSensor):
devices.append(HomematicipAccelerationSensor(hap, device))
entities.append(HomematicipAccelerationSensor(hap, device))
if isinstance(device, (AsyncContactInterface, AsyncFullFlushContactInterface)):
devices.append(HomematicipContactInterface(hap, device))
entities.append(HomematicipContactInterface(hap, device))
if isinstance(
device,
(AsyncShutterContact, AsyncShutterContactMagnetic, AsyncRotaryHandleSensor),
):
devices.append(HomematicipShutterContact(hap, device))
entities.append(HomematicipShutterContact(hap, device))
if isinstance(
device,
(
@ -101,31 +104,31 @@ async def async_setup_entry(
AsyncMotionDetectorPushButton,
),
):
devices.append(HomematicipMotionDetector(hap, device))
entities.append(HomematicipMotionDetector(hap, device))
if isinstance(device, AsyncPresenceDetectorIndoor):
devices.append(HomematicipPresenceDetector(hap, device))
entities.append(HomematicipPresenceDetector(hap, device))
if isinstance(device, AsyncSmokeDetector):
devices.append(HomematicipSmokeDetector(hap, device))
entities.append(HomematicipSmokeDetector(hap, device))
if isinstance(device, AsyncWaterSensor):
devices.append(HomematicipWaterDetector(hap, device))
entities.append(HomematicipWaterDetector(hap, device))
if isinstance(device, (AsyncWeatherSensorPlus, AsyncWeatherSensorPro)):
devices.append(HomematicipRainSensor(hap, device))
entities.append(HomematicipRainSensor(hap, device))
if isinstance(
device, (AsyncWeatherSensor, AsyncWeatherSensorPlus, AsyncWeatherSensorPro)
):
devices.append(HomematicipStormSensor(hap, device))
devices.append(HomematicipSunshineSensor(hap, device))
entities.append(HomematicipStormSensor(hap, device))
entities.append(HomematicipSunshineSensor(hap, device))
if isinstance(device, AsyncDevice) and device.lowBat is not None:
devices.append(HomematicipBatterySensor(hap, device))
entities.append(HomematicipBatterySensor(hap, device))
for group in hap.home.groups:
if isinstance(group, AsyncSecurityGroup):
devices.append(HomematicipSecuritySensorGroup(hap, group))
entities.append(HomematicipSecuritySensorGroup(hap, group))
elif isinstance(group, AsyncSecurityZoneGroup):
devices.append(HomematicipSecurityZoneSensorGroup(hap, group))
entities.append(HomematicipSecurityZoneSensorGroup(hap, group))
if devices:
async_add_entities(devices)
if entities:
async_add_entities(entities)
class HomematicipAccelerationSensor(HomematicipGenericDevice, BinarySensorDevice):
@ -142,7 +145,7 @@ class HomematicipAccelerationSensor(HomematicipGenericDevice, BinarySensorDevice
return self._device.accelerationSensorTriggered
@property
def device_state_attributes(self):
def device_state_attributes(self) -> Dict[str, Any]:
"""Return the state attributes of the acceleration sensor."""
state_attr = super().device_state_attributes
@ -296,7 +299,7 @@ class HomematicipSunshineSensor(HomematicipGenericDevice, BinarySensorDevice):
return self._device.sunshine
@property
def device_state_attributes(self):
def device_state_attributes(self) -> Dict[str, Any]:
"""Return the state attributes of the illuminance sensor."""
state_attr = super().device_state_attributes
@ -346,7 +349,7 @@ class HomematicipSecurityZoneSensorGroup(HomematicipGenericDevice, BinarySensorD
return True
@property
def device_state_attributes(self):
def device_state_attributes(self) -> Dict[str, Any]:
"""Return the state attributes of the security zone group."""
state_attr = super().device_state_attributes
@ -390,7 +393,7 @@ class HomematicipSecuritySensorGroup(
super().__init__(hap, device, "Sensors")
@property
def device_state_attributes(self):
def device_state_attributes(self) -> Dict[str, Any]:
"""Return the state attributes of the security group."""
state_attr = super().device_state_attributes

View File

@ -1,6 +1,6 @@
"""Support for HomematicIP Cloud climate devices."""
import logging
from typing import Awaitable
from typing import Any, Dict, List, Optional, Union
from homematicip.aio.device import AsyncHeatingThermostat, AsyncHeatingThermostatCompact
from homematicip.aio.group import AsyncHeatingGroup
@ -41,7 +41,9 @@ HMIP_MANUAL_CM = "MANUAL"
HMIP_ECO_CM = "ECO"
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
async def async_setup_platform(
hass, config, async_add_entities, discovery_info=None
) -> None:
"""Set up the HomematicIP Cloud climate devices."""
pass
@ -51,13 +53,13 @@ async def async_setup_entry(
) -> None:
"""Set up the HomematicIP climate from a config entry."""
hap = hass.data[HMIPC_DOMAIN][config_entry.data[HMIPC_HAPID]]
devices = []
entities = []
for device in hap.home.groups:
if isinstance(device, AsyncHeatingGroup):
devices.append(HomematicipHeatingGroup(hap, device))
entities.append(HomematicipHeatingGroup(hap, device))
if devices:
async_add_entities(devices)
if entities:
async_add_entities(entities)
class HomematicipHeatingGroup(HomematicipGenericDevice, ClimateDevice):
@ -74,10 +76,10 @@ class HomematicipHeatingGroup(HomematicipGenericDevice, ClimateDevice):
super().__init__(hap, device)
self._simple_heating = None
if device.actualTemperature is None:
self._simple_heating = self._get_first_radiator_thermostat()
self._simple_heating = self._first_radiator_thermostat
@property
def device_info(self):
def device_info(self) -> Dict[str, Any]:
"""Return device specific attributes."""
return {
"identifiers": {(HMIPC_DOMAIN, self._device.id)},
@ -127,7 +129,7 @@ class HomematicipHeatingGroup(HomematicipGenericDevice, ClimateDevice):
return HVAC_MODE_AUTO
@property
def hvac_modes(self):
def hvac_modes(self) -> List[str]:
"""Return the list of available hvac operation modes."""
if self._disabled_by_cooling_mode and not self._has_switch:
return [HVAC_MODE_OFF]
@ -139,7 +141,7 @@ class HomematicipHeatingGroup(HomematicipGenericDevice, ClimateDevice):
)
@property
def preset_mode(self):
def preset_mode(self) -> Optional[str]:
"""Return the current preset mode."""
if self._device.boostMode:
return PRESET_BOOST
@ -162,7 +164,7 @@ class HomematicipHeatingGroup(HomematicipGenericDevice, ClimateDevice):
)
@property
def preset_modes(self):
def preset_modes(self) -> List[str]:
"""Return a list of available preset modes incl. hmip profiles."""
# Boost is only available if a radiator thermostat is in the room,
# and heat mode is enabled.
@ -190,7 +192,7 @@ class HomematicipHeatingGroup(HomematicipGenericDevice, ClimateDevice):
"""Return the maximum temperature."""
return self._device.maxTemperature
async def async_set_temperature(self, **kwargs):
async def async_set_temperature(self, **kwargs) -> None:
"""Set new target temperature."""
temperature = kwargs.get(ATTR_TEMPERATURE)
if temperature is None:
@ -199,7 +201,7 @@ class HomematicipHeatingGroup(HomematicipGenericDevice, ClimateDevice):
if self.min_temp <= temperature <= self.max_temp:
await self._device.set_point_temperature(temperature)
async def async_set_hvac_mode(self, hvac_mode: str) -> Awaitable[None]:
async def async_set_hvac_mode(self, hvac_mode: str) -> None:
"""Set new target hvac mode."""
if hvac_mode not in self.hvac_modes:
return
@ -209,7 +211,7 @@ class HomematicipHeatingGroup(HomematicipGenericDevice, ClimateDevice):
else:
await self._device.set_control_mode(HMIP_MANUAL_CM)
async def async_set_preset_mode(self, preset_mode: str) -> Awaitable[None]:
async def async_set_preset_mode(self, preset_mode: str) -> None:
"""Set new preset mode."""
if preset_mode not in self.preset_modes:
return
@ -225,7 +227,7 @@ class HomematicipHeatingGroup(HomematicipGenericDevice, ClimateDevice):
await self._device.set_active_profile(profile_idx)
@property
def device_state_attributes(self):
def device_state_attributes(self) -> Dict[str, Any]:
"""Return the state attributes of the access point."""
state_attr = super().device_state_attributes
@ -242,12 +244,12 @@ class HomematicipHeatingGroup(HomematicipGenericDevice, ClimateDevice):
return state_attr
@property
def _indoor_climate(self):
def _indoor_climate(self) -> IndoorClimateHome:
"""Return the hmip indoor climate functional home of this group."""
return self._home.get_functionalHome(IndoorClimateHome)
@property
def _device_profiles(self):
def _device_profiles(self) -> List[str]:
"""Return the relevant profiles."""
return [
profile
@ -258,11 +260,11 @@ class HomematicipHeatingGroup(HomematicipGenericDevice, ClimateDevice):
]
@property
def _device_profile_names(self):
def _device_profile_names(self) -> List[str]:
"""Return a collection of profile names."""
return [profile.name for profile in self._device_profiles]
def _get_profile_idx_by_name(self, profile_name):
def _get_profile_idx_by_name(self, profile_name: str) -> int:
"""Return a profile index by name."""
relevant_index = self._relevant_profile_group
index_name = [
@ -274,19 +276,19 @@ class HomematicipHeatingGroup(HomematicipGenericDevice, ClimateDevice):
return relevant_index[index_name[0]]
@property
def _heat_mode_enabled(self):
def _heat_mode_enabled(self) -> bool:
"""Return, if heating mode is enabled."""
return not self._device.cooling
@property
def _disabled_by_cooling_mode(self):
def _disabled_by_cooling_mode(self) -> bool:
"""Return, if group is disabled by the cooling mode."""
return self._device.cooling and (
self._device.coolingIgnored or not self._device.coolingAllowed
)
@property
def _relevant_profile_group(self):
def _relevant_profile_group(self) -> List[str]:
"""Return the relevant profile groups."""
if self._disabled_by_cooling_mode:
return []
@ -305,9 +307,12 @@ class HomematicipHeatingGroup(HomematicipGenericDevice, ClimateDevice):
@property
def _has_radiator_thermostat(self) -> bool:
"""Return, if a radiator thermostat is in the hmip heating group."""
return bool(self._get_first_radiator_thermostat())
return bool(self._first_radiator_thermostat)
def _get_first_radiator_thermostat(self):
@property
def _first_radiator_thermostat(
self,
) -> Optional[Union[AsyncHeatingThermostat, AsyncHeatingThermostatCompact]]:
"""Return the first radiator thermostat from the hmip heating group."""
for device in self._device.devices:
if isinstance(

View File

@ -1,5 +1,5 @@
"""Config flow to configure the HomematicIP Cloud component."""
from typing import Set
from typing import Any, Dict, Set
import voluptuous as vol
@ -34,15 +34,15 @@ class HomematicipCloudFlowHandler(config_entries.ConfigFlow):
VERSION = 1
CONNECTION_CLASS = config_entries.CONN_CLASS_CLOUD_PUSH
def __init__(self):
def __init__(self) -> None:
"""Initialize HomematicIP Cloud config flow."""
self.auth = None
async def async_step_user(self, user_input=None):
async def async_step_user(self, user_input=None) -> Dict[str, Any]:
"""Handle a flow initialized by the user."""
return await self.async_step_init(user_input)
async def async_step_init(self, user_input=None):
async def async_step_init(self, user_input=None) -> Dict[str, Any]:
"""Handle a flow start."""
errors = {}
@ -69,7 +69,7 @@ class HomematicipCloudFlowHandler(config_entries.ConfigFlow):
errors=errors,
)
async def async_step_link(self, user_input=None):
async def async_step_link(self, user_input=None) -> Dict[str, Any]:
"""Attempt to link with the HomematicIP Cloud access point."""
errors = {}
@ -91,7 +91,7 @@ class HomematicipCloudFlowHandler(config_entries.ConfigFlow):
return self.async_show_form(step_id="link", errors=errors)
async def async_step_import(self, import_info):
async def async_step_import(self, import_info) -> Dict[str, Any]:
"""Import a new access point as a config entry."""
hapid = import_info[HMIPC_HAPID]
authtoken = import_info[HMIPC_AUTHTOKEN]

View File

@ -22,7 +22,9 @@ HMIP_SLATS_OPEN = 0
HMIP_SLATS_CLOSED = 1
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
async def async_setup_platform(
hass, config, async_add_entities, discovery_info=None
) -> None:
"""Set up the HomematicIP Cloud cover devices."""
pass
@ -32,15 +34,15 @@ async def async_setup_entry(
) -> None:
"""Set up the HomematicIP cover from a config entry."""
hap = hass.data[HMIPC_DOMAIN][config_entry.data[HMIPC_HAPID]]
devices = []
entities = []
for device in hap.home.devices:
if isinstance(device, AsyncFullFlushBlind):
devices.append(HomematicipCoverSlats(hap, device))
entities.append(HomematicipCoverSlats(hap, device))
elif isinstance(device, AsyncFullFlushShutter):
devices.append(HomematicipCoverShutter(hap, device))
entities.append(HomematicipCoverShutter(hap, device))
if devices:
async_add_entities(devices)
if entities:
async_add_entities(entities)
class HomematicipCoverShutter(HomematicipGenericDevice, CoverDevice):
@ -51,7 +53,7 @@ class HomematicipCoverShutter(HomematicipGenericDevice, CoverDevice):
"""Return current position of cover."""
return int((1 - self._device.shutterLevel) * 100)
async def async_set_cover_position(self, **kwargs):
async def async_set_cover_position(self, **kwargs) -> None:
"""Move the cover to a specific position."""
position = kwargs[ATTR_POSITION]
# HmIP cover is closed:1 -> open:0
@ -65,15 +67,15 @@ class HomematicipCoverShutter(HomematicipGenericDevice, CoverDevice):
return self._device.shutterLevel == HMIP_COVER_CLOSED
return None
async def async_open_cover(self, **kwargs):
async def async_open_cover(self, **kwargs) -> None:
"""Open the cover."""
await self._device.set_shutter_level(HMIP_COVER_OPEN)
async def async_close_cover(self, **kwargs):
async def async_close_cover(self, **kwargs) -> None:
"""Close the cover."""
await self._device.set_shutter_level(HMIP_COVER_CLOSED)
async def async_stop_cover(self, **kwargs):
async def async_stop_cover(self, **kwargs) -> None:
"""Stop the device if in motion."""
await self._device.set_shutter_stop()
@ -86,21 +88,21 @@ class HomematicipCoverSlats(HomematicipCoverShutter, CoverDevice):
"""Return current tilt position of cover."""
return int((1 - self._device.slatsLevel) * 100)
async def async_set_cover_tilt_position(self, **kwargs):
async def async_set_cover_tilt_position(self, **kwargs) -> None:
"""Move the cover to a specific tilt position."""
position = kwargs[ATTR_TILT_POSITION]
# HmIP slats is closed:1 -> open:0
level = 1 - position / 100.0
await self._device.set_slats_level(level)
async def async_open_cover_tilt(self, **kwargs):
async def async_open_cover_tilt(self, **kwargs) -> None:
"""Open the slats."""
await self._device.set_slats_level(HMIP_SLATS_OPEN)
async def async_close_cover_tilt(self, **kwargs):
async def async_close_cover_tilt(self, **kwargs) -> None:
"""Close the slats."""
await self._device.set_slats_level(HMIP_SLATS_CLOSED)
async def async_stop_cover_tilt(self, **kwargs):
async def async_stop_cover_tilt(self, **kwargs) -> None:
"""Stop the device if in motion."""
await self._device.set_shutter_stop()

View File

@ -1,6 +1,6 @@
"""Generic device for the HomematicIP Cloud component."""
import logging
from typing import Optional
from typing import Any, Dict, Optional
from homematicip.aio.device import AsyncDevice
from homematicip.aio.group import AsyncGroup
@ -79,7 +79,7 @@ class HomematicipGenericDevice(Entity):
_LOGGER.info("Setting up %s (%s)", self.name, self._device.modelType)
@property
def device_info(self):
def device_info(self) -> Dict[str, Any]:
"""Return device specific attributes."""
# Only physical devices should be HA devices.
if isinstance(self._device, AsyncDevice):
@ -96,14 +96,14 @@ class HomematicipGenericDevice(Entity):
}
return None
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Register callbacks."""
self._hap.hmip_device_by_entity_id[self.entity_id] = self._device
self._device.on_update(self._async_device_changed)
self._device.on_remove(self._async_device_removed)
@callback
def _async_device_changed(self, *args, **kwargs):
def _async_device_changed(self, *args, **kwargs) -> None:
"""Handle device state changes."""
# Don't update disabled entities
if self.enabled:
@ -152,7 +152,7 @@ class HomematicipGenericDevice(Entity):
entity_registry.async_remove(entity_id)
@callback
def _async_device_removed(self, *args, **kwargs):
def _async_device_removed(self, *args, **kwargs) -> None:
"""Handle hmip device removal."""
# Set marker showing that the HmIP device hase been removed.
self.hmip_device_removed = True
@ -193,7 +193,7 @@ class HomematicipGenericDevice(Entity):
return None
@property
def device_state_attributes(self):
def device_state_attributes(self) -> Dict[str, Any]:
"""Return the state attributes of the generic device."""
state_attr = {}

View File

@ -22,13 +22,13 @@ _LOGGER = logging.getLogger(__name__)
class HomematicipAuth:
"""Manages HomematicIP client registration."""
def __init__(self, hass, config):
def __init__(self, hass, config) -> None:
"""Initialize HomematicIP Cloud client registration."""
self.hass = hass
self.config = config
self.auth = None
async def async_setup(self):
async def async_setup(self) -> bool:
"""Connect to HomematicIP for registration."""
try:
self.auth = await self.get_auth(
@ -38,7 +38,7 @@ class HomematicipAuth:
except HmipcConnectionError:
return False
async def async_checkbutton(self):
async def async_checkbutton(self) -> bool:
"""Check blue butten has been pressed."""
try:
return await self.auth.isRequestAcknowledged()
@ -82,7 +82,7 @@ class HomematicipHAP:
self._accesspoint_connected = True
self.hmip_device_by_entity_id = {}
async def async_setup(self, tries: int = 0):
async def async_setup(self, tries: int = 0) -> bool:
"""Initialize connection."""
try:
self.home = await self.get_hap(
@ -108,7 +108,7 @@ class HomematicipHAP:
return True
@callback
def async_update(self, *args, **kwargs):
def async_update(self, *args, **kwargs) -> None:
"""Async update the home device.
Triggered when the HMIP HOME_CHANGED event has fired.
@ -141,23 +141,23 @@ class HomematicipHAP:
self.home.update_home_only(args[0])
@callback
def async_create_entity(self, *args, **kwargs):
def async_create_entity(self, *args, **kwargs) -> None:
"""Create a device or a group."""
is_device = EventType(kwargs["event_type"]) == EventType.DEVICE_ADDED
self.hass.async_create_task(self.async_create_entity_lazy(is_device))
async def async_create_entity_lazy(self, is_device=True):
async def async_create_entity_lazy(self, is_device=True) -> None:
"""Delay entity creation to allow the user to enter a device name."""
if is_device:
await asyncio.sleep(30)
await self.hass.config_entries.async_reload(self.config_entry.entry_id)
async def get_state(self):
async def get_state(self) -> None:
"""Update HMIP state and tell Home Assistant."""
await self.home.get_current_state()
self.update_all()
def get_state_finished(self, future):
def get_state_finished(self, future) -> None:
"""Execute when get_state coroutine has finished."""
try:
future.result()
@ -167,18 +167,18 @@ class HomematicipHAP:
_LOGGER.error("Updating state after HMIP access point reconnect failed")
self.hass.async_create_task(self.home.disable_events())
def set_all_to_unavailable(self):
def set_all_to_unavailable(self) -> None:
"""Set all devices to unavailable and tell Home Assistant."""
for device in self.home.devices:
device.unreach = True
self.update_all()
def update_all(self):
def update_all(self) -> None:
"""Signal all devices to update their state."""
for device in self.home.devices:
device.fire_update_event()
async def async_connect(self):
async def async_connect(self) -> None:
"""Start WebSocket connection."""
tries = 0
while True:
@ -210,7 +210,7 @@ class HomematicipHAP:
except asyncio.CancelledError:
break
async def async_reset(self):
async def async_reset(self) -> bool:
"""Close the websocket connection."""
self._ws_close_requested = True
if self._retry_task is not None:

View File

@ -1,5 +1,6 @@
"""Support for HomematicIP Cloud lights."""
import logging
from typing import Any, Dict
from homematicip.aio.device import (
AsyncBrandDimmer,
@ -33,7 +34,9 @@ ATTR_TODAY_ENERGY_KWH = "today_energy_kwh"
ATTR_CURRENT_POWER_W = "current_power_w"
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
async def async_setup_platform(
hass, config, async_add_entities, discovery_info=None
) -> None:
"""Old way of setting up HomematicIP Cloud lights."""
pass
@ -43,16 +46,16 @@ async def async_setup_entry(
) -> None:
"""Set up the HomematicIP Cloud lights from a config entry."""
hap = hass.data[HMIPC_DOMAIN][config_entry.data[HMIPC_HAPID]]
devices = []
entities = []
for device in hap.home.devices:
if isinstance(device, AsyncBrandSwitchMeasuring):
devices.append(HomematicipLightMeasuring(hap, device))
entities.append(HomematicipLightMeasuring(hap, device))
elif isinstance(device, AsyncBrandSwitchNotificationLight):
devices.append(HomematicipLight(hap, device))
devices.append(
entities.append(HomematicipLight(hap, device))
entities.append(
HomematicipNotificationLight(hap, device, device.topLightChannelIndex)
)
devices.append(
entities.append(
HomematicipNotificationLight(
hap, device, device.bottomLightChannelIndex
)
@ -61,10 +64,10 @@ async def async_setup_entry(
device,
(AsyncDimmer, AsyncPluggableDimmer, AsyncBrandDimmer, AsyncFullFlushDimmer),
):
devices.append(HomematicipDimmer(hap, device))
entities.append(HomematicipDimmer(hap, device))
if devices:
async_add_entities(devices)
if entities:
async_add_entities(entities)
class HomematicipLight(HomematicipGenericDevice, Light):
@ -79,11 +82,11 @@ class HomematicipLight(HomematicipGenericDevice, Light):
"""Return true if device is on."""
return self._device.on
async def async_turn_on(self, **kwargs):
async def async_turn_on(self, **kwargs) -> None:
"""Turn the device on."""
await self._device.turn_on()
async def async_turn_off(self, **kwargs):
async def async_turn_off(self, **kwargs) -> None:
"""Turn the device off."""
await self._device.turn_off()
@ -92,7 +95,7 @@ class HomematicipLightMeasuring(HomematicipLight):
"""Representation of a HomematicIP Cloud measuring light device."""
@property
def device_state_attributes(self):
def device_state_attributes(self) -> Dict[str, Any]:
"""Return the state attributes of the generic device."""
state_attr = super().device_state_attributes
@ -127,14 +130,14 @@ class HomematicipDimmer(HomematicipGenericDevice, Light):
"""Flag supported features."""
return SUPPORT_BRIGHTNESS
async def async_turn_on(self, **kwargs):
async def async_turn_on(self, **kwargs) -> None:
"""Turn the light on."""
if ATTR_BRIGHTNESS in kwargs:
await self._device.set_dim_level(kwargs[ATTR_BRIGHTNESS] / 255.0)
else:
await self._device.set_dim_level(1)
async def async_turn_off(self, **kwargs):
async def async_turn_off(self, **kwargs) -> None:
"""Turn the light off."""
await self._device.set_dim_level(0)
@ -184,7 +187,7 @@ class HomematicipNotificationLight(HomematicipGenericDevice, Light):
return self._color_switcher.get(simple_rgb_color, [0.0, 0.0])
@property
def device_state_attributes(self):
def device_state_attributes(self) -> Dict[str, Any]:
"""Return the state attributes of the generic device."""
state_attr = super().device_state_attributes
@ -208,7 +211,7 @@ class HomematicipNotificationLight(HomematicipGenericDevice, Light):
"""Return a unique ID."""
return f"{self.__class__.__name__}_{self.post}_{self._device.id}"
async def async_turn_on(self, **kwargs):
async def async_turn_on(self, **kwargs) -> None:
"""Turn the light on."""
# Use hs_color from kwargs,
# if not applicable use current hs_color.
@ -236,7 +239,7 @@ class HomematicipNotificationLight(HomematicipGenericDevice, Light):
rampTime=transition,
)
async def async_turn_off(self, **kwargs):
async def async_turn_off(self, **kwargs) -> None:
"""Turn the light off."""
simple_rgb_color = self._func_channel.simpleRGBColorState
transition = kwargs.get(ATTR_TRANSITION, 0.5)
@ -250,7 +253,7 @@ class HomematicipNotificationLight(HomematicipGenericDevice, Light):
)
def _convert_color(color) -> RGBColorState:
def _convert_color(color: tuple) -> RGBColorState:
"""
Convert the given color to the reduced RGBColorState color.

View File

@ -1,5 +1,6 @@
"""Support for HomematicIP Cloud sensors."""
import logging
from typing import Any, Dict
from homematicip.aio.device import (
AsyncBrandSwitchMeasuring,
@ -55,7 +56,9 @@ ILLUMINATION_DEVICE_ATTRIBUTES = {
}
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
async def async_setup_platform(
hass, config, async_add_entities, discovery_info=None
) -> None:
"""Set up the HomematicIP Cloud sensors devices."""
pass
@ -65,11 +68,11 @@ async def async_setup_entry(
) -> None:
"""Set up the HomematicIP Cloud sensors from a config entry."""
hap = hass.data[HMIPC_DOMAIN][config_entry.data[HMIPC_HAPID]]
devices = [HomematicipAccesspointStatus(hap)]
entities = [HomematicipAccesspointStatus(hap)]
for device in hap.home.devices:
if isinstance(device, (AsyncHeatingThermostat, AsyncHeatingThermostatCompact)):
devices.append(HomematicipHeatingThermostat(hap, device))
devices.append(HomematicipTemperatureSensor(hap, device))
entities.append(HomematicipHeatingThermostat(hap, device))
entities.append(HomematicipTemperatureSensor(hap, device))
if isinstance(
device,
(
@ -81,8 +84,8 @@ async def async_setup_entry(
AsyncWeatherSensorPro,
),
):
devices.append(HomematicipTemperatureSensor(hap, device))
devices.append(HomematicipHumiditySensor(hap, device))
entities.append(HomematicipTemperatureSensor(hap, device))
entities.append(HomematicipHumiditySensor(hap, device))
if isinstance(
device,
(
@ -96,7 +99,7 @@ async def async_setup_entry(
AsyncWeatherSensorPro,
),
):
devices.append(HomematicipIlluminanceSensor(hap, device))
entities.append(HomematicipIlluminanceSensor(hap, device))
if isinstance(
device,
(
@ -105,18 +108,18 @@ async def async_setup_entry(
AsyncFullFlushSwitchMeasuring,
),
):
devices.append(HomematicipPowerSensor(hap, device))
entities.append(HomematicipPowerSensor(hap, device))
if isinstance(
device, (AsyncWeatherSensor, AsyncWeatherSensorPlus, AsyncWeatherSensorPro)
):
devices.append(HomematicipWindspeedSensor(hap, device))
entities.append(HomematicipWindspeedSensor(hap, device))
if isinstance(device, (AsyncWeatherSensorPlus, AsyncWeatherSensorPro)):
devices.append(HomematicipTodayRainSensor(hap, device))
entities.append(HomematicipTodayRainSensor(hap, device))
if isinstance(device, AsyncPassageDetector):
devices.append(HomematicipPassageDetectorDeltaCounter(hap, device))
entities.append(HomematicipPassageDetectorDeltaCounter(hap, device))
if devices:
async_add_entities(devices)
if entities:
async_add_entities(entities)
class HomematicipAccesspointStatus(HomematicipGenericDevice):
@ -127,7 +130,7 @@ class HomematicipAccesspointStatus(HomematicipGenericDevice):
super().__init__(hap, hap.home)
@property
def device_info(self):
def device_info(self) -> Dict[str, Any]:
"""Return device specific attributes."""
# Adds a sensor to the existing HAP device
return {
@ -158,7 +161,7 @@ class HomematicipAccesspointStatus(HomematicipGenericDevice):
return "%"
@property
def device_state_attributes(self):
def device_state_attributes(self) -> Dict[str, Any]:
"""Return the state attributes of the access point."""
state_attr = super().device_state_attributes
@ -246,7 +249,7 @@ class HomematicipTemperatureSensor(HomematicipGenericDevice):
return TEMP_CELSIUS
@property
def device_state_attributes(self):
def device_state_attributes(self) -> Dict[str, Any]:
"""Return the state attributes of the windspeed sensor."""
state_attr = super().device_state_attributes
@ -283,7 +286,7 @@ class HomematicipIlluminanceSensor(HomematicipGenericDevice):
return "lx"
@property
def device_state_attributes(self):
def device_state_attributes(self) -> Dict[str, Any]:
"""Return the state attributes of the wind speed sensor."""
state_attr = super().device_state_attributes
@ -336,7 +339,7 @@ class HomematicipWindspeedSensor(HomematicipGenericDevice):
return "km/h"
@property
def device_state_attributes(self):
def device_state_attributes(self) -> Dict[str, Any]:
"""Return the state attributes of the wind speed sensor."""
state_attr = super().device_state_attributes
@ -378,7 +381,7 @@ class HomematicipPassageDetectorDeltaCounter(HomematicipGenericDevice):
return self._device.leftRightCounterDelta
@property
def device_state_attributes(self):
def device_state_attributes(self) -> Dict[str, Any]:
"""Return the state attributes of the delta counter."""
state_attr = super().device_state_attributes

View File

@ -1,5 +1,6 @@
"""Support for HomematicIP Cloud switches."""
import logging
from typing import Any, Dict
from homematicip.aio.device import (
AsyncBrandSwitchMeasuring,
@ -24,7 +25,9 @@ from .hap import HomematicipHAP
_LOGGER = logging.getLogger(__name__)
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
async def async_setup_platform(
hass, config, async_add_entities, discovery_info=None
) -> None:
"""Set up the HomematicIP Cloud switch devices."""
pass
@ -34,7 +37,7 @@ async def async_setup_entry(
) -> None:
"""Set up the HomematicIP switch from a config entry."""
hap = hass.data[HMIPC_DOMAIN][config_entry.data[HMIPC_HAPID]]
devices = []
entities = []
for device in hap.home.devices:
if isinstance(device, AsyncBrandSwitchMeasuring):
# BrandSwitchMeasuring inherits PlugableSwitchMeasuring
@ -44,27 +47,27 @@ async def async_setup_entry(
elif isinstance(
device, (AsyncPlugableSwitchMeasuring, AsyncFullFlushSwitchMeasuring)
):
devices.append(HomematicipSwitchMeasuring(hap, device))
entities.append(HomematicipSwitchMeasuring(hap, device))
elif isinstance(
device, (AsyncPlugableSwitch, AsyncPrintedCircuitBoardSwitchBattery)
):
devices.append(HomematicipSwitch(hap, device))
entities.append(HomematicipSwitch(hap, device))
elif isinstance(device, AsyncOpenCollector8Module):
for channel in range(1, 9):
devices.append(HomematicipMultiSwitch(hap, device, channel))
entities.append(HomematicipMultiSwitch(hap, device, channel))
elif isinstance(device, AsyncMultiIOBox):
for channel in range(1, 3):
devices.append(HomematicipMultiSwitch(hap, device, channel))
entities.append(HomematicipMultiSwitch(hap, device, channel))
elif isinstance(device, AsyncPrintedCircuitBoardSwitch2):
for channel in range(1, 3):
devices.append(HomematicipMultiSwitch(hap, device, channel))
entities.append(HomematicipMultiSwitch(hap, device, channel))
for group in hap.home.groups:
if isinstance(group, AsyncSwitchingGroup):
devices.append(HomematicipGroupSwitch(hap, group))
entities.append(HomematicipGroupSwitch(hap, group))
if devices:
async_add_entities(devices)
if entities:
async_add_entities(entities)
class HomematicipSwitch(HomematicipGenericDevice, SwitchDevice):
@ -79,11 +82,11 @@ class HomematicipSwitch(HomematicipGenericDevice, SwitchDevice):
"""Return true if device is on."""
return self._device.on
async def async_turn_on(self, **kwargs):
async def async_turn_on(self, **kwargs) -> None:
"""Turn the device on."""
await self._device.turn_on()
async def async_turn_off(self, **kwargs):
async def async_turn_off(self, **kwargs) -> None:
"""Turn the device off."""
await self._device.turn_off()
@ -111,7 +114,7 @@ class HomematicipGroupSwitch(HomematicipGenericDevice, SwitchDevice):
return True
@property
def device_state_attributes(self):
def device_state_attributes(self) -> Dict[str, Any]:
"""Return the state attributes of the switch-group."""
state_attr = super().device_state_attributes
@ -120,11 +123,11 @@ class HomematicipGroupSwitch(HomematicipGenericDevice, SwitchDevice):
return state_attr
async def async_turn_on(self, **kwargs):
async def async_turn_on(self, **kwargs) -> None:
"""Turn the group on."""
await self._device.turn_on()
async def async_turn_off(self, **kwargs):
async def async_turn_off(self, **kwargs) -> None:
"""Turn the group off."""
await self._device.turn_off()
@ -148,7 +151,7 @@ class HomematicipSwitchMeasuring(HomematicipSwitch):
class HomematicipMultiSwitch(HomematicipGenericDevice, SwitchDevice):
"""Representation of a HomematicIP Cloud multi switch device."""
def __init__(self, hap: HomematicipHAP, device, channel: int):
def __init__(self, hap: HomematicipHAP, device, channel: int) -> None:
"""Initialize the multi switch device."""
self.channel = channel
super().__init__(hap, device, f"Channel{channel}")
@ -163,10 +166,10 @@ class HomematicipMultiSwitch(HomematicipGenericDevice, SwitchDevice):
"""Return true if device is on."""
return self._device.functionalChannels[self.channel].on
async def async_turn_on(self, **kwargs):
async def async_turn_on(self, **kwargs) -> None:
"""Turn the device on."""
await self._device.turn_on(self.channel)
async def async_turn_off(self, **kwargs):
async def async_turn_off(self, **kwargs) -> None:
"""Turn the device off."""
await self._device.turn_off(self.channel)

View File

@ -37,7 +37,9 @@ HOME_WEATHER_CONDITION = {
}
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
async def async_setup_platform(
hass, config, async_add_entities, discovery_info=None
) -> None:
"""Set up the HomematicIP Cloud weather sensor."""
pass
@ -47,17 +49,17 @@ async def async_setup_entry(
) -> None:
"""Set up the HomematicIP weather sensor from a config entry."""
hap = hass.data[HMIPC_DOMAIN][config_entry.data[HMIPC_HAPID]]
devices = []
entities = []
for device in hap.home.devices:
if isinstance(device, AsyncWeatherSensorPro):
devices.append(HomematicipWeatherSensorPro(hap, device))
entities.append(HomematicipWeatherSensorPro(hap, device))
elif isinstance(device, (AsyncWeatherSensor, AsyncWeatherSensorPlus)):
devices.append(HomematicipWeatherSensor(hap, device))
entities.append(HomematicipWeatherSensor(hap, device))
devices.append(HomematicipHomeWeather(hap))
entities.append(HomematicipHomeWeather(hap))
if devices:
async_add_entities(devices)
if entities:
async_add_entities(entities)
class HomematicipWeatherSensor(HomematicipGenericDevice, WeatherEntity):