Blacked and removed yamil config. Added types to coordinator.py

Signed-off-by: [email protected] <[email protected]>

custom_components/mqtt_vacuum_camera/__init__.py

@@ -6,21 +6,22 @@
 
 from homeassistant import config_entries, core
 from homeassistant.components import mqtt
+from homeassistant.config_entries import ConfigEntry, ConfigEntryState
 from homeassistant.const import (
     CONF_UNIQUE_ID,
     EVENT_HOMEASSISTANT_FINAL_WRITE,
     SERVICE_RELOAD,
     Platform,
 )
 from homeassistant.core import ServiceCall
-from homeassistant.config_entries import ConfigEntryState, ConfigEntry
 from homeassistant.exceptions import ConfigEntryNotReady
+from homeassistant.helpers import config_validation as cv
 from homeassistant.helpers.reload import async_register_admin_service
 from homeassistant.helpers.storage import STORAGE_DIR
 
 from .common import (
-    get_vacuum_device_info,
     get_entity_identifier_from_mqtt,
+    get_vacuum_device_info,
     get_vacuum_mqtt_topic,
     get_vacuum_unique_id_from_mqtt_topic,
     update_options,
@@ -40,6 +41,7 @@
 )
 
 PLATFORMS = [Platform.CAMERA]
+CONFIG_SCHEMA = cv.config_entry_only_config_schema
 _LOGGER = logging.getLogger(__name__)
 
 

custom_components/mqtt_vacuum_camera/camera.py

@@ -19,43 +19,31 @@
 
 from PIL import Image
 from homeassistant import config_entries, core
-from homeassistant.components.camera import PLATFORM_SCHEMA, Camera, CameraEntityFeature
-from homeassistant.helpers.update_coordinator import CoordinatorEntity
-from homeassistant.const import CONF_NAME, CONF_UNIQUE_ID, MATCH_ALL
+from homeassistant.components.camera import Camera, CameraEntityFeature
+from homeassistant.const import CONF_UNIQUE_ID, MATCH_ALL
 from homeassistant.helpers import config_validation as cv
 from homeassistant.helpers.storage import STORAGE_DIR
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
 from psutil_home_assistant import PsutilWrapper as ProcInsp
-import voluptuous as vol
 
+from .camera_processing import CameraProcessor
 from .common import get_vacuum_unique_id_from_mqtt_topic
 from .const import (
     ATTR_FRIENDLY_NAME,
     ATTR_JSON_DATA,
-    ATTR_ROTATE,
     ATTR_SNAPSHOT_PATH,
     ATTR_VACUUM_TOPIC,
     CAMERA_STORAGE,
-    CONF_VACUUM_CONNECTION_STRING,
-    CONF_VACUUM_ENTITY_ID,
     CONF_VACUUM_IDENTIFIERS,
-    DEFAULT_NAME,
     DOMAIN,
     NOT_STREAMING_STATES,
 )
+from .snapshots.snapshot import Snapshots
 from .types import SnapshotStore
 from .utils.colors_man import ColorsManagment
 from .utils.files_operations import async_get_active_user_language, is_auth_updated
-from .snapshots.snapshot import Snapshots
-from .camera_processing import CameraProcessor
 
-PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
-    {
-        vol.Required(CONF_VACUUM_CONNECTION_STRING): cv.string,
-        vol.Required(CONF_VACUUM_ENTITY_ID): cv.string,
-        vol.Required(ATTR_ROTATE, default="0"): cv.string,
-        vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.entity_id,
-    }
-)
+CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
 
 SCAN_INTERVAL = timedelta(seconds=3)
 _LOGGER = logging.getLogger(__name__)

custom_components/mqtt_vacuum_camera/coordinator.py

@@ -2,21 +2,19 @@
 
 from datetime import timedelta
 import logging
+from typing import Optional
 
 import async_timeout
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
 from homeassistant.exceptions import ConfigEntryAuthFailed
+from homeassistant.helpers.device_registry import DeviceInfo
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
 
-from homeassistant.helpers.update_coordinator import (
-    DataUpdateCoordinator,
-    UpdateFailed,
-)
-
-from .camera_shared import CameraSharedManager
-from .const import DEFAULT_NAME
+from .camera_shared import CameraShared, CameraSharedManager
 from .common import get_camera_device_info
+from .const import DEFAULT_NAME
 from .valetudo.MQTT.connector import ValetudoConnector
-# from .snapshots.snapshot import Snapshots
-# from .camera_processing import CameraProcessor
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -26,8 +24,8 @@ class MQTTVacuumCoordinator(DataUpdateCoordinator):
 
     def __init__(
         self,
-        hass,
-        entry,
+        hass: HomeAssistant,
+        entry: ConfigEntry,
         vacuum_topic: str,
         polling_interval=timedelta(seconds=3),
     ):
@@ -38,59 +36,82 @@ def __init__(
             name=DEFAULT_NAME,
             update_interval=polling_interval,
         )
