syphon.server_directory

  1from dataclasses import dataclass
  2from enum import Enum
  3from typing import Callable, Any, List, Optional
  4
  5import objc
  6from Cocoa import NSRunLoop, NSDefaultRunLoopMode, NSDate, NSImage
  7
  8
  9class SyphonServerNotification(Enum):
 10    """
 11    Enum representing Syphon server notifications.
 12
 13    Enum Values:
 14    - Announce: A new SyphonServer is available on the system.
 15    - Update: An existing SyphonServer instance has changed its description.
 16    - Retire: A SyphonServer instance will no longer be available.
 17    """
 18    Announce = "SyphonServerAnnounceNotification"
 19    Update = "SyphonServerUpdateNotification"
 20    Retire = "SyphonServerRetireNotification"
 21
 22
 23@dataclass
 24class SyphonServerDescription:
 25    """
 26    Data class representing the description of a Syphon server.
 27
 28    Attributes:
 29    - uuid (str): The UUID of the Syphon server.
 30    - name (str): The name of the Syphon server.
 31    - app_name (str): The name of the application associated with the Syphon server.
 32    - icon (NSImage): The icon image of the Syphon server.
 33    - raw (Any): The raw server information.
 34    """
 35    uuid: str
 36    name: str
 37    app_name: str
 38    icon: NSImage
 39    raw: Any
 40
 41
 42class SyphonServerDirectory:
 43    """
 44    Class for interacting with the Syphon server directory.
 45
 46    Attributes:
 47    - run_loop_interval (float): The interval for the run loop in seconds.
 48    """
 49
 50    def __init__(self):
 51        """
 52        Initialize a SyphonServerDirectory.
 53        """
 54        self._syphonServerDirectoryObjC = objc.lookUpClass("SyphonServerDirectory")
 55        self._notification_center = objc.lookUpClass("NSNotificationCenter").defaultCenter()
 56
 57        self.run_loop_interval: float = 1.0
 58
 59    def add_observer(self, notification: SyphonServerNotification, handler: Callable[[Any], None]):
 60        """
 61        Add an observer for a Syphon server notification.
 62
 63        Parameters:
 64        - notification (SyphonServerNotification): The notification to observe.
 65        - handler (Callable[[Any], None]): The handler function to be called when the notification is received.
 66        """
 67        self._notification_center.addObserverForName_object_queue_usingBlock_(
 68            notification.value,
 69            None,
 70            None,
 71            handler
 72        )
 73
 74    @property
 75    def servers(self) -> List[SyphonServerDescription]:
 76        """
 77        Get a list of Syphon servers in the directory.
 78
 79        Returns:
 80        - List[SyphonServerDescription]: A list of SyphonServerDescription objects.
 81        """
 82        self.update_run_loop()
 83        directory = self._syphonServerDirectoryObjC.sharedDirectory()
 84        servers = directory.servers()
 85
 86        return [
 87            SyphonServerDescription(
 88                str(s["SyphonServerDescriptionUUIDKey"]),
 89                str(s["SyphonServerDescriptionNameKey"]),
 90                str(s["SyphonServerDescriptionAppNameKey"]),
 91                s["SyphonServerDescriptionIconKey"],
 92                s
 93            )
 94            for s in servers
 95        ]
 96
 97    def update_run_loop(self):
 98        """
 99        Update the run loop to process events.
100        """
101        NSRunLoop.currentRunLoop().runMode_beforeDate_(
102            NSDefaultRunLoopMode,
103            NSDate.dateWithTimeIntervalSinceNow_(self.run_loop_interval)
104        )
105
106    def servers_matching_name(self,
107                              name: Optional[str] = None,
108                              app_name: Optional[str] = None) -> List[SyphonServerDescription]:
109        """
110        Get a list of Syphon servers that match the specified name or application name.
111
112        Parameters:
113        - name (Optional[str]): The name to match.
114        - app_name (Optional[str]): The application name to match.
115
116        Returns:
117        - List[SyphonServerDescription]: A list of SyphonServerDescription objects that match the criteria.
118        """
119        filtered_servers = []
120
121        for server in self.servers:
122            if name is not None and name == server.name:
123                filtered_servers.append(server)
124                continue
125
126            if app_name is not None and app_name == server.app_name:
127                filtered_servers.append(server)
128                continue
129
130        return filtered_servers