tessl/pypi-python-osc

Open Sound Control server and client implementations in pure Python for networked music and multimedia applications

—

Pending

Overview

Eval results

Files

UDP Networking

Name: tessl/pypi-python-osc
Author: tessl

UDP client implementations for sending OSC messages and bundles with support for broadcast transmissions, timeout handling, message reception, and response dispatching. UDP provides low-latency communication suitable for real-time applications.

Capabilities

Simple UDP Client

High-level UDP client with message building helpers for quick OSC communication.

class SimpleUDPClient:
    """UDP client with message building helpers."""
    
    def __init__(self, address: str, port: int, allow_broadcast: bool = False):
        """Initialize UDP client.
        
        Parameters:
        - address: Target IP address or hostname
        - port: Target UDP port number
        - allow_broadcast: Enable broadcast transmissions
        """
    
    def send_message(self, address: str, value: ArgValue):
        """Build and send OSC message.
        
        Parameters:
        - address: OSC address pattern string
        - value: Single argument or list of arguments
        """
    
    def get_messages(self, timeout: int = 30) -> Generator[OscMessage, None, None]:
        """Receive and parse OSC messages.
        
        Parameters:
        - timeout: Receive timeout in seconds
        
        Yields:
        OscMessage objects from received datagrams
        
        Raises:
        socket.timeout: If no data received within timeout
        """

Basic UDP Client

Low-level UDP client for sending pre-built messages and bundles.

class UDPClient:
    """Basic OSC UDP client for sending messages and bundles."""
    
    def __init__(self, address: str, port: int, allow_broadcast: bool = False, 
                 family: socket.AddressFamily = socket.AF_UNSPEC):
        """Initialize UDP client.
        
        Parameters:
        - address: Target IP address or hostname
        - port: Target UDP port number  
        - allow_broadcast: Enable broadcast transmissions
        - family: Address family (AF_INET, AF_INET6, AF_UNSPEC)
        """
    
    def send(self, content: Union[OscMessage, OscBundle]):
        """Send OSC message or bundle.
        
        Parameters:
        - content: OscMessage or OscBundle to transmit
        
        Raises:
        OSError: If transmission fails
        """
    
    def receive(self, timeout: int = 30) -> bytes:
        """Receive raw datagram bytes.
        
        Parameters:
        - timeout: Receive timeout in seconds
        
        Returns:
        Raw datagram bytes
        
        Raises:
        socket.timeout: If no data received within timeout
        """

Dispatcher Client

UDP client with automatic response handling using dispatcher pattern for bidirectional communication.

class DispatchClient:
    """UDP client with dispatcher for handling responses."""
    
    dispatcher: Dispatcher  # Class-level dispatcher instance
    
    def __init__(self, address: str, port: int, allow_broadcast: bool = False):
        """Initialize dispatcher client.
        
        Parameters:
        - address: Target IP address or hostname
        - port: Target UDP port number
        - allow_broadcast: Enable broadcast transmissions
        """
    
    def send_message(self, address: str, value: ArgValue):
        """Build and send OSC message.
        
        Parameters:
        - address: OSC address pattern string
        - value: Single argument or list of arguments
        """
    
    def handle_messages(self, timeout: int = 30):
        """Process received messages with dispatcher.
        
        Parameters:
        - timeout: Receive timeout in seconds
        
        Processes incoming messages using the class dispatcher,
        calling mapped handlers for matching addresses.
        """

Usage Examples

Basic Message Sending

from pythonosc import udp_client

# Create simple client
client = udp_client.SimpleUDPClient("127.0.0.1", 5005)

# Send various message types
client.send_message("/filter", 0.75)
client.send_message("/volume", [0.8, 0.6])  # Stereo volume
client.send_message("/synth/freq", 440.0)
client.send_message("/trigger", True)
client.send_message("/reset", [])  # No arguments

Broadcast Communication

from pythonosc import udp_client

# Enable broadcast for network-wide messages
broadcast_client = udp_client.SimpleUDPClient("255.255.255.255", 9000, 
                                             allow_broadcast=True)
broadcast_client.send_message("/sync/beat", [120, 4, 4])  # BPM, time signature

Sending Pre-built Messages

from pythonosc import udp_client
from pythonosc.osc_message_builder import OscMessageBuilder
from pythonosc.osc_bundle_builder import OscBundleBuilder, IMMEDIATELY

