shared updated

SCR/valetudo_map_parser/config/drawable.py

@@ -15,7 +15,7 @@
 
 import numpy as np
 from mvcrender.blend import get_blended_color, sample_and_blend_color
-from mvcrender.draw import circle_u8, line_u8
+from mvcrender.draw import circle_u8, line_u8, polygon_u8
 from PIL import Image, ImageDraw, ImageFont
 
 from .types import Color, NumpyArray, PilPNG, Point, Tuple, Union
@@ -129,7 +129,7 @@ async def go_to_flag(
         """
         Draw a flag centered at specified coordinates on the input layer.
         It uses the rotation angle of the image to orient the flag.
-        Includes color blending for better visual integration.
+        Uses mvcrender's polygon_u8 for efficient triangle drawing.
         """
         # Check if coordinates are within bounds
         height, width = layer.shape[:2]
@@ -194,9 +194,12 @@ async def go_to_flag(
             xp1, yp1 = center[0] - (pole_width // 2), y1
             xp2, yp2 = center[0] - (pole_width // 2), center[1] + flag_size
 
-        # Draw flag outline using _polygon_outline
-        points = [(x1, y1), (x2, y2), (x3, y3)]
-        layer = Drawable._polygon_outline(layer, points, 1, flag_color, flag_color)
+        # Draw flag triangle using mvcrender's polygon_u8 (much faster than _polygon_outline)
+        xs = np.array([x1, x2, x3], dtype=np.int32)
+        ys = np.array([y1, y2, y3], dtype=np.int32)
+        # Draw filled triangle with thin outline
+        polygon_u8(layer, xs, ys, flag_color, 1, flag_color)
+
         # Draw pole using _line
         layer = Drawable._line(layer, xp1, yp1, xp2, yp2, pole_color, pole_width)
         return layer
@@ -378,24 +381,26 @@ def _polygon_outline(
     @staticmethod
     async def zones(layers: NumpyArray, coordinates, color: Color) -> NumpyArray:
         """
-        Draw zones as solid filled polygons with alpha blending using a per-zone mask.
-        Keeps API the same; no dotted rendering.
+        Draw zones as filled polygons with alpha blending using mvcrender.
+        Creates a mask with polygon_u8 and blends it onto the image with proper alpha.
+        This eliminates PIL dependency for zone drawing.
         """
         if not coordinates:
             return layers
 
         height, width = layers.shape[:2]
-        # Precompute color and alpha
         r, g, b, a = color
         alpha = a / 255.0
         inv_alpha = 1.0 - alpha
+        # Pre-allocate color array once (avoid creating it in every iteration)
         color_rgb = np.array([r, g, b], dtype=np.float32)
 
         for zone in coordinates:
             try:
                 pts = zone["points"]
             except (KeyError, TypeError):
                 continue
+
             if not pts or len(pts) < 6:
                 continue
 
@@ -407,29 +412,48 @@ async def zones(layers: NumpyArray, coordinates, color: Color) -> NumpyArray:
             if min_x >= max_x or min_y >= max_y:
                 continue
 
-            # Adjust polygon points to local bbox coordinates
-            poly_xy = [
-                (int(pts[i] - min_x), int(pts[i + 1] - min_y))
-                for i in range(0, len(pts), 2)
-            ]
             box_w = max_x - min_x + 1
             box_h = max_y - min_y + 1
 
-            # Build mask via PIL polygon fill (fast, C-impl)
-            mask_img = Image.new("L", (box_w, box_h), 0)
-            draw = ImageDraw.Draw(mask_img)
-            draw.polygon(poly_xy, fill=255)
-            zone_mask = np.array(mask_img, dtype=bool)
+            # Create mask using mvcrender's polygon_u8
+            mask_rgba = np.zeros((box_h, box_w, 4), dtype=np.uint8)
+
+            # Convert points to xs, ys arrays (adjusted to local bbox coordinates)
+            xs = np.array([int(pts[i] - min_x) for i in range(0, len(pts), 2)], dtype=np.int32)
+            ys = np.array([int(pts[i] - min_y) for i in range(1, len(pts), 2)], dtype=np.int32)
+
+            # Draw filled polygon on mask
+            polygon_u8(mask_rgba, xs, ys, (0, 0, 0, 0), 0, (255, 255, 255, 255))
+
+            # Extract boolean mask from first channel
+            zone_mask = (mask_rgba[:, :, 0] > 0)
+            del mask_rgba
+            del xs
+            del ys
+
             if not np.any(zone_mask):
+                del zone_mask
                 continue
 
-            # Vectorized alpha blend on RGB channels only
+            # Optimized alpha blend - minimize temporary allocations
             region = layers[min_y : max_y + 1, min_x : max_x + 1]
-            rgb = region[..., :3].astype(np.float32)
-            mask3 = zone_mask[:, :, None]
-            blended_rgb = np.where(mask3, rgb * inv_alpha + color_rgb * alpha, rgb)
-            region[..., :3] = blended_rgb.astype(np.uint8)
-            # Leave alpha channel unchanged to avoid stacking transparency
+
+            # Work directly on the region's RGB channels
+            rgb_region = region[..., :3]
+
+            # Apply blending only where mask is True
+            # Use boolean indexing to avoid creating full-size temporary arrays
+            rgb_masked = rgb_region[zone_mask].astype(np.float32)
+
+            # Blend: new_color = old_color * (1 - alpha) + zone_color * alpha
+            rgb_masked *= inv_alpha
+            rgb_masked += color_rgb * alpha
+
+            # Write back (convert to uint8)
+            rgb_region[zone_mask] = rgb_masked.astype(np.uint8)
+
+            del zone_mask
+            del rgb_masked
 
         return layers
 

SCR/valetudo_map_parser/config/shared.py

@@ -40,6 +40,7 @@
     CONF_VAC_STAT_SIZE,
     CONF_ZOOM_LOCK_RATIO,
     DEFAULT_VALUES,
+    NOT_STREAMING_STATES,
     CameraModes,
     Colors,
     PilPNG,
@@ -119,10 +120,17 @@ def __init__(self, file_name):
         self.trims = TrimsData.from_dict(DEFAULT_VALUES["trims_data"])
         self.skip_room_ids: List[str] = []
         self.device_info = None
+        self._battery_state = None
 
     def vacuum_bat_charged(self) -> bool:
         """Check if the vacuum is charging."""
-        return (self.vacuum_state == "docked") and (int(self.vacuum_battery) < 100)
+        if self.vacuum_state != "docked":
+                self._battery_state = "not_charging"
+        elif (self._battery_state == "charging_done") and (int(self.vacuum_battery) == 100):
+            self._battery_state = "charged"
+        else:
+            self._battery_state = "charging" if int(self.vacuum_battery) < 100 else "charging_done"
+        return (self.vacuum_state == "docked") and (self._battery_state == "charging")
 
     @staticmethod
     def _compose_obstacle_links(vacuum_host_ip: str, obstacles: list) -> list | None:
@@ -209,17 +217,25 @@ def generate_attributes(self) -> dict:
 
         return attrs
 
+    def is_streaming(self) -> bool:
+        """Return true if the device is streaming."""
+        updated_status = self.vacuum_state
+        attr_is_streaming = ((updated_status not in NOT_STREAMING_STATES 
+                              or self.vacuum_bat_charged()) 
+                             or not self.binary_image)
+        return attr_is_streaming
+
     def to_dict(self) -> dict:
         """Return a dictionary with image and attributes data."""
-
-        return {
+        data = {
             "image": {
                 "binary": self.binary_image,
-                "pil_image": self.new_image,
                 "size": pil_size_rotation(self.image_rotate, self.new_image),
+                "streaming": self.is_streaming()
             },
             "attributes": self.generate_attributes(),
         }
+        return data
 
 
 class CameraSharedManager:

SCR/valetudo_map_parser/config/utils.py

@@ -111,6 +111,12 @@ async def async_get_image(
         try:
             # Backup current image to last_image before processing new one
             if hasattr(self.shared, "new_image") and self.shared.new_image is not None:
+                # Close old last_image to free memory before replacing it
+                if hasattr(self.shared, "last_image") and self.shared.last_image is not None:
+                    try:
+                        self.shared.last_image.close()
+                    except Exception:
+                        pass  # Ignore errors if image is already closed
                 self.shared.last_image = self.shared.new_image
 
             # Call the appropriate handler method based on handler type
@@ -171,13 +177,11 @@ async def async_get_image(
                 LOGGER.warning(
                     "%s: Failed to generate image from JSON data", self.file_name
                 )
-                if bytes_format and hasattr(self.shared, "last_image"):
-                    return pil_to_png_bytes(self.shared.last_image), {}
                 return (
                     self.shared.last_image
                     if hasattr(self.shared, "last_image")
                     else None
-                ), {}
+                ), self.shared.to_dict()
 
         except Exception as e:
             LOGGER.warning(

SCR/valetudo_map_parser/hypfer_handler.py

@@ -254,7 +254,12 @@ async def async_get_image_from_json(
                         )
                     LOGGER.info("%s: Completed base Layers", self.file_name)
                     # Copy the new array in base layer.
+                    # Delete old base layer before creating new one to free memory
+                    if self.img_base_layer is not None:
+                        del self.img_base_layer
                     self.img_base_layer = await self.async_copy_array(img_np_array)
+                    # Delete source array after copying to free memory
+                    del img_np_array
 
                 self.shared.frame_number = self.frame_number
                 self.frame_number += 1
@@ -268,6 +273,9 @@ async def async_get_image_from_json(
                     or self.img_work_layer.shape != self.img_base_layer.shape
                     or self.img_work_layer.dtype != self.img_base_layer.dtype
                 ):
+                    # Delete old buffer before creating new one to free memory
+                    if self.img_work_layer is not None:
+                        del self.img_work_layer
                     self.img_work_layer = np.empty_like(self.img_base_layer)
 
                 # Copy the base layer into the persistent working buffer (no new allocation per frame)

SCR/valetudo_map_parser/rand256_handler.py

@@ -58,6 +58,7 @@ def __init__(self, shared_data):
         self.drawing_config, self.draw = initialize_drawing_config(self)
         self.go_to = None  # Go to position data
         self.img_base_layer = None  # Base image layer
+        self.img_work_layer = None  # Persistent working buffer (reused across frames)
         self.img_rotate = shared_data.image_rotate  # Image rotation
         self.room_propriety = None  # Room propriety data
         self.active_zones = None  # Active zones
@@ -156,18 +157,32 @@ async def get_image_from_rrm(
 
                 # Increment frame number
                 self.frame_number += 1
-                img_np_array = await self.async_copy_array(self.img_base_layer)
                 if self.frame_number > 5:
                     self.frame_number = 0
 
+                # Ensure persistent working buffer exists and matches base (allocate only when needed)
+                if (
+                    self.img_work_layer is None
+                    or self.img_work_layer.shape != self.img_base_layer.shape
+                    or self.img_work_layer.dtype != self.img_base_layer.dtype
+                ):
+                    # Delete old buffer before creating new one to free memory
+                    if self.img_work_layer is not None:
+                        del self.img_work_layer
+                    self.img_work_layer = np.empty_like(self.img_base_layer)
+
+                # Copy the base layer into the persistent working buffer (no new allocation per frame)
+                np.copyto(self.img_work_layer, self.img_base_layer)
+                img_np_array = self.img_work_layer
+
                 # Draw map elements
                 img_np_array = await self._draw_map_elements(
                     img_np_array, m_json, colors, robot_position, robot_position_angle
                 )
 
                 # Return PIL Image using async utilities
                 pil_img = await AsyncPIL.async_fromarray(img_np_array, mode="RGBA")
-                del img_np_array  # free memory
+                # Note: Don't delete img_np_array here as it's the persistent work buffer
                 return await self._finalize_image(pil_img)
 
         except (RuntimeError, RuntimeWarning) as e:
@@ -228,7 +243,12 @@ async def _setup_robot_and_image(
                         (robot_position[1] * 10),
                         robot_position_angle,
                     )
+                # Delete old base layer before creating new one to free memory
+                if self.img_base_layer is not None:
+                    del self.img_base_layer
                 self.img_base_layer = await self.async_copy_array(img_np_array)
+                # Delete source array after copying to free memory
+                del img_np_array
             else:
                 # If floor is disabled, create an empty image
                 background_color = self.drawing_config.get_property(
@@ -237,7 +257,12 @@ async def _setup_robot_and_image(
                 img_np_array = await self.draw.create_empty_image(
                     size_x, size_y, background_color
                 )
+                # Delete old base layer before creating new one to free memory
+                if self.img_base_layer is not None:
+                    del self.img_base_layer
                 self.img_base_layer = await self.async_copy_array(img_np_array)
+                # Delete source array after copying to free memory
+                del img_np_array
 
         # Check active zones BEFORE auto-crop to enable proper zoom functionality
         # This needs to run on every frame, not just frame 0

SCR/valetudo_map_parser/rooms_handler.py

@@ -161,8 +161,17 @@ async def _process_room_layer(
             np.uint8
         )
 
+        # Free intermediate arrays to reduce memory usage
+        del local_mask
+        del struct_elem
+        del eroded
+
         # Extract contour from the mask
         outline = self.convex_hull_outline(mask)
+
+        # Free mask after extracting outline
+        del mask
+
         if not outline:
             return None, None
 

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "valetudo-map-parser"
-version = "0.1.10"
+version = "0.1.11b1"
 description = "A Python library to parse Valetudo map data returning a PIL Image object."
 authors = ["Sandro Cantarella <gsca075@gmail.com>"]
 license = "Apache-2.0"
@@ -18,7 +18,7 @@ python = ">=3.13"
 numpy = ">=1.26.4"
 Pillow = ">=10.3.0"
 scipy = ">=1.12.0"
-mvcrender = ">=0.0.5"
+mvcrender = "==0.0.6"
 
 [tool.poetry.group.dev.dependencies]
 ruff = "*"