Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Water Heater. Ability to send only mandatory parameters #14

Merged
merged 11 commits into from
Jul 18, 2023
11 changes: 11 additions & 0 deletions pyhon/appliances/wh.py
Original file line number Diff line number Diff line change
@@ -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
25 changes: 23 additions & 2 deletions pyhon/commands.py
Original file line number Diff line number Diff line change
Expand Up @@ -76,6 +76,14 @@ def parameter_groups(self) -> Dict[str, Dict[str, Union[str, float]]]:
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()}
Expand Down Expand Up @@ -111,8 +119,21 @@ def _create_parameters(
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)
Expand Down