Add plans buttons

.vscode/settings.json

@@ -1,3 +1,4 @@
 {
-    "makefile.extensionOutputFolder": "./.vscode"
+    "makefile.extensionOutputFolder": "./.vscode",
+    "python.formatting.provider": "black"
 }

Pipfile

@@ -6,6 +6,8 @@ name = "pypi"
 [packages]
 boto3 = "*"
 python-dotenv = "*"
+homeassistant = "*"
+pycognito = "*"
 
 [dev-packages]
 

custom_components/cecotec_conga/__init__.py

@@ -1,218 +1,43 @@
 import logging
-import random
-import requests
-import string
-import json
-import boto3
-import datetime
-from pycognito import Cognito
-from pycognito.utils import RequestsSrpAuth
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
+from homeassistant.helpers.entity import DeviceInfo
 
-_LOGGER = logging.getLogger(__name__)
+from .conga import Conga
+from .const import (
+    CONF_USERNAME,
+    CONF_PASSWORD,
+    DOMAIN,
+)
 
-CECOTEC_API_BASE_URL = "https://qafbskf2ug.execute-api.eu-west-2.amazonaws.com"
-AWS_IOT_ENDPOINT = "https://a39k27k2ztga9m-ats.iot.eu-west-2.amazonaws.com"
+_LOGGER = logging.getLogger(__name__)
 
 
 async def async_setup_entry(hass, entry):
     """Set up Cecotec Conga sensors based on a config entry."""
     _LOGGER.info("Setting up Cecotec Conga integration")
+    hass.data.setdefault(DOMAIN, {})
+
+    conga_client = Conga(entry.data[CONF_USERNAME], entry.data[CONF_PASSWORD])
+    await hass.async_add_executor_job(
+        conga_client.update_shadows, entry.data["devices"][0]["sn"]
+    )
+    plans = await hass.async_add_executor_job(conga_client.list_plans)
+
+    hass.data[DOMAIN][entry.entry_id] = {
+        "controller": conga_client,
+        "devices": entry.data["devices"],
+        "plans": plans,
+        "lastTimeSync": 0,
+        "lastFirmwareCheck": 0,
+        "latestFirmwareVersion": False,
+        "entities": [],
+        "name": "test",
+    }
+
     hass.async_create_task(
         hass.config_entries.async_forward_entry_setup(entry, "vacuum")
     )
+    hass.async_create_task(
+        hass.config_entries.async_forward_entry_setup(entry, "button")
+    )
     return True
