Support for manually entered devices (#126)

* Update dev to version 2.0.2

* Feature/new devices (#117)

* Add H612B
* Add H61C2
* Add H605D
* Add H6167
* Update issue template

* add segment number to h61e5 according to app (#116)

* Add support for H6609 (#119)

* Feature/new devices (#124)

* Add H6008
* Add H61B3
* Add H8072

* Manual devices (#125)

Add the capability to manually add a device by it's IP address, bypassing discovery.
Device must still be reachable by unicast scan and update messages.

---------

Co-authored-by: Niklas K. <46166944+NIREKI@users.noreply.github.com>
Co-authored-by: Jan Novotný <novotny.jan23+github@gmail.com>

.github/ISSUE_TEMPLATE/feature_request.md

@@ -1,22 +1,26 @@
 ---
 name: Feature request
 about: Request support for a light model
-title: ''
+title: ""
 labels: enhancement
 assignees: Galorhallen
-
 ---
 
 ## 🚀 Light Support Template for Govee LAN API Library
+
 ---
 
 ### 💡 Light Details
+
 Provide information about the Govee light(s) features:
-- **Model**: 
+
+- **Model**:
 - **Supported Functionalities**:
   - [ ] RGB
   - [ ] Brightness
   - [ ] White Temperature
   - [ ] Scenes
-- **Number of Segments**: 
+  - [ ] Segments (Please, specify the number of segments showed in the app, otherwise the feature will not be added to the device)
+- **Number of Segments**:
+
 ---

example/main.py

@@ -18,35 +18,34 @@ def discovered_callback(device: GoveeDevice, is_new: bool) -> bool:
     return True
 
 
-async def create_controller() -> GoveeController:
+async def create_controller(
+    discovery_enabled: bool, manual_device_ip: str | None = None
+) -> GoveeController:
     controller = GoveeController(
         loop=asyncio.get_event_loop(),
         listening_address="0.0.0.0",
-        discovery_enabled=True,
+        discovery_enabled=discovery_enabled,
         discovered_callback=discovered_callback,
         evicted_callback=lambda device: print(f"Evicted {device}"),
     )
-    await controller.start()
-    while not controller.devices:
-        print("Waiting for devices... ")
-        await asyncio.sleep(1)
-    print("Devices: ", [str(d) for d in controller.devices])
-    return controller
 
+    await controller.start()
 
-async def menu(device: GoveeDevice) -> None:
-    print("\nDevice: ", device)
-    print("Select an option:")
-    print("0. Exit")
-    print("1. Turn on")
-    print("2. Turn off")
-    print("3. Set brightness (0-100)")
-    print("4. Set color (R G B)")
-    print("5. Set segment color (Segment, R G B)")
-    print("6. Set scene")
-    print("7. Send raw hex")
-    print("8. Clear screen")
-    print("9. Clear Device")
+    if discovery_enabled:
+        while not controller.devices:
+            print("Waiting for devices... ")
+            await asyncio.sleep(1)
+    else:
+        if not manual_device_ip:
+            raise ValueError(
+                "Manual device IP must be provided if discovery is disabled."
+            )
+        print(f"Discovery not enabled. Adding {manual_device_ip} to discovery.")
+        controller.add_device_to_discovery(manual_device_ip)
+        while not controller.devices:
+            print(f"Waiting for device {manual_device_ip} to be discovered...")
+            await asyncio.sleep(1)
+    return controller
 
 
 async def handle_turn_on(device: GoveeDevice) -> None:
@@ -170,6 +169,29 @@ async def handle_send_hex(device: GoveeDevice, session: PromptSession) -> None:
     await device.send_raw_command(hex_data)
 
 
+async def handle_manual_device(
+    device: GoveeDevice, controller: GoveeController, session: PromptSession
+) -> None:
+    ip = await session.prompt_async("Enter device IP: ")
+    controller.add_device_to_discovery(ip)
+
+
+async def menu(device: GoveeDevice) -> None:
+    print("\nDevice: ", device)
+    print("Select an option:")
+    print("0. Exit")
+    print("1. Turn on")
+    print("2. Turn off")
+    print("3. Set brightness (0-100)")
+    print("4. Set color (R G B)")
+    print("5. Set segment color (Segment, R G B)")
+    print("6. Set scene")
+    print("7. Send raw hex")
+    print("8. Clear screen")
+    print("9. Clear Device")
+    print("10. Add device")
+
+
 async def repl() -> None:
     session = PromptSession()
     print("Welcome to the LED Control REPL.")
@@ -185,18 +207,29 @@ async def repl() -> None:
         "6": lambda device: handle_set_scene(device, session),
         "7": lambda device: handle_send_hex(device, session),
         "8": handle_clear_screen,
+        "10": lambda device: handle_manual_device(device, controller, session),
     }
 
-    controller: GoveeController = await create_controller()
+    enable_discovery = await session.prompt_async("Enable device discovery? (y/n): ")
+    discovery_enabled: bool = enable_discovery.lower() == "y"
     selected_device: GoveeDevice | None = None
 
+    manual_device_ip = None
+    if not discovery_enabled:
+        manual_device_ip = await session.prompt_async("Enter device IP: ")
+    controller: GoveeController = await create_controller(
+        discovery_enabled, manual_device_ip
+    )
+
     while True:
         if not selected_device:
             selected_device = await devices_menu(session, controller.devices)
 
         await menu(selected_device)
         with patch_stdout():
-            user_choice = await session.prompt_async("Choose an option (1-10): ")
+            user_choice = await session.prompt_async(
+                f"Choose an option (0-{len(command_handlers)}): "
+            )
             if user_choice == "9":
                 selected_device = None
                 continue

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "govee-local-api"
-version = "2.0.1"
+version = "2.1.0"
 license = "Apache-2.0"
 authors = [{ name = "Galorhallen", email = "andrea.ponte1987@gmail.com" }]
 description = "Library to communicate with Govee local API"

src/govee_local_api/__init__.py

@@ -10,4 +10,4 @@
     "GoveeLightCapabilities",
 ]
 
-__version__ = "2.0.1"
+__version__ = "2.1.0"

src/govee_local_api/controller.py

@@ -9,10 +9,10 @@
 from typing import Any, cast
 
 from .device import GoveeDevice
+from .device_registry import DeviceRegistry
 from .light_capabilities import (
     GOVEE_LIGHT_CAPABILITIES,
     ON_OFF_CAPABILITIES,
-    GoveeLightCapabilities,
     GoveeLightFeatures,
 )
 from .message import (
@@ -76,6 +76,7 @@ def __init__(
             discovered_callback (Callable[GoveeDevice, bool]): An optional function to call when a device is discovered (or rediscovered). Default None
             evicted_callback (Callable[GoveeDevice]): An optional function to call when a device is evicted.
         """
+        self._logger = logger or logging.getLogger(__name__)
 
         self._transport: Any = None
         self._protocol = None
@@ -88,7 +89,7 @@ def __init__(
         self._loop = loop or asyncio.get_running_loop()
         self._cleanup_done: asyncio.Event = asyncio.Event()
         self._message_factory = MessageResponseFactory()
-        self._devices: dict[str, GoveeDevice] = {}
+        self._registry: DeviceRegistry = DeviceRegistry(self._logger)
 
         self._discovery_enabled = discovery_enabled
         self._discovery_interval = discovery_interval
@@ -100,17 +101,20 @@ def __init__(
         self._device_discovered_callback = discovered_callback
         self._device_evicted_callback = evicted_callback
 
-        self._logger = logger or logging.getLogger(__name__)
-
         self._discovery_handle: asyncio.TimerHandle | None = None
         self._update_handle: asyncio.TimerHandle | None = None
 
+        self._response_handler: dict[str, Callable] = {
+            ScanResponse.command: self._handle_scan_response,
+            DevStatusResponse.command: self._handle_status_update_response,
+        }
+
     async def start(self):
         self._transport, self._protocol = await self._loop.create_datagram_endpoint(
             lambda: self, local_addr=(self._listening_address, self._listening_port)
         )
 
-        if self._discovery_enabled:
+        if self._discovery_enabled or self._registry.has_queued_devices:
             self.send_discovery_message()
         if self._update_enabled:
             self.send_update_message()
@@ -122,24 +126,26 @@ def cleanup(self) -> asyncio.Event:
 
         if self._transport:
             self._transport.close()
-        self._devices.clear()
+        self._registry.cleanup()
         return self._cleanup_done
 
-    def add_device(
-        self,
-        ip: str,
-        sku: str,
-        fingerprint,
-        capabilities: GoveeLightCapabilities,
-    ) -> None:
-        device: GoveeDevice = GoveeDevice(self, ip, fingerprint, sku, capabilities)
-        self._devices[fingerprint] = device
+    def add_device_to_discovery_queue(self, ip: str) -> bool:
+        ip_added: bool = self._registry.add_device_to_queue(ip)
+        if not self._discovery_enabled and ip_added:
+            self.send_discovery_message()
+        return ip_added
+
+    def remove_device_from_discovery_queue(self, ip: str) -> bool:
+        return self._registry.remove_device_from_queue(ip)
+
+    @property
+    def discovery_queue(self) -> set[str]:
+        return self._registry.devices_queue
 
     def remove_device(self, device: str | GoveeDevice) -> None:
         if isinstance(device, GoveeDevice):
             device = device.fingerprint
-        if device in self._devices:
-            del self._devices[device]
+        self._registry.remove_discovered_device(device)
 
     @property
     def evict_enabled(self) -> bool:
@@ -191,24 +197,41 @@ def update_enabled(self) -> bool:
         return self._update_enabled
 
     def send_discovery_message(self) -> None:
-        message: ScanMessage = ScanMessage()
-        if self._transport:
+        message: bytes = bytes(ScanMessage())
+        call_later: bool = False
+        if not self._transport:
+            return
+
+        if self._discovery_enabled:
+            call_later = True
             self._transport.sendto(
-                bytes(message), (self._broadcast_address, self._broadcast_port)
+                message, (self._broadcast_address, self._broadcast_port)
             )
 
-            if self._discovery_enabled:
-                self._discovery_handle = self._loop.call_later(
-                    self._discovery_interval, self.send_discovery_message
-                )
+        if self._registry.has_queued_devices:
+            call_later = True
+            for ip in self._registry.devices_queue:
+                self._transport.sendto(message, (ip, self._broadcast_port))
+
+        manually_added_devices = [
+            device.ip
+            for device in self._registry.discovered_devices.values()
+            if device.is_manual
+        ]
+        if manually_added_devices:
+            call_later = True
+            for ip in manually_added_devices:
+                self._transport.sendto(message, (ip, self._broadcast_port))
+
+        if call_later:
+            self._discovery_handle = self._loop.call_later(
+                self._discovery_interval, self.send_discovery_message
+            )
 
-    def send_update_message(self, device: GoveeDevice | None = None) -> None:
+    def send_update_message(self) -> None:
         if self._transport:
-            if device:
-                self._send_update_message(device=device)
-            else:
-                for d in self._devices.values():
-                    self._send_update_message(device=d)
+            for d in self._registry.discovered_devices.values():
+                self._send_update_message(device=d)
 
             if self._update_enabled:
                 self._update_handle = self._loop.call_later(
@@ -282,22 +305,17 @@ async def send_raw_command(self, device: GoveeDevice, command: str) -> None:
         self._send_message(HexMessage([command]), device)
 
     def get_device_by_ip(self, ip: str) -> GoveeDevice | None:
-        return next(
-            (device for device in self._devices.values() if device.ip == ip),
-            None,
-        )
+        return self._registry.get_device_by_ip(ip)
 
     def get_device_by_sku(self, sku: str) -> GoveeDevice | None:
-        return next(
-            (device for device in self._devices.values() if device.sku == sku), None
-        )
+        return self._registry.get_device_by_sku(sku)
 
     def get_device_by_fingerprint(self, fingerprint: str) -> GoveeDevice | None:
-        return self._devices.get(fingerprint, None)
+        return self._registry.get_device_by_fingerprint(fingerprint)
 
     @property
     def devices(self) -> list[GoveeDevice]:
-        return list(self._devices.values())
+        return list(self._registry.discovered_devices.values())
 
     def connection_made(self, transport):
         self._transport = transport
@@ -385,7 +403,7 @@ async def _handle_scan_response(self, message: ScanResponse) -> None:
                 self, message.ip, fingerprint, message.sku, capabilities
             )
             if self._call_discovered_callback(device, True):
-                self._devices[fingerprint] = device
+                device = self._registry.add_discovered_device(device)
                 self._logger.debug("Device discovered: %s", device)
             else:
                 self._logger.debug("Device %s ignored", device)
@@ -406,15 +424,14 @@ def _send_update_message(self, device: GoveeDevice):
 
     def _evict(self) -> None:
         now = datetime.now()
-        devices = dict(self._devices)
+        devices = dict(self._registry.discovered_devices)
         for fingerprint, device in devices.items():
             diff: timedelta = now - device.lastseen
             if diff.total_seconds() >= self._evict_interval:
                 device._controller = None
-                del self._devices[fingerprint]
+                self._registry.remove_discovered_device(fingerprint)
+                self._logger.debug("Device evicted: %s", device)
                 if self._device_evicted_callback and callable(
                     self._device_evicted_callback
                 ):
-                    self._logger.debug("Device evicted: %s", device)
                     self._device_evicted_callback(device)
-        self._devices = devices

src/govee_local_api/device.py

@@ -41,6 +41,7 @@ def __init__(
         self._temperature_color = 0
         self._brightness = 0
         self._update_callback: Callable[[GoveeDevice], None] | None = None
+        self.is_manual: bool = False
 
     @property
     def controller(self):