Mirrors
/
ha_gehome
mirror of https://github.com/simbaja/ha_gehome.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Resync with HA PR

This commit is contained in:
Andrew Marks 2020-08-12 09:31:39 -04:00
parent 87e087b357
commit 8b7de57434
12 changed files with 523 additions and 335 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

269
.gitignore vendored
View File

@ -1,133 +1,136 @@
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
pip-wheel-metadata/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
.pytest_cache/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
db.sqlite3-journal
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
.python-version
# pipenv
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
# However, in case of collaboration, if having platform-specific dependencies or dependencies
# having no cross-platform support, pipenv may install dependencies that don't work, or not
# install all needed dependencies.
#Pipfile.lock
# PEP 582; used by e.g. github.com/David-OConnor/pyflow
__pypackages__/
# Celery stuff
celerybeat-schedule
celerybeat.pid
# SageMath parsed files
*.sage.py
# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# IDE files
.idea/
.vscode/
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
pip-wheel-metadata/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
.pytest_cache/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
db.sqlite3-journal
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
.python-version
# pipenv
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
# However, in case of collaboration, if having platform-specific dependencies or dependencies
# having no cross-platform support, pipenv may install dependencies that don't work, or not
# install all needed dependencies.
#Pipfile.lock
# PEP 582; used by e.g. github.com/David-OConnor/pyflow
__pypackages__/
# Celery stuff
celerybeat-schedule
celerybeat.pid
# SageMath parsed files
*.sage.py
# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# IDE files
.idea/
.vscode/
# Secrets
*/secrets.py

0
geappliances/__init__.py
View File

7
geappliances/const.py
View File

@ -1,7 +0,0 @@
"""Constants"""
# OAuth2 credentials
OAUTH2_CLIENT_ID = '564c31616c4f7474434b307435412b4d'
OAUTH2_CLIENT_SECRET = '6476512b5246446d452f697154444941387052645938466e5671746e5847593d'
OAUTH2_REDIRECT_URI = 'brillion.4e617a766474657344444e562b5935566e51324'
OAUTH2_APP_ID = 'com.ge.kitchen.w'

55
geappliances/utils.py
View File

@ -1,55 +0,0 @@
"""Utility functions provided for documentation purposes"""
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import padding
# These constants were lifted directly from the config.properties in the Android APK
# Yes, the encryption really was stored in the same file. Yes, they really should be embarrassed.
CRYPT_KEY = "geKitchenAndroid".encode('ascii')
# Client id
CLIENT_ID_PRODUCTION = bytes.fromhex(
"23BAE464CCEAB1BE85F83FB71362B458A0C11C246975D02768F66100B52A8DD3B7E557958C8F93623965A6648F11CA49"
)
CLIENT_ID_FIELD = bytes.fromhex(
"47211A6E64A1860512D89CAB11995CCBE5148D5ADBEBC2A746918F5FC57E4F3C07A551217CE2BE8D533EDBC2068CC238"
)
# client secret
CLIENT_SECRET_PRODUCTION = bytes.fromhex(
"833248D2530FC0593CDE6D880DB17574BE27B3DCEF148A98853511AD3D96A5BE2B9B52FCF654F1126C3BABC88076B3DE"
"6C110BC2D8B4F50226DF1FF012E888F8F09EF93FEF6E451D364F74F6B39DD98F"
)
CLIENT_SECRET_FIELD = bytes.fromhex(
"10461990F551C3DF1AB1CEA0ED0C3D5B0C339205B29E97F7C869F92280DD92DCBD1FBC9A35A7AC5FBB606662A67E1148"
"B92CD0C588B8C22F48E46B44AA3FC582F09EF93FEF6E451D364F74F6B39DD98F"
)
# Redirect URI
REDIRECT_URI_PRODUCTION = bytes.fromhex(
"4934CC358DA408C26F948248AF7DD434145E3BEEA860EFB81017F9E4E93A4CB44DA6E9B0218BD7E3759537FDDD0C03A35"
"812860067FE1DD1A40831B101546DEF"
)
REDIRECT_URI_FIELD = bytes.fromhex(
"B2290735A24D5F5FAF099822BECA736E5F43D83837E0D0A63B23500C86AEE6CB87E2327CF482BD5F2612911666F4DD138"
"88B056E1A5E2509F828FD1CF64C1497"
)
# App ID
APP_ID_PRODUCTION = bytes.fromhex("4F73FCB16F54313979BCFB6DE1CB66794B5D28FA89B367C201A318C770976515")
APP_ID_FIELD = bytes.fromhex("4F73FCB16F54313979BCFB6DE1CB667995D405E80D40D247BC849B36AA7F0B5F")
def ge_decrypt(cypher_bytes: bytes) -> str:
"""Decrypt the obfuscated oauth2 parameters"""
backend = default_backend()
cipher = Cipher(algorithms.AES(CRYPT_KEY), modes.ECB(), backend=backend)
decryptor = cipher.decryptor()
plain_bytes = decryptor.update(cypher_bytes) + decryptor.finalize()
unpadder = padding.PKCS7(128).unpadder()
unpadded_bytes = unpadder.update(plain_bytes)
return unpadded_bytes.decode('ascii')

