get_content_type as property, def_context_type from device_info — v0.3.0

SCR/valetudo_map_parser/config/shared.py

@@ -156,7 +156,7 @@ def set_content_type(self, new_image_format: str = "pil") -> None:
         """Set image format / content type.
 
         Accepts either a short key ('pil', 'png', 'jpeg') or the full MIME
-        value returned by get_content_type() ('image/pil', 'image/png',
+        value returned by content_type ('image/pil', 'image/png',
         'image/jpeg'), so that a round-trip set→get→set preserves the format.
         Unknown values fall back to 'image/pil'.
         """
@@ -167,6 +167,7 @@ def set_content_type(self, new_image_format: str = "pil") -> None:
         else:
             self._image_format = ALLOWED_IMAGE_FORMAT["pil"]
 
+    @property
     def get_content_type(self) -> str:
         """Return the current set _image_format"""
         return self._image_format
@@ -393,6 +394,9 @@ def update_shared_data(self, device_info):
             instance.vacuum_status_position = device_info.get(
                 CONF_VAC_STAT_POS, DEFAULT_VALUES["vac_status_position"]
             )
+            instance.set_content_type(
+                device_info.get("def_context_type", "pil")
+            )
             # Robot size
             robot_size = device_info.get("robot_size", 25)
             try:

SCR/valetudo_map_parser/config/utils.py

