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

Add support for Soma Tilt devices (#49734)

Browse source
This commit is contained in:
sebfortier2288 2022-01-20 15:21:54 -05:00 • committed by GitHub
parent 8b99adc1dc
commit a9785f1b41
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
7 changed files with 213 additions and 78 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.coveragerc
View file

@ -1032,6 +1032,7 @@ omit =
homeassistant/components/soma/__init__.py
homeassistant/components/soma/cover.py
homeassistant/components/soma/sensor.py
homeassistant/components/soma/utils.py
homeassistant/components/somfy/__init__.py
homeassistant/components/somfy/api.py
homeassistant/components/somfy/climate.py

4
CODEOWNERS
View file

@ -863,8 +863,8 @@ homeassistant/components/solaredge_local/* @drobtravels @scheric
homeassistant/components/solarlog/* @Ernst79
tests/components/solarlog/* @Ernst79
homeassistant/components/solax/* @squishykid
homeassistant/components/soma/* @ratsept
tests/components/soma/* @ratsept
homeassistant/components/soma/* @ratsept @sebfortier2288
tests/components/soma/* @ratsept @sebfortier2288
homeassistant/components/somfy/* @tetienne
tests/components/somfy/* @tetienne
homeassistant/components/sonarr/* @ctalkington

60
homeassistant/components/soma/__init__.py
View file

@ -1,5 +1,8 @@
"""Support for Soma Smartshades."""
import logging
from api.soma_api import SomaApi
from requests import RequestException
import voluptuous as vol
from homeassistant import config_entries
@ -11,6 +14,9 @@ from homeassistant.helpers.entity import DeviceInfo, Entity
from homeassistant.helpers.typing import ConfigType
from .const import API, DOMAIN, HOST, PORT
from .utils import is_api_response_success
_LOGGER = logging.getLogger(__name__)
DEVICES = "devices"
@ -62,6 +68,40 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
def soma_api_call(api_call):
"""Soma api call decorator."""
async def inner(self) -> dict:
response = {}
try:
response_from_api = await api_call(self)
except RequestException:
if self.api_is_available:
_LOGGER.warning("Connection to SOMA Connect failed")
self.api_is_available = False
else:
if not self.api_is_available:
self.api_is_available = True
_LOGGER.info("Connection to SOMA Connect succeeded")
if not is_api_response_success(response_from_api):
if self.is_available:
self.is_available = False
_LOGGER.warning(
"Device is unreachable (%s). Error while fetching the state: %s",
self.name,
response_from_api["msg"],
)
else:
if not self.is_available:
self.is_available = True
_LOGGER.info("Device %s is now reachable", self.name)
response = response_from_api
return response
return inner
class SomaEntity(Entity):
"""Representation of a generic Soma device."""
@ -72,6 +112,7 @@ class SomaEntity(Entity):
self.current_position = 50
self.battery_state = 0
self.is_available = True
self.api_is_available = True
@property
def available(self):
@ -99,3 +140,22 @@ class SomaEntity(Entity):
manufacturer="Wazombi Labs",
name=self.name,
)
def set_position(self, position: int) -> None:
"""Set the current device position."""
self.current_position = position
self.schedule_update_ha_state()
@soma_api_call
async def get_shade_state_from_api(self) -> dict:
"""Return the shade state from the api."""
return await self.hass.async_add_executor_job(
self.api.get_shade_state, self.device["mac"]
)
@soma_api_call
async def get_battery_level_from_api(self) -> dict:
"""Return the battery level from the api."""
return await self.hass.async_add_executor_job(
self.api.get_battery_level, self.device["mac"]
)

192
homeassistant/components/soma/cover.py
View file

@ -1,16 +1,28 @@
"""Support for Soma Covers."""
import logging
from __future__ import annotations
from requests import RequestException
from homeassistant.components.cover import ATTR_POSITION, CoverEntity
from homeassistant.components.cover import (
ATTR_POSITION,
ATTR_TILT_POSITION,
DEVICE_CLASS_BLIND,
DEVICE_CLASS_SHADE,
SUPPORT_CLOSE,
SUPPORT_CLOSE_TILT,
SUPPORT_OPEN,
SUPPORT_OPEN_TILT,
SUPPORT_SET_POSITION,
SUPPORT_SET_TILT_POSITION,
SUPPORT_STOP,
SUPPORT_STOP_TILT,
CoverEntity,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import API, DEVICES, DOMAIN, SomaEntity
_LOGGER = logging.getLogger(__name__)
from .utils import is_api_response_success
async def async_setup_entry(
@ -20,56 +32,105 @@ async def async_setup_entry(
) -> None:
"""Set up the Soma cover platform."""
api = hass.data[DOMAIN][API]
devices = hass.data[DOMAIN][DEVICES]
entities: list[SomaTilt | SomaShade] = []
async_add_entities(
[SomaCover(cover, hass.data[DOMAIN][API]) for cover in devices], True
for device in devices:
# Assume a shade device if the type is not present in the api response (Connect <2.2.6)
if "type" in device and device["type"].lower() == "tilt":
entities.append(SomaTilt(device, api))
else:
entities.append(SomaShade(device, api))
async_add_entities(entities, True)
class SomaTilt(SomaEntity, CoverEntity):
"""Representation of a Soma Tilt device."""
_attr_device_class = DEVICE_CLASS_BLIND
_attr_supported_features = (
SUPPORT_OPEN_TILT
| SUPPORT_CLOSE_TILT
| SUPPORT_STOP_TILT
| SUPPORT_SET_TILT_POSITION
)
@property
def current_cover_tilt_position(self):
"""Return the current cover tilt position."""
return self.current_position
class SomaCover(SomaEntity, CoverEntity):
"""Representation of a Soma cover device."""
@property
def is_closed(self):
"""Return if the cover tilt is closed."""
return self.current_position == 0
def close_cover(self, **kwargs):
"""Close the cover."""
def close_cover_tilt(self, **kwargs):
"""Close the cover tilt."""
response = self.api.set_shade_position(self.device["mac"], 100)
if response["result"] != "success":
_LOGGER.error(
"Unable to reach device %s (%s)", self.device["name"], response["msg"]
if not is_api_response_success(response):
raise HomeAssistantError(
f'Error while closing the cover ({self.name}): {response["msg"]}'
)
self.set_position(0)
def open_cover(self, **kwargs):
"""Open the cover."""
response = self.api.set_shade_position(self.device["mac"], 0)
if response["result"] != "success":
_LOGGER.error(
"Unable to reach device %s (%s)", self.device["name"], response["msg"]
def open_cover_tilt(self, **kwargs):
"""Open the cover tilt."""
response = self.api.set_shade_position(self.device["mac"], -100)
if not is_api_response_success(response):
raise HomeAssistantError(
f'Error while opening the cover ({self.name}): {response["msg"]}'
)
self.set_position(100)
def stop_cover(self, **kwargs):
"""Stop the cover."""
# Set cover position to some value where up/down are both enabled
self.current_position = 50
def stop_cover_tilt(self, **kwargs):
"""Stop the cover tilt."""
response = self.api.stop_shade(self.device["mac"])
if response["result"] != "success":
_LOGGER.error(
"Unable to reach device %s (%s)", self.device["name"], response["msg"]
if not is_api_response_success(response):
raise HomeAssistantError(
f'Error while stopping the cover ({self.name}): {response["msg"]}'
)
# Set cover position to some value where up/down are both enabled
self.set_position(50)
def set_cover_position(self, **kwargs):
"""Move the cover shutter to a specific position."""
self.current_position = kwargs[ATTR_POSITION]
response = self.api.set_shade_position(
self.device["mac"], 100 - kwargs[ATTR_POSITION]
)
if response["result"] != "success":
_LOGGER.error(
"Unable to reach device %s (%s)", self.device["name"], response["msg"]
def set_cover_tilt_position(self, **kwargs):
"""Move the cover tilt to a specific position."""
# 0 -> Closed down (api: 100)
# 50 -> Fully open (api: 0)
# 100 -> Closed up (api: -100)
target_api_position = 100 - ((kwargs[ATTR_TILT_POSITION] / 50) * 100)
response = self.api.set_shade_position(self.device["mac"], target_api_position)
if not is_api_response_success(response):
raise HomeAssistantError(
f'Error while setting the cover position ({self.name}): {response["msg"]}'
)
self.set_position(kwargs[ATTR_TILT_POSITION])
async def async_update(self):
"""Update the entity with the latest data."""
response = await self.get_shade_state_from_api()
api_position = int(response["position"])
if "closed_upwards" in response.keys():
self.current_position = 50 + ((api_position * 50) / 100)
else:
self.current_position = 50 - ((api_position * 50) / 100)
class SomaShade(SomaEntity, CoverEntity):
"""Representation of a Soma Shade device."""
_attr_device_class = DEVICE_CLASS_SHADE
_attr_supported_features = (
SUPPORT_OPEN | SUPPORT_CLOSE | SUPPORT_STOP | SUPPORT_SET_POSITION
)
@property
def current_cover_position(self):
"""Return the current position of cover shutter."""
"""Return the current cover position."""
return self.current_position
@property
@ -77,22 +138,45 @@ class SomaCover(SomaEntity, CoverEntity):
"""Return if the cover is closed."""
return self.current_position == 0
def close_cover(self, **kwargs):
"""Close the cover."""
response = self.api.set_shade_position(self.device["mac"], 100)
if not is_api_response_success(response):
raise HomeAssistantError(
f'Error while closing the cover ({self.name}): {response["msg"]}'
)
def open_cover(self, **kwargs):
"""Open the cover."""
response = self.api.set_shade_position(self.device["mac"], 0)
if not is_api_response_success(response):
raise HomeAssistantError(
f'Error while opening the cover ({self.name}): {response["msg"]}'
)
def stop_cover(self, **kwargs):
"""Stop the cover."""
response = self.api.stop_shade(self.device["mac"])
if not is_api_response_success(response):
raise HomeAssistantError(
f'Error while stopping the cover ({self.name}): {response["msg"]}'
)
# Set cover position to some value where up/down are both enabled
self.set_position(50)
def set_cover_position(self, **kwargs):
"""Move the cover shutter to a specific position."""
self.current_position = kwargs[ATTR_POSITION]
response = self.api.set_shade_position(
self.device["mac"], 100 - kwargs[ATTR_POSITION]
)
if not is_api_response_success(response):
raise HomeAssistantError(
f'Error while setting the cover position ({self.name}): {response["msg"]}'
)
async def async_update(self):
"""Update the cover with the latest data."""
try:
_LOGGER.debug("Soma Cover Update")
response = await self.hass.async_add_executor_job(
self.api.get_shade_state, self.device["mac"]
)
except RequestException:
_LOGGER.error("Connection to SOMA Connect failed")
self.is_available = False
return
if response["result"] != "success":
_LOGGER.error(
"Unable to reach device %s (%s)", self.device["name"], response["msg"]
)
self.is_available = False
return
response = await self.get_shade_state_from_api()
self.current_position = 100 - int(response["position"])
self.is_available = True

5
homeassistant/components/soma/manifest.json
View file

@ -3,7 +3,10 @@
"name": "Soma Connect",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/soma",
"codeowners": ["@ratsept"],
"codeowners": [
"@ratsept",
"@sebfortier2288"
],
"requirements": ["pysoma==0.0.10"],
"iot_class": "local_polling"
}

23
homeassistant/components/soma/sensor.py
View file

@ -1,8 +1,5 @@
"""Support for Soma sensors."""
from datetime import timedelta
import logging
from requests import RequestException
from homeassistant.components.sensor import SensorDeviceClass, SensorEntity
from homeassistant.config_entries import ConfigEntry
@ -16,8 +13,6 @@ from .const import API, DOMAIN
MIN_TIME_BETWEEN_UPDATES = timedelta(minutes=30)
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
hass: HomeAssistant,
@ -52,21 +47,8 @@ class SomaSensor(SomaEntity, SensorEntity):
@Throttle(MIN_TIME_BETWEEN_UPDATES)
async def async_update(self):
"""Update the sensor with the latest data."""
try:
_LOGGER.debug("Soma Sensor Update")
response = await self.hass.async_add_executor_job(
self.api.get_battery_level, self.device["mac"]
)
except RequestException:
_LOGGER.error("Connection to SOMA Connect failed")
self.is_available = False
return
if response["result"] != "success":
_LOGGER.error(
"Unable to reach device %s (%s)", self.device["name"], response["msg"]
)
self.is_available = False
return
response = await self.get_battery_level_from_api()
# https://support.somasmarthome.com/hc/en-us/articles/360026064234-HTTP-API
# battery_level response is expected to be min = 360, max 410 for
# 0-100% levels above 410 are consider 100% and below 360, 0% as the
@ -74,4 +56,3 @@ class SomaSensor(SomaEntity, SensorEntity):
_battery = round(2 * (response["battery_level"] - 360))
battery = max(min(100, _battery), 0)
self.battery_state = battery
self.is_available = True

6
homeassistant/components/soma/utils.py Normal file
View file

@ -0,0 +1,6 @@
"""Soma helpers functions."""
def is_api_response_success(api_response: dict) -> bool:
"""Check if the response returned from the Connect API is a success or not."""
return "result" in api_response and api_response["result"].lower() == "success"