works with live updates

custom_components/ha_behringer_mixer/__init__.py

@@ -19,8 +19,6 @@
     Platform.SENSOR,
 ]
 
-
-# https://developers.home-assistant.io/docs/config_entries_index/#setting-up-an-entry
 async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     """Set up this integration using UI."""
     hass.data.setdefault(DOMAIN, {})
@@ -34,8 +32,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
         hass=hass,
         client=client,
     )
-    # https://developers.home-assistant.io/docs/integration_fetching_data#coordinated-single-api-poll-for-data-for-all-entities
     await coordinator.async_config_entry_first_refresh()
+    client.register_coordinator(coordinator)
 
     await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
     entry.async_on_unload(entry.add_update_listener(async_reload_entry))

custom_components/ha_behringer_mixer/api.py

@@ -1,6 +1,7 @@
 """Sample API Client."""
 from __future__ import annotations
 import logging
+import asyncio
 from .behringer_mixer import mixer_api
 
 
@@ -29,6 +30,8 @@ def __init__(self, mixer_ip: str, mixer_type: str) -> None:
         self._num_matrix = 0
         self._num_dca = 0
         self._mixer = None
+        self.tasks = set()
+        self.coordinator = None
 
     async def setup(self):
         """Setup the server"""
@@ -37,20 +40,29 @@ async def setup(self):
             self._mixer_type, ip=self._mixer_ip, logLevel=logging.WARNING
         )
         await self._mixer.connectserver()
-        # await self._mixer.subscribe(self.new_data_callback)
+        # Get Initial state first
+        await self._mixer.reload()
+        # Setup subscription for live updates
+        task = asyncio.create_task(self._mixer.subscribe(self.new_data_callback))
+        self.tasks.add(task)
+        task.add_done_callback(self.tasks.discard)
+
         return True
 
     async def async_get_data(self) -> any:
         """Get data from the API."""
-        await self._mixer.reload()
         return self._mixer.state()
 
     async def async_set_value(self, address: str, value: str) -> any:
         """Set data"""
         return await self._mixer.set_value(address, value)
 
-    async def new_data_callback(self, data: dict):
+    def new_data_callback(self, data: dict):
         """Callback function to receive new data from the mixer"""
-        print("CB")
-        print(data)
+        if self.coordinator:
+            self.coordinator.async_update_listeners()
         return True
+
+    def register_coordinator(self, coordinator):
+        """ register the coordinator object """
+        self.coordinator = coordinator

custom_components/ha_behringer_mixer/behringer_mixer/MixerBase.py

@@ -22,6 +22,7 @@ class MixerBase:
     delay: float = 0.02
     addresses_to_load = []
     cmd_scene_load = ""
+    tasks = set()
 
     def __init__(self, **kwargs):
         self.ip = kwargs.get("ip")
@@ -63,23 +64,22 @@ async def connectserver(self):
 
     async def startup(self):
         """Startup the server"""
-        print("in Startup")
         if not self.server:
-            print("starting server")
             dispatcher = Dispatcher()
             dispatcher.set_default_handler(self.msg_handler)
             self.server = OSCClientServer(
                 (self.ip, self.port), dispatcher, asyncio.get_event_loop()
             )
             transport, protocol = await self.server.create_serve_endpoint()
-            self.server.register_transport(transport)
+            self.server.register_transport(transport, protocol)
             await self.validate_connection()
 
     def msg_handler(self, addr, *data):
         """Handle callback response"""
         self.logger.debug(f"received: {addr} {data if data else ''}")
-        updates = self._update_state(addr, data)
-        if self._callback_function:
+        handling_subscriptions = bool(self._callback_function)
+        updates = self._update_state(addr, data, handling_subscriptions)
+        if handling_subscriptions:
             for row in updates:
                 self._callback_function(row)
         else:
@@ -93,22 +93,16 @@ async def send(self, addr: str, param: Optional[str] = None):
         await asyncio.sleep(self._delay)
 
     async def query(self, address):
+        """Send an receive the value of an OSC message"""
         await self.send(address)
         return self.info_response
 
     async def subscribe(self, callback_function):
         await self._subscribe_worker("/xremote", callback_function)
 
-    def _subscribe(self, parameter_string, callback_function):
-        self.subscription = threading.Thread(
-            target=self._subscribe_worker,
-            args=(
-                parameter_string,
-                callback_function,
-            ),
-            daemon=True,
-        )
-        self.subscription.start()
+    # async def _subscribe(self, parameter_string, callback_function):
+    #    self._subscribe_worker(parameter_string, callback_function)
+    #    return True
 
     async def _subscribe_worker(self, parameter_string, callback_function):
         self._callback_function = callback_function
@@ -173,26 +167,25 @@ async def _load_initial(self):
         for address in expanded_addresses:
             await self.send(address)
 
