0
# Data Types and Parsing
1

2
Low-level OSC data type encoding/decoding, NTP timestamp handling, and SLIP protocol implementation. These modules provide the foundation for OSC message formatting and reliable TCP communication.
3

4
## Capabilities
5

6
### OSC Type Conversion
7

8
Functions for converting between Python types and OSC binary format with full support for all standard OSC data types.
9

10
```python { .api }
11
# String handling
12
def write_string(val: str) -> bytes:
13
    """Convert Python string to OSC string format.
14
    
15
    Parameters:
16
    - val: Python string to encode
17
    
18
    Returns:
19
    OSC-formatted string bytes (UTF-8 encoded, null-terminated, padded to 4-byte boundary)
20
    
21
    Raises:
22
    BuildError: If string cannot be encoded
23
    """
24

25
def get_string(dgram: bytes, start_index: int) -> Tuple[str, int]:
26
    """Parse OSC string from datagram.
27
    
28
    Parameters:
29
    - dgram: Datagram bytes containing OSC string
30
    - start_index: Starting position in datagram
31
    
32
    Returns:
33
    Tuple of (parsed_string, next_index)
34
    
35
    Raises:
36
    ParseError: If string cannot be parsed
37
    """
38

39
# Integer handling
40
def write_int(val: int) -> bytes:
41
    """Convert Python int to OSC int32 format.
42
    
43
    Parameters:
44
    - val: 32-bit integer value
45
    
46
    Returns:
47
    4-byte big-endian integer representation
48
    
49
    Raises:
50
    BuildError: If value exceeds int32 range
51
    """
52

53
def get_int(dgram: bytes, start_index: int) -> Tuple[int, int]:
54
    """Parse OSC int32 from datagram.
55
    
56
    Parameters:
57
    - dgram: Datagram bytes containing OSC int32
58
    - start_index: Starting position in datagram
59
    
60
    Returns:
61
    Tuple of (parsed_int, next_index)
62
    
63
    Raises:
64
    ParseError: If integer cannot be parsed
65
    """
66

67
def write_int64(val: int) -> bytes:
68
    """Convert Python int to OSC int64 format.
69
    
70
    Parameters:
71
    - val: 64-bit integer value
72
    
73
    Returns:
74
    8-byte big-endian integer representation
75
    """
76

77
def get_int64(dgram: bytes, start_index: int) -> Tuple[int, int]:
78
    """Parse OSC int64 from datagram.
79
    
80
    Parameters:
81
    - dgram: Datagram bytes containing OSC int64
82
    - start_index: Starting position in datagram
83
    
84
    Returns:
85
    Tuple of (parsed_int64, next_index)
86
    """
87

88
def get_uint64(dgram: bytes, start_index: int) -> Tuple[int, int]:
89
    """Parse OSC uint64 from datagram.
90
    
91
    Parameters:
92
    - dgram: Datagram bytes containing OSC uint64
93
    - start_index: Starting position in datagram
94
    
95
    Returns:
96
    Tuple of (parsed_uint64, next_index)
97
    """
98

99
# Float handling
100
def write_float(val: float) -> bytes:
101
    """Convert Python float to OSC float32 format.
102
    
103
    Parameters:
104
    - val: Float value for 32-bit representation
105
    
106
    Returns:
107
    4-byte IEEE 754 single-precision representation
108
    """
109

110
def get_float(dgram: bytes, start_index: int) -> Tuple[float, int]:
111
    """Parse OSC float32 from datagram.
112
    
113
    Parameters:
114
    - dgram: Datagram bytes containing OSC float32
115
    - start_index: Starting position in datagram
116
    
117
    Returns:
118
    Tuple of (parsed_float, next_index)
119
    """
120

121
def write_double(val: float) -> bytes:
122
    """Convert Python float to OSC double/float64 format.
123
    
124
    Parameters:
125
    - val: Float value for 64-bit representation
126
    
127
    Returns:
128
    8-byte IEEE 754 double-precision representation
129
    """
130

131
def get_double(dgram: bytes, start_index: int) -> Tuple[float, int]:
132
    """Parse OSC double/float64 from datagram.
133
    
134
    Parameters:
135
    - dgram: Datagram bytes containing OSC double
136
    - start_index: Starting position in datagram
137
    
138
    Returns:
139
    Tuple of (parsed_double, next_index)
140
    """
141

142
# Binary data handling
143
def write_blob(val: bytes) -> bytes:
144
    """Convert Python bytes to OSC blob format.
145
    
146
    Parameters:
147
    - val: Binary data to encode
148
    
149
    Returns:
150
    OSC blob with 4-byte size prefix and padded data
151
    """
152

153
def get_blob(dgram: bytes, start_index: int) -> Tuple[bytes, int]:
154
    """Parse OSC blob from datagram.
155
    
156
    Parameters:
157
    - dgram: Datagram bytes containing OSC blob
158
    - start_index: Starting position in datagram
159
    
160
    Returns:
161
    Tuple of (parsed_bytes, next_index)
162
    """
163

164
# Special type handling
165
def write_rgba(val: Tuple[int, int, int, int]) -> bytes:
166
    """Convert RGBA tuple to OSC RGBA color format.
167
    
168
    Parameters:
169
    - val: (red, green, blue, alpha) tuple with values 0-255
170
    
171
    Returns:
172
    4-byte RGBA color representation
173
    """
174

175
def get_rgba(dgram: bytes, start_index: int) -> Tuple[Tuple[int, int, int, int], int]:
176
    """Parse OSC RGBA color from datagram.
177
    
178
    Parameters:
179
    - dgram: Datagram bytes containing OSC RGBA
180
    - start_index: Starting position in datagram
181
    
182
    Returns:
183
    Tuple of ((r, g, b, a), next_index)
184
    """
185

186
def write_midi(val: MidiPacket) -> bytes:
187
    """Convert MIDI packet to OSC MIDI format.
188
    
189
    Parameters:
190
    - val: (port_id, status_byte, data1, data2) MIDI packet
191
    
192
    Returns:
193
    4-byte OSC MIDI representation
194
    """
195

196
def get_midi(dgram: bytes, start_index: int) -> Tuple[MidiPacket, int]:
197
    """Parse OSC MIDI message from datagram.
198
    
199
    Parameters:
200
    - dgram: Datagram bytes containing OSC MIDI
201
    - start_index: Starting position in datagram
202
    
203
    Returns:
204
    Tuple of (midi_packet, next_index)
205
    """
206
```
207

