load available programs from api and add service calls for appliance programs

custom_components/homeconnect/api.py

@@ -2,20 +2,30 @@
 
 from asyncio import run_coroutine_threadsafe
 import logging
+import re
 
 import homeconnect
 from homeconnect.api import HomeConnectError
+import voluptuous as vol
 
 from homeassistant import config_entries, core
 from homeassistant.core import callback
 from homeassistant.helpers import config_entry_oauth2_flow
 from homeassistant.helpers.entity import Entity
-
-from .const import DOMAIN
+from homeassistant.helpers import config_validation as cv
+from homeassistant.components.sensor import DEVICE_CLASS_TEMPERATURE
+from homeassistant.components.binary_sensor import DEVICE_CLASS_DOOR
+
+from .const import (
+    DOMAIN,
+    SERVICE_STARTPROGRAM,
+    SERVICE_STOPPROGRAM,
+    PROGRAM_NAMES,
+    PROGRAM_OPTIONS,
+)
 
 _LOGGER = logging.getLogger(__name__)
 
-
 class ConfigEntryAuth(homeconnect.HomeConnectAPI):
     """Provide Home Connect authentication tied to an OAuth2 based config entry."""
 
@@ -77,6 +87,7 @@ class HomeConnectDevice:
     # for some devices, this is instead 'BSH.Common.EnumType.PowerState.Standby'
     # see https://developer.home-connect.com/docs/settings/power_state
     power_off_state = "BSH.Common.EnumType.PowerState.Off"
+    has_programs = False
 
     def __init__(self, appliance):
         """Initialize the device class."""
@@ -152,23 +163,110 @@ def async_entity_update(self):
         _LOGGER.debug("Entity update triggered on %s", self)
         self.async_schedule_update_ha_state(True)
 
+def convert_to_snake(camel):
+    """Convert from CamelCase to snake_case.
+
+    Taken from https://stackoverflow.com/questions/1175208/elegant-python-function-to-convert-camelcase-to-snake-case
+    """
+    snake = re.sub("(.)([A-Z][a-z]+)", r"\1_\2", camel)
+    return re.sub("([a-z0-9])([A-Z])", r"\1_\2", snake).lower()
+
+
+def format_key(key):
+    """Format Home Connect keys like `BSH.Something.SomeValue` to a simple `some_value`."""
+    if not isinstance(key, str):
+        return key
+    return convert_to_snake(key.split(".")[-1])
+
 
 class DeviceWithPrograms(HomeConnectDevice):
     """Device with programs."""
 
-    _programs = []
+    has_programs = True
 
     def get_programs_available(self):
         """Get the available programs."""
-        return self._programs
+        programs = self.appliance.get_programs_available()
+        #_LOGGER.debug("available programs: {}".format(programs))
+        return [{"name": p} for p in programs]
 
     def get_program_switches(self):
         """Get a dictionary with info about program switches.
 
         There will be one switch for each program.
         """
         programs = self.get_programs_available()
