vmcp.osc.channel

15class Sender(AbstractChannel):
16    """Sender."""
17
18    def open(self) -> 'Sender':
19        """Open channel.
20
21        Returns:
22            Class:
23                Object instance.
24
25        """
26        if not self.is_open:
27            super().open()
28            self.system.backend.client(self.host, self.port, self.name)
29        return self
30
31    def send(
32        self,
33        messages: Message | tuple[Message, ...] | list[Message],
34        delay: TimeTagType | None = None
35    ) -> None:
36        """Send message(s).
37
38        Args:
39            messages (Message | tuple[Message, ...] | list[Message]):
40                One or many ``Message``(s).
41            delay (TimeTagType | None):
42                Message processing delay (Default: Depends on backend).
43
44        """
45        if delay is None:
46            delay = self.system.backend.TIME_DEFAULT
47        packet: tuple[Message, ...] = ()
48        if isinstance(messages, Message):
49            packet = (messages,)
50        else:
51            packet = tuple(messages)
52        self.system.backend.send(
53            self.name,
54            packet,
55            delay
56        )
57
58    def __repr__(self) -> str:
59        """Return a string, representing the object in a reconstructable way.
60
61        Returns:
62            str:
63                Representing the object in a reconstructable way.
64
65        """
66        return f"Sender({self})"
67
68    def close(self) -> None:
69        """Close channel."""
70        if self.is_open:
71            self.system.backend.get_channel(self.name).terminate()
72            super().close()

 20class Receiver(AbstractChannel):
 21    """Receiver."""
 22
 23    _handler_methods: dict[UUID, object]
 24
 25    def __init__(
 26        self,
 27        system: 'OSC',
 28        host: str,
 29        port: int,
 30        name: str
 31    ) -> None:
 32        """Channel constructor.
 33
 34        Args:
 35            system (OSC):
 36                OSC system instance.
 37            host (str):
 38                Host DNS or IP (recommended).
 39            port (int):
 40                Host port.
 41            name (str):
 42                Channel name (arbitrary, but needs to be unique).
 43
 44        Raises:
 45            ValueError:
 46                If ``name`` is already used.
 47
 48        """
 49        super().__init__(system, host, port, name)
 50        self._handler_methods = {}
 51
 52    def open(self) -> 'Receiver':
 53        """Open channel.
 54
 55        Returns:
 56            Class:
 57                Object instance.
 58
 59        """
 60        if not self.is_open:
 61            super().open()
 62            self.system.backend.server(self.host, self.port, self.name)
 63        return self
 64
 65    def register_handler(
 66        self,
 67        address: str,
 68        handler: Callable[..., None],
 69        scheme: ArgumentsScheme | None = None,
 70        extra: Any | None = None
 71    ) -> UUID:
 72        """Register receiving handler.
 73
 74        Args:
 75            address (str):
 76                OSC address pattern filter string expression.
 77            handler (Callable[..., None]):
 78                Handler function.
 79            scheme (ArgumentsScheme | None):
 80                Argument schemes (Default: Depends on backend).
 81            extra (Any | None):
 82                Extra parameter(s) for your handler function.
 83
 84        Return:
 85            UUID:
 86                UUID object, which identifies the handler.
 87
 88        """
 89        if scheme is None:
 90            scheme = self.system.backend.ARG_DEFAULT
 91        uuid = uuid4()
 92        method = self.system.backend.MethodFilter(
 93            addrpattern=address,
 94            function=lambda *args: (
 95                handler(*args[1:]) if args[0] is self.name else ()
 96            ),
 97            argscheme=(self.system.backend.ARG_READERNAME + scheme).scheme,
 98            extra=extra
 99        )
100        self.system.backend.register_method(method)
101        self._handler_methods[uuid] = method
102        return uuid
103
104    def unregister_handler(
105        self,
106        uuid: UUID
107    ) -> None:
108        """Unregister receiving handler.
109
110        Args:
111            uuid (UUID):
112                UUID object, which identifies the handler.
113
114        """
115        self.system.backend.unregister_method(self._handler_methods.pop(uuid))
116
117    def __repr__(self) -> str:
118        """Return a string, representing the object in a reconstructable way.
119
120        Returns:
121            str:
122                Representing the object in a reconstructable way.
123
124        """
125        return f"Receiver({self})"
126
127    def close(self) -> None:
128        """Close channel."""
129        if self.is_open:
130            for uuid in self._handler_methods:
131                self.unregister_handler(uuid)
132            self.system.backend.get_channel(self.name).terminate()
133            super().close()