-    def _update_state(self, address, values):
+    def _update_state(self, address, values, updates_only=False):
         # update internal state representation
         # State looks like
-        #    /ch/02/mix_fader = Value
-        #    /ch/02/config_name = Value
+        #    /ch/2/mix_fader = Value
+        #    /ch/2/config_name = Value
         rewrite_key = self._rewrites.get(address)
         if rewrite_key:
             address = rewrite_key
         state_key = self._generate_state_key(address)
         value = values[0]
         updates = []
         if state_key:
-            if self._callback_function and state_key not in self._state:
-                # we are processing updates and data captured is not in initial state
-                # Therefore we want to ignore data
-                return updates
-
             if state_key.endswith("_on"):
                 value = bool(value)
             state_key = re.sub(r"/0+(\d+)/", r"/\1/", state_key)
+            if updates_only and state_key not in self._state:
+                # we are processing updates and data captured is not in initial state
+                # Therefore we want to ignore data
+                return updates
             self._state[state_key] = value
             updates.append({"property": state_key, "value": value})
             if state_key.endswith("_fader"):
@@ -227,7 +220,6 @@ def _build_reverse_rewrite(self):
 
     async def set_value(self, address, value):
         """Set the value in the mixer"""
-        print(f"SET VALUE {address} {value}")
         if address.endswith("_db"):
             address = address.replace("_db", "")
             value = db_to_fader(value)

custom_components/ha_behringer_mixer/behringer_mixer/mixer_osc.py

@@ -1,4 +1,3 @@
-import asyncio
 from pythonosc.dispatcher import Dispatcher
 from pythonosc.osc_message_builder import OscMessageBuilder
 from pythonosc.osc_server import AsyncIOOSCUDPServer
@@ -10,6 +9,7 @@ def __init__(self, address: str, dispatcher: Dispatcher, event_loop):
         self.mixer_address = address
         self.event_loop = event_loop
         self.transport = None
+        self.protocol = None
 
     def send_message(self, address: str, vals):
         builder = OscMessageBuilder(address=address)
@@ -21,8 +21,9 @@ def send_message(self, address: str, vals):
         msg = builder.build()
         self.transport.sendto(msg.dgram, self.mixer_address)
 
-    def register_transport(self, transport):
+    def register_transport(self, transport, protocol):
         self.transport = transport
+        self.protocol = protocol
 
     def shutdown(self):
         self.transport.close()

custom_components/ha_behringer_mixer/behringer_mixer/mixer_types.py

@@ -4,7 +4,8 @@
 
 
 class MixerTypeBase(MixerBase):
-    """ Base class for mixer type configuration """
+    """Base class for mixer type configuration"""
+
     mixer_type: str = ""
     port_number: int = 10023
     delay: float = 0.02
@@ -23,9 +24,9 @@ class MixerTypeBase(MixerBase):
         ["/bus/{num_bus}/mix/fader"],
         ["/bus/{num_bus}/mix/on"],
         ["/bus/{num_bus}/config/name"],
-        ["/mtx/{num_matrix:2}/mix/fader"],
-        ["/mtx/{num_matrix:2}/mix/on"],
-        ["/mtx/{num_matrix:2}/config/name"],
+        ["/mtx/{num_matrix}/mix/fader"],
+        ["/mtx/{num_matrix}/mix/on"],
+        ["/mtx/{num_matrix}/config/name"],
         ["/dca/{num_dca}/fader"],
         ["/dca/{num_dca}/on"],
         ["/dca/{num_dca}/config/name"],
@@ -39,7 +40,7 @@ class MixerTypeBase(MixerBase):
     cmd_scene_load = "/-action/goscene"
 
     def info(self):