-        return [{"device": self, "program_name": p["name"]} for p in programs]
+        return [{"device": self, "program_name": p['name']} for p in programs]
+
+    def get_programs_services(self):
+        """Get a dictionary with info about program services."""
+
+        def start_program(service):
+            """Service to change current home schedule."""
+            program = ''
+            for p_key, p_name in PROGRAM_NAMES.items():
+                if p_name == service.service.replace(self.appliance.type.lower() + '_' + SERVICE_STARTPROGRAM + '_', ''):
+                    program = p_key
+                    options = {}
+                    for d in service.data:
+                        #_LOGGER.debug(f"d: {d}")
+                        for o_key, o_name in PROGRAM_OPTIONS.items():
+                            if o_name == d:
+                                options[o_key] = service.data.get(d)
+                    break
+            _LOGGER.info(f"Starting progra {program} with options {options}")
+            self.appliance.start_program(program, options)
+
+        def stop_program(service):
+            """Service to stop the current active program."""
+            self.appliance.stop_program()
+            _LOGGER.debug("stop program service called")
+
+        programs = self.appliance.get_programs_available()
+        #_LOGGER.debug("available programs: {}".format(programs))
+        _services = []
+        for p in programs:
+            options = self.appliance.get_program_options(p)
+            if options is not None:
+                #_LOGGER.debug(f"program {p}")
+                #_LOGGER.debug(f"options: {options}")
+                _schema = {}
+                i = 0
+                for o in options:
+                    #_LOGGER.debug(f"paramters: {o}")
+                    for v in o.keys():
+                        #_LOGGER.debug(f"v {v}")
+                        values = options[i][v]
+                        #_LOGGER.debug(f"values: {values}")
+                        param_key = values['key']
+                        if param_key in PROGRAM_OPTIONS:
+                            param_key = PROGRAM_OPTIONS[param_key]
+                        if values['type'] in ["Int", "Double"]:
+                            _schema[vol.Required(param_key)] = cv.positive_int
+                        elif values['type'] in ["Boolean"]:
+                            _schema[vol.Required(param_key)] = cv.boolean
+                        else: 
+                            _LOGGER.error(f"option type {values['type']} not handled")
+                    i+=1
+                _service_schema = vol.Schema(_schema)
+                if p in PROGRAM_NAMES:
+                    program_name = PROGRAM_NAMES[p]
+                else:
+                    program_name = format_key(p)
+                _services.append({  'service_domain': DOMAIN,
+                                'service_name': self.appliance.type.lower() + '_' + SERVICE_STARTPROGRAM + '_' + program_name,
+                                'service_callback': start_program,
+                                'service_schema': _service_schema,
+                            })
+            else:
+                _LOGGER.debug(f"no options found for {p}")
+        _services.append({  'service_domain': DOMAIN,
+                        'service_name': self.appliance.type.lower() + '_' + SERVICE_STOPPROGRAM,
+                        'service_callback': stop_program,
+                        'service_schema': vol.Schema({}),
+                    })
+        return _services
+
 
     def get_program_sensors(self):
         """Get a dictionary with info about program sensors.
@@ -201,32 +299,63 @@ def get_door_entity(self):
         return {
             "device": self,
             "name": self.appliance.name + " Door",
-            "device_class": "door",
+            "key": "BSH.Common.Status.DoorState",
+            "device_class": DEVICE_CLASS_DOOR,
         }
 
 
+class DeviceWithCustomSensors(HomeConnectDevice):
+    """Device that has custom specific sensors."""
+
+    _appliance_binary_sensors = []
+    _appliance_sensors = []
+
+    def get_appliance_sensors(self):
+        """Get a dictionary with info about appliance sensors."""
+
+        if not self.appliance.status:
+            _status = self.appliance.get_status()
+        else:
+            _status = self.appliance.status
+
+        _sensors = []
+        for name, object_class, device_class in self._appliance_sensors:
+            if object_class in _status:
+                _unit = ""
+                if 'unit' in _status[object_class]:
+                    _unit = _status[object_class]['unit']
+
+                _sensors.append(
+                    {
+                        "device": self,
+                        "name": " ".join((self.appliance.name, name)),
+                        "unit": _unit,
+                        "key": object_class,
+                        "device_class": device_class,
+                    }
+                )
+
+        #_LOGGER.debug(f"sensors to be created: {_sensors}")
+
+        _binary_sensors = []
+        for name, object_class, device_class in self._appliance_binary_sensors:
+            if object_class in _status:
+                _binary_sensors.append(
+                    {
+                        "device": self,
+                        "name": " ".join((self.appliance.name, name)),
+                        "key": object_class,
+                        "device_class": device_class,
+                    }
+                )
+
+        #_LOGGER.debug(f"binary sensors to be created: {_binary_sensors}")
+        return _sensors, _binary_sensors
+
+
 class Dryer(DeviceWithDoor, DeviceWithPrograms):
     """Dryer class."""
 
-    _programs = [
-        {"name": "LaundryCare.Dryer.Program.Cotton"},
-        {"name": "LaundryCare.Dryer.Program.Synthetic"},
-        {"name": "LaundryCare.Dryer.Program.Mix"},
-        {"name": "LaundryCare.Dryer.Program.Blankets"},
-        {"name": "LaundryCare.Dryer.Program.BusinessShirts"},
-        {"name": "LaundryCare.Dryer.Program.DownFeathers"},
-        {"name": "LaundryCare.Dryer.Program.Hygiene"},
-        {"name": "LaundryCare.Dryer.Program.Jeans"},
-        {"name": "LaundryCare.Dryer.Program.Outdoor"},
-        {"name": "LaundryCare.Dryer.Program.SyntheticRefresh"},
-        {"name": "LaundryCare.Dryer.Program.Towels"},
-        {"name": "LaundryCare.Dryer.Program.Delicates"},
-        {"name": "LaundryCare.Dryer.Program.Super40"},
-        {"name": "LaundryCare.Dryer.Program.Shirts15"},
-        {"name": "LaundryCare.Dryer.Program.Pillow"},
-        {"name": "LaundryCare.Dryer.Program.AntiShrink"},
-    ]
-
     def get_entities(self):
         """Get a dictionary with infos about the associated entities."""
         door_entity = self.get_door_entity()
@@ -242,31 +371,6 @@ def get_entities(self):
 class Dishwasher(DeviceWithDoor, DeviceWithPrograms):
     """Dishwasher class."""
 
-    _programs = [
-        {"name": "Dishcare.Dishwasher.Program.Auto1"},
-        {"name": "Dishcare.Dishwasher.Program.Auto2"},
-        {"name": "Dishcare.Dishwasher.Program.Auto3"},
-        {"name": "Dishcare.Dishwasher.Program.Eco50"},
-        {"name": "Dishcare.Dishwasher.Program.Quick45"},
-        {"name": "Dishcare.Dishwasher.Program.Intensiv70"},
-        {"name": "Dishcare.Dishwasher.Program.Normal65"},
-        {"name": "Dishcare.Dishwasher.Program.Glas40"},
-        {"name": "Dishcare.Dishwasher.Program.GlassCare"},
-        {"name": "Dishcare.Dishwasher.Program.NightWash"},
-        {"name": "Dishcare.Dishwasher.Program.Quick65"},
-        {"name": "Dishcare.Dishwasher.Program.Normal45"},
-        {"name": "Dishcare.Dishwasher.Program.Intensiv45"},
-        {"name": "Dishcare.Dishwasher.Program.AutoHalfLoad"},
-        {"name": "Dishcare.Dishwasher.Program.IntensivPower"},
-        {"name": "Dishcare.Dishwasher.Program.MagicDaily"},
-        {"name": "Dishcare.Dishwasher.Program.Super60"},
-        {"name": "Dishcare.Dishwasher.Program.Kurz60"},
-        {"name": "Dishcare.Dishwasher.Program.ExpressSparkle65"},
-        {"name": "Dishcare.Dishwasher.Program.MachineCare"},
-        {"name": "Dishcare.Dishwasher.Program.SteamFresh"},
-        {"name": "Dishcare.Dishwasher.Program.MaximumCleaning"},
-    ]
-
     def get_entities(self):
         """Get a dictionary with infos about the associated entities."""
         door_entity = self.get_door_entity()
@@ -279,58 +383,44 @@ def get_entities(self):
         }
 
 
-class Oven(DeviceWithDoor, DeviceWithPrograms):
+class Oven(DeviceWithDoor, DeviceWithPrograms, DeviceWithCustomSensors):
     """Oven class."""
 
-    _programs = [
-        {"name": "Cooking.Oven.Program.HeatingMode.PreHeating"},
-        {"name": "Cooking.Oven.Program.HeatingMode.HotAir"},
-        {"name": "Cooking.Oven.Program.HeatingMode.TopBottomHeating"},
-        {"name": "Cooking.Oven.Program.HeatingMode.PizzaSetting"},
-        {"name": "Cooking.Oven.Program.Microwave.600Watt"},
+    power_off_state = "BSH.Common.EnumType.PowerState.Standby"
+
+    _appliance_binary_sensors = [
+        [ "Local Control Active", "BSH.Common.Status.LocalControlActive", None ],
+        [ "Remote Control Start Allowed", "BSH.Common.Status.RemoteControlStartAllowed", None ],
+        [ "Remote Control Active", "BSH.Common.Status.RemoteControlActive", None ],
+        [ "Fast PreHeat", "Cooking.Oven.Option.FastPreHeat", None ],
     ]
 
-    power_off_state = "BSH.Common.EnumType.PowerState.Standby"
+    _appliance_sensors = [
+        [ "Current Cavity Temperature", "Cooking.Oven.Status.CurrentCavityTemperature", DEVICE_CLASS_TEMPERATURE ],
+        [ "Operation State", "BSH.Common.Status.OperationState", None ],
+        [ "Power State", "BSH.Common.Setting.PowerState", None ],
+        [ "Setpoint Temperature", "Cooking.Oven.Option.SetpointTemperature", DEVICE_CLASS_TEMPERATURE ],
+        [ "Active Program", "BSH.Common.Root.ActiveProgram", None ],
+    ]
 
     def get_entities(self):
         """Get a dictionary with infos about the associated entities."""
         door_entity = self.get_door_entity()
-        program_sensors = self.get_program_sensors()
-        program_switches = self.get_program_switches()
+        _sensors, _binary_sensors = self.get_appliance_sensors()
+        binary_sensors = [door_entity] + _binary_sensors
+        sensors = self.get_program_sensors() + _sensors
+        switches = self.get_program_switches()
+        self.get_programs_services()
         return {
-            "binary_sensor": [door_entity],
-            "switch": program_switches,
-            "sensor": program_sensors,
+            "binary_sensor": binary_sensors,
+            "switch": switches,
+            "sensor": sensors,
         }
 
 
 class Washer(DeviceWithDoor, DeviceWithPrograms):
     """Washer class."""
 
-    _programs = [
-        {"name": "LaundryCare.Washer.Program.Cotton"},
-        {"name": "LaundryCare.Washer.Program.Cotton.CottonEco"},
-        {"name": "LaundryCare.Washer.Program.EasyCare"},
-        {"name": "LaundryCare.Washer.Program.Mix"},
-        {"name": "LaundryCare.Washer.Program.DelicatesSilk"},
-        {"name": "LaundryCare.Washer.Program.Wool"},
-        {"name": "LaundryCare.Washer.Program.Sensitive"},
-        {"name": "LaundryCare.Washer.Program.Auto30"},
-        {"name": "LaundryCare.Washer.Program.Auto40"},
-        {"name": "LaundryCare.Washer.Program.Auto60"},
-        {"name": "LaundryCare.Washer.Program.Chiffon"},
-        {"name": "LaundryCare.Washer.Program.Curtains"},
-        {"name": "LaundryCare.Washer.Program.DarkWash"},
-        {"name": "LaundryCare.Washer.Program.Dessous"},
-        {"name": "LaundryCare.Washer.Program.Monsoon"},
-        {"name": "LaundryCare.Washer.Program.Outdoor"},
-        {"name": "LaundryCare.Washer.Program.PlushToy"},
-        {"name": "LaundryCare.Washer.Program.ShirtsBlouses"},
-        {"name": "LaundryCare.Washer.Program.SportFitness"},
-        {"name": "LaundryCare.Washer.Program.Towels"},
-        {"name": "LaundryCare.Washer.Program.WaterProof"},
-    ]
-
     def get_entities(self):
         """Get a dictionary with infos about the associated entities."""
         door_entity = self.get_door_entity()
@@ -346,15 +436,6 @@ def get_entities(self):
 class CoffeeMaker(DeviceWithPrograms):
     """Coffee maker class."""
 
-    _programs = [
-        {"name": "ConsumerProducts.CoffeeMaker.Program.Beverage.Espresso"},
-        {"name": "ConsumerProducts.CoffeeMaker.Program.Beverage.EspressoMacchiato"},
-        {"name": "ConsumerProducts.CoffeeMaker.Program.Beverage.Coffee"},
-        {"name": "ConsumerProducts.CoffeeMaker.Program.Beverage.Cappuccino"},
-        {"name": "ConsumerProducts.CoffeeMaker.Program.Beverage.LatteMacchiato"},
-        {"name": "ConsumerProducts.CoffeeMaker.Program.Beverage.CaffeLatte"},
-    ]
-
     power_off_state = "BSH.Common.EnumType.PowerState.Standby"
 
     def get_entities(self):
@@ -367,12 +448,6 @@ def get_entities(self):
 class Hood(DeviceWithPrograms):
     """Hood class."""
 
-    _programs = [
-        {"name": "Cooking.Common.Program.Hood.Automatic"},
-        {"name": "Cooking.Common.Program.Hood.Venting"},
-        {"name": "Cooking.Common.Program.Hood.DelayedShutOff"},
-    ]
-
     def get_entities(self):
         """Get a dictionary with infos about the associated entities."""
         program_sensors = self.get_program_sensors()
@@ -392,8 +467,6 @@ def get_entities(self):
 class Hob(DeviceWithPrograms):
     """Hob class."""
 
-    _programs = [{"name": "Cooking.Hob.Program.PowerLevelMode"}]
-
     def get_entities(self):
         """Get a dictionary with infos about the associated entities."""
         program_sensors = self.get_program_sensors()