Make PyStageLinQ listen to broadcasts on all interfaces

The patch also updates the local IP detection method, as
it did not work on Linux. This also added the psutils
as a dependency.

Solves #47

Author: Jaxc

PyStageLinQ/PyStageLinQ.py

@@ -8,6 +8,7 @@
 import time
 import asyncio
 import logging
+import psutil
 from typing import Callable
 
 from . import Device
@@ -63,11 +64,13 @@ def __init__(
 
         self.device_list = Device.DeviceList()
 
+        self.ip = []
         if ip is None:
-            interfaces = socket.getaddrinfo(
-                host=socket.gethostname(), port=None, family=socket.AF_INET
-            )
-            self.ip = [ip[-1][0] for ip in interfaces]
+            interfaces = psutil.net_if_addrs()
+            for interface in interfaces.items():
+                for interface_address in interface[1]:
+                    if socket.AF_INET == interface_address.family:
+                        self.ip.append(interface_address.address)
 
         else:
             self.ip = [ip]
@@ -155,7 +158,7 @@ async def _discover_stagelinq_device(self, host_ip, timeout=10):
         discover_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
         try:
             discover_socket.bind(
-                (host_ip, self.StageLinQ_discovery_port)
+                ("255.255.255.255", self.StageLinQ_discovery_port)
             )  # bind socket StageLinQ interface
         except:
             # Cannot bind to socket, check if IP is correct and link is up

changelog.md

@@ -2,6 +2,11 @@
 Here follows a log of released versions of PyStageLinQ.
 
 ## [0.2.2]
+### Fixed
+Problems on Linux should now be solved.
+PyStageLinQ will now listen to address "255.255.255.255" on all interfaces, and transmit discovery frames on either the
+interface specified by `PyStageLinQ(..., ip=)` or on all interfaces if `ip` is not set.
+
 ### Added
 More logging output in PyStageLinQ.py.
 

pyproject.toml

@@ -21,6 +21,10 @@ keywords = ["StageLinQ", "DJ", "VJ", "Denon DJ", "Prime Go", "Streaming"]
 
 version = "0.0.0"
 
+dependencies = [
+"psutil",
+]
+
 [project.urls]
 "Homepage" = "https://github.com/Jaxc/PyStageLinQ"
 "Bug Tracker" = "https://github.com/Jaxc/PyStageLinQ/issues"

test_requirements.txt

@@ -1,3 +1,4 @@
 pytest==8.3.2
 pytest-cov==5.0.0
-pytest-asyncio==0.23.8
+pytest-asyncio==0.23.8
+psutil==6.1.0

tests/unit/test_unit_PyStageLinQ.py

@@ -3,6 +3,7 @@
 from PyStageLinQ.ErrorCodes import *
 from unittest.mock import AsyncMock, Mock, MagicMock
 from unittest import mock
+from socket import AF_INET
 
 import random
 
@@ -77,13 +78,26 @@ def test_init_values(dummy_pystagelinq, dummy_ip):
 
 
 def test_init_values_ip_none(monkeypatch, dummy_socket):
-    dummy_ips = [[["1,2,3,4"]], [["5.6.7.8"]]]
-    dummy_socket.getaddrinfo.return_value = dummy_ips
-    monkeypatch.setattr(PyStageLinQ.PyStageLinQ, "socket", dummy_socket)
+    class ifutils_net_if_addrs:
+        def __init__(self, ip=[]):
+            self.address = ip
+            self.family = AF_INET
+
+    dummy_psutil = MagicMock()
+    dummy_ips = {
+        "interface1": [ifutils_net_if_addrs(ip="1.2.3.4")],
+        "interface2": [ifutils_net_if_addrs(ip="5.6.7.8")],
+        "interface3": [ifutils_net_if_addrs(ip="9.10.11.12")],
+    }
+
+    dummy_ips["interface3"][0].family = None
+
+    dummy_psutil.net_if_addrs.return_value = dummy_ips
+    monkeypatch.setattr(PyStageLinQ.PyStageLinQ, "psutil", dummy_psutil)
 
     dummy_pystagelinq = PyStageLinQ.PyStageLinQ.PyStageLinQ(None, ip=None)
 
-    assert dummy_pystagelinq.ip == [dummy_ips[0][0][0], dummy_ips[1][0][0]]
+    assert dummy_pystagelinq.ip == ["1.2.3.4", "5.6.7.8"]
 
 
 def test_start_standalone(dummy_pystagelinq, monkeypatch):
@@ -243,7 +257,7 @@ async def test_discover_stagelinq_check_initialization(
         dummy_socket.AF_INET, dummy_socket.SOCK_DGRAM
     )
     dummy_socket.socket.return_value.bind.assert_called_once_with(
-        (dummy_pystagelinq.ip[0], dummy_pystagelinq.StageLinQ_discovery_port)
+        ("255.255.255.255", dummy_pystagelinq.StageLinQ_discovery_port)
     )
     dummy_socket.socket.return_value.setblocking.assert_called_once_with(False)