208
### Timestamp Handling
209

210
OSC timetag conversion functions for precise timing control with NTP timestamp format.
211

212
```python { .api }
213
def write_date(system_time: Union[float, int]) -> bytes:
214
    """Convert system time to OSC timetag format.
215
    
216
    Parameters:
217
    - system_time: System time in seconds since epoch, or IMMEDIATELY constant
218
    
219
    Returns:
220
    8-byte NTP timestamp for OSC timetag
221
    """
222

223
def get_date(dgram: bytes, start_index: int) -> Tuple[float, int]:
224
    """Parse OSC timetag from datagram as system time.
225
    
226
    Parameters:
227
    - dgram: Datagram bytes containing OSC timetag
228
    - start_index: Starting position in datagram
229
    
230
    Returns:
231
    Tuple of (system_time, next_index)
232
    """
233

234
def get_timetag(dgram: bytes, start_index: int) -> Tuple[datetime.datetime, int]:
235
    """Parse OSC timetag from datagram as datetime object.
236
    
237
    Parameters:
238
    - dgram: Datagram bytes containing OSC timetag
239
    - start_index: Starting position in datagram
240
    
241
    Returns:
242
    Tuple of (datetime_object, next_index)
243
    """
244

245
# Special timing constant
246
IMMEDIATELY: int = 0  # Special value for immediate execution
247
```
248

249
### NTP Timestamp Functions
250

251
Low-level NTP timestamp manipulation for precise timing operations.
252

