A custom DISPLAY to determinate the socket_path.

Add `display` argument to sync and async `Connetion()` classes
to determineate the socket_path. Useful when you work with many
Xephyr servers.

Author: rysson

i3ipc/aio/connection.py

@@ -179,7 +179,7 @@ def _unpack_header(data: bytes) -> Tuple[bytes, int, int]:
     return struct.unpack(_struct_header, data[:_struct_header_size])
 
 
-async def _find_socket_path() -> Optional[str]:
+async def _find_socket_path(disp: Optional[str] = None) -> Optional[str]:
     socket_path = None
 
     def exists(path):
@@ -205,7 +205,7 @@ def exists(path):
 
     # next try the root window property
     try:
-        d = display.Display()
+        d = display.Display(disp)
         atom = d.get_atom('I3_SOCKET_PATH')
         root = d.screen().root
         prop = root.get_full_property(atom, X.AnyPropertyType)
@@ -269,17 +269,21 @@ class Connection:
     :param auto_reconnect: Whether to attempt to reconnect if the connection to
         the socket is broken when i3 restarts.
     :type auto_reconnect: bool
+    :param display: A custom DISPLAY to determinate the socket_path.
+    :type display: str
 
     :raises Exception: If the connection to i3 cannot be established.
     """
-    def __init__(self, socket_path: Optional[str] = None, auto_reconnect: bool = False):
+    def __init__(self, socket_path: Optional[str] = None, auto_reconnect: bool = False, *,
+                 display: Optional[str] = None):
         self._socket_path = socket_path
         self._auto_reconnect = auto_reconnect
         self._pubsub = _AIOPubSub(self)
         self._subscriptions = set()
         self._main_future = None
         self._reconnect_future = None
         self._synchronizer = None
+        self._display = display
 
     def _sync(self):
         if self._synchronizer is None:
@@ -390,7 +394,7 @@ async def connect(self) -> 'Connection':
             logger.info('using user provided socket path: {}', self._socket_path)
 
         if not self._socket_path:
-            self._socket_path = await _find_socket_path()
+            self._socket_path = await _find_socket_path(self._display)
 
         if not self.socket_path:
             raise Exception('Failed to retrieve the i3 or sway IPC socket path')

i3ipc/connection.py

@@ -44,6 +44,8 @@ class Connection:
     :param auto_reconnect: Whether to attempt to reconnect if the connection to
         the socket is broken when i3 restarts.
     :type auto_reconnect: bool
+    :param display: A custom DISPLAY to determinate the socket_path.
+    :type display: str
 
     :raises Exception: If the connection to i3 cannot be established.
     """
@@ -53,7 +55,7 @@ class Connection:
     _struct_header = '=%dsII' % len(_MAGIC.encode('utf-8'))
     _struct_header_size = struct.calcsize(_struct_header)
 
-    def __init__(self, socket_path=None, auto_reconnect=False):
+    def __init__(self, socket_path=None, auto_reconnect=False, display=None):
 
         if socket_path:
             logger.info('using user provided socket path: %s', socket_path)
@@ -74,6 +76,7 @@ def __init__(self, socket_path=None, auto_reconnect=False):
         self._auto_reconnect = auto_reconnect
         self._quitting = False
         self._synchronizer = None
+        self._display = display
 
     def _find_socket_path(self):
         socket_path = os.environ.get("I3SOCK")
@@ -87,7 +90,7 @@ def _find_socket_path(self):
             return socket_path
 
         try:
-            disp = Xlib.display.Display()
+            disp = Xlib.display.Display(self._display)
             root = disp.screen().root
             i3atom = disp.intern_atom("I3_SOCKET_PATH")
             prop = root.get_full_property(i3atom, Xlib.X.AnyPropertyType)