# Create client
client = udp_client.UDPClient("192.168.1.100", 8000)

# Send pre-built message
builder = OscMessageBuilder("/instrument/piano")
builder.add_arg(60)    # MIDI note
builder.add_arg(127)   # Velocity
builder.add_arg(1.5)   # Duration
message = builder.build()
client.send(message)

# Send bundle
bundle_builder = OscBundleBuilder(IMMEDIATELY)
bundle_builder.add_content(OscMessageBuilder("/note/on").add_arg(60).build())
bundle_builder.add_content(OscMessageBuilder("/note/on").add_arg(64).build())
bundle = bundle_builder.build()
client.send(bundle)

Receiving Messages

from pythonosc import udp_client
import socket

# Create receiving client
receiver = udp_client.SimpleUDPClient("127.0.0.1", 5006)

try:
    # Receive messages with timeout
    for message in receiver.get_messages(timeout=10):
        print(f"Received: {message.address} -> {message.params}")
except socket.timeout:
    print("No messages received within timeout")

Bidirectional Communication with Dispatcher

from pythonosc import udp_client
from pythonosc.dispatcher import Dispatcher

# Set up response handlers
def handle_ack(address, *args):
    print(f"Acknowledgment: {address} {args}")

def handle_error(address, *args):
    print(f"Error response: {address} {args}")

# Configure class dispatcher
DispatchClient.dispatcher = Dispatcher()
DispatchClient.dispatcher.map("/ack", handle_ack)
DispatchClient.dispatcher.map("/error", handle_error)

# Create dispatcher client
client = udp_client.DispatchClient("127.0.0.1", 5005)

# Send message and handle responses
client.send_message("/request/data", ["sensor1", "temperature"])

# Process responses
try:
    client.handle_messages(timeout=5)
except socket.timeout:
    print("No response received")

IPv6 Support

from pythonosc import udp_client
import socket

# IPv6 client
ipv6_client = udp_client.UDPClient("::1", 5005, family=socket.AF_INET6)
ipv6_client.send_message("/ipv6/test", "Hello IPv6")

# Auto-detect family
auto_client = udp_client.UDPClient("localhost", 5005, family=socket.AF_UNSPEC)
auto_client.send_message("/auto/detect", "Hello")

Raw Datagram Handling

from pythonosc import udp_client
from pythonosc.osc_message import OscMessage
from pythonosc.osc_bundle import OscBundle

# Low-level client
client = udp_client.UDPClient("127.0.0.1", 5005)

# Receive and parse manually
try:
    raw_data = client.receive(timeout=30)
    
    if OscMessage.dgram_is_message(raw_data):
        message = OscMessage(raw_data)
        print(f"Message: {message.address} {message.params}")
    elif OscBundle.dgram_is_bundle(raw_data):
        bundle = OscBundle(raw_data)
        print(f"Bundle: {bundle.num_contents} items at {bundle.timestamp}")
        
except socket.timeout:
    print("Receive timeout")

Network Configuration Examples

Multi-interface Setup

from pythonosc import udp_client
import socket

# Bind to specific interface on multi-homed systems
interfaces = {
    "local": udp_client.SimpleUDPClient("127.0.0.1", 5005),
    "lan": udp_client.SimpleUDPClient("192.168.1.100", 5005),
    "wan": udp_client.SimpleUDPClient("203.0.113.10", 5005)
}

# Send to different networks
interfaces["local"].send_message("/local/test", "localhost")
interfaces["lan"].send_message("/lan/broadcast", "network")
interfaces["wan"].send_message("/wan/remote", "internet")

Performance Tuning

from pythonosc import udp_client
import socket

# Create client with socket options
client = udp_client.UDPClient("127.0.0.1", 5005)

# Access underlying socket for tuning (if needed)
# Note: This requires accessing private attributes and should be used carefully
sock = client._sock
sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 65536)  # Send buffer
sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 65536)  # Receive buffer

Types and Imports

from typing import Union, Generator
import socket

from pythonosc.osc_message import OscMessage
from pythonosc.osc_bundle import OscBundle
from pythonosc.osc_message_builder import ArgValue
from pythonosc.dispatcher import Dispatcher

Install with Tessl CLI