70
sharkiq/__init__.py
View File

@ -1,35 +1,39 @@
"""Shark IQ Integration"""
"""Shark IQ Integration."""
import asyncio
import async_timeout
import asyncio
import logging
import voluptuous as vol
from homeassistant import config_entries, exceptions
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from sharkiqpy import (
AylaApi,
SharkIqAuthError,
SharkIqAuthExpiringError,
SharkIqNotAuthedError,
get_ayla_api,
)
from .const import COMPONENTS, DOMAIN, SHARKIQ_SESSION
import voluptuous as vol
_LOGGER = logging.getLogger(__name__)
from homeassistant import config_entries, exceptions
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from .const import API_TIMEOUT, COMPONENTS, DOMAIN, LOGGER
from .update_coordinator import SharkIqUpdateCoordinator
CONFIG_SCHEMA = vol.Schema({DOMAIN: vol.Schema({})}, extra=vol.ALLOW_EXTRA)
class CannotConnect(exceptions.HomeAssistantError):
"""Error to indicate we cannot connect."""
pass
async def async_setup(hass, config):
"""Set up the sharkiq environment"""
"""Set up the sharkiq environment."""
hass.data.setdefault(DOMAIN, {})
if DOMAIN not in config:
return True
for index, conf in enumerate(config[DOMAIN]):
_LOGGER.debug("Importing Shark IQ #%d (Username: %s)", index, conf[CONF_USERNAME])
LOGGER.debug(
"Importing Shark IQ #%d (Username: %s)", index, conf[CONF_USERNAME]
)
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=conf,
@ -40,27 +44,25 @@ async def async_setup(hass, config):
async def async_connect_or_timeout(ayla_api: AylaApi) -> AylaApi:
"""Connect to vacuum."""
try:
with async_timeout.timeout(10):
_LOGGER.debug("Initialize connection to Ayla networks API")
with async_timeout.timeout(API_TIMEOUT):
LOGGER.debug("Initialize connection to Ayla networks API")
await ayla_api.async_sign_in()
except SharkIqAuthError as exc:
_LOGGER.error("Error to connect to Shark IQ api")
LOGGER.error("Error to connect to Shark IQ api")
raise CannotConnect from exc
except asyncio.TimeoutError as exc:
_LOGGER.error("Timeout expired")
LOGGER.error("Timeout expired")
raise CannotConnect from exc
return ayla_api
async def async_setup_entry(hass, config_entry):
"""Set the config entry up."""
# Set up sharkiq platforms with config entry
"""Initialize the sharkiq platform via config entry."""
ayla_api = get_ayla_api(
username=config_entry.data[CONF_USERNAME],
password=config_entry.data[CONF_PASSWORD],
websession=hass.helpers.aiohttp_client.async_get_clientsession()
websession=hass.helpers.aiohttp_client.async_get_clientsession(),
)
try:
@ -69,9 +71,17 @@ async def async_setup_entry(hass, config_entry):
except CannotConnect as exc:
raise exceptions.ConfigEntryNotReady from exc
hass.data[DOMAIN][config_entry.entry_id] = {
SHARKIQ_SESSION: ayla_api
}
shark_vacs = await ayla_api.async_get_devices(False)
device_names = ", ".join([d.name for d in shark_vacs])
LOGGER.info("Found %d Shark IQ device(s): %s", len(device_names), device_names)
coordinator = SharkIqUpdateCoordinator(hass, config_entry, ayla_api, shark_vacs)
await coordinator.async_refresh()
if not coordinator.last_update_success:
raise exceptions.ConfigEntryNotReady
hass.data[DOMAIN][config_entry.entry_id] = coordinator
for component in COMPONENTS:
hass.async_create_task(
@ -84,11 +94,14 @@ async def async_setup_entry(hass, config_entry):
return True
async def async_disconnect_or_timeout(ayla_api):
async def async_disconnect_or_timeout(coordinator: SharkIqUpdateCoordinator):
"""Disconnect to vacuum."""
_LOGGER.debug("Disconnecting from Ayla Api")
with async_timeout.timeout(3):
await ayla_api.async_sign_out()
LOGGER.debug("Disconnecting from Ayla Api")
with async_timeout.timeout(5):
try:
await coordinator.ayla_api.async_sign_out()
except (SharkIqAuthError, SharkIqAuthExpiringError, SharkIqNotAuthedError):
pass
return True
@ -109,7 +122,10 @@ async def async_unload_entry(hass, config_entry):
)
if unload_ok:
domain_data = hass.data[DOMAIN][config_entry.entry_id]
await async_disconnect_or_timeout(ayla_api=domain_data[SHARKIQ_SESSION])
try:
await async_disconnect_or_timeout(coordinator=domain_data)
except SharkIqAuthError:
pass
hass.data[DOMAIN].pop(config_entry.entry_id)
return unload_ok

