support ipv6 in OSCUDPServer

Author: attwad

pythonosc/osc_message_builder.py

@@ -72,17 +72,19 @@ def args(self) -> List[Tuple[str, Union[ArgValue, None]]]:
         """Returns the (type, value) arguments list of this message."""
         return self._args
 
-    def _valid_type(self, arg_type: str) -> bool:
-        if arg_type in self._SUPPORTED_ARG_TYPES:
-            return True
+    def _valid_type(self, arg_type: Union[str, List[Any]]) -> bool:
+        if isinstance(arg_type, str):
+            return arg_type in self._SUPPORTED_ARG_TYPES
         elif isinstance(arg_type, list):
             for sub_type in arg_type:
                 if not self._valid_type(sub_type):
                     return False
             return True
         return False
 
-    def add_arg(self, arg_value: ArgValue, arg_type: Optional[str] = None) -> None:
+    def add_arg(
+        self, arg_value: ArgValue, arg_type: Optional[Union[str, List[Any]]] = None
+    ) -> None:
         """Add a typed argument to this message.
 
         Args:
@@ -100,7 +102,7 @@ def add_arg(self, arg_value: ArgValue, arg_type: Optional[str] = None) -> None:
             arg_type = self._get_arg_type(arg_value)
         if isinstance(arg_type, list):
             self._args.append((self.ARG_TYPE_ARRAY_START, None))
-            for v, t in zip(arg_value, arg_type):  # type: ignore[arg-type, var-annotated]
+            for v, t in zip(arg_value, arg_type):  # type: ignore[arg-type]
                 self.add_arg(v, t)
             self._args.append((self.ARG_TYPE_ARRAY_STOP, None))
         else:

pythonosc/osc_server.py

@@ -2,6 +2,7 @@
 
 import asyncio
 import os
+import socket
 import socketserver
 from socket import socket as _socket
 from typing import Any, Coroutine, Tuple, Union, cast
@@ -64,6 +65,7 @@ def __init__(
         dispatcher: Dispatcher,
         bind_and_activate: bool = True,
         timeout: float | None = None,
+        family: socket.AddressFamily | None = None,
     ) -> None:
         """Initialize
 
@@ -72,7 +74,25 @@ def __init__(
             dispatcher: Dispatcher this server will use
             (optional) bind_and_activate: default=True defines if the server has to start on call of constructor
             (optional) timeout: Default timeout in seconds for socket operations
+            (optional) family: socket.AF_INET or socket.AF_INET6. If None, it will be inferred from server_address.
         """
+        if family is not None:
+            self.address_family = family
+        else:
+            # Try to infer address family from server_address
+            try:
+                infos = socket.getaddrinfo(
+                    server_address[0],
+                    server_address[1],
+                    type=socket.SOCK_DGRAM,
+                    family=socket.AF_UNSPEC,
+                )
+                if infos:
+                    self.address_family = infos[0][0]
+            except (socket.gaierror, IndexError):
+                # Fallback to default if resolution fails
+                pass
+
         super().__init__(server_address, _UDPHandler, bind_and_activate)
         self._dispatcher = dispatcher
         self.timeout = timeout

pythonosc/osc_tcp_server.py

@@ -35,6 +35,7 @@
 import asyncio
 import logging
 import os
+import socket
 import socketserver
 import struct
 from typing import List, Tuple
@@ -146,11 +147,30 @@ def __init__(
         server_address: Tuple[str | bytes | bytearray, int],
         dispatcher: Dispatcher,
         mode: str = MODE_1_1,
+        family: socket.AddressFamily | None = None,
     ):
         self.request_queue_size = 300
         self.mode = mode
         if mode not in [MODE_1_0, MODE_1_1]:
             raise ValueError("OSC Mode must be '1.0' or '1.1'")
+
+        if family is not None:
+            self.address_family = family
+        elif isinstance(server_address[0], str):
+            # Try to infer address family from server_address
+            try:
+                infos = socket.getaddrinfo(
+                    server_address[0],
+                    server_address[1],
+                    type=socket.SOCK_STREAM,
+                    family=socket.AF_UNSPEC,
+                )
+                if infos:
+                    self.address_family = infos[0][0]
+            except (socket.gaierror, IndexError):
+                # Fallback to default if resolution fails
+                pass
+
         if self.mode == MODE_1_0:
             super().__init__(server_address, _TCPHandler1_0)
         else:

pythonosc/test/test_osc_server.py

@@ -1,3 +1,4 @@
+import socket
 import unittest
 import unittest.mock
 
@@ -119,6 +120,21 @@ def test_init_timeout(self, mock_socket_ctor):
         server = osc_server.OSCUDPServer(("127.0.0.1", 0), dispatcher, timeout=10.0)
         self.assertEqual(server.timeout, 10.0)
 
+    @unittest.mock.patch("socket.socket")
+    def test_init_family_inference_ipv4(self, mock_socket_ctor):
+        dispatcher = unittest.mock.Mock()
+        server = osc_server.OSCUDPServer(("127.0.0.1", 0), dispatcher)
+        self.assertEqual(server.address_family, socket.AF_INET)
+
+    @unittest.mock.patch("socket.socket")
+    def test_init_family_inference_ipv6(self, mock_socket_ctor):
+        dispatcher = unittest.mock.Mock()
+        # Mock getaddrinfo to return IPv6 for this test to be environment-independent
+        with unittest.mock.patch("socket.getaddrinfo") as mock_getaddrinfo:
+            mock_getaddrinfo.return_value = [(socket.AF_INET6, None, None, None, None)]
+            server = osc_server.OSCUDPServer(("::1", 0), dispatcher)
+            self.assertEqual(server.address_family, socket.AF_INET6)
+
 
 if __name__ == "__main__":
     unittest.main()