253
```python { .api }
254
def parse_timestamp(timestamp: int) -> Timestamp:
255
    """Parse NTP timestamp into seconds and fraction components.
256
    
257
    Parameters:
258
    - timestamp: 64-bit NTP timestamp
259
    
260
    Returns:
261
    Timestamp namedtuple with seconds and fraction fields
262
    """
263

264
def ntp_to_system_time(timestamp: bytes) -> float:
265
    """Convert NTP timestamp bytes to system time.
266
    
267
    Parameters:
268
    - timestamp: 8-byte NTP timestamp
269
    
270
    Returns:
271
    System time in seconds since Unix epoch
272
    
273
    Raises:
274
    NtpError: If timestamp cannot be converted
275
    """
276

277
def system_time_to_ntp(seconds: float) -> bytes:
278
    """Convert system time to NTP timestamp bytes.
279
    
280
    Parameters:
281
    - seconds: System time in seconds since Unix epoch
282
    
283
    Returns:
284
    8-byte NTP timestamp representation
285
    """
286

287
def ntp_time_to_system_epoch(seconds: float) -> float:
288
    """Convert NTP epoch time to system epoch time.
289
    
290
    Parameters:
291
    - seconds: Seconds since NTP epoch (1900-01-01)
292
    
293
    Returns:
294
    Seconds since Unix epoch (1970-01-01)
295
    """
296

297
def system_time_to_ntp_epoch(seconds: float) -> float:
298
    """Convert system epoch time to NTP epoch time.
299
    
300
    Parameters:
301
    - seconds: Seconds since Unix epoch
302
    
303
    Returns:
304
    Seconds since NTP epoch
305
    """
306

307
class Timestamp:
308
    """NTP timestamp representation."""
309
    seconds: int    # Integer seconds component
310
    fraction: int   # Fractional seconds component (32-bit)
311

312
# NTP constants
313
IMMEDIATELY: bytes  # Special NTP timestamp for immediate execution
314
```
315

316
### SLIP Protocol
317

318
Serial Line Internet Protocol implementation for reliable TCP OSC communication (OSC 1.1).
319

320
```python { .api }
321
def encode(msg: bytes) -> bytes:
322
    """Encode message bytes into SLIP packet format.
323
    
324
    Parameters:
325
    - msg: Message bytes to encode
326
    
327
    Returns:
328
    SLIP-encoded packet with proper framing and escaping
329
    """
330

331
def decode(packet: bytes) -> bytes:
332
    """Decode SLIP packet to retrieve original message.
333
    
334
    Parameters:
335
    - packet: SLIP-encoded packet bytes
336
    
337
    Returns:
338
    Original message bytes
339
    
340
    Raises:
341
    ProtocolError: If packet contains invalid SLIP sequences
342
    """
343

344
def is_valid(packet: bytes) -> bool:
345
    """Check if packet conforms to SLIP specification.
346
    
347
    Parameters:
348
    - packet: Packet bytes to validate
349
    
350
    Returns:
351
    True if packet is valid SLIP format
352
    """
353

354
# SLIP protocol constants
355
END: bytes = b"\xc0"        # Frame delimiter
356
ESC: bytes = b"\xdb"        # Escape character
357
ESC_END: bytes = b"\xdc"    # Escaped END
358
ESC_ESC: bytes = b"\xdd"    # Escaped ESC
359
END_END: bytes = b"\xc0\xc0"  # Double END sequence
360
```
361

362
## Usage Examples
363

364
### Manual Message Construction
365

366
```python
367
from pythonosc.parsing import osc_types
368

369
# Build message datagram manually
370
address_bytes = osc_types.write_string("/synth/freq")
371
type_tag_bytes = osc_types.write_string(",f")  # One float argument
372
freq_bytes = osc_types.write_float(440.0)
373

374
# Combine into complete message
375
message_dgram = address_bytes + type_tag_bytes + freq_bytes
376
print(f"Message size: {len(message_dgram)} bytes")
377
```
378

379
### Parsing Custom Data Types
380

