Cleanup hub entity name if channel exists (#1966)

changelog.md

@@ -1,5 +1,7 @@
 # Version 2025.1.2 (2025-01-06)
 
+- Add legacy name for hub entities
+- Cleanup hub entity name if channel exists
 - Identify channel of a system variable:
   - name ends with channel address
   - name contains channel/device id

hahomematic/central/__init__.py

@@ -349,16 +349,16 @@ def version(self) -> str | None:
 
     def add_sysvar_data_point(self, sysvar_data_point: GenericSysvarDataPoint) -> None:
         """Add new program button."""
-        if (name := sysvar_data_point.name) is not None:
-            self._sysvar_data_points[name] = sysvar_data_point
+        if (vid := sysvar_data_point.vid) is not None:
+            self._sysvar_data_points[vid] = sysvar_data_point
         if sysvar_data_point.state_path not in self._sysvar_data_point_event_subscriptions:
             self._sysvar_data_point_event_subscriptions[sysvar_data_point.state_path] = sysvar_data_point.event
 
-    def remove_sysvar_data_point(self, name: str) -> None:
+    def remove_sysvar_data_point(self, vid: str) -> None:
         """Remove a sysvar data_point."""
-        if (sysvar_dp := self.get_sysvar_data_point(name=name)) is not None:
+        if (sysvar_dp := self.get_sysvar_data_point(vid=vid)) is not None:
             sysvar_dp.fire_device_removed_callback()
-            del self._sysvar_data_points[name]
+            del self._sysvar_data_points[vid]
             if sysvar_dp.state_path in self._sysvar_data_point_event_subscriptions:
                 del self._sysvar_data_point_event_subscriptions[sysvar_dp.state_path]
 
@@ -1207,18 +1207,18 @@ async def load_and_refresh_data_point_data(
             paramset_key=paramset_key, interface=interface, direct_call=direct_call
         )
 
-    async def get_system_variable(self, name: str) -> Any | None:
+    async def get_system_variable(self, legacy_name: str) -> Any | None:
         """Get system variable from CCU / Homegear."""
         if client := self.primary_client:
-            return await client.get_system_variable(name)
+            return await client.get_system_variable(legacy_name)
         return None
 
-    async def set_system_variable(self, name: str, value: Any) -> None:
+    async def set_system_variable(self, legacy_name: str, value: Any) -> None:
         """Set variable value on CCU/Homegear."""
-        if dp := self.get_sysvar_data_point(name=name):
+        if dp := self.get_sysvar_data_point(legacy_name=legacy_name):
             await dp.send_variable(value=value)
         else:
-            _LOGGER.warning("Variable %s not found on %s", name, self.name)
+            _LOGGER.warning("Variable %s not found on %s", legacy_name, self.name)
 
     def get_parameters(
         self,
@@ -1300,18 +1300,27 @@ def get_custom_data_point(self, address: str, channel_no: int) -> CustomDataPoin
             return device.get_custom_data_point(channel_no=channel_no)
         return None
 
-    def get_sysvar_data_point(self, name: str) -> GenericSysvarDataPoint | None:
+    def get_sysvar_data_point(
+        self, vid: str | None = None, legacy_name: str | None = None
+    ) -> GenericSysvarDataPoint | None:
         """Return the sysvar data_point."""
-        if sysvar := self._sysvar_data_points.get(name):
+        if vid and (sysvar := self._sysvar_data_points.get(vid)):
             return sysvar
-        for sysvar in self._sysvar_data_points.values():
-            if sysvar.name == name:
-                return sysvar
+        if legacy_name:
+            for sysvar in self._sysvar_data_points.values():
+                if sysvar.legacy_name == legacy_name:
+                    return sysvar
         return None
 
-    def get_program_data_point(self, pid: str) -> ProgramDpType | None:
+    def get_program_data_point(self, pid: str | None = None, legacy_name: str | None = None) -> ProgramDpType | None:
         """Return the program data points."""
-        return self._program_data_points.get(pid)
+        if pid and (program := self._program_data_points.get(pid)):
+            return program
+        if legacy_name:
+            for program in self._program_data_points.values():
+                if legacy_name in (program.button.legacy_name, program.switch.legacy_name):
+                    return program
+        return None
 
     def get_data_point_path(self) -> tuple[str, ...]:
         """Return the registered state path."""

hahomematic/client/__init__.py

@@ -363,7 +363,7 @@ async def set_program_state(self, pid: str, state: bool) -> bool:
         """Set the program state on CCU / Homegear."""
 
     @abstractmethod
-    async def set_system_variable(self, name: str, value: Any) -> bool:
+    async def set_system_variable(self, legacy_name: str, value: Any) -> bool:
         """Set a system variable on CCU / Homegear."""
 
     @abstractmethod
@@ -1068,9 +1068,9 @@ async def report_value_usage(self, address: str, value_id: str, ref_counter: int
             ) from ex
 
     @service(measure_performance=True)
-    async def set_system_variable(self, name: str, value: Any) -> bool:
+    async def set_system_variable(self, legacy_name: str, value: Any) -> bool:
         """Set a system variable on CCU / Homegear."""
-        return await self._json_rpc_client.set_system_variable(name=name, value=value)
+        return await self._json_rpc_client.set_system_variable(legacy_name=legacy_name, value=value)
 
     @service()
     async def delete_system_variable(self, name: str) -> bool:
@@ -1391,9 +1391,9 @@ async def set_program_state(self, pid: str, state: bool) -> bool:
         return True
 
     @service(measure_performance=True)
-    async def set_system_variable(self, name: str, value: Any) -> bool:
+    async def set_system_variable(self, legacy_name: str, value: Any) -> bool:
         """Set a system variable on CCU / Homegear."""
-        await self._proxy.setSystemVariable(name, value)
+        await self._proxy.setSystemVariable(legacy_name, value)
         return True
 
     @service()
@@ -1415,7 +1415,7 @@ async def get_all_system_variables(
         variables: list[SystemVariableData] = []
         if hg_variables := await self._proxy.getAllSystemVariables():
             for name, value in hg_variables.items():
-                variables.append(SystemVariableData(vid=name, name=name, value=value))
+                variables.append(SystemVariableData(vid=name, legacy_name=name, value=value))
         return tuple(variables)
 
     @service(re_raise=False, no_raise_return=())

hahomematic/client/json_rpc.py

@@ -510,9 +510,9 @@ async def set_program_state(self, pid: str, state: bool) -> bool:
 
         return True
 
-    async def set_system_variable(self, name: str, value: Any) -> bool:
+    async def set_system_variable(self, legacy_name: str, value: Any) -> bool:
         """Set a system variable on CCU / Homegear."""
-        params = {_JsonKey.NAME: name, _JsonKey.VALUE: value}
+        params = {_JsonKey.NAME: legacy_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)
@@ -579,10 +579,10 @@ async def get_all_system_variables(
                 enabled_default = False
                 extended_sysvar = False
                 var_id = var[_JsonKey.ID]
-                name = var[_JsonKey.NAME]
+                legacy_name = var[_JsonKey.NAME]
                 is_internal = var[_JsonKey.IS_INTERNAL]
-                if new_name := RENAME_SYSVAR_BY_NAME.get(name):
-                    name = new_name
+                if new_name := RENAME_SYSVAR_BY_NAME.get(legacy_name):
+                    legacy_name = new_name
                 if var_id in ALWAYS_ENABLE_SYSVARS_BY_ID:
                     enabled_default = True
 
@@ -634,7 +634,7 @@ async def get_all_system_variables(
                     variables.append(
                         SystemVariableData(
                             vid=var_id,
-                            name=name,
+                            legacy_name=legacy_name,
                             data_type=data_type,
                             description=description,
                             unit=unit,
@@ -651,7 +651,7 @@ async def get_all_system_variables(
                         "GET_ALL_SYSTEM_VARIABLES failed: %s [%s] Failed to parse SysVar %s ",
                         vterr.__class__.__name__,
                         reduce_args(args=vterr.args),
-                        name,
+                        legacy_name,
                     )
 
         return tuple(variables)
@@ -999,7 +999,7 @@ async def get_all_programs(self, markers: tuple[DescriptionMarker | str, ...]) -
                 all_programs.append(
                     ProgramData(
                         pid=pid,
-                        name=name,
+                        legacy_name=name,
                         description=description,
                         is_active=is_active,
                         is_internal=is_internal,

hahomematic/const.py

@@ -633,7 +633,7 @@ class DataPointKey(NamedTuple):
 class HubData:
     """Dataclass for hub data points."""
 
-    name: str
+    legacy_name: str
     enabled_default: bool = False
     description: str | None = None
 

hahomematic/model/hub/__init__.py

@@ -145,17 +145,14 @@ async def _update_sysvar_data_points(self) -> None:
         if self._central.model is Backend.CCU:
             variables = _clean_variables(variables)
 
-        if missing_variable_names := self._identify_missing_variable_names(variables=variables):
-            self._remove_sysvar_data_point(del_data_points=missing_variable_names)
+        if missing_variable_ids := self._identify_missing_variable_ids(variables=variables):
+            self._remove_sysvar_data_point(del_data_point_ids=missing_variable_ids)
 
         new_sysvars: list[GenericSysvarDataPoint] = []
 
         for sysvar in variables:
-            name = sysvar.name
-            value = sysvar.value
-
-            if dp := self._central.get_sysvar_data_point(name=name):
-                dp.write_value(value)
+            if dp := self._central.get_sysvar_data_point(vid=sysvar.vid):
+                dp.write_value(sysvar.value)
             else:
                 new_sysvars.append(self._create_system_variable(data=sysvar))
 
@@ -204,10 +201,10 @@ def _remove_program_data_point(self, ids: set[str]) -> None:
         for pid in ids:
             self._central.remove_program_button(pid=pid)
 
-    def _remove_sysvar_data_point(self, del_data_points: set[str]) -> None:
+    def _remove_sysvar_data_point(self, del_data_point_ids: set[str]) -> None:
         """Remove sysvar data_point from hub."""
-        for name in del_data_points:
-            self._central.remove_sysvar_data_point(name=name)
+        for vid in del_data_point_ids:
+            self._central.remove_sysvar_data_point(vid=vid)
 
     def _identify_missing_program_ids(self, programs: tuple[ProgramData, ...]) -> set[str]:
         """Identify missing programs."""
@@ -217,18 +214,18 @@ def _identify_missing_program_ids(self, programs: tuple[ProgramData, ...]) -> se
             if program_dp.pid not in [x.pid for x in programs]
         }
 
-    def _identify_missing_variable_names(self, variables: tuple[SystemVariableData, ...]) -> set[str]:
+    def _identify_missing_variable_ids(self, variables: tuple[SystemVariableData, ...]) -> set[str]:
         """Identify missing variables."""
-        variable_names: dict[str, bool] = {x.name: x.extended_sysvar for x in variables}
-        missing_variables: list[str] = []
+        variable_ids: dict[str, bool] = {x.vid: x.extended_sysvar for x in variables}
+        missing_variable_ids: list[str] = []
         for sysvar_data_point in self._central.sysvar_data_points:
             if sysvar_data_point.data_type == SysvarType.STRING:
                 continue
-            if (name := sysvar_data_point.name) is not None and (
-                name not in variable_names or (sysvar_data_point.is_extended is not variable_names.get(name))
+            if (vid := sysvar_data_point.vid) is not None and (
+                vid not in variable_ids or (sysvar_data_point.is_extended is not variable_ids.get(vid))
             ):
-                missing_variables.append(name)
-        return set(missing_variables)
+                missing_variable_ids.append(vid)
+        return set(missing_variable_ids)
 
 
 def _is_excluded(variable: str, excludes: list[str]) -> bool:
@@ -238,7 +235,7 @@ def _is_excluded(variable: str, excludes: list[str]) -> bool:
 
 def _clean_variables(variables: tuple[SystemVariableData, ...]) -> tuple[SystemVariableData, ...]:
     """Clean variables by removing excluded."""
-    return tuple(sv for sv in variables if not _is_excluded(sv.name, _EXCLUDED))
+    return tuple(sv for sv in variables if not _is_excluded(sv.legacy_name, _EXCLUDED))
 
 
 def _get_new_hub_data_points(

hahomematic/model/hub/data_point.py

@@ -20,7 +20,14 @@
 from hahomematic.model.data_point import CallbackDataPoint
 from hahomematic.model.decorators import config_property, state_property
 from hahomematic.model.device import Channel
-from hahomematic.model.support import PathData, PayloadMixin, ProgramPathData, SysvarPathData, generate_unique_id
+from hahomematic.model.support import (
+    PathData,
+    PayloadMixin,
+    ProgramPathData,
+    SysvarPathData,
+    generate_unique_id,
+    get_hub_data_point_name_data,
+)
 from hahomematic.support import parse_sys_var
 
 
@@ -38,21 +45,28 @@ def __init__(
         unique_id: Final = generate_unique_id(
             central=central,
             address=address,
-            parameter=slugify(data.name),
+            parameter=slugify(data.legacy_name),
+        )
+        self._legacy_name = data.legacy_name
+        self._channel = central.identify_channel(text=data.legacy_name)
+        self._name_data: Final = get_hub_data_point_name_data(
+            channel=self._channel, legacy_name=data.legacy_name, central_name=central.name
         )
-        self._name: Final = data.name
         self._description = data.description
         super().__init__(central=central, unique_id=unique_id)
-        self._full_name: Final = f"{self._central.name}_{self._name}"
         self._enabled_default: Final = data.enabled_default
-        self._channel = self._central.identify_channel(text=data.name)
         self._state_uncertain: bool = True
 
     @state_property
     def available(self) -> bool:
         """Return the availability of the device."""
         return self.central.available
 
+    @property
+    def legacy_name(self) -> str | None:
+        """Return the original sysvar name."""
+        return self._legacy_name
+
     @property
     def channel(self) -> Channel | None:
         """Return the identified channel."""
@@ -66,7 +80,7 @@ def description(self) -> str | None:
     @property
     def full_name(self) -> str:
         """Return the fullname of the data_point."""
-        return self._full_name
+        return self._name_data.full_name
 
     @property
     def enabled_default(self) -> bool:
@@ -76,7 +90,7 @@ def enabled_default(self) -> bool:
     @config_property
     def name(self) -> str:
         """Return the name of the data_point."""
-        return self._name
+        return self._name_data.name
 
     @property
     def state_uncertain(self) -> bool:
@@ -226,7 +240,7 @@ def _convert_value(self, old_value: Any, new_value: Any) -> Any:
     async def send_variable(self, value: Any) -> None:
         """Set variable value on CCU/Homegear."""
         if client := self.central.primary_client:
-            await client.set_system_variable(name=self._name, value=parse_sys_var(self._data_type, value))
+            await client.set_system_variable(legacy_name=self._legacy_name, value=parse_sys_var(self._data_type, value))
         self._write_temporary_value(value=value)
 
 

hahomematic/model/hub/sensor.py

@@ -26,7 +26,7 @@ def value(self) -> Any | None:
             and (value := get_value_from_value_list(value=self._value, value_list=self.values)) is not None
         ):
             return value
-        return _check_length_and_log(name=self._name, value=self._value)
+        return _check_length_and_log(name=self._legacy_name, value=self._value)
 
 
 def _check_length_and_log(name: str | None, value: Any) -> Any: