From 4f7d4860dbcba0c69c8e4c5f3f722c3158fcaf4a Mon Sep 17 00:00:00 2001 From: Vadym Date: Tue, 18 Jul 2023 22:26:11 +0300 Subject: [PATCH] Water Heater. Ability to send only mandatory parameters (#14) * Added water heater appliance. Added ability to send only mandatory parameters * fixed build * formatting * cleanup * cleanup * reformatting * Added ability to send specific parameters. Useful in case the command has many not mandatory parameters and you want to send only one/few * cleanup * Fixed code style --------- Co-authored-by: Vadym Melnychuk --- pyhon/appliances/wh.py | 11 +++++++++++ pyhon/commands.py | 25 +++++++++++++++++++++++-- 2 files changed, 34 insertions(+), 2 deletions(-) create mode 100644 pyhon/appliances/wh.py diff --git a/pyhon/appliances/wh.py b/pyhon/appliances/wh.py new file mode 100644 index 0000000..4aa1a3c --- /dev/null +++ b/pyhon/appliances/wh.py @@ -0,0 +1,11 @@ +from typing import Any, Dict + +from pyhon.appliances.base import ApplianceBase + + +class Appliance(ApplianceBase): + def attributes(self, data: Dict[str, Any]) -> Dict[str, Any]: + data = super().attributes(data) + data["active"] = data["parameters"]["onOffStatus"] == "1" + + return data diff --git a/pyhon/commands.py b/pyhon/commands.py index 442dc10..80fcb93 100644 --- a/pyhon/commands.py +++ b/pyhon/commands.py @@ -75,6 +75,14 @@ class HonCommand: result.setdefault(parameter.group, {})[name] = parameter.intern_value return result + @property + def mandatory_parameter_groups(self) -> Dict[str, Dict[str, Union[str, float]]]: + result: Dict[str, Dict[str, Union[str, float]]] = {} + for name, parameter in self._parameters.items(): + if parameter.mandatory: + result.setdefault(parameter.group, {})[name] = parameter.intern_value + return result + @property def parameter_value(self) -> Dict[str, Union[str, float]]: return {n: p.value for n, p in self._parameters.items()} @@ -110,8 +118,21 @@ class HonCommand: name = "program" if "PROGRAM" in self._category_name else "category" self._parameters[name] = HonParameterProgram(name, self, "custom") - async def send(self) -> bool: - params = self.parameter_groups.get("parameters", {}) + async def send(self, only_mandatory: bool = False) -> bool: + grouped_params = ( + self.mandatory_parameter_groups if only_mandatory else self.parameter_groups + ) + params = grouped_params.get("parameters", {}) + return await self.send_parameters(params) + + async def send_specific(self, param_names: List[str]) -> bool: + params: Dict[str, str | float] = {} + for key, parameter in self._parameters.items(): + if key in param_names: + params[key] = parameter.value + return await self.send_parameters(params) + + async def send_parameters(self, params: Dict[str, str | float]) -> bool: ancillary_params = self.parameter_groups.get("ancillaryParameters", {}) ancillary_params.pop("programRules", None) self.appliance.sync_command_to_params(self.name)