Decode received sysvar/program descriptions

.pre-commit-config.yaml

@@ -1,6 +1,6 @@
 repos:
   - repo: https://github.com/charliermarsh/ruff-pre-commit
-    rev: v0.8.2
+    rev: v0.8.3
     hooks:
       - id: ruff
         args:

changelog.md

@@ -1,7 +1,8 @@
 # Version 2024.12.3 (2024-12-14)
 
-- Ignore sysvar/program descriptions with problematic character(s)
-- Replace tabs in sysvar/program descriptions
+- Add method cleanup_text_from_html_tags
+- Decode received sysvar/program descriptions
+- Replace special character replacement by simple UriEncode() method use by @jens-maus
 
 # Version 2024.12.2 (2024-12-10)
 

hahomematic/client/json_rpc.py

@@ -10,6 +10,7 @@
 from pathlib import Path
 from ssl import SSLContext
 from typing import Any, Final
+from urllib.parse import unquote
 
 from aiohttp import (
     ClientConnectorCertificateError,
@@ -25,7 +26,6 @@
 from hahomematic.const import (
     DESCRIPTIONS_ERROR_MESSAGE,
     EXTENDED_SYSVAR_MARKER,
-    HTMLTAG_PATTERN,
     PATH_JSON_RPC,
     REGA_SCRIPT_PATH,
     UTF8,
@@ -48,7 +48,12 @@
     UnsupportedException,
 )
 from hahomematic.model.support import convert_value
-from hahomematic.support import get_tls_context, parse_sys_var, reduce_args
+from hahomematic.support import (
+    cleanup_text_from_html_tags,
+    get_tls_context,
+    parse_sys_var,
+    reduce_args,
+)
 
 _LOGGER: Final = logging.getLogger(__name__)
 
@@ -459,21 +464,18 @@ async def execute_program(self, pid: str) -> bool:
 
     async def set_system_variable(self, name: str, value: Any) -> bool:
         """Set a system variable on CCU / Homegear."""
-        params = {
-            _JsonKey.NAME: name,
-            _JsonKey.VALUE: value,
-        }
+        params = {_JsonKey.NAME: name, _JsonKey.VALUE: value}
         if isinstance(value, bool):
             params[_JsonKey.VALUE] = int(value)
             response = await self._post(method=_JsonRpcMethod.SYSVAR_SET_BOOL, extra_params=params)
         elif isinstance(value, str):
-            if HTMLTAG_PATTERN.findall(value):
+            if (clean_text := cleanup_text_from_html_tags(text=value)) != value:
+                params[_JsonKey.VALUE] = clean_text
                 _LOGGER.warning(
-                    "SET_SYSTEM_VARIABLE failed: "
-                    "Value (%s) contains html tags. This is not allowed",
+                    "SET_SYSTEM_VARIABLE: "
+                    "Value (%s) contains html tags. These are filtered out when writing.",
                     value,
                 )
-                return False
             response = await self._post_script(
                 script_name=RegaScript.SET_SYSTEM_VARIABLE, extra_params=params
             )
@@ -592,7 +594,10 @@ async def _get_program_descriptions(self) -> dict[str, str]:
             _LOGGER.debug("GET_PROGRAM_DESCRIPTIONS: Getting program descriptions")
             if json_result := response[_JsonKey.RESULT]:
                 for data in json_result:
-                    descriptions[data[_JsonKey.ID]] = data[_JsonKey.DESCRIPTION]
+                    decoded_text = unquote(string=data[_JsonKey.DESCRIPTION])
+                    descriptions[data[_JsonKey.ID]] = cleanup_text_from_html_tags(
+                        text=decoded_text
+                    )
         except JSONDecodeError as err:
             _LOGGER.error(
                 "GET_PROGRAM_DESCRIPTIONS failed: Unable to decode json: %s. %s",
@@ -612,8 +617,10 @@ async def _get_system_variable_descriptions(self) -> dict[str, str]:
             _LOGGER.debug("GET_SYSTEM_VARIABLE_DESCRIPTIONS: Getting system variable descriptions")
             if json_result := response[_JsonKey.RESULT]:
                 for data in json_result:
-                    descriptions[data[_JsonKey.ID]] = data[_JsonKey.DESCRIPTION]
-
+                    decoded_text = unquote(string=data[_JsonKey.DESCRIPTION])
+                    descriptions[data[_JsonKey.ID]] = cleanup_text_from_html_tags(
+                        text=decoded_text
+                    )
         except JSONDecodeError as err:
             _LOGGER.error(
                 "GET_SYSTEM_VARIABLE_DESCRIPTIONS failed: Unable to decode json: %s. %s",

hahomematic/support.py

@@ -29,6 +29,7 @@
     DP_KEY,
     FILE_DEVICES,
     FILE_PARAMSETS,
+    HTMLTAG_PATTERN,
     IDENTIFIER_SEPARATOR,
     INIT_DATETIME,
     MAX_CACHE_AGE,
@@ -468,3 +469,8 @@ def supports_rx_mode(command_rx_mode: CommandRxMode, rx_modes: tuple[RxMode, ...
     return (command_rx_mode == CommandRxMode.BURST and RxMode.BURST in rx_modes) or (
         command_rx_mode == CommandRxMode.WAKEUP and RxMode.WAKEUP in rx_modes
     )
+
+
+def cleanup_text_from_html_tags(text: str) -> str:
+    """Cleanup text from html tags."""
+    return re.sub(HTMLTAG_PATTERN, "", text)

requirements_test.txt

@@ -6,18 +6,18 @@ freezegun==1.5.1
 mypy-dev==1.14.0a4
 pip==24.3.1
 pre-commit==4.0.1
-pur==7.3.2
+pur==7.3.3
 pydantic==2.10.3
 pydevccu==0.1.8
 pylint-per-file-ignores==1.3.2
 pylint-strict-informational==0.1
 pylint==3.3.2
 pytest-aiohttp==1.0.5
-pytest-asyncio==0.24.0
+pytest-asyncio==0.25.0
 pytest-cov==6.0.0
 pytest-rerunfailures==15.0
 pytest-socket==0.7.0
 pytest-timeout==2.3.1
 pytest==8.3.4
 types-python-slugify==8.0.2.20240310
-uv==0.5.7
+uv==0.5.9

requirements_test_pre_commit.txt

@@ -1,4 +1,4 @@
 bandit==1.8.0
 codespell==2.3.0
-ruff==0.8.2
+ruff==0.8.3
 yamllint==1.35.1

tests/test_json_rpc.py

@@ -7,6 +7,8 @@
 import orjson
 import pytest
 
+from hahomematic.support import cleanup_text_from_html_tags
+
 SUCCESS = '{"HmIP-RF.0001D3C99C3C93%3A0.CONFIG_PENDING":false,\r\n"VirtualDevices.INT0000001%3A1.SET_POINT_TEMPERATURE":4.500000,\r\n"VirtualDevices.INT0000001%3A1.SWITCH_POINT_OCCURED":false,\r\n"VirtualDevices.INT0000001%3A1.VALVE_STATE":4,\r\n"VirtualDevices.INT0000001%3A1.WINDOW_STATE":0,\r\n"HmIP-RF.001F9A49942EC2%3A0.CARRIER_SENSE_LEVEL":10.000000,\r\n"HmIP-RF.0003D7098F5176%3A0.UNREACH":false,\r\n"BidCos-RF.OEQ1860891%3A0.UNREACH":true,\r\n"BidCos-RF.OEQ1860891%3A0.STICKY_UNREACH":true,\r\n"BidCos-RF.OEQ1860891%3A1.INHIBIT":false,\r\n"HmIP-RF.000A570998B3FB%3A0.CONFIG_PENDING":false,\r\n"HmIP-RF.000A570998B3FB%3A0.UPDATE_PENDING":false,\r\n"HmIP-RF.000A5A4991BDDC%3A0.CONFIG_PENDING":false,\r\n"HmIP-RF.000A5A4991BDDC%3A0.UPDATE_PENDING":false,\r\n"BidCos-RF.NEQ1636407%3A1.STATE":0,\r\n"BidCos-RF.NEQ1636407%3A2.STATE":false,\r\n"BidCos-RF.NEQ1636407%3A2.INHIBIT":false,\r\n"CUxD.CUX2800001%3A12.TS":"0"}'
 FAILURE = '{"HmIP-RF.0001D3C99C3C93%3A0.CONFIG_PENDING":false,\r\n"VirtualDevices.INT0000001%3A1.SET_POINT_TEMPERATURE":4.500000,\r\n"VirtualDevices.INT0000001%3A1.SWITCH_POINT_OCCURED":false,\r\n"VirtualDevices.INT0000001%3A1.VALVE_STATE":4,\r\n"VirtualDevices.INT0000001%3A1.WINDOW_STATE":0,\r\n"HmIP-RF.001F9A49942EC2%3A0.CARRIER_SENSE_LEVEL":10.000000,\r\n"HmIP-RF.0003D7098F5176%3A0.UNREACH":false,\r\n,\r\n,\r\n"BidCos-RF.OEQ1860891%3A0.UNREACH":true,\r\n"BidCos-RF.OEQ1860891%3A0.STICKY_UNREACH":true,\r\n"BidCos-RF.OEQ1860891%3A1.INHIBIT":false,\r\n"HmIP-RF.000A570998B3FB%3A0.CONFIG_PENDING":false,\r\n"HmIP-RF.000A570998B3FB%3A0.UPDATE_PENDING":false,\r\n"HmIP-RF.000A5A4991BDDC%3A0.CONFIG_PENDING":false,\r\n"HmIP-RF.000A5A4991BDDC%3A0.UPDATE_PENDING":false,\r\n"BidCos-RF.NEQ1636407%3A1.STATE":0,\r\n"BidCos-RF.NEQ1636407%3A2.STATE":false,\r\n"BidCos-RF.NEQ1636407%3A2.INHIBIT":false,\r\n"CUxD.CUX2800001%3A12.TS":"0"}'
 
@@ -34,3 +36,18 @@ def test_defect_json() -> None:
         json = "{" + '"name": "Text mit Wert ' + sc + '"' + "}"
         with pytest.raises(orjson.JSONDecodeError):
             orjson.loads(json)
+
+
+@pytest.mark.parametrize(
+    (
+        "test_tag",
+        "expected_result",
+    ),
+    [
+        (" <>", " "),
+        ("Test1", "Test1"),
+    ],
+)
+def test_cleanup_html_tags(test_tag: str, expected_result: str) -> None:
+    """Test cleanup html tags."""
+    assert cleanup_text_from_html_tags(text=test_tag) == expected_result