Fix logic of constructing device name if cartridge type is not available

Author: elad-bar

CHANGELOG.md

@@ -1,5 +1,10 @@
 # Changelog
 
+## 2.0.0b7
+
+- Safe code blocks (try / catch / log) for generating entities
+- Fix logic of constructing device name if cartridge type is not available
+
 ## 2.0.0b6
 
 - When constructing device name, avoid null parts of it [#113](https://github.com/elad-bar/ha-hpprinter/issues/113)

custom_components/hpprinter/managers/ha_coordinator.py

@@ -1,4 +1,5 @@
 import logging
+import sys
 
 from homeassistant.core import Event
 from homeassistant.helpers.dispatcher import (
@@ -92,128 +93,178 @@ async def initialize(self):
         await self.async_config_entry_first_refresh()
 
     def create_main_device(
-        self, device_key: str, device_data: dict, _device_config: dict
+        self, device_key: str, device_data: dict, device_config: dict
     ):
-        self._main_device_data = device_data
-        self._main_device_id = device_key
+        try:
+            self._main_device_data = device_data
+            self._main_device_id = device_key
 
-        model = device_data.get("make_and_model")
-        serial_number = device_data.get("serial_number")
-        manufacturer = device_data.get("manufacturer_name")
+            model = device_data.get("make_and_model")
+            serial_number = device_data.get("serial_number")
+            manufacturer = device_data.get("manufacturer_name")
 
-        device_identifier = (DOMAIN, self._main_device_id)
+            device_identifier = (DOMAIN, self._main_device_id)
 
-        device_info = DeviceInfo(
-            identifiers={device_identifier},
-            name=self.entry_title,
-            model=model,
-            serial_number=serial_number,
-            manufacturer=manufacturer,
-        )
+            device_info = DeviceInfo(
+                identifiers={device_identifier},
+                name=self.entry_title,
+                model=model,
+                serial_number=serial_number,
+                manufacturer=manufacturer,
+            )
 
-        self._devices[device_key] = device_info
+            self._devices[device_key] = device_info
+
+        except Exception as ex:
+            exc_type, exc_obj, tb = sys.exc_info()
+            line_number = tb.tb_lineno
+            _LOGGER.error(
+                f"Failed to create main device, "
+                f"Device Key: {device_key}, "
+                f"Data: {device_data}, "
+                f"Config: {device_config}, "
+                f"Error: {ex}, "
+                f"Line: {line_number}"
+            )
 
     def create_sub_unit_device(
-        self, device_key: str, _device_data: dict, device_config: dict
+        self, device_key: str, device_data: dict, device_config: dict
     ):
-        model = self._main_device_data.get("make_and_model")
-        serial_number = self._main_device_data.get("serial_number")
-        manufacturer = self._main_device_data.get("manufacturer_name")
+        try:
+            model = self._main_device_data.get("make_and_model")
+            serial_number = self._main_device_data.get("serial_number")
+            manufacturer = self._main_device_data.get("manufacturer_name")
 
-        device_type = device_config.get("device_type")
+            device_type = device_config.get("device_type")
 
-        device_unique_id = slugify(f"{self.entry_id}.{device_key}")
+            device_unique_id = slugify(f"{self.entry_id}.{device_key}")
 
-        sub_unit_device_name = f"{self.entry_title} {device_type}"
+            sub_unit_device_name = f"{self.entry_title} {device_type}"
 
-        device_identifier = (DOMAIN, device_unique_id)
+            device_identifier = (DOMAIN, device_unique_id)
 
-        device_info = DeviceInfo(
-            identifiers={device_identifier},
-            name=sub_unit_device_name,
-            model=model,
-            serial_number=serial_number,
-            manufacturer=manufacturer,
-            via_device=(DOMAIN, self._main_device_id),
-        )
+            device_info = DeviceInfo(
+                identifiers={device_identifier},
+                name=sub_unit_device_name,
+                model=model,
+                serial_number=serial_number,
+                manufacturer=manufacturer,
+                via_device=(DOMAIN, self._main_device_id),
+            )
 
-        self._devices[device_key] = device_info
+            self._devices[device_key] = device_info
+
+        except Exception as ex:
+            exc_type, exc_obj, tb = sys.exc_info()
+            line_number = tb.tb_lineno
+            _LOGGER.error(
+                f"Failed to sub unit device, "
+                f"Device Key: {device_key}, "
+                f"Data: {device_data}, "
+                f"Config: {device_config}, "
+                f"Error: {ex}, "
+                f"Line: {line_number}"
+            )
 
     def create_consumable_device(
-        self, device_key: str, device_data: dict, _device_config: dict
+        self, device_key: str, device_data: dict, device_config: dict
     ):
-        printer_device_unique_id = slugify(f"{self.entry_id}.printer")
+        try:
+            printer_device_unique_id = slugify(f"{self.entry_id}.printer")
 
-        device_name_parts = [self.entry_title]
-        cartridge_type: str = device_data.get("consumable_type_enum")
-        cartridge_color = device_data.get("marker_color")
-        manufacturer = device_data.get("consumable_life_state_brand")
-        serial_number = device_data.get("serial_number")
+            device_name_parts = [self.entry_title]
+            cartridge_type: str = device_data.get("consumable_type_enum")
+            cartridge_color = device_data.get("marker_color")
+            manufacturer = device_data.get("consumable_life_state_brand")
+            serial_number = device_data.get("serial_number")
 
-        model = device_data.get("consumable_selectibility_number")
+            model = device_data.get("consumable_selectibility_number")
 
-        if cartridge_type == "printhead":
-            device_name_parts.append(cartridge_type.capitalize())
-            model = cartridge_type.capitalize()
+            if cartridge_type == "printhead":
+                model = cartridge_type.capitalize()
+            else:
+                device_name_parts.append(cartridge_color)
 
-        else:
-            device_name_parts.append(cartridge_color)
-            device_name_parts.append(cartridge_type.capitalize())
+            if cartridge_type is not None:
+                device_name_parts.append(cartridge_type.capitalize())
 
-        device_name_parts = [
-            device_name_part
-            for device_name_part in device_name_parts
-            if device_name_part is not None
-        ]
+            device_name_parts = [
+                device_name_part
+                for device_name_part in device_name_parts
+                if device_name_part is not None
+            ]
 
-        device_unique_id = slugify(f"{self.entry_id}.{device_key}")
+            device_unique_id = slugify(f"{self.entry_id}.{device_key}")
 
-        cartridge_device_name = " ".join(device_name_parts)
+            cartridge_device_name = " ".join(device_name_parts)
 
-        device_identifier = (DOMAIN, device_unique_id)
+            device_identifier = (DOMAIN, device_unique_id)
 
-        device_info = DeviceInfo(
-            identifiers={device_identifier},
-            name=cartridge_device_name,
-            model=model,
-            serial_number=serial_number,
-            manufacturer=manufacturer,
-            via_device=(DOMAIN, printer_device_unique_id),
-        )
+            device_info = DeviceInfo(
+                identifiers={device_identifier},
+                name=cartridge_device_name,
+                model=model,
+                serial_number=serial_number,
+                manufacturer=manufacturer,
+                via_device=(DOMAIN, printer_device_unique_id),
+            )
 
-        self._devices[device_key] = device_info
+            self._devices[device_key] = device_info
+
+        except Exception as ex:
+            exc_type, exc_obj, tb = sys.exc_info()
+            line_number = tb.tb_lineno
+            _LOGGER.error(
+                f"Failed to consumable device, "
+                f"Device Key: {device_key}, "
+                f"Data: {device_data}, "
+                f"Config: {device_config}, "
+                f"Error: {ex}, "
+                f"Line: {line_number}"
+            )
 
     def create_adapter_device(
         self, device_key: str, device_data: dict, device_config: dict
     ):
-        serial_number = self._main_device_data.get("serial_number")
-        manufacturer = self._main_device_data.get("manufacturer_name")
-
-        adapter_name = device_data.get("hardware_config_name").upper()
-        model = (
-            device_data.get("hardware_config_device_connectivity_port_type")
-            .replace("Embedded", "")
-            .upper()
-        )
+        try:
+            serial_number = self._main_device_data.get("serial_number")
+            manufacturer = self._main_device_data.get("manufacturer_name")
 
-        device_type = device_config.get("device_type")
+            adapter_name = device_data.get("hardware_config_name").upper()
 
-        device_unique_id = slugify(f"{self.entry_id}.{device_key}")
+            port_type_key = "hardware_config_device_connectivity_port_type"
+            model = device_data.get(port_type_key).replace("Embedded", "").upper()
 
-        adapter_device_name = f"{self.entry_title} {device_type} {adapter_name}"
+            device_type = device_config.get("device_type")
 
-        device_identifier = (DOMAIN, device_unique_id)
+            device_unique_id = slugify(f"{self.entry_id}.{device_key}")
 
-        device_info = DeviceInfo(
-            identifiers={device_identifier},
-            name=adapter_device_name,
-            model=model,
-            serial_number=serial_number,
-            manufacturer=manufacturer,
-            via_device=(DOMAIN, self._main_device_id),
-        )
+            adapter_device_name = f"{self.entry_title} {device_type} {adapter_name}"
+
+            device_identifier = (DOMAIN, device_unique_id)
 
-        self._devices[device_key] = device_info
+            device_info = DeviceInfo(
+                identifiers={device_identifier},
+                name=adapter_device_name,
+                model=model,
+                serial_number=serial_number,
+                manufacturer=manufacturer,
+                via_device=(DOMAIN, self._main_device_id),
+            )
+
+            self._devices[device_key] = device_info
+
+        except Exception as ex:
+            exc_type, exc_obj, tb = sys.exc_info()
+            line_number = tb.tb_lineno
+            _LOGGER.error(
+                f"Failed to adapter device, "
+                f"Device Key: {device_key}, "
+                f"Data: {device_data}, "
+                f"Config: {device_config}, "
+                f"Error: {ex}, "
+                f"Line: {line_number}"
+            )
 
     def get_device(self, device_key: str) -> DeviceInfo | None:
         result = self._devices.get(device_key)

custom_components/hpprinter/manifest.json

@@ -9,5 +9,5 @@
   "iot_class": "local_polling",
   "issue_tracker": "https://github.com/elad-bar/ha-hpprinter/issues",
   "requirements": ["xmltodict~=0.13.0", "flatten_json", "defusedxml"],
-  "version": "2.0.0b6"
+  "version": "2.0.0b7"
 }