An easier way to send and receive from the same port ?

[hoh]

Some devices such as the Behringer XR12-XR16-XR18 are using a communication protocol that is compatible to standard OSC with MUSIC Group specific extensions (e.g. parameter enquiry, subscriptions). As detailed in the specification, OSC packets are received on UDP port 10024 and replies are sent back to the requester's IP/port.

The codes below allow to do just that, but they contain some complexity that would be better located inside the python-osc library.

1. Blocking server that sends and receives on the same UDP port

from typing import Optional, Union

from pythonosc.dispatcher import Dispatcher
from pythonosc.osc_message_builder import OscMessageBuilder
from pythonosc.osc_server import BlockingOSCUDPServer


def handler(*args, **kwargs):
    print("handler", args, kwargs)


dispatcher = Dispatcher()
dispatcher.set_default_handler(handler)


class OSCClientServer(BlockingOSCUDPServer):
    def __init__(self, address: str, dispatcher: Dispatcher):
        super().__init__(("", 0), dispatcher)
        self.xr_address = address

    def send_message(self, address: str, vals: Optional[Union[str, list]]):
        builder = OscMessageBuilder(address=address)
        vals = vals if vals is not None else []
        if not isinstance(vals, list):
            vals = [vals]
        for val in vals:
            builder.add_arg(val)
        msg = builder.build()
        self.socket.sendto(msg.dgram, self.xr_address)


server = OSCClientServer(("192.168.1.1", 10024), dispatcher)

import threading
thread = threading.Thread(target=server.serve_forever, args=tuple())
thread.start()

server.send_message("/info", None)

2. Asyncio server that sends and receives on the same UDP port

import asyncio
from typing import Optional, Union

from pythonosc.dispatcher import Dispatcher
from pythonosc.osc_message_builder import OscMessageBuilder
from pythonosc.osc_server import AsyncIOOSCUDPServer


def handler(*args, **kwargs):
    print("handler", args, kwargs)


dispatcher = Dispatcher()
dispatcher.set_default_handler(handler)


class OSCClientServer(AsyncIOOSCUDPServer):
    def __init__(self, address: str, dispatcher: Dispatcher):
        super().__init__(server_address=("0.0.0.0", 0), dispatcher=dispatcher, loop=asyncio.get_event_loop())
        self.xr_address = address

    def send_message(self, address: str, transport: asyncio.DatagramTransport, vals: Optional[Union[str, list]]):
        builder = OscMessageBuilder(address=address)
        vals = vals if vals is not None else []
        if not isinstance(vals, list):
            vals = [vals]
        for val in vals:
            builder.add_arg(val)
        msg = builder.build()
        transport.sendto(msg.dgram, self.xr_address)


async def main():
    server = OSCClientServer(("192.168.1.1", 10024), dispatcher)
    transport, protocol = await server.create_serve_endpoint()
    server.send_message("/info", transport, None)

    await asyncio.sleep(10)

asyncio.run(main())