Finish sACN receiver

Breina · Breina · commit 0b0c3980cd1c · 2025-08-24T15:11:58.000+02:00
diff --git a/custom_components/dmx/__init__.py b/custom_components/dmx/__init__.py
@@ -23,7 +23,7 @@
 from custom_components.dmx.io.dmx_io import DmxUniverse
 from custom_components.dmx.server import PortAddress, ArtPollReply
 from custom_components.dmx.server.artnet_server import ArtNetServer, Node, ManualNode
-from custom_components.dmx.server.sacn_server import SacnServer, SacnServerConfig
+from custom_components.dmx.server.sacn_server import SacnServer, SacnServerConfig, create_sacn_receiver
 from custom_components.dmx.util.rate_limiter import RateLimiter
 
 log = logging.getLogger(__name__)
@@ -222,6 +222,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
     entities: list[Entity] = []
     universes: dict[PortAddress, DmxUniverse] = {}
     sacn_server = None
+    sacn_receiver = None
 
     # Initialize sACN server if configured
     if (sacn_yaml := dmx_yaml.get(CONF_NODE_TYPE_SACN)) is not None:
@@ -237,6 +238,59 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
         sacn_server.start_server()
         log.info(f"sACN server started with source name: {sacn_config.source_name}")
 
+        # Set up rate limiting for sACN reception (similar to Art-Net)
+        rate_limit = sacn_yaml.get(CONF_RATE_LIMIT, CONF_RATE_LIMIT_DEFAULT)
+        sacn_universe_rate_limiters = {}
+
+        # Function to process sACN universe updates with rate limiting
+        def sacn_state_callback(port_address: PortAddress, data: bytearray, source: str | None = None):
+            log.info(f"sACN state callback triggered for {port_address} from source '{source}' with {len(data)} channels")
+            
+            callback_universe: DmxUniverse = universes.get(port_address)
+            if callback_universe is None:
+                log.warning(f"Received sACN data for unknown universe: {port_address}")
+                return
+
+            if port_address not in sacn_universe_rate_limiters:
+                updates_dict = {}
+
+                async def process_updates():
+                    nonlocal updates_dict
+                    updates_to_process = updates_dict.copy()
+                    updates_dict.clear()
+
+                    if updates_to_process:
+                        log.debug(f"Processing {len(updates_to_process)} sACN channel updates for {port_address}")
+                        await callback_universe.update_multiple_values(updates_to_process, source, send_update=False)
+
+                limiter = RateLimiter(
+                    hass,
+                    update_method=lambda: hass.async_create_task(process_updates()),
+                    update_interval=rate_limit,
+                    force_update_after=rate_limit * 4
+                )
+
+                sacn_universe_rate_limiters[port_address] = (limiter, updates_dict)
+
+            limiter, updates_dict = sacn_universe_rate_limiters[port_address]
+
+            changes_detected = False
+
+            for channel, value in enumerate(data, start=1):  # DMX channels are 1-based
+                if value > 0 or callback_universe.get_channel_value(channel) != value:
+                    updates_dict[channel] = value
+                    changes_detected = True
+
+            if changes_detected:
+                log.debug(f"Detected changes in {len([k for k, v in updates_dict.items()])} channels for {port_address}")
+                limiter.schedule_update()
+            else:
+                log.debug(f"No changes detected for {port_address}")
+
+        # Create sACN receiver for incoming multicast data
+        sacn_receiver = await create_sacn_receiver(hass, sacn_state_callback)
+        log.info("sACN receiver started for multicast reception")
+
         # Process sACN universes and devices
         for universe_dict in sacn_yaml[CONF_UNIVERSES]:
             (universe_value, universe_yaml), = universe_dict.items()
@@ -257,6 +311,10 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
             # Add universe to sACN server with unicast addresses
             sacn_server.add_universe(universe_id, unicast_addresses)
             
+            # Subscribe receiver to this universe for incoming multicast data
+            sacn_receiver.subscribe_universe(universe_id)
+            log.info(f"Subscribed sACN receiver to universe {universe_id}")
+            
             # Create universe with sACN support, passing the actual sACN universe ID
             universe = DmxUniverse(port_address, None, True, sacn_server, universe_id)
             universes[port_address] = universe
@@ -502,6 +560,7 @@ def node_lost_callback(node: Node):
                         vol.Optional(CONF_SYNC_ADDRESS, default=0): vol.All(vol.Coerce(int), vol.Range(min=0, max=63999)),
                         vol.Optional(CONF_MULTICAST_TTL, default=CONF_MULTICAST_TTL_DEFAULT): vol.All(vol.Coerce(int), vol.Range(min=1, max=255)),
                         vol.Optional(CONF_ENABLE_PREVIEW_DATA, default=False): cv.boolean,