-        """ Return information about the mixer """
+        """Return information about the mixer"""
         return {
             "channel": self.num_channel,
             "bus": self.num_bus,
@@ -52,7 +53,8 @@ def info(self):
 
 
 class MixerTypeXAir(MixerTypeBase):
-    """ Base Mixer class for the XAir type mixers """
+    """Base Mixer class for the XAir type mixers"""
+
     port_number: int = 10024
 
     cmd_scene_load = "/-snap/load"
@@ -69,7 +71,8 @@ def __init__(self, *args):
 
 
 class MixerTypeX32(MixerTypeBase):
-    """ Class for Behringer X32 Mixer """
+    """Class for Behringer X32 Mixer"""
+
     mixer_type: str = "X32"
     num_channel: int = 32
     num_bus: int = 16
@@ -80,7 +83,8 @@ class MixerTypeX32(MixerTypeBase):
 
 
 class MixerTypeXR12(MixerTypeXAir):
-    """ Class for Behringer XR-12 Mixer """
+    """Class for Behringer XR-12 Mixer"""
+
     mixer_type: str = "XR12"
     num_channel: int = 12
     num_bus: int = 2
@@ -89,7 +93,8 @@ class MixerTypeXR12(MixerTypeXAir):
 
 
 class MixerTypeXR16(MixerTypeXAir):
-    """ Class for Behringer XR-16 Mixer """
+    """Class for Behringer XR-16 Mixer"""
+
     mixer_type: str = "XR16"
     num_channel: int = 16
     num_bus: int = 4
@@ -98,7 +103,7 @@ class MixerTypeXR16(MixerTypeXAir):
 
 
 class MixerTypeXR18(MixerTypeXAir):
-    """ Class for Behringer XR-18 Mixer """
+    """Class for Behringer XR-18 Mixer"""
 
     mixer_type: str = "XR18"
     num_channel: int = 16
@@ -117,7 +122,7 @@ class MixerTypeXR18(MixerTypeXAir):
 
 
 def make_mixer(mixer_type, **kwargs):
-    """ Make the actual mixer object based on the type """
+    """Make the actual mixer object based on the type"""
     if mixer_type in _supported_mixers:
         mixer_class_name = "MixerType" + mixer_type
         # module_ = importlib.import_module(".mixer_types", package="behringer_mixer")

custom_components/ha_behringer_mixer/coordinator.py

@@ -14,7 +14,7 @@
 from .api import (
     BehringerMixerApiClient,
     BehringerMixerApiClientAuthenticationError,
-    BehringerMixerApiClientError
+    BehringerMixerApiClientError,
 )
 from .const import DOMAIN, LOGGER
 
@@ -36,10 +36,11 @@ def __init__(
             hass=hass,
             logger=LOGGER,
             name=DOMAIN,
-            update_interval=timedelta(minutes=5),
         )
-    # async def async_set_updated_data(self):
-    #    pass
+
+#    async def async_set_updated_data(self):
+#        print("SET UPDATED DATA")
+#        pass
 
     async def _async_update_data(self):
         """Update data via library."""
@@ -49,3 +50,4 @@ async def _async_update_data(self):
             raise ConfigEntryAuthFailed(exception) from exception
         except BehringerMixerApiClientError as exception:
             raise UpdateFailed(exception) from exception
+

custom_components/ha_behringer_mixer/entity.py

@@ -12,18 +12,7 @@ class BehringerMixerEntity(CoordinatorEntity):
     """BlueprintEntity class."""
 
     _attr_attribution = ATTRIBUTION
-
-    def __init2__(self, coordinator: BlueprintDataUpdateCoordinator) -> None:
-        """Initialize."""
-        super().__init__(coordinator)
-        self._attr_unique_id = coordinator.config_entry.entry_id
-        self._attr_device_info = DeviceInfo(
-            identifiers={(DOMAIN, coordinator.config_entry.entry_id)},
-            name=f"Mixer - {coordinator.config_entry.data['MIXER_TYPE']} - {coordinator.config_entry.data['MIXER_IP']}",
-            model=VERSION,
-            manufacturer="Behringer",
-        )
-        self.base_address = ""
+    _attr_should_poll = False
 
     def __init__(
         self,

custom_components/ha_behringer_mixer/number.py

@@ -4,7 +4,6 @@
 from homeassistant.components.number import NumberEntity, NumberEntityDescription
 
 from .const import DOMAIN
-from .coordinator import BlueprintDataUpdateCoordinator
 from .entity import BehringerMixerEntity
 
 
@@ -61,4 +60,4 @@ async def async_set_native_value(self, value: float) -> None:
         await self.coordinator.client.async_set_value(
             self.base_address + "/mix_fader", value
         )
-        await self.coordinator.async_request_refresh()
+        #await self.coordinator.async_request_refresh()

custom_components/ha_behringer_mixer/sensor.py

@@ -4,7 +4,6 @@
 from homeassistant.components.sensor import SensorEntity, SensorEntityDescription
 
 from .const import DOMAIN
-from .coordinator import BlueprintDataUpdateCoordinator
 from .entity import BehringerMixerEntity
 
 

custom_components/ha_behringer_mixer/switch.py

@@ -4,7 +4,6 @@
 from homeassistant.components.switch import SwitchEntity, SwitchEntityDescription
 
 from .const import DOMAIN
-from .coordinator import BlueprintDataUpdateCoordinator
 from .entity import BehringerMixerEntity
 
 
@@ -60,11 +59,9 @@ async def async_turn_on(self, **_: any) -> None:
         await self.coordinator.client.async_set_value(
             self.base_address + "/mix_on", True
         )
-        await self.coordinator.async_request_refresh()
 
     async def async_turn_off(self, **_: any) -> None:
         """Turn off the switch."""
         await self.coordinator.client.async_set_value(
             self.base_address + "/mix_on", False
         )
-        await self.coordinator.async_request_refresh()