Enhanced initialization of shared data #250 from sca075/refactoring_camera

Final Adjustments camera.py

Author: sca075

custom_components/mqtt_vacuum_camera/camera.py

@@ -33,29 +33,15 @@
 from .camera_shared import CameraSharedManager
 from .common import get_vacuum_unique_id_from_mqtt_topic
 from .const import (
-    ATTR_MARGINS,
-    ATTR_ROTATE,
-    ATTR_VACUUM_TOPIC,
+    ATTR_FRIENDLY_NAME,
     ATTR_JSON_DATA,
+    ATTR_ROTATE,
     ATTR_SNAPSHOT_PATH,
-    ATTR_FRIENDLY_NAME,
+    ATTR_VACUUM_TOPIC,
     CAMERA_STORAGE,
-    CONF_ASPECT_RATIO,
-    CONF_AUTO_ZOOM,
-    CONF_OFFSET_BOTTOM,
-    CONF_OFFSET_LEFT,
-    CONF_OFFSET_RIGHT,
-    CONF_OFFSET_TOP,
-    CONF_SNAPSHOTS_ENABLE,
-    CONF_VAC_STAT,
-    CONF_VAC_STAT_FONT,
-    CONF_VAC_STAT_POS,
-    CONF_VAC_STAT_SIZE,
     CONF_VACUUM_CONNECTION_STRING,
     CONF_VACUUM_ENTITY_ID,
     CONF_VACUUM_IDENTIFIERS,
-    CONF_ZOOM_LOCK_RATIO,
-    DEFAULT_VALUES,
     DEFAULT_NAME,
     DOMAIN,
     NOT_STREAMING_STATES,
@@ -128,15 +114,11 @@ def __init__(self, hass, device_info):
         self._directory_path = self.hass.config.path()  # get Home Assistant path
         self._mqtt_listen_topic = str(device_info.get(CONF_VACUUM_CONNECTION_STRING))
         self._shared, self._file_name = self._handle_init_shared_data(
-            self._mqtt_listen_topic
+            self._mqtt_listen_topic,
+            device_info,
         )
         self._start_up_logs()
-        self._init_shared_data(device_info)
-        self._storage_path = f"{self.hass.config.path(STORAGE_DIR)}/{CAMERA_STORAGE}"
-        if not os.path.exists(self._storage_path):
-            self._storage_path = f"{self._directory_path}/{STORAGE_DIR}"
-        self.snapshot_img = f"{self._storage_path}/{self._file_name}.png"
-        self.log_file = f"{self._storage_path}/{self._file_name}.zip"
+        self._storage_path, self.snapshot_img, self.log_file = self._init_paths()
         self._attr_unique_id = device_info.get(
             CONF_UNIQUE_ID,
             get_vacuum_unique_id_from_mqtt_topic(self._mqtt_listen_topic),
@@ -153,9 +135,7 @@ def __init__(self, hass, device_info):
         self._attr_frame_interval = 6
         self._vac_json_available = None
         self._cpu_percent = None
-        # If there is a log zip in www remove it
-        if os.path.isfile(self.log_file):
-            os.remove(self.log_file)
+        self._init_clear_www_folder()
         self._last_image = None
         self._update_time = None
         self._rrm_data = False  # Check for rrm data
@@ -167,11 +147,13 @@ def __init__(self, hass, device_info):
         self._attr_brand = "MQTT Vacuum Camera"
 
     @staticmethod
-    def _handle_init_shared_data(mqtt_listen_topic: str):
+    def _handle_init_shared_data(mqtt_listen_topic: str, device_info):
         """Handle the shared data initialization."""
         manager, shared, file_name = None, None, None
         if mqtt_listen_topic:
-            manager = CameraSharedManager(mqtt_listen_topic.split("/")[1].lower())
+            manager = CameraSharedManager(
+                mqtt_listen_topic.split("/")[1].lower(), device_info
+            )
             shared = manager.get_instance()
             file_name = shared.file_name
             _LOGGER.debug(f"Camera {file_name} Starting up..")
@@ -189,80 +171,24 @@ def _start_up_logs():
             f" and In Use: {round((ProcInsp().psutil.virtual_memory().used / (1024 * 1024)), 1)}"
         )
 
-    def _init_shared_data(self, device_info):
-        if self._shared:
-            try:
-                self._shared.attr_calibration_points = None
-
-                # Initialize shared data with defaults from DEFAULT_VALUES
-                self._shared.offset_top = device_info.get(
-                    CONF_OFFSET_TOP, DEFAULT_VALUES["offset_top"]
-                )
-                self._shared.offset_down = device_info.get(
-                    CONF_OFFSET_BOTTOM, DEFAULT_VALUES["offset_bottom"]
-                )
-                self._shared.offset_left = device_info.get(
-                    CONF_OFFSET_LEFT, DEFAULT_VALUES["offset_left"]
-                )
-                self._shared.offset_right = device_info.get(
-                    CONF_OFFSET_RIGHT, DEFAULT_VALUES["offset_right"]
-                )
-                self._shared.image_auto_zoom = device_info.get(
-                    CONF_AUTO_ZOOM, DEFAULT_VALUES["auto_zoom"]
-                )
-                self._shared.image_zoom_lock_ratio = device_info.get(
-                    CONF_ZOOM_LOCK_RATIO, DEFAULT_VALUES["zoom_lock_ratio"]
-                )
-                self._shared.image_aspect_ratio = device_info.get(
-                    CONF_ASPECT_RATIO, DEFAULT_VALUES["aspect_ratio"]
-                )
-                self._shared.image_rotate = int(
-                    device_info.get(ATTR_ROTATE, DEFAULT_VALUES["rotate_image"])
-                )
-                self._shared.margins = int(
-                    device_info.get(ATTR_MARGINS, DEFAULT_VALUES["margins"])
-                )
-                self._shared.show_vacuum_state = device_info.get(
-                    CONF_VAC_STAT, DEFAULT_VALUES["show_vac_status"]
-                )
-                self._shared.vacuum_status_font = device_info.get(
-                    CONF_VAC_STAT_FONT, DEFAULT_VALUES["vac_status_font"]
-                )
-                self._shared.vacuum_status_size = device_info.get(
-                    CONF_VAC_STAT_SIZE, DEFAULT_VALUES["vac_status_size"]
-                )
-                self._shared.vacuum_status_position = device_info.get(
-                    CONF_VAC_STAT_POS, DEFAULT_VALUES["vac_status_position"]
-                )
-
-                # If enable_snapshots check if for png in www.
-                self._shared.enable_snapshots = device_info.get(
-                    CONF_SNAPSHOTS_ENABLE, DEFAULT_VALUES["enable_www_snapshots"]
-                )
-
-                if not self._shared.enable_snapshots and os.path.isfile(
-                    f"{self._directory_path}/www/snapshot_{self._file_name}.png"
-                ):
-                    os.remove(
-                        f"{self._directory_path}/www/snapshot_{self._file_name}.png"
-                    )
+    def _init_clear_www_folder(self):
+        # If enable_snapshots check if for png in www
+        if not self._shared.enable_snapshots and os.path.isfile(
+            f"{self._directory_path}/www/snapshot_{self._file_name}.png"
+        ):
+            os.remove(f"{self._directory_path}/www/snapshot_{self._file_name}.png")
+        # If there is a log zip in www remove it
+        if os.path.isfile(self.log_file):
+            os.remove(self.log_file)
 
-            except TypeError as ex:
-                _LOGGER.error(
-                    f"Shared data can't be initialized due to a TypeError! {ex}"
-                )
-            except AttributeError as ex:
-                _LOGGER.error(
-                    f"Shared data can't be initialized due to an AttributeError! Possibly _shared is not properly initialized: {ex}"
-                )
-            except Exception as ex:
-                _LOGGER.error(
-                    f"An unexpected error occurred while initializing shared data: {ex}"
-                )
-        else:
-            _LOGGER.error(
-                "Shared data initialization failed because _shared is not defined."
-            )
+    def _init_paths(self):
+        """Initialize Camera Paths"""
+        storage_path = f"{self.hass.config.path(STORAGE_DIR)}/{CAMERA_STORAGE}"
+        if not os.path.exists(storage_path):
+            storage_path = f"{self._directory_path}/{STORAGE_DIR}"
+        snapshot_img = f"{storage_path}/{self._file_name}.png"
+        log_file = f"{storage_path}/{self._file_name}.zip"
+        return storage_path, snapshot_img, log_file
 
     async def async_added_to_hass(self) -> None:
         """Handle entity added to Home Assistant."""

custom_components/mqtt_vacuum_camera/camera_shared.py

@@ -5,20 +5,39 @@
 """
 
 import asyncio
+import logging
 
 from custom_components.mqtt_vacuum_camera.types import Colors
+
 from .const import (
     ATTR_CALIBRATION_POINTS,
-    ATTR_ROOMS,
-    ATTR_ZONES,
+    ATTR_MARGINS,
     ATTR_POINTS,
+    ATTR_ROOMS,
+    ATTR_ROTATE,
     ATTR_SNAPSHOT,
     ATTR_VACUUM_BATTERY,
     ATTR_VACUUM_JSON_ID,
     ATTR_VACUUM_POSITION,
     ATTR_VACUUM_STATUS,
+    ATTR_ZONES,
+    CONF_ASPECT_RATIO,
+    CONF_AUTO_ZOOM,
+    CONF_OFFSET_BOTTOM,
+    CONF_OFFSET_LEFT,
+    CONF_OFFSET_RIGHT,
+    CONF_OFFSET_TOP,
+    CONF_SNAPSHOTS_ENABLE,
+    CONF_VAC_STAT,
+    CONF_VAC_STAT_FONT,
+    CONF_VAC_STAT_POS,
+    CONF_VAC_STAT_SIZE,
+    CONF_ZOOM_LOCK_RATIO,
+    DEFAULT_VALUES,
 )
 
+_LOGGER = logging.getLogger(__name__)
+
 
 class CameraShared(object):
     """
@@ -134,10 +153,78 @@ def generate_attributes(self) -> dict:
 class CameraSharedManager:
     """Camera Shared Manager class."""
 
-    def __init__(self, file_name):
+    def __init__(self, file_name, device_info):
         self._instances = {}
         self._lock = asyncio.Lock()
         self.file_name = file_name
+        self.device_info = device_info
+
+        # Automatically initialize shared data for the instance
+        self._init_shared_data(device_info)
+
+    def _init_shared_data(self, device_info):
+        """Initialize the shared data with device_info."""
+        instance = self.get_instance()  # Retrieve the correct instance
+
+        try:
+            instance.attr_calibration_points = None
+
+            # Initialize shared data with defaults from DEFAULT_VALUES
+            instance.offset_top = device_info.get(
+                CONF_OFFSET_TOP, DEFAULT_VALUES["offset_top"]
+            )
+            instance.offset_down = device_info.get(
+                CONF_OFFSET_BOTTOM, DEFAULT_VALUES["offset_bottom"]
+            )
+            instance.offset_left = device_info.get(
+                CONF_OFFSET_LEFT, DEFAULT_VALUES["offset_left"]
+            )
+            instance.offset_right = device_info.get(
+                CONF_OFFSET_RIGHT, DEFAULT_VALUES["offset_right"]
+            )
+            instance.image_auto_zoom = device_info.get(
+                CONF_AUTO_ZOOM, DEFAULT_VALUES["auto_zoom"]
+            )
+            instance.image_zoom_lock_ratio = device_info.get(
+                CONF_ZOOM_LOCK_RATIO, DEFAULT_VALUES["zoom_lock_ratio"]
+            )
+            instance.image_aspect_ratio = device_info.get(
+                CONF_ASPECT_RATIO, DEFAULT_VALUES["aspect_ratio"]
+            )
+            instance.image_rotate = int(
+                device_info.get(ATTR_ROTATE, DEFAULT_VALUES["rotate_image"])
+            )
+            instance.margins = int(
+                device_info.get(ATTR_MARGINS, DEFAULT_VALUES["margins"])
+            )
+            instance.show_vacuum_state = device_info.get(
+                CONF_VAC_STAT, DEFAULT_VALUES["show_vac_status"]
+            )
+            instance.vacuum_status_font = device_info.get(
+                CONF_VAC_STAT_FONT, DEFAULT_VALUES["vac_status_font"]
+            )
+            instance.vacuum_status_size = device_info.get(
+                CONF_VAC_STAT_SIZE, DEFAULT_VALUES["vac_status_size"]
+            )
+            instance.vacuum_status_position = device_info.get(
+                CONF_VAC_STAT_POS, DEFAULT_VALUES["vac_status_position"]
+            )
+
+            # If enable_snapshots, check for png in www.
+            instance.enable_snapshots = device_info.get(
+                CONF_SNAPSHOTS_ENABLE, DEFAULT_VALUES["enable_www_snapshots"]
+            )
+
+        except TypeError as ex:
+            _LOGGER.error(f"Shared data can't be initialized due to a TypeError! {ex}")
+        except AttributeError as ex:
+            _LOGGER.error(
+                f"Shared data can't be initialized due to an AttributeError! Possibly _shared is not properly initialized: {ex}"
+            )
+        except Exception as ex:
+            _LOGGER.error(
+                f"An unexpected error occurred while initializing shared data: {ex}"
+            )
 
     def get_instance(self):
         """Get the shared instance."""