@@ -216,7 +216,7 @@ def _convert_to_binary(self, new_image: PilPNG, bytes_format: bool):
             - "image/jpeg" → JPEG bytes
         """
         if bytes_format:
-            match self.shared.get_content_type():
+            match self.shared.get_content_type:
                 case "image/jpeg":
                     self.shared.binary_image = pil_to_jpeg_bytes(new_image)
                 case "image/png":

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "valetudo-map-parser"
-version = "0.3.0b2"
+version = "0.3.0"
 description = "A Python library to parse Valetudo map data returning a PIL Image object."
 authors = ["Sandro Cantarella <gsca075@gmail.com>"]
 license = "Apache-2.0"

tests/test.py

@@ -36,14 +36,14 @@
 _LOGGER = logging.getLogger(__name__)
 
 # ----- Test Configuration -----
-TEST_FILE = "x40_carpet.json"  # Changed to test.json which has path data
-FRAME_COUNT = 100  # Set to 1/10/25/50/100 as needed
+TEST_FILE = "X40_carpet.json"  # Changed to test.json which has path data
+FRAME_COUNT = 10  # Set to 1/10/25/50/100 as needed
 ENABLE_PROFILER = True  # Master switch for profiler usage
 ENABLE_CPU_TIMING = False  # Lightweight per-frame CPU timing (process CPU time)
 ENABLE_MEMORY_PROFILING = True  # Use tracemalloc snapshots
 SNAPSHOT_EVERY_FRAME = False  # If False, snapshot only first and last frame
 ENABLE_LEGACY_CPROFILE = False  # Legacy cProfile around the whole run
-ENABLE_PIL_SHOW = False  # Display images with PIL.show() (can cause memory overhead)
+ENABLE_PIL_SHOW = True  # Display images with PIL.show() (can cause memory overhead)
 SAVE_FIRST_FRAME = True  # Save first frame to /tmp for inspection
 # ------------------------------
 
@@ -346,10 +346,10 @@ async def test_image_handler(self, frame_queue=None):
             "get_svg_file": False,
             "trims_data": {
                 "floor": "floor_0",
-                "trim_up": 2400,
-                "trim_left": 2945,
-                "trim_down": 3654,
-                "trim_right": 3740,
+                "trim_up": 0,
+                "trim_left": 0,
+                "trim_down": 0,
+                "trim_right": 0,
             },
             "disable_floor": False,
             "disable_wall": False,
@@ -395,6 +395,11 @@ async def test_image_handler(self, frame_queue=None):
                 40,
             ],  # Bright GREEN for tile grout lines (testing)
             "alpha_material_tile": 40,  # High alpha for very visible grout
+            # Obstacle link configuration (optional custom endpoint)
+            "obstacle_link_ip": "192.168.1.59",  # Custom IP for obstacle images
+            "obstacle_link_port": 10,  # Custom port for obstacle images
+            "obstacle_link_protocol": "https",  # Custom protocol for obstacle images
+            "def_context_type": "png",
         }
 
         # 'trims_data': {
@@ -426,6 +431,11 @@ async def test_image_handler(self, frame_queue=None):
         # Active zones will be populated from the JSON data automatically
         # No need to manually set them here
 
+        _LOGGER.info(
+            "Content type loaded from device_info: %s (expected: image/%s)",
+            shared.get_content_type,
+            device_info.get("def_context_type", "pil"),
+        )
         _LOGGER.debug(f"Shared instance trims: {shared.trims}")
         _LOGGER.info(f"Mop mode enabled: {shared.mop_mode}")
 

tests/test_shared_attributes.py

@@ -1,16 +1,20 @@
 """Test to check shared.to_dict()['attributes'] output."""
+
 from __future__ import annotations
 
 import json
 import logging
 import os
 import sys
 
+import pytest
+
 # Add the project root directory to the Python path
 sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
 
 from SCR.valetudo_map_parser.config.shared import CameraSharedManager
 
+
 # Configure logging
 logging.basicConfig(
     level=logging.INFO,
@@ -22,7 +26,7 @@
 
 def test_shared_attributes():
     """Test shared.to_dict()['attributes'] output."""
-    
+
     # Device info from test.py
     device_info = {
         "platform": "mqtt_vacuum_camera",
@@ -79,25 +83,25 @@ def test_shared_attributes():
         },
         "image_format": "image/jpeg",  # TEST: Set image format via device_info
     }
-    
+
     _LOGGER.info("=" * 80)
     _LOGGER.info("TEST: image_format from device_info")
     _LOGGER.info("=" * 80)
-    
+
     # Initialize shared data
     file_name = "test_shared_attributes"
     shared_data = CameraSharedManager(file_name, device_info)
     shared = shared_data.get_instance()
-    
+
     # Set some vacuum state
     shared.vacuum_state = "docked"
     shared.dock_state = "mop cleaning"
     shared.vacuum_connection = True
     shared.vacuum_battery = 100
     shared.set_content_type("jpeg")
-    
+
     # Check what image_format was set to
-    _LOGGER.info(f"shared.get_content_type() = {shared.get_content_type()}")
+    _LOGGER.info(f"shared.get_content_type = {shared.get_content_type}")
     _LOGGER.info(f"Expected: 'image/jpeg'")
 
     # Get the to_dict output
@@ -111,17 +115,84 @@ def test_shared_attributes():
     _LOGGER.info("\n" + "=" * 80)
     _LOGGER.info("VERIFICATION")
     _LOGGER.info("=" * 80)
-    content_type = result['attributes']['content_type']
+    content_type = result["attributes"]["content_type"]
     _LOGGER.info(f"content_type in attributes: {content_type}")
 
     if content_type == "image/jpeg":
         _LOGGER.info("✅ SUCCESS: set_content_type('jpeg') is working!")
     else:
         _LOGGER.error(f"❌ FAIL: Expected 'image/jpeg', got '{content_type}'")
-    
+
     return result
 
 
+_MINIMAL_DEVICE_INFO = {
+    "platform": "mqtt_vacuum_camera",
+    "unique_id": "test_camera",
+    "vacuum_config_entry": "abc123",
+    "vacuum_map": "valetudo/test",
+    "vacuum_identifiers": {("mqtt", "test")},
+    "is_rand256": False,
+}
+
+
+@pytest.fixture()
+def shared():
+    """Return a fresh CameraShared instance for each test."""
+    mgr = CameraSharedManager("test_content_type", _MINIMAL_DEVICE_INFO)
+    return mgr.get_instance()
+
+
+class TestSetGetContentType:
+    """Tests for set_content_type / get_content_type property round-trips."""
+
+    def test_set_by_key_pil(self, shared):
+        shared.set_content_type("pil")
+        assert shared.get_content_type == "image/pil"
+
+    def test_set_by_key_png(self, shared):
+        shared.set_content_type("png")
+        assert shared.get_content_type == "image/png"
+
+    def test_set_by_key_jpeg(self, shared):
+        shared.set_content_type("jpeg")
+        assert shared.get_content_type == "image/jpeg"
+
+    def test_set_by_mime_pil(self, shared):
+        """Passing a MIME value returned by get_content_type must be accepted."""
+        shared.set_content_type("image/pil")
+        assert shared.get_content_type == "image/pil"
+
+    def test_set_by_mime_png(self, shared):
+        shared.set_content_type("image/png")
+        assert shared.get_content_type == "image/png"
+
+    def test_set_by_mime_jpeg(self, shared):
+        shared.set_content_type("image/jpeg")
+        assert shared.get_content_type == "image/jpeg"
+
+    def test_round_trip_jpeg(self, shared):
+        """set → get → set again must preserve the format."""
+        shared.set_content_type("jpeg")
+        mime = shared.get_content_type           # "image/jpeg"
+        shared.set_content_type(mime)            # must NOT reset to "image/pil"
+        assert shared.get_content_type == "image/jpeg"
+
+    def test_round_trip_png(self, shared):
+        shared.set_content_type("png")
+        shared.set_content_type(shared.get_content_type)
+        assert shared.get_content_type == "image/png"
+
+    def test_invalid_input_falls_back_to_pil(self, shared):
+        shared.set_content_type("webp")
+        assert shared.get_content_type == "image/pil"
+
+    def test_default_parameter_is_pil(self, shared):
+        """Default value of set_content_type() must produce image/pil."""
+        shared.set_content_type("jpeg")          # change away from default
+        shared.set_content_type()                # call with no argument
+        assert shared.get_content_type == "image/pil"
+
+
 if __name__ == "__main__":
     test_shared_attributes()
-