+                        vol.Optional(CONF_RATE_LIMIT, default=CONF_RATE_LIMIT_DEFAULT): cv.positive_float,
 
                         vol.Required(CONF_UNIVERSES): vol.Schema(
                             [{
diff --git a/custom_components/dmx/entity/light/light_state.py b/custom_components/dmx/entity/light/light_state.py
@@ -17,28 +17,62 @@ def __init__(self, color_mode: ColorMode, converter: ColorConverter, channels: D
 
         self.is_on = False
         self.brightness = 255
-        self.rgb = (255, 255, 255)
         self.cold_white = 255
         self.warm_white = 255
         self.color_temp_kelvin = 3000
         self.color_temp_dmx = 255
 
         self.last_brightness = 255
-        self.last_rgb = (255, 255, 255)
         self.last_cold_white = 255
         self.last_warm_white = 255
         self.last_color_temp_kelvin = 3000
         self.last_color_temp_dmx = 255
-
-        self._channel_handlers = {
-            ChannelType.DIMMER: self._handle_dimmer_update,
-            ChannelType.RED: lambda v: self._handle_rgb_component_update(0, v),
-            ChannelType.GREEN: lambda v: self._handle_rgb_component_update(1, v),
-            ChannelType.BLUE: lambda v: self._handle_rgb_component_update(2, v),
-            ChannelType.COLD_WHITE: self._handle_cold_white_update,
-            ChannelType.WARM_WHITE: self._handle_warm_white_update,
-            ChannelType.COLOR_TEMPERATURE: self.update_color_temp_dmx,
-        }
+        
+        # Only initialize RGB for color modes that use it
+        if color_mode in [ColorMode.RGB, ColorMode.RGBW, ColorMode.RGBWW]:
+            self.rgb = (255, 255, 255)
+            self.last_rgb = (255, 255, 255)
+        else:
+            self.rgb = None
+            self.last_rgb = None
+
+        self._channel_handlers = {}
+
+        if ChannelType.DIMMER in channels:
+            self._channel_handlers[ChannelType.DIMMER] = self._handle_dimmer_update
+            
+        if ChannelType.RED in channels:
+            self._channel_handlers[ChannelType.RED] = lambda v: self._handle_rgb_component_update(0, v)
+        if ChannelType.GREEN in channels:
+            self._channel_handlers[ChannelType.GREEN] = lambda v: self._handle_rgb_component_update(1, v)
+        if ChannelType.BLUE in channels:
+            self._channel_handlers[ChannelType.BLUE] = lambda v: self._handle_rgb_component_update(2, v)
+            
+        if ChannelType.COLD_WHITE in channels:
+            self._channel_handlers[ChannelType.COLD_WHITE] = self._handle_cold_white_update
+        if ChannelType.WARM_WHITE in channels:
+            self._channel_handlers[ChannelType.WARM_WHITE] = self._handle_warm_white_update
+        if ChannelType.COLOR_TEMPERATURE in channels:
+            self._channel_handlers[ChannelType.COLOR_TEMPERATURE] = self.update_color_temp_dmx
+            
+        # Ensure state consistency after init/restoration
+        self._validate_and_fix_state()
+
+    def _validate_and_fix_state(self):
+        """Ensure state fields are consistent with color mode after init/restoration"""
+        if self.color_mode in [ColorMode.RGB, ColorMode.RGBW, ColorMode.RGBWW]:
+            # RGB color modes must have valid RGB values
+            if self.rgb is None:
+                self.rgb = (255, 255, 255)
+                log.warning(f"Fixed None RGB value for {self.color_mode} color mode")
+            if self.last_rgb is None:
+                self.last_rgb = (255, 255, 255)
+        else:
+            # Non-RGB color modes should have RGB as None
+            if self.rgb is not None:
+                self.rgb = None
+            if self.last_rgb is not None:
+                self.last_rgb = None
 
     def has_channel(self, t: ChannelType) -> bool:
         return t in self.channels
@@ -62,6 +96,16 @@ def _handle_dimmer_update(self, value: int):
         self.is_on = value > 0
 
     def _handle_rgb_component_update(self, component_index: int, value: int):
+        # This should only be called for RGB color modes
+        if self.color_mode not in [ColorMode.RGB, ColorMode.RGBW, ColorMode.RGBWW]:
+            log.warning(f"RGB component update called for non-RGB color mode {self.color_mode}")
+            return
+        
+        # Safety check: ensure RGB is valid (in case of state restoration corruption)
+        if self.rgb is None:
+            self.rgb = (255, 255, 255)
+            log.warning(f"Fixed None RGB during RGB update for {self.color_mode} color mode")
+            
         new_rgb = list(self.rgb)
         new_rgb[component_index] = value
         self._update_rgb_based_on_color_mode(*new_rgb)
@@ -254,7 +298,6 @@ def update_color_temp_kelvin(self, kelvin: int):
 
     def reset(self):
         self.brightness = 0
-        self.rgb = (0, 0, 0)
         self.cold_white = 0
         self.warm_white = 0
         self.is_on = False
@@ -269,7 +312,7 @@ def is_all_zero(self, has_dimmer: bool) -> bool:
     def _update_brightness_from_channels(self):
         values = []
 
-        if self.has_rgb():
+        if self.has_rgb() and self.rgb is not None:
             values.extend(self.rgb)
 
         if self.has_channel(ChannelType.COLD_WHITE):
diff --git a/custom_components/dmx/server/sacn_server.py b/custom_components/dmx/server/sacn_server.py
@@ -295,17 +295,27 @@ def connection_lost(self, exc: Optional[Exception]):
     
     def datagram_received(self, data: bytes, addr):
         try:
+            log.debug(f"Received {len(data)} bytes from {addr[0]}:{addr[1]}")
             packet = SacnPacket.deserialize(data)
             
             if packet.universe in self.subscribed_universes:
-                log.debug(f"Received sACN data for universe {packet.universe} from {addr}")
+                log.debug(f"Received sACN data for universe {packet.universe} from {addr[0]} "
+                        f"(source: '{packet.source_name}', seq: {packet.sequence_number}, "
+                        f"priority: {packet.priority}, channels: {len(packet.dmx_data)})")
                 
                 if self.data_callback:
                     port_address = PortAddress(0, 0, packet.universe)
                     self.data_callback(port_address, packet.dmx_data, packet.source_name)
+                else:
+                    log.warning("No data callback configured for sACN receiver")
+            else:
+                log.debug(f"Ignoring sACN data for universe {packet.universe} from {addr[0]} "
+                         f"(not subscribed, subscribed universes: {self.subscribed_universes})")
             
         except Exception as e:
-            log.debug(f"Error processing sACN packet from {addr}: {e}")
+            log.warning(f"Error processing sACN packet from {addr[0]}:{addr[1]} "
+                       f"({len(data)} bytes): {e}")
+            log.debug(f"Raw packet data: {data[:50].hex()}{'...' if len(data) > 50 else ''}")
     
     def subscribe_universe(self, universe_id: int):
         if not (1 <= universe_id <= 63999):
@@ -340,10 +350,12 @@ def _manage_multicast_group(self, universe_id: int, join: bool):
                 action = "Joined" if join else "Left"
                 sock.setsockopt(socket.IPPROTO_IP, operation, mreq)
                 log.debug(f"{action} multicast group {multicast_addr} for universe {universe_id}")
+            else:
+                log.error(f"No socket available to {'join' if join else 'leave'} multicast group for universe {universe_id}")
             
         except Exception as e:
             action = "join" if join else "leave"
-            log.error(f"Failed to {action} multicast group for universe {universe_id}: {e}")
+            log.error(f"Failed to {action} multicast group {multicast_addr} for universe {universe_id}: {e}")
     
     def _join_multicast_group(self, universe_id: int):
         self._manage_multicast_group(universe_id, True)
@@ -357,11 +369,21 @@ async def create_sacn_receiver(hass: HomeAssistant,
     receiver = SacnReceiver(hass, data_callback)
     
     loop = hass.loop
-    transport, protocol = await loop.create_datagram_endpoint(
-        lambda: receiver,
-        local_addr=('0.0.0.0', SACN_PORT),
-        reuse_address=True
-    )
-    
-    log.info("sACN receiver started")
-    return receiver
+    try:
+        transport, protocol = await loop.create_datagram_endpoint(
+            lambda: receiver,
+            local_addr=('0.0.0.0', SACN_PORT)
+        )
+        
+        # Set socket options for multicast reception
+        sock = transport.get_extra_info('socket')
+        if sock:
+            sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+            log.debug("Set socket reuse option for sACN receiver")
+        
+        log.info("sACN receiver started successfully")
+        return receiver
+        
+    except Exception as e:
+        log.error(f"Failed to create sACN receiver: {e}")
+        raise
