vmcp.osc

 23class OSC:
 24    """Open Sound Control (OSC) network protocol system."""
 25
 26    _senders: dict[str, Sender]
 27    """dict[str, Sender]: Holds sender channel references."""
 28
 29    _receivers: dict[str, Receiver]
 30    """dict[str, Receiver]: Holds receiver channel references."""
 31
 32    @property
 33    def backend(self) -> BackendType:
 34        """BackendType: Returns system's currently used backend module."""
 35        return self._backend
 36
 37    @property
 38    def is_open(self) -> bool:
 39        """bool: Returns whether the system is currently ready to use."""
 40        return self._opened
 41
 42    def __init__(
 43        self,
 44        backend: BackendType,
 45        logger: Logger = None
 46    ) -> None:
 47        """OSC constructor.
 48
 49        Args:
 50            backend (BackendType):
 51                Backend module.
 52            logger (Optional[Logger]):
 53                Logger instance.
 54
 55        """
 56        self._senders = {}
 57        self._receivers = {}
 58        self._opened = False
 59        self._backend = backend
 60        self._logger = logger
 61
 62    def open(self) -> 'OSC':
 63        """Start system.
 64
 65        Returns:
 66            OSC:
 67                Object instance.
 68
 69        """
 70        if not self.is_open:
 71            self.backend.startup(
 72                logger=self._logger
 73            )
 74            self._opened = True
 75        return self
 76
 77    def __enter__(self: OSCClass) -> OSCClass:
 78        """Return the object instance for usage of the ``with`` statement.
 79
 80        Returns:
 81            OSCClass:
 82                Object instance.
 83
 84        """
 85        self.open()
 86        return self
 87
 88    def create_sender(
 89        self,
 90        host: str,
 91        port: int,
 92        name: str
 93    ) -> Sender:
 94        """Create sender channel.
 95
 96        Args:
 97            host (str):
 98                Host DNS or IP (recommended).
 99            port (int):
100                Port.
101            name (str):
102                Channel name (arbitrary, but need to be unique).
103
104        Returns:
105            Sender:
106                Channel instance.
107
108        Raises:
109            ValueError:
110                If ``name`` is already used.
111
112        """
113        sender = Sender(self, host, port, name)
114        self._senders[name] = sender
115        return sender
116
117    def get_sender(
118        self,
119        name: str
120    ) -> Sender:
121        """Get sender channel instance by it's name.
122
123        Args:
124            name (str):
125                Channel name.
126
127        Returns:
128            Sender:
129                Channel instance.
130
131        Raises:
132            KeyError:
133                If ``name`` does not exists.
134
135        """
136        return self._senders[name]
137
138    def create_receiver(
139        self,
140        host: str,
141        port: int,
142        name: str
143    ) -> Receiver:
144        """Create receiver channel.
145
146        Args:
147            host (str):
148                Host DNS or IP (recommended).
149            port (int):
150                Port.
151            name (str):
152                Channel name (arbitrary, but need to be unique).
153
154        Returns:
155            Receiver:
156                Channel instance.
157
158        Raises:
159            ValueError:
160                If ``name`` is already used.
161
162        """
163        receiver = Receiver(self, host, port, name)
164        self._receivers[name] = receiver
165        return receiver
166
167    def get_receiver(
168        self,
169        name: str
170    ) -> Receiver:
171        """Get receiver channel instance by it's name.
172
173        Args:
174            name (str):
175                Channel name.
176
177        Returns:
178            Receiver:
179                Channel instance.
180
181        Raises:
182            KeyError:
183                If ``name`` does not exists.
184
185        """
186        return self._receivers[name]
187
188    def run(self) -> None:
189        """Process on every invoke."""
190        self.backend.process()
191
192    def __str__(self) -> str:
193        """Return string representation of the object.
194
195        Returns:
196            str:
197                Representation of the object.
198
199        """
200        return (
201            f"{{'sender': {list(self._senders.values())},"
202            f" 'receiver': {list(self._receivers.values())}}}"
203        )
204
205    def close(self) -> None:
206        """Shut down system."""
207        if self.is_open:
208            self.backend.terminate()
209            self._opened = False
210
211    def __exit__(
212        self,
213        exc_type: type[BaseException] = None,
214        exc_value: BaseException = None,
215        traceback: TracebackType = None
216    ) -> Literal[False]:
217        """Return everytime ``False`` if exiting the runtime context.
218
219        Args:
220            exc_type (Optional[type[BaseException]]):
221                Exception type.
222            exc_value (Optional[BaseException]):
223                Exception instance.
224            traceback (Optional[TracebackType]):
225                Exception context.
226
227        Returns:
228            bool:
229                Everytime ``False``.
230
231        """
232        self.close()
233        return False