381
```python
382
from pythonosc.parsing import osc_types
383

384
def parse_custom_message(dgram):
385
    """Parse a message with known structure."""
386
    index = 0
387
    
388
    # Parse address
389
    address, index = osc_types.get_string(dgram, index)
390
    print(f"Address: {address}")
391
    
392
    # Parse type tag
393
    type_tag, index = osc_types.get_string(dgram, index)
394
    print(f"Type tag: {type_tag}")
395
    
396
    # Parse arguments based on type tag
397
    args = []
398
    for arg_type in type_tag[1:]:  # Skip comma
399
        if arg_type == 'i':
400
            value, index = osc_types.get_int(dgram, index)
401
        elif arg_type == 'f':
402
            value, index = osc_types.get_float(dgram, index)
403
        elif arg_type == 's':
404
            value, index = osc_types.get_string(dgram, index)
405
        elif arg_type == 'b':
406
            value, index = osc_types.get_blob(dgram, index)
407
        elif arg_type == 'd':
408
            value, index = osc_types.get_double(dgram, index)
409
        elif arg_type == 'h':
410
            value, index = osc_types.get_int64(dgram, index)
411
        elif arg_type == 'm':
412
            value, index = osc_types.get_midi(dgram, index)
413
        elif arg_type == 'r':
414
            value, index = osc_types.get_rgba(dgram, index)
415
        else:
416
            print(f"Unknown type: {arg_type}")
417
            continue
418
        args.append(value)
419
    
420
    return address, args
421

422
# Test with sample datagram
423
sample_dgram = (osc_types.write_string("/test") + 
424
                osc_types.write_string(",ifs") +
425
                osc_types.write_int(42) +
426
                osc_types.write_float(3.14) +
427
                osc_types.write_string("hello"))
428

429
address, args = parse_custom_message(sample_dgram)
430
print(f"Parsed: {address} -> {args}")
431
```
432

433
### Precise Timing with NTP
434

435
```python
436
from pythonosc.parsing import ntp, osc_types
437
import time
438

439
# Current time as NTP timestamp
440
current_time = time.time()
441
ntp_timestamp = ntp.system_time_to_ntp(current_time)
442
print(f"NTP timestamp: {ntp_timestamp.hex()}")
443

444
# Schedule for 2 seconds in the future
445
future_time = current_time + 2.0
446
future_ntp = ntp.system_time_to_ntp(future_time)
447

448
# Create timetag for bundle
449
timetag_bytes = osc_types.write_date(future_time)
450
print(f"Future timetag: {timetag_bytes.hex()}")
451

452
# Parse timestamp components
453
timestamp_int = int.from_bytes(future_ntp, 'big')
454
parsed = ntp.parse_timestamp(timestamp_int)
455
print(f"Seconds: {parsed.seconds}, Fraction: {parsed.fraction}")
456
```
457

458
### SLIP Encoding for TCP
459

460
```python
461
from pythonosc import slip
462
from pythonosc.parsing import osc_types
463

464
# Create OSC message
465
message_data = (osc_types.write_string("/tcp/test") +
466
                osc_types.write_string(",s") +
467
                osc_types.write_string("TCP message"))
468

469
# Encode for TCP transmission (OSC 1.1)
470
slip_packet = slip.encode(message_data)
471
print(f"SLIP packet: {slip_packet.hex()}")
472

473
# Validate packet
474
is_valid = slip.is_valid(slip_packet)
475
print(f"Valid SLIP packet: {is_valid}")
476

477
# Decode received packet
478
try:
479
    decoded_message = slip.decode(slip_packet)
480
    print(f"Decoded message length: {len(decoded_message)}")
481
    print(f"Original matches decoded: {message_data == decoded_message}")
482
except slip.ProtocolError as e:
483
    print(f"SLIP decode error: {e}")
484
```
485

486
### Working with MIDI Data
487

488
```python
489
from pythonosc.parsing import osc_types
490

491
# Create MIDI note on message
492
midi_packet = (0, 0x90, 60, 127)  # Channel 1, Note On, Middle C, Velocity 127
493
midi_bytes = osc_types.write_midi(midi_packet)
494

495
# Create OSC message with MIDI data
496
message_dgram = (osc_types.write_string("/midi/note") +
497
                 osc_types.write_string(",m") +
498
                 midi_bytes)
499

500
# Parse MIDI data back
501
index = len(osc_types.write_string("/midi/note") + osc_types.write_string(",m"))
502
parsed_midi, _ = osc_types.get_midi(message_dgram, index)
503

504
print(f"Original MIDI: {midi_packet}")
505
print(f"Parsed MIDI: {parsed_midi}")
506
print(f"Port: {parsed_midi[0]}, Status: 0x{parsed_midi[1]:02x}, Data1: {parsed_midi[2]}, Data2: {parsed_midi[3]}")
507
```
508