-        self.hass = hass
-        self.vacuum_topic = vacuum_topic
-        self.device_entity = entry
-        self.device_info = get_camera_device_info(hass, self.device_entity)
-        self.shared_manager = None
+        self.hass: HomeAssistant = hass
+        self.vacuum_topic: str = vacuum_topic
+        self.device_entity: ConfigEntry = entry
+        self.device_info: DeviceInfo = get_camera_device_info(hass, self.device_entity)
+        self.shared_manager: Optional[CameraSharedManager] = None
+        self.shared: Optional[CameraShared] = None
+        self.file_name: str = ""
+        self.connector: Optional[ValetudoConnector] = None
+        self.in_sync_with_camera: bool = False
+
+        # Initialize shared data and MQTT connector
         self.shared, self.file_name = self._init_shared_data(self.vacuum_topic)
-        self.connector = None
-        # self.snapshot = Snapshots(hass, self.shared)
-        # self.camera_processor = CameraProcessor(hass, self.shared)
         self.stat_up_mqtt()
 
+    def _init_shared_data(self, mqtt_listen_topic: str) -> tuple[CameraShared, str]:
+        """
+        Initialize the shared data.
 
+        Args:
+            mqtt_listen_topic (str): The topic to listen for MQTT messages.
+
+        Returns:
+            tuple[CameraShared, str]: The CameraShared instance and file name.
+        """
+        shared = None
+        file_name = None
 
-    def _init_shared_data(self, mqtt_listen_topic: str):
-        """Initialize the shared data."""
-        shared, file_name = None, None
         if mqtt_listen_topic and not self.shared_manager:
             file_name = mqtt_listen_topic.split("/")[1].lower()
             self.shared_manager = CameraSharedManager(file_name, self.device_info)
             shared = self.shared_manager.get_instance()
             _LOGGER.debug(f"Camera {file_name} Starting up..")
+
         return shared, file_name
 
-    async def _async_setup(self):
-        """Set up the coordinator."""
-        self._device = self.device_info
-        _LOGGER.info(f"Setting up coordinator for {self.file_name}.")
+    def stat_up_mqtt(self) -> ValetudoConnector:
+        """
+        Initialize the MQTT Connector.
 
-    def stat_up_mqtt(self):
-        """Init the MQTT Connector"""
+        Returns:
+            ValetudoConnector: The initialized MQTT connector.
+        """
         self.connector = ValetudoConnector(self.vacuum_topic, self.hass, self.shared)
         return self.connector
 
-    def update_shared_data(self, dev_info):
-        """Create / update instance of the shared data"""
+    def update_shared_data(self, dev_info: DeviceInfo) -> tuple[CameraShared, str]:
+        """
+        Create or update the instance of the shared data.
+
+        Args:
+            dev_info (DeviceInfo): The device information to update.
+
+        Returns:
+            tuple[CameraShared, str]: Updated shared data and file name.
+        """
         self.shared_manager.update_shared_data(dev_info)
         self.shared = self.shared_manager.get_instance()
+        self.in_sync_with_camera = True
         return self.shared, self.file_name
 
     async def _async_update_data(self, process: bool = True):
-        """Fetch data from the MQTT topics."""
+        """
+        Fetch data from the MQTT topics.
+
+        Args:
+            process (bool): Whether to process the data (default: True).
+
+        Returns:
+            The fetched data from MQTT.
+        """
         try:
-            # conside adding shared updates here ***
             async with async_timeout.timeout(10):
-                # Fetch and process data from the MQTT connector
+                # Fetch and process maps data from the MQTT connector
                 return await self.connector.update_data(process)
         except ConfigEntryAuthFailed as err:
             raise ConfigEntryAuthFailed from err
         except Exception as err:
             _LOGGER.error(f"Error communicating with MQTT or processing data: {err}")
             raise UpdateFailed(f"Error communicating with MQTT: {err}") from err
-
-    async def async_will_remove_from_hass(self):
-        """Handle cleanup when the coordinator is removed."""
-        _LOGGER.info(f"Cleaning up {self.file_name} coordinator.")
-        await self.connector.async_unsubscribe_from_topics()

custom_components/mqtt_vacuum_camera/valetudo/MQTT/connector.py

@@ -427,5 +427,4 @@ async def publish_to_broker(self, cust_topic: str, cust_payload: dict) -> None:
             self._mqtt_topic + cust_topic,
             payload,
             _QOS,
-            encoding="utf-8",
         )

custom_components/mqtt_vacuum_camera/valetudo/hypfer/image_handler.py

@@ -208,7 +208,9 @@ async def async_get_image_from_json(
                     self.img_base_layer = await self.imd.async_copy_array(img_np_array)
                 self.shared.frame_number = self.frame_number
                 self.frame_number += 1
-                if (self.frame_number >= self.max_frames) or (new_frame_hash != self.img_hash):
+                if (self.frame_number >= self.max_frames) or (
+                    new_frame_hash != self.img_hash
+                ):
                     self.frame_number = 0
                 _LOGGER.debug(
                     f"{self.file_name}: {self.json_id} at Frame Number: {self.frame_number}"