67
sharkiq/config_flow.py
View File

@ -1,36 +1,36 @@
"""Config flow for Shark IQ integration."""
import async_timeout
import asyncio
import logging
from typing import Dict, Optional
import aiohttp
import async_timeout
from sharkiqpy import SharkIqAuthError, get_ayla_api
import voluptuous as vol
from homeassistant import config_entries, core, exceptions
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from sharkiqpy import get_ayla_api, SharkIqAuthError
from .const import DOMAIN
from .const import DOMAIN, LOGGER # pylint:disable=unused-import
SHARKIQ_SCHEMA = vol.Schema(
{vol.Required(CONF_USERNAME): str, vol.Required(CONF_PASSWORD): str}
)
_LOGGER = logging.getLogger(__name__)
async def validate_input(hass: core.HomeAssistant, data):
"""Validate the user input allows us to connect."""
ayla_api = get_ayla_api(
username=data[CONF_USERNAME],
password=data[CONF_PASSWORD],
websession=hass.helpers.aiohttp_client.async_get_clientsession(hass)
websession=hass.helpers.aiohttp_client.async_get_clientsession(hass),
)
try:
with async_timeout.timeout(10):
_LOGGER.debug("Initialize connection to Ayla networks API")
LOGGER.debug("Initialize connection to Ayla networks API")
await ayla_api.async_sign_in()
except asyncio.TimeoutError:
except (asyncio.TimeoutError, aiohttp.ClientError):
raise CannotConnect
except SharkIqAuthError:
raise InvalidAuth
@ -39,34 +39,65 @@ async def validate_input(hass: core.HomeAssistant, data):
return {"title": f"Shark IQ ({data[CONF_USERNAME]:s})"}
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
class SharkIqConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Shark IQ."""
VERSION = 1
# TODO pick one of the available connection classes in homeassistant/config_entries.py
CONNECTION_CLASS = config_entries.CONN_CLASS_UNKNOWN
CONNECTION_CLASS = config_entries.CONN_CLASS_CLOUD_POLL
async def async_step_user(self, user_input=None):
"""Handle the initial step."""
async def _async_validate_input(self, user_input):
"""Validate form input."""
errors = {}
info = None
if user_input is not None:
# noinspection PyBroadException
try:
info = await validate_input(self.hass, user_input)
return self.async_create_entry(title=info["title"], data=user_input)
except CannotConnect:
errors["base"] = "cannot_connect"
except InvalidAuth:
errors["base"] = "invalid_auth"
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Unexpected exception")
LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
return info, errors
async def async_step_user(self, user_input: Optional[Dict] = None):
"""Handle the initial step."""
errors = {}
if user_input is not None:
info, errors = await self._async_validate_input(user_input)
if info:
return self.async_create_entry(title=info["title"], data=user_input)
return self.async_show_form(
step_id="user", data_schema=SHARKIQ_SCHEMA, errors=errors
)
async def async_step_reauth(self, user_input: Optional[dict] = None):
"""Handle re-auth if login is invalid."""
errors = {}
if user_input is not None:
_, errors = await self._async_validate_input(user_input)
if not errors:
for entry in self._async_current_entries():
if entry.unique_id == self.unique_id:
self.hass.config_entries.async_update_entry(
entry, data=user_input
)
return self.async_abort(reason="reauth_successful")
if errors["base"] != "invalid_auth":
return self.async_abort(reason=errors["base"])
return self.async_show_form(
step_id="reauth", data_schema=SHARKIQ_SCHEMA, errors=errors,
)
class CannotConnect(exceptions.HomeAssistantError):
"""Error to indicate we cannot connect."""

11
sharkiq/const.py
View File

@ -1,6 +1,11 @@
"""Shark IQ Constants"""
"""Shark IQ Constants."""
DOMAIN = "sharkiq"
from datetime import timedelta
import logging
API_TIMEOUT = 20
COMPONENTS = ["vacuum"]
SHARKIQ_SESSION = "sharkiq_session"
DOMAIN = "sharkiq"
LOGGER = logging.getLogger(__package__)
SHARK = "Shark"
UPDATE_INTERVAL = timedelta(seconds=30)

2
sharkiq/manifest.json
View File

@ -3,7 +3,7 @@
"name": "Shark IQ",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/sharkiq",
"requirements": ["sharkiqpy==0.1.3"],
"requirements": ["sharkiqpy==0.1.8"],
"dependencies": [],
"codeowners": ["@ajmarks"]
}

63
sharkiq/sensor.py Normal file
View File

@ -0,0 +1,63 @@
"""Sensor for checking the battery level of Shark IQ"""
import logging
from homeassistant.const import DEVICE_CLASS_BATTERY, UNIT_PERCENTAGE
from homeassistant.helpers.icon import icon_for_battery_level
from typing import Callable, List, TYPE_CHECKING
from sharkiqpy import Properties
from .const import DOMAIN, SHARKIQ_SESSION
from .sharkiq import SharkVacuumEntity
if TYPE_CHECKING:
from homeassistant.core import HomeAssistant
from sharkiqpy import AylaApi, SharkIqVacuum
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass: "HomeAssistant", config_entry, async_add_entities: Callable):
"""Set up the Shark IQ battery sensor"""
domain_data = hass.data[DOMAIN][config_entry.entry_id]
ayla_api = domain_data[SHARKIQ_SESSION] # type: AylaApi
devices = await ayla_api.async_get_devices() # type: List[SharkIqVacuum]
device_names = ', '.join([d.name for d in devices])
_LOGGER.debug("Found % Shark IQ device(s): %s", len(devices), device_names)
async_add_entities([SharkIqBattery(d) for d in devices], True)
class SharkIqBattery(SharkVacuumEntity):
"""Class to hold Shark IQ battery info"""
@property
def name(self):
"""Return the name of the sensor."""
return f"{self.sharkiq.name} Battery Level"
@property
def unique_id(self) -> str:
"""Return the unique id of the vacuum cleaner."""
return "sharkiq-{:s}-battery".format(self.serial_number)
@property
def device_class(self):
"""Return the device class of the sensor."""
return DEVICE_CLASS_BATTERY
@property
def unit_of_measurement(self):
"""Return the unit_of_measurement of the device."""
return UNIT_PERCENTAGE
@property
def icon(self):
"""Return the icon for the battery."""
charging = self.sharkiq.get_property_value(Properties.CHARGING_STATUS)
return icon_for_battery_level(battery_level=self.battery_level, charging=charging)
@property
def state(self):
"""Return the state of the sensor."""
return self.battery_level

191
sharkiq/sharkiq.py
View File

@ -1,12 +1,14 @@
"""Shark IQ Wrapper"""
"""Shark IQ Wrapper."""
import logging
from typing import Dict, Optional
from sharkiqpy import OperatingModes, PowerModes, Properties, SharkIqVacuum
from homeassistant.components.vacuum import (
STATE_CLEANING,
STATE_DOCKED,
STATE_ERROR,
STATE_IDLE,
STATE_PAUSED,
STATE_RETURNING,
@ -21,14 +23,13 @@ from homeassistant.components.vacuum import (
SUPPORT_STOP,
StateVacuumEntity,
)
from sharkiqpy import SharkIqVacuum, PowerModes, OperatingModes, Properties
from typing import Dict, Optional
from .const import DOMAIN, SHARK
_LOGGER = logging.getLogger(__name__)
from .const import DOMAIN, SHARK
from .update_coordinator import SharkIqUpdateCoordinator
LOGGER = logging.getLogger(__name__)
# Supported features
# TODO: Add support for mapping
SUPPORT_SHARKIQ = (
SUPPORT_BATTERY
| SUPPORT_FAN_SPEED
@ -54,51 +55,74 @@ FAN_SPEEDS_MAP = {
"Max": PowerModes.MAX,
}
STATE_RECHARGING_TO_RESUME = 'recharging_to_resume' # TODO: Add strings for this
STATE_RECHARGING_TO_RESUME = "recharging_to_resume"
# Attributes to expose
ATTR_ERROR_CODE = "last_error_code"
ATTR_ERROR_MSG = "last_error_message"
ATTR_LOW_LIGHT = "low_light"
ATTR_RECHARGE_RESUME = "recharge_and_resume"
ATTR_RSSI = "rssi"
class SharkVacuumEntity(StateVacuumEntity):
"""Shark IQ vacuum entity"""
"""Shark IQ vacuum entity."""
def __init__(self, sharkiq: SharkIqVacuum):
def __init__(self, sharkiq: SharkIqVacuum, coordinator: SharkIqUpdateCoordinator):
"""Create a new SharkVacuumEntity."""
if sharkiq.serial_number not in coordinator.shark_vacs:
raise RuntimeError(
f"Shark IQ robot {sharkiq.serial_number} is not known to the coordinator"
)
self.coordinator = coordinator
self.sharkiq = sharkiq
@property
def should_poll(self):
"""Don't poll this entity. Polling is done via the coordinator."""
return False
def clean_spot(self, **kwargs):
"""Clean a spot. Not yet implemented."""
raise NotImplementedError()
def send_command(self, command, params=None, **kwargs):
"""Send a command to the vacuum. Not yet implemented."""
raise NotImplementedError()
@property
def is_online(self) -> bool:
"""Tell us if the device is online."""
return self.coordinator.is_online(self.sharkiq.serial_number)
@property
def name(self) -> str:
"""Device name."""
return self.sharkiq.name
@property
def serial_number(self) -> str:
"""Device DSN"""
"""Vacuum API serial number (DSN)."""
return self.sharkiq.serial_number
@property
def model(self) -> str:
"""Vacuum model number."""
if self.sharkiq.vac_model_number:
return self.sharkiq.vac_model_number
else:
return self.sharkiq.oem_model_number
return self.sharkiq.oem_model_number
@property
def device_info(self) -> Dict:
"""Device info dictionary."""
return {
"identifiers": {
(DOMAIN, self.serial_number)
},
"identifiers": {(DOMAIN, self.serial_number)},
"name": self.name,
"manufacturer": SHARK,
"model": self.model,
"sw_version": self.sharkiq.get_property_value(Properties.ROBOT_FIRMWARE_VERSION)
"sw_version": self.sharkiq.get_property_value(
Properties.ROBOT_FIRMWARE_VERSION
),
}
@property
@ -108,111 +132,104 @@ class SharkVacuumEntity(StateVacuumEntity):
@property
def is_docked(self) -> Optional[bool]:
"""Is vacuum docked"""
"""Is vacuum docked."""
return self.sharkiq.get_property_value(Properties.DOCKED_STATUS)
@property
def error_code(self) -> Optional[int]:
"""Error code or None"""
# Errors remain for a while, so we should only show an error if the device is stopped
if self.sharkiq.get_property_value(Properties.OPERATING_MODE) == OperatingModes.STOP and not self.is_docked:
return self.sharkiq.get_property_value(Properties.ERROR_CODE)
else:
"""Return the last observed error code (or None)."""
return self.sharkiq.error_code
@property
def error_message(self) -> Optional[str]:
"""Return the last observed error message (or None)."""
if not self.error_code:
return None
return self.sharkiq.error_code
@property
def operating_mode(self) -> Optional[str]:
"""Operating mode"""
"""Operating mode.."""
op_mode = self.sharkiq.get_property_value(Properties.OPERATING_MODE)
return OPERATING_STATE_MAP.get(op_mode)
@property
def recharging_to_resume(self) -> Optional[int]:
"""Return True if vacuum set to recharge and resume cleaning."""
return self.sharkiq.get_property_value(Properties.RECHARGING_TO_RESUME)
@property
def state(self):
"""Current state"""
"""
Get the current vacuum state.
NB: Currently, we do not return an error state because they can be very, very stale.
In the app, these are (usually) handled by showing the robot as stopped and sending the
user a notification.
"""
if self.recharging_to_resume:
return STATE_RECHARGING_TO_RESUME
elif self.is_docked:
if self.is_docked:
return STATE_DOCKED
elif self.error_code:
return STATE_ERROR
else:
return self.operating_mode
@property
def sharkiq_unique_id(self) -> str:
"""Return the uniqueid of the vacuum cleaner."""
return "sharkiq-{:s}-vacuum".format(self.serial_number)
return self.operating_mode
@property
def unique_id(self) -> str:
"""Return the unique id of the vacuum cleaner."""
return self.sharkiq_unique_id
return f"sharkiq-{self.serial_number:s}-vacuum"
@property
def available(self) -> bool:
"""Return True if entity is available."""
return True # Always available, otherwise setup will fail
"""Determine if the sensor is available based on API results."""
# If the last update was successful...
if self.coordinator.last_update_success and self.is_online:
return True
# Otherwise, we are not.
return False
@property
def battery_level(self):
"""Current battery level"""
"""Get the current battery level."""
return self.sharkiq.get_property_value(Properties.BATTERY_CAPACITY)
def update(self, property_list=None):
"""Update the known properties"""
self.sharkiq.update(property_list)
async def async_update(self):
"""Update the known properties asynchronously."""
await self.coordinator.async_request_refresh()
async def async_update(self, property_list=None):
"""Update the known properties asynchronously"""
await self.sharkiq.async_update(property_list)
def return_to_base(self, **kwargs):
"""Have the device return to base"""
self.sharkiq.set_operating_mode(OperatingModes.RETURN)
def pause(self):
"""Pause the cleaning task."""
self.sharkiq.set_operating_mode(OperatingModes.PAUSE)
def start(self):
"""Start the device"""
self.sharkiq.set_operating_mode(OperatingModes.START)
def stop(self, **kwargs):
"""Stop the device"""
self.sharkiq.set_operating_mode(OperatingModes.STOP)
def locate(self, **kwargs):
"""Cause the device to generate a loud chirp"""
self.sharkiq.find_device()
async def async_added_to_hass(self) -> None:
"""Connect to dispatcher listening for entity data notifications."""
self.async_on_remove(
self.coordinator.async_add_listener(self.async_write_ha_state)
)
async def async_return_to_base(self, **kwargs):
"""Have the device return to base"""
"""Have the device return to base."""
await self.sharkiq.async_set_operating_mode(OperatingModes.RETURN)
await self.coordinator.async_refresh()
async def async_pause(self):
"""Pause the cleaning task."""
await self.sharkiq.async_set_operating_mode(OperatingModes.PAUSE)
await self.coordinator.async_refresh()
async def async_start(self):
"""Start the device"""
"""Start the device."""
await self.sharkiq.async_set_operating_mode(OperatingModes.START)
await self.coordinator.async_refresh()
async def async_stop(self, **kwargs):
"""Stop the device"""
"""Stop the device."""
await self.sharkiq.async_set_operating_mode(OperatingModes.STOP)
await self.coordinator.async_refresh()
async def async_locate(self, **kwargs):
"""Cause the device to generate a loud chirp"""
"""Cause the device to generate a loud chirp."""
await self.sharkiq.async_find_device()
@property
def fan_speed(self) -> str:
"""Return the current fan speed"""
"""Return the current fan speed."""
fan_speed = None
speed_level = self.sharkiq.get_property_value(Properties.POWER_MODE)
for k, val in FAN_SPEEDS_MAP.items():
@ -220,11 +237,12 @@ class SharkVacuumEntity(StateVacuumEntity):
fan_speed = k
return fan_speed
def set_fan_speed(self, fan_speed: str, **kwargs):
"""Set the fan speed"""
self.sharkiq.set_property_value(
async def async_set_fan_speed(self, fan_speed: str, **kwargs):
"""Set the fan speed."""
await self.sharkiq.async_set_property_value(
Properties.POWER_MODE, FAN_SPEEDS_MAP.get(fan_speed.capitalize())
)
await self.coordinator.async_refresh()
@property
def fan_speed_list(self):
@ -234,17 +252,32 @@ class SharkVacuumEntity(StateVacuumEntity):
# Various attributes we want to expose
@property
def recharge_resume(self) -> Optional[bool]:
"""Recharge and resume mode active"""
"""Recharge and resume mode active."""
return self.sharkiq.get_property_value(Properties.RECHARGE_RESUME)
@property
def rssi(self) -> Optional[int]:
"""WiFi RSSI"""
"""Get the WiFi RSSI."""
return self.sharkiq.get_property_value(Properties.RSSI)
@property
def state_attributes(self) -> Dict:
data = super().state_attributes
data[ATTR_RSSI] = self.rssi
data[ATTR_RECHARGE_RESUME] = self.recharge_resume
def low_light(self):
"""Let us know if the robot is operating in low-light mode."""
return self.sharkiq.get_property_value(Properties.LOW_LIGHT_MISSION)
@property
def shark_state_attributes(self) -> Dict:
"""Return a dictionary of device state attributes specific to sharkiq."""
data = {
ATTR_ERROR_CODE: self.error_code,
ATTR_ERROR_MSG: self.sharkiq.error_text,
ATTR_LOW_LIGHT: self.low_light,
ATTR_RECHARGE_RESUME: self.recharge_resume,
ATTR_RSSI: self.rssi,
}
return data
@property
def state_attributes(self) -> Dict:
"""Return a dictionary of device state attributes."""
return dict(**super().state_attributes, **self.shark_state_attributes)

93
sharkiq/update_coordinator.py Normal file
View File

@ -0,0 +1,93 @@
"""Data update coordinator for shark iq vacuums."""
from typing import Dict, List
from async_timeout import timeout
from sharkiqpy import (
AylaApi,
SharkIqAuthError,
SharkIqAuthExpiringError,
SharkIqNotAuthedError,
SharkIqVacuum,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import API_TIMEOUT, DOMAIN, LOGGER, UPDATE_INTERVAL
class SharkIqUpdateCoordinator(DataUpdateCoordinator):
"""Define a wrapper class to update Shark IQ data."""
def __init__(
self,
hass: HomeAssistant,
config_entry,
ayla_api: AylaApi,
shark_vacs: List[SharkIqVacuum],
) -> None:
"""Set up the SharkIqUpdateCoordinator class."""
self.ayla_api = ayla_api
self.shark_vacs = {
sharkiq.serial_number: sharkiq for sharkiq in shark_vacs
} # type: Dict[SharkIqVacuum]
self._config_entry = config_entry
self._online_dsns = {}
super().__init__(hass, LOGGER, name=DOMAIN, update_interval=UPDATE_INTERVAL)
def is_online(self, dsn: str) -> bool:
"""Return the online state of a given vacuum dsn."""
return dsn in self._online_dsns
@staticmethod
async def _async_update_vacuum(sharkiq: SharkIqVacuum) -> None:
"""Asynchronously update the data for a single vacuum."""
dsn = sharkiq.serial_number
LOGGER.info("Updating sharkiq data for device DSN %s", dsn)
with timeout(API_TIMEOUT):
await sharkiq.async_update()
async def _async_update_data(self) -> bool:
"""Update data via Awair client library."""
try:
all_vacuums = await self.ayla_api.async_list_devices()
self._online_dsns = {
v["dsn"]
for v in all_vacuums
if v["connection_status"] == "Online" and v["dsn"] in self.shark_vacs
}
LOGGER.info("Updating sharkiq data")
for dsn in self._online_dsns:
await self._async_update_vacuum(self.shark_vacs[dsn])
except (
SharkIqAuthError,
SharkIqNotAuthedError,
SharkIqAuthExpiringError,
) as err:
LOGGER.info("Bad auth state", exc_info=err)
flow_context = {
"source": "reauth",
"unique_id": self._config_entry.unique_id,
}
matching_flows = [
flow
for flow in self.hass.config_entries.flow.async_progress()
if flow["context"] == flow_context
]
if not matching_flows:
self.hass.async_create_task(
self.hass.config_entries.flow.async_init(
DOMAIN, context=flow_context, data=self._config_entry.data,
)
)
raise UpdateFailed(err)
except Exception as err: # pylint: disable=broad-except
raise UpdateFailed(err)
return True

30
sharkiq/vacuum.py
View File

@ -1,22 +1,28 @@
"""Shark IQ robot vacuums"""
"""Shark IQ robot vacuums."""
import logging
from typing import List, TYPE_CHECKING
from .const import DOMAIN, SHARKIQ_SESSION
from typing import TYPE_CHECKING, Iterable
from .const import DOMAIN
from .sharkiq import SharkVacuumEntity
from .update_coordinator import SharkIqUpdateCoordinator
if TYPE_CHECKING:
from sharkiqpy import AylaApi, SharkIqVacuum
from sharkiqpy import SharkIqVacuum
_LOGGER = logging.getLogger(__name__)
LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up the Shark IQ vacuum cleaner."""
domain_data = hass.data[DOMAIN][config_entry.entry_id]
ayla_api = domain_data[SHARKIQ_SESSION] # type: AylaApi
devices = await ayla_api.async_get_devices() # type: List[SharkIqVacuum]
device_names = ', '.join([d.name for d in devices])
_LOGGER.debug("Found % Shark IQ device(s): %s", len(devices), device_names)
async_add_entities([SharkVacuumEntity(d) for d in devices], True)
coordinator = hass.data[DOMAIN][
config_entry.entry_id
] # type: SharkIqUpdateCoordinator
devices = coordinator.shark_vacs.values() # type: Iterable[SharkIqVacuum]
device_names = [d.name for d in devices]
LOGGER.debug(
"Found %d Shark IQ device(s): %s",
len(device_names),
", ".join([d.name for d in devices]),
)
async_add_entities([SharkVacuumEntity(d, coordinator) for d in devices])