509
### Color Data Handling
510

511
```python
512
from pythonosc.parsing import osc_types
513

514
# Create RGBA color (red with 50% alpha)
515
color = (255, 0, 0, 128)
516
color_bytes = osc_types.write_rgba(color)
517

518
# Create message with color
519
message_dgram = (osc_types.write_string("/light/color") +
520
                 osc_types.write_string(",r") +
521
                 color_bytes)
522

523
# Parse color back
524
index = len(osc_types.write_string("/light/color") + osc_types.write_string(",r"))
525
parsed_color, _ = osc_types.get_rgba(message_dgram, index)
526

527
print(f"Original color: RGBA{color}")
528
print(f"Parsed color: RGBA{parsed_color}")
529
```
530

531
### Binary Data (Blobs)
532

533
```python
534
from pythonosc.parsing import osc_types
535
import struct
536

537
# Create binary data (e.g., audio samples)
538
audio_samples = struct.pack('>10f', *[0.1 * i for i in range(10)])
539
blob_bytes = osc_types.write_blob(audio_samples)
540

541
# Create message with blob
542
message_dgram = (osc_types.write_string("/audio/samples") +
543
                 osc_types.write_string(",b") +
544
                 blob_bytes)
545

546
# Parse blob back
547
index = len(osc_types.write_string("/audio/samples") + osc_types.write_string(",b"))
548
parsed_blob, _ = osc_types.get_blob(message_dgram, index)
549

550
# Unpack audio samples
551
parsed_samples = struct.unpack('>10f', parsed_blob)
552
print(f"Blob size: {len(parsed_blob)} bytes")
553
print(f"Audio samples: {parsed_samples}")
554
```
555

556
### Performance Optimization
557

558
```python
559
from pythonosc.parsing import osc_types
560
import itertools
561

562
# Pre-compute common type tags for performance
563
common_types = {
564
    'f': osc_types.write_string(",f"),
565
    'i': osc_types.write_string(",i"),
566
    's': osc_types.write_string(",s"),
567
    'ff': osc_types.write_string(",ff"),
568
    'ifs': osc_types.write_string(",ifs"),
569
}
570

571
def fast_build_message(address, args):
572
    """Optimized message building for known patterns."""
573
    address_bytes = osc_types.write_string(address)
574
    
575
    # Determine type pattern
576
    type_pattern = ''.join('f' if isinstance(arg, float) else
577
                          'i' if isinstance(arg, int) else
578
                          's' if isinstance(arg, str) else 'x' 
579
                          for arg in args)
580
    
581
    # Use pre-computed type tag if available
582
    if type_pattern in common_types:
583
        type_tag_bytes = common_types[type_pattern]
584
    else:
585
        type_tag_bytes = osc_types.write_string(',' + type_pattern)
586
    
587
    # Build argument bytes
588
    arg_bytes = b''
589
    for arg in args:
590
        if isinstance(arg, float):
591
            arg_bytes += osc_types.write_float(arg)
592
        elif isinstance(arg, int):
593
            arg_bytes += osc_types.write_int(arg)
594
        elif isinstance(arg, str):
595
            arg_bytes += osc_types.write_string(arg)
596
    
597
    return address_bytes + type_tag_bytes + arg_bytes
598

599
# Test optimized building
600
fast_msg = fast_build_message("/fast/test", [440.0, 127, "hello"])
601
print(f"Fast message size: {len(fast_msg)} bytes")
602
```
603

604
## Types and Exceptions
605

606
```python { .api }
607
from typing import Tuple, Union
608
from datetime import datetime
609

610
MidiPacket = Tuple[int, int, int, int]  # (port_id, status_byte, data1, data2)
611

612
class ParseError(Exception):
613
    """Raised when OSC data parsing fails."""
614

615
class BuildError(Exception):
616
    """Raised when OSC data building fails."""
617

618
class ProtocolError(ValueError):
619
    """Raised when SLIP protocol error occurs."""
620

621
class NtpError(Exception):
622
    """Raised when NTP timestamp conversion fails."""
623
```