Skip to content

Improve xml readability #707

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.

Sign up for GitHub

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

Merged
bouwew merged 3 commits into from
Feb 16, 2025
Merged

Improve xml readability #707

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
0f96c53
Improve xml readability
bouwew Feb 16, 2025
5e5e9e3
Update CHANGELOG
bouwew Feb 16, 2025
3de0348
Add missing .strip()
bouwew Feb 16, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,9 @@
# Changelog

## Ongoing

- Improve readability of xml-data in POST/PUT requests

## v1.7.2

- Bugfix for Plugwise-beta issue [833](https://github.com/plugwise/plugwise-beta/issues/833) solving relay- and lock-switches not switching for the Stretch.
Expand Down
45 changes: 27 additions & 18 deletions plugwise/legacy/smile.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -172,9 +172,14 @@ async def set_preset(self, _: str, preset: str) -> None:

locator = f'rule/directives/when/then[@icon="{preset}"].../.../...'
rule = self._domain_objects.find(locator)
data = f'<rules><rule id="{rule.attrib["id"]}"><active>true</active></rule></rules>'

await self.call_request(RULES, method="put", data=data)
data = f'''
<rules>
<rule id="{rule.attrib["id"]}">
<active>true</active>
</rule>
</rules>
'''
await self.call_request(RULES, method="put", data=data.strip())

async def set_regulation_mode(self, mode: str) -> None:
"""Set-function placeholder for legacy devices."""
Expand Down Expand Up @@ -219,14 +224,17 @@ async def set_schedule_state(
for rule in self._domain_objects.findall(locator):
template_id = rule.attrib["id"]

data = f'''
<rules>
<rule id="{schedule_rule_id}">
<name><![CDATA[{name}]]></name>
<template id="{template_id}" />
<active>{new_state}</active>
</rule>
</rules>
'''
uri = f"{RULES};id={schedule_rule_id}"
data = (
"<rules><rule"
f' id="{schedule_rule_id}"><name><![CDATA[{name}]]></name><template'
f' id="{template_id}" /><active>{new_state}</active></rule></rules>'
)

await self.call_request(uri, method="put", data=data)
await self.call_request(uri, method="put", data=data.strip())

async def set_switch_state(
self, appl_id: str, members: list[str] | None, model: str, state: str
Expand Down Expand Up @@ -262,8 +270,9 @@ async def set_switch_state(
</{switch.func_type}>
</{switch.actuator}>
</appliance>
</appliances>'''.strip()
await self.call_request(APPLIANCES, method="post", data=data)
</appliances>
'''
await self.call_request(APPLIANCES, method="post", data=data.strip())
return

# Handle group of switches
Expand Down Expand Up @@ -308,13 +317,13 @@ async def set_temperature(self, _: str, items: dict[str, float]) -> None:
) # pragma: no cover"

temperature = str(setpoint)
data = f"""
<thermostat_functionality>
<setpoint>{temperature}</setpoint>
</thermostat_functionality>
"""
uri = self._thermostat_uri()
data = (
"<thermostat_functionality><setpoint>"
f"{temperature}</setpoint></thermostat_functionality>"
)

await self.call_request(uri, method="put", data=data)
await self.call_request(uri, method="put", data=data.strip())

async def call_request(self, uri: str, **kwargs: Any) -> None:
"""ConnectionFailedError wrapper for calling request()."""
Expand Down
111 changes: 71 additions & 40 deletions plugwise/smile.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -174,9 +174,13 @@ async def set_number(
if thermostat_id is None:
raise PlugwiseError(f"Plugwise: cannot change setpoint, {key} not found.")

data = f"""
<thermostat_functionality>
<setpoint>{temp}</setpoint>
</thermostat_functionality>
"""
uri = f"{APPLIANCES};id={self._heater_id}/thermostat;id={thermostat_id}"
data = f"<thermostat_functionality><setpoint>{temp}</setpoint></thermostat_functionality>"
await self.call_request(uri, method="put", data=data)
await self.call_request(uri, method="put", data=data.strip())

async def set_offset(self, dev_id: str, offset: float) -> None:
"""Set the Temperature offset for thermostats that support this feature."""
Expand All @@ -186,10 +190,13 @@ async def set_offset(self, dev_id: str, offset: float) -> None:
)

value = str(offset)
data = f"""
<offset_functionality>
<offset>{value}</offset>
</offset_functionality>
"""
uri = f"{APPLIANCES};id={dev_id}/offset;type=temperature_offset"
data = f"<offset_functionality><offset>{value}</offset></offset_functionality>"

await self.call_request(uri, method="put", data=data)
await self.call_request(uri, method="put", data=data.strip())

async def set_preset(self, loc_id: str, preset: str) -> None:
"""Set the given Preset on the relevant Thermostat - from LOCATIONS."""
Expand All @@ -201,15 +208,17 @@ async def set_preset(self, loc_id: str, preset: str) -> None:
current_location = self._domain_objects.find(f'location[@id="{loc_id}"]')
location_name = current_location.find("name").text
location_type = current_location.find("type").text

data = f'''
<locations>
<location id="{loc_id}">
<name>{location_name}</name>
<type>{location_type}</type>
<preset>{preset}</preset>
</location>
</locations>
'''
uri = f"{LOCATIONS};id={loc_id}"
data = (
"<locations><location"
f' id="{loc_id}"><name>{location_name}</name><type>{location_type}'
f"</type><preset>{preset}</preset></location></locations>"
)

await self.call_request(uri, method="put", data=data)
await self.call_request(uri, method="put", data=data.strip())

async def set_select(
self, key: str, loc_id: str, option: str, state: str | None
Expand All @@ -231,10 +240,13 @@ async def set_dhw_mode(self, mode: str) -> None:
if mode not in self._dhw_allowed_modes:
raise PlugwiseError("Plugwise: invalid dhw mode.")

data = f"""
<domestic_hot_water_mode_control_functionality>
<mode>{mode}</mode>
</domestic_hot_water_mode_control_functionality>
"""
uri = f"{APPLIANCES};type=heater_central/domestic_hot_water_mode_control"
data = f"<domestic_hot_water_mode_control_functionality><mode>{mode}</mode></domestic_hot_water_mode_control_functionality>"

await self.call_request(uri, method="put", data=data)
await self.call_request(uri, method="put", data=data.strip())

async def set_gateway_mode(self, mode: str) -> None:
"""Set the gateway mode."""
Expand All @@ -259,23 +271,32 @@ async def set_gateway_mode(self, mode: str) -> None:
vacation_time = time_2 + "T23:00:00.000Z"
valid = f"<valid_from>{vacation_time}</valid_from><valid_to>{end_time}</valid_to>"

data = f"""
<gateway_mode_control_functionality>
<mode>{mode}</mode>
{valid}
</gateway_mode_control_functionality>
"""
uri = f"{APPLIANCES};id={self._smile_props['gateway_id']}/gateway_mode_control"
data = f"<gateway_mode_control_functionality><mode>{mode}</mode>{valid}</gateway_mode_control_functionality>"

await self.call_request(uri, method="put", data=data)
await self.call_request(uri, method="put", data=data.strip())

async def set_regulation_mode(self, mode: str) -> None:
"""Set the heating regulation mode."""
if mode not in self._reg_allowed_modes:
raise PlugwiseError("Plugwise: invalid regulation mode.")

uri = f"{APPLIANCES};type=gateway/regulation_mode_control"
duration = ""
if "bleeding" in mode:
duration = "<duration>300</duration>"
data = f"<regulation_mode_control_functionality>{duration}<mode>{mode}</mode></regulation_mode_control_functionality>"

await self.call_request(uri, method="put", data=data)
data = f"""
<regulation_mode_control_functionality>
{duration}
<mode>{mode}</mode>
</regulation_mode_control_functionality>
"""
uri = f"{APPLIANCES};type=gateway/regulation_mode_control"
await self.call_request(uri, method="put", data=data.strip())

async def set_schedule_state(
self,
Expand Down Expand Up @@ -323,13 +344,17 @@ async def set_schedule_state(
template = f'<template id="{template_id}" />'

contexts = self.determine_contexts(loc_id, name, new_state, schedule_rule_id)
data = f'''
<rules>
<rule id="{schedule_rule_id}">
<name><![CDATA[{name}]]></name>
{template}
{contexts}
</rule>
</rules>
'''
uri = f"{RULES};id={schedule_rule_id}"
data = (
f'<rules><rule id="{schedule_rule_id}"><name><![CDATA[{name}]]></name>'
f"{template}{contexts}</rule></rules>"
)

await self.call_request(uri, method="put", data=data)
await self.call_request(uri, method="put", data=data.strip())
self._schedule_old_states[loc_id][name] = new_state

def determine_contexts(
Expand Down Expand Up @@ -387,9 +412,12 @@ async def set_switch_state(
switch_id = item.attrib["id"]
break

data = f"""
<{switch.func_type}>
<{switch.func}>{state}</{switch.func}>
</{switch.func_type}>
"""
uri = f"{APPLIANCES};id={appl_id}/{switch.device};id={switch_id}"
data = f"<{switch.func_type}><{switch.func}>{state}</{switch.func}></{switch.func_type}>"

if model == "relay":
locator = (
f'appliance[@id="{appl_id}"]/{switch.actuator}/{switch.func_type}/lock'
Expand All @@ -398,7 +426,7 @@ async def set_switch_state(
if self._domain_objects.find(locator).text == "true":
raise PlugwiseError("Plugwise: the locked Relay was not switched.")

await self.call_request(uri, method="put", data=data)
await self.call_request(uri, method="put", data=data.strip())

async def _set_groupswitch_member_state(
self, members: list[str], state: str, switch: Munch
Expand All @@ -411,9 +439,12 @@ async def _set_groupswitch_member_state(
locator = f'appliance[@id="{member}"]/{switch.actuator}/{switch.func_type}'
switch_id = self._domain_objects.find(locator).attrib["id"]
uri = f"{APPLIANCES};id={member}/{switch.device};id={switch_id}"
data = f"<{switch.func_type}><{switch.func}>{state}</{switch.func}></{switch.func_type}>"

await self.call_request(uri, method="put", data=data)
data = f"""
<{switch.func_type}>
<{switch.func}>{state}</{switch.func}>
</{switch.func_type}>
"""
await self.call_request(uri, method="put", data=data.strip())

async def set_temperature(self, loc_id: str, items: dict[str, float]) -> None:
"""Set the given Temperature on the relevant Thermostat."""
Expand Down Expand Up @@ -448,13 +479,13 @@ async def set_temperature(self, loc_id: str, items: dict[str, float]) -> None:
) # pragma: no cover"

temperature = str(setpoint)
data = f"""
<thermostat_functionality>
<setpoint>{temperature}</setpoint>
</thermostat_functionality>
"""
uri = self._thermostat_uri(loc_id)
data = (
"<thermostat_functionality><setpoint>"
f"{temperature}</setpoint></thermostat_functionality>"
)

await self.call_request(uri, method="put", data=data)
await self.call_request(uri, method="put", data=data.strip())

async def call_request(self, uri: str, **kwargs: Any) -> None:
"""ConnectionFailedError wrapper for calling request()."""
Expand Down
Loading