-
-
-class Conga():
-    def __init__(self, username, password):
-        self._username = username
-        self._password = password
-        self._devices = []
-        self._shadow = {}
-        self._tactics = {}
-        self._api_token = None
-        self._iot_client = None
-        self._iot_token_expiration = None
-
-    def list_vacuums(self):
-        self._refresh_api_token()
-        devices = requests.post(
-            f'{CECOTEC_API_BASE_URL}/api/user_machine/list',
-            json={},
-            auth=self._api_token
-        )
-        devices.raise_for_status()
-        self._devices = devices.json()["data"]["page_items"]
-        _LOGGER.warn(self._devices)
-        return self._devices
-
-    def list_plans(self):
-        return self._plan_names
-
-    def update_shadows(self, sn):
-        self._refresh_iot_client()
-        shadow = self._iot_client.get_thing_shadow(thingName=sn)
-        shadow_service = self._iot_client.get_thing_shadow(
-            thingName=sn, shadowName='service')
-
-        shadow = json.load(shadow['payload'])['state']['reported']
-        shadow_service = json.load(shadow_service['payload'])[
-            'state']['reported']
-
-        self._shadow = shadow
-        self._tactics = shadow_service['getTimeTactics']['body']['timeTactics']
-
-        # Fill plans variables
-        plans = []
-        plan_names = []
-        for tactic in json.loads(self._tactics)['value']:
-            if "planName" in tactic:
-                plans.append(tactic)
-                plan_names.append(tactic['planName'])
-
-        self._plans = plans
-        self._plan_names = plan_names
-
-        return self._shadow
-
-    def get_status(self):
-        return self._shadow
-
-    def start(self, sn, fan_speed):
-        payload = {
-            "state": {
-                "desired": {
-                    "startClean": {
-                        "state": 1,
-                        "body": {
-                            "mode": "Auto",
-                            "deepClean": 0,
-                            "fanLevel": fan_speed,
-                            "water": 1,
-                            "autoBoost": 0,
-                            "params": "[]"
-                        }
-                    }
-                }
-            }
-        }
-
-        self._send_payload(sn, payload)
-
-    def set_fan_speed(self, sn, level):
-        payload = {
-            "state": {
-                "desired": {
-                    "workNoisy": level
-                }
-            }
-        }
-        _LOGGER.debug(payload)
-        self._refresh_iot_client()
-        self._iot_client.update_thing_shadow(
-            thingName=sn,
-            payload=bytes(json.dumps(payload), "ascii")
-        )
-
-    def set_water_level(self, sn, level):
-        payload = {
-            "state": {
-                "desired": {
-                    "water": level
-                }
-            }
-        }
-        _LOGGER.debug(payload)
-        self._refresh_iot_client()
-        self._iot_client.update_thing_shadow(
-            thingName=sn,
-            payload=bytes(json.dumps(payload), "ascii")
-        )
-
-    def start_plan(self, sn, plan_name):
-        allowed_chars = string.ascii_lowercase + \
-            string.ascii_uppercase + string.digits
-        result_str = ''.join(random.choice(allowed_chars) for i in range(10))
-        plan = self._get_plan_details(plan_name)
-        payload = {
-            "state": {
-                "desired": {
-                    "StartTimedCleanTask": {
-                        "id": result_str,
-                        "params": json.dumps(plan)
-                    }
-                }
-            }
-        }
-        _LOGGER.warn(payload)
-        self._send_payload(sn, payload)
-
-    def home(self, sn):
-        payload = {
-            "state": {
-                "desired": {
-                    "startFindCharge": {"state": 1}
-                }
-            }
-        }
-
-        self._send_payload(sn, payload)
-
-    def _get_plan_details(self, plan_name):
-        for plan in self._plans:
-            if plan['planName'] == plan_name:
-                return plan
-        return ""
-
-    def _send_payload(self, sn, payload):
-        _LOGGER.debug(payload)
-        self._refresh_iot_client()
-        self._iot_client.update_thing_shadow(
-            thingName=sn,
-            shadowName='service',
-            payload=bytes(json.dumps(payload), "ascii")
-        )
-
-    def _refresh_api_token(self):
-        if self._api_token != None:
-            return self._api_token
-
-        self._api_token = RequestsSrpAuth(
-            username=self._username,
-            password=self._password,
-            user_pool_id='eu-west-2_L5T0M5yrf',
-            client_id='6iep27ce22ojt8bgb2vji3d387',
-            user_pool_region='eu-west-2',
-        )
-
-    def _refresh_iot_client(self):
-        if self._iot_client != None and self._iot_token_expiration != None and datetime.datetime.now().timestamp() < self._iot_token_expiration.timestamp():
-            return self._iot_client
-
-        _LOGGER.info("Refreshing Cecotec Conga token")
-
-        u = Cognito('eu-west-2_L5T0M5yrf', '6iep27ce22ojt8bgb2vji3d387',
-                    username=self._username)
-        u.authenticate(password=self._password)
-        cognito = boto3.client('cognito-identity', 'eu-west-2')
-        response = cognito.get_id(
-            IdentityPoolId='eu-west-2:0cdeb155-55bb-45f8-9710-4895bd40d605',
-            Logins={
-                'cognito-idp.eu-west-2.amazonaws.com/eu-west-2_L5T0M5yrf': u.id_token
-            }
-        )
-        creds = cognito.get_credentials_for_identity(
-            IdentityId=response["IdentityId"],
-            Logins={
-                'cognito-idp.eu-west-2.amazonaws.com/eu-west-2_L5T0M5yrf': u.id_token
-            }
-        )
-        self._iot_client = boto3.client(
-            'iot-data',
-            region_name='eu-west-2',
-            endpoint_url=AWS_IOT_ENDPOINT,
-            aws_access_key_id=creds["Credentials"]["AccessKeyId"],
-            aws_secret_access_key=creds["Credentials"]["SecretKey"],
-            aws_session_token=creds["Credentials"]["SessionToken"],
-        )
-        self._iot_token_expiration = creds["Credentials"]["Expiration"]

custom_components/cecotec_conga/button.py

@@ -0,0 +1,101 @@
+import logging
+from homeassistant.core import HomeAssistant
+from homeassistant.components.button import ButtonEntity
+from homeassistant.helpers.entity import DeviceInfo, Entity
+
+from .utils import build_device_info
+from .const import (
+    BRAND,
+    CONF_DEVICES,
+    DOMAIN,
+    MODEL,
+)
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def async_setup_entry(hass, config_entry, async_add_entities):
+    """Set up the Cecotec Conga sensor from a config entry."""
+    entities = []
+
+    devices = hass.data[DOMAIN][config_entry.entry_id]["devices"]
+    plans = hass.data[DOMAIN][config_entry.entry_id]["plans"]
+
+    for device in devices:
+        for plan in plans:
+            conga_data = hass.data[DOMAIN][config_entry.entry_id]
+            button = CongaVacuumPlanButton(
+                hass, conga_data, plan, device["sn"], device["note_name"]
+            )
+            entities.append(button)
+            # hass.data[DOMAIN][config_entry.entry_id]["entities"].append(button)
+
+    async_add_entities(entities, update_before_add=True)
+
+
+class CongaEntity(Entity):
+    def __init__(
+        self,
+        conga_data: dict,
+        device_name: str,
+        sn: str,
+    ):
+        self._enabled = False
+        self._device_name = device_name
+        self._conga_data = conga_data
+        self._conga_client = conga_data["controller"]
+        self._sn = sn
+
+    @property
+    def device_info(self) -> DeviceInfo:
+        return build_device_info(self._device_name, self._sn)
+
+    @property
+    def model(self):
+        return MODEL
+
+    @property
+    def brand(self):
+        return BRAND
+
+    async def async_added_to_hass(self) -> None:
+        self._enabled = True
+
+    async def async_will_remove_from_hass(self) -> None:
+        self._enabled = False
+
+
+class CongaVacuumPlanButton(ButtonEntity, CongaEntity):
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        conga_data: dict,
+        plan_name: str,
+        sn: str,
+        device_name: str,
+    ):
+        self._hass = hass
+        self._conga_data = conga_data
+        self._conga_client = conga_data["controller"]
+        self._plan_name = plan_name
+        self._device_name = device_name
+        self._name = f"{self._device_name} Start {self._plan_name}"
+        self._sn = sn
+        self._unique_id = f"{self._device_name}_{self._plan_name}"
+        CongaEntity.__init__(self, conga_data, device_name, sn)
+        ButtonEntity.__init__(self)
+
+    @property
+    def name(self):
+        """Return the name of the sensor."""
+        return self._name
+
+    @property
+    def unique_id(self) -> str:
+        return self._unique_id
+
+    async def async_press(self) -> None:
+        _LOGGER.info(f"Running plan {self._plan_name} on {self._device_name}")
+        await self._hass.async_add_executor_job(
+            self._conga_client.start_plan, self._sn, self._plan_name
+        )

custom_components/cecotec_conga/config_flow.py

@@ -2,7 +2,7 @@
 import logging
 
 from requests.models import HTTPError
-from . import Conga
+from .conga import Conga
 import voluptuous as vol
 
 from homeassistant import config_entries
@@ -38,9 +38,7 @@ async def async_step_login(self, user_input=None):
         if user_input is not None:
             try:
                 # Validate credentials
-                c = Conga(
-                    user_input[CONF_USERNAME], user_input[CONF_PASSWORD]
-                )
+                c = Conga(user_input[CONF_USERNAME], user_input[CONF_PASSWORD])
                 vacuums = await self.hass.async_add_executor_job(c.list_vacuums)
 
                 # Create devices
@@ -49,7 +47,7 @@ async def async_step_login(self, user_input=None):
                     data={
                         CONF_USERNAME: user_input[CONF_USERNAME],
                         CONF_PASSWORD: user_input[CONF_PASSWORD],
-                        CONF_DEVICES: vacuums
+                        CONF_DEVICES: vacuums,
                     },
                 )