uwacomm

Underwater Communications Codec – DCCL-inspired compact binary encoding for Python

Inspiration & Credits

uwacomm is inspired by DCCL (Dynamic Compact Control Language) from GobySoft. DCCL is a mature, battle-tested C++ library used extensively in underwater robotics and autonomous vehicle communications.

Key differences:

• uwacomm: Python-native, Pydantic-based, designed for ease of use in Python/ROS2 ecosystems
• DCCL: C++ implementation with Protobuf integration, part of the larger Goby underwater autonomy framework

We are NOT affiliated with or claiming to replace DCCL. If you need:

• Production-grade C++ implementation
• Full Goby framework integration
• Official DCCL Python bindings

→ Use the official DCCL project: https://github.com/GobySoft/dccl

uwacomm implements similar compact encoding concepts but with a pure-Python, Pydantic-first approach for Python developers who want DCCL-inspired functionality without C++ dependencies.

Why uwacomm exists

While DCCL is excellent, Python developers often want:

• Native Pydantic integration for modern Python codebases
• Simpler installation (pip install, no compilation)
• Pythonic API design
• Easy integration with Python-based robotics stacks

Standing on the shoulders of giants: This project wouldn't exist without the pioneering work of the DCCL team at GobySoft. Thank you!

Overview

uwacomm is a Python library for schema-based compact binary encoding designed for bandwidth-constrained communications, particularly underwater acoustic modems. Inspired by DCCL from GobySoft, uwacomm uses Pydantic models for message definition and provides DCCL-style bounded field optimization to minimize transmitted bytes.

Key Features

• 🎯 Schema-first design: Define messages using Pydantic's intuitive field syntax
• 📦 Compact encoding: Bounded fields use only the minimum required bits
• 🔀 Multi-mode encoding: Three modes for different use cases (point-to-point, self-describing, multi-vehicle routing)
• 📐 Float support: DCCL-style bounded floats with precision control (50-85% bandwidth savings vs IEEE 754)
• 🚀 Multi-vehicle routing: Built-in source/dest addressing, priority levels, ACK support
• 🏗️ Nested messages: Compose BaseMessage fields inline with zero overhead (v0.4.0)
• 📝 Variable-length fields: VarBytes, VarStr, VarList — pay only for bytes actually sent (v0.4.0)
• 🔒 Type-safe: Full type hints and mypy strict mode compliance
• ✅ Deterministic: Platform-independent, reproducible encodings
• 🛡️ Error detection: Built-in CRC-16/CRC-32 and framing utilities
• 🔄 Protobuf interop: Generate .proto schemas from Pydantic models
• 📊 Size analysis: Calculate encoded sizes before transmission
• 🧪 Well-tested: 224+ passing tests, 81% code coverage
• ⚡ Benchmarked: pytest-benchmark suite for codec, float, routing, and fragmentation throughput

Installation

pip install uwacomm

Optional dependencies

# For Protobuf schema generation
pip install uwacomm[protobuf]

# For development (tests, docs, linting)
pip install uwacomm[dev]

# All optional dependencies
pip install uwacomm[all]

Quick Start

1. Define a Message

from uwacomm import BaseMessage, BoundedInt, BoundedFloat

class VehicleStatus(BaseMessage):
    """Underwater vehicle status with efficient float encoding."""

    # Position (GPS coordinates with 6 decimal places = ~11cm accuracy)
    position_lat: float = BoundedFloat(min=-90.0, max=90.0, precision=6)   # 28 bits
    position_lon: float = BoundedFloat(min=-180.0, max=180.0, precision=6)  # 29 bits

    # Depth in meters (centimeter precision)
    depth_m: float = BoundedFloat(min=0.0, max=5000.0, precision=2)  # 20 bits

    # Vehicle state
    heading_deg: float = BoundedFloat(min=0.0, max=360.0, precision=1)  # 12 bits
    battery_pct: int = BoundedInt(ge=0, le=100)  # 7 bits

    uwacomm_id: int = 10
    uwacomm_max_bytes: int = 64

2. Mode 1: Point-to-Point (Maximum Compression)

from uwacomm import encode, decode

# Create a message
msg = VehicleStatus(
    position_lat=42.358894,
    position_lon=-71.063611,
    depth_m=125.75,
    heading_deg=45.5,
    battery_pct=78
)

# Encode (Mode 1 - no ID, minimal overhead)
data = encode(msg)  # ~14 bytes (8.2% smaller than DCCL!)

# Decode
decoded = decode(VehicleStatus, data)
assert decoded.depth_m == msg.depth_m

3. Mode 2: Self-Describing Messages (Logging/Replay)

from uwacomm import encode, decode, register_message, decode_by_id

# Encode with message ID
data = encode(msg, include_id=True)  # +1 byte for ID

# Register for auto-decode
register_message(VehicleStatus)

# Auto-decode without knowing message type!
decoded = decode_by_id(data)
print(f"Auto-decoded: {type(decoded).__name__}")

4. Mode 3: Multi-Vehicle Routing (Swarm Robotics)

from uwacomm import encode_with_routing, decode_with_routing

# Vehicle 3 sends high-priority status to topside (ID 0)
data = encode_with_routing(
    msg,
    source_id=3,
    dest_id=0,
    priority=2,          # 0=low, 3=high
    ack_requested=True
)

# Topside receives and processes
routing, decoded = decode_with_routing(VehicleStatus, data)
print(f"From vehicle {routing.source_id}, priority {routing.priority}")

if routing.ack_requested:
    # Send acknowledgment
    pass

5. Broadcast Messages (Swarm Coordination)

# Lead vehicle broadcasts formation update to all vehicles
data = encode_with_routing(
    formation_update,
    source_id=1,
    dest_id=255,  # 255 = broadcast to all
    priority=3    # Urgent
)

# All vehicles receive and process
routing, update = decode_with_routing(FormationUpdate, data)
if routing.dest_id == 255:  # Broadcast
    print(f"Formation update from vehicle {routing.source_id}")

Hardware-in-the-Loop (HITL) Simulation

Test acoustic modem communication without physical hardware using the mock modem driver. Perfect for CI/CD, development, and debugging before deploying to real underwater systems.

Mock Modem Driver

The MockModemDriver simulates underwater acoustic communication with configurable channel characteristics:

from uwacomm import encode, decode, BaseMessage, BoundedInt
from uwacomm.modem import MockModemDriver, MockModemConfig
from typing import ClassVar

class Heartbeat(BaseMessage):
    """Vehicle heartbeat message."""
    depth: int = BoundedInt(ge=0, le=1000)
    battery: int = BoundedInt(ge=0, le=100)
    uwacomm_id: ClassVar[int | None] = 10

# Configure realistic underwater channel
config = MockModemConfig(
    transmission_delay=1.5,      # 1.5 second round-trip (1 km range)
    packet_loss_probability=0.1,  # 10% packet loss
    bit_error_rate=0.0005,        # 0.05% BER (acoustic noise)
    max_frame_size=64,            # 64 byte max (typical modem limit)
    data_rate=80,                 # 80 bps (long range, low frequency)
)

# Create and connect mock modem
modem = MockModemDriver(config)
modem.connect("/dev/null", 19200)  # Fake port (simulation mode)

# Register RX callback
def on_receive(data: bytes, src_id: int):
    msg = decode(Heartbeat, data)
    print(f"Received from {src_id}: depth={msg.depth}m, battery={msg.battery}%")

modem.attach_rx_callback(on_receive)

# Send frame (will echo back after transmission_delay seconds)
heartbeat = Heartbeat(depth=250, battery=87)
modem.send_frame(encode(heartbeat), dest_id=0)

# Wait for loopback
import time
time.sleep(2.0)
modem.disconnect()

Channel Simulation Features

The mock modem simulates realistic acoustic channel conditions:

• Transmission delay: Acoustic propagation time (speed of sound in seawater ≈ 1500 m/s)

  • Short range (< 1 km): 0.5 - 2.0 seconds
  • Medium range (1-5 km): 2.0 - 7.0 seconds
  • Long range (> 5 km): 7.0 - 15.0 seconds

• Packet loss: Unreliable underwater channel

  • Good conditions: 1-5% loss
  • Moderate conditions: 5-15% loss
  • Poor conditions: 15-30% loss

• Bit errors: Acoustic noise and multipath

  • Good SNR: 0.01-0.1% BER
  • Moderate SNR: 0.1-1% BER
  • Poor SNR: 1-10% BER

• Loopback testing: Sent frames echo back to RX callbacks after simulated delay

Vendor-Agnostic Abstraction

The ModemDriver interface is completely vendor-agnostic:

from uwacomm.modem import ModemDriver

# Abstract interface works with ANY acoustic modem:
# - MockModemDriver (simulation)
# - WhoiModemDriver (WHOI MicroModem 2) - future
# - EvoLogicsModemDriver (EvoLogics S2C) - future
# - SonarbyneModemDriver (Sonardyne) - future
# - Your custom driver (subclass ModemDriver)

Key Benefits:

• ✅ Test without physical hardware (CI/CD, development)
• ✅ Switch modem vendors without changing application code
• ✅ Reproducible test scenarios (controlled channel conditions)
• ✅ Third-party driver support (extensible design)

See examples/hitl_simulation.py for a complete demo.

Message Fragmentation

Split large messages across multiple acoustic modem frames. Acoustic modems typically have strict frame size limits (32-64 bytes), requiring larger messages to be fragmented for transmission.

Automatic Fragmentation

from uwacomm import encode, decode, BaseMessage, FixedBytes
from uwacomm.fragmentation import fragment_message, reassemble_fragments
from typing import ClassVar

class LargeMessage(BaseMessage):
    """Large telemetry message (150+ bytes)."""
    sensor_data: bytes = FixedBytes(length=150)
    uwacomm_id: ClassVar[int | None] = 20

# Encode message
msg = LargeMessage(sensor_data=b'x' * 150)
encoded = encode(msg)  # ~153 bytes

# Fragment for 64-byte modem frames
fragments = fragment_message(encoded, max_fragment_size=64)
# Returns: 3 fragments (64 + 64 + 25 bytes)

# Send fragments over acoustic modem...
for frag in fragments:
    modem.send_frame(frag, dest_id=0)

# Receiver collects fragments and reassembles
reassembled = reassemble_fragments(fragments)
decoded = decode(LargeMessage, reassembled)  # ✓ Perfect reconstruction

Fragment Header Format

Each fragment includes a 4-byte header for reliable reassembly:

┌─────────────┬─────────┬─────────┬─────────────────┐
│ Fragment ID │ Seq Num │  Total  │  Data Chunk     │
│  16 bits    │  8 bits │ 8 bits  │  N bytes        │
└─────────────┴─────────┴─────────┴─────────────────┘

• Fragment ID (16 bits): Unique identifier for this message (0-65535)
• Sequence Number (8 bits): Fragment index (0-255)
• Total (8 bits): Total number of fragments (1-255)
• Data Chunk (N bytes): Actual payload data

Robust Error Detection

• Out-of-order delivery: Fragments can arrive in any order
• Missing fragments: Detected with clear error messages
• Duplicate fragments: Detected and rejected
• Concurrent messages: Fragment IDs distinguish multiple simultaneous messages

from uwacomm.exceptions import FragmentationError

# Simulate packet loss
fragments = fragment_message(data, max_fragment_size=64)
del fragments[1]  # Lost fragment 1

try:
    reassemble_fragments(fragments)
except FragmentationError as e:
    print(f"Missing fragments: {e}")
    # Error: Missing fragments: [1]. Expected 4 fragments, got 3

Integration with Mock Modem

from uwacomm.modem import MockModemDriver, MockModemConfig
from uwacomm.fragmentation import fragment_message, reassemble_fragments

# Fragment large message
encoded = encode(large_message)
fragments = fragment_message(encoded, max_fragment_size=64)

# Send over mock modem
modem = MockModemDriver()
modem.connect("/dev/null", 19200)

for frag in fragments:
    modem.send_frame(frag, dest_id=0)

# Receiver reassembles
received_fragments = []
def on_receive(data: bytes, src_id: int):
    received_fragments.append(data)

modem.attach_rx_callback(on_receive)
# ... wait for all fragments ...

reassembled = reassemble_fragments(received_fragments)
decoded = decode(MessageClass, reassembled)

Memory-Efficient Iteration

For very large messages, use iter_fragments() to avoid storing all fragments in memory:

from uwacomm.fragmentation import iter_fragments

large_data = b'x' * 10000  # 10 KB message

# Send fragments without creating full list
for fragment in iter_fragments(large_data, max_fragment_size=64):
    modem.send_frame(fragment, dest_id=0)

See examples/fragmentation_demo.py for complete examples including out-of-order delivery, missing fragment detection, and concurrent fragmented messages.

Nested Messages (v0.4.0)

Compose messages by nesting one BaseMessage inside another. Fields are packed inline — no length prefix, no ID, zero overhead.

from uwacomm import BaseMessage, BoundedInt, encode, decode, encoded_bits
from uwacomm.models.fields import BoundedFloat
from typing import ClassVar

class GPSPosition(BaseMessage):
    lat: float = BoundedFloat(min=-90.0, max=90.0, precision=6)   # 28 bits
    lon: float = BoundedFloat(min=-180.0, max=180.0, precision=6)  # 29 bits

class VehicleStatus(BaseMessage):
    vehicle_id: int = BoundedInt(ge=0, le=255)   # 8 bits
    position: GPSPosition                         # 57 bits inline
    depth_cm: int = BoundedInt(ge=0, le=50000)   # 16 bits
    battery: int = BoundedInt(ge=0, le=100)       # 7 bits
    uwacomm_id: ClassVar[int | None] = 20

print(encoded_bits(VehicleStatus))  # 88 bits = 11 bytes

msg = VehicleStatus(
    vehicle_id=7,
    position=GPSPosition(lat=44.648766, lon=-63.575237),
    depth_cm=1250,
    battery=83,
)
decoded = decode(VehicleStatus, encode(msg))
print(decoded.position.lat)   # 44.648766

Two-level nesting works too — see examples/nested_messages.py.

Variable-Length Fields (v0.4.0)

Three new field helpers let on-wire size scale with actual content rather than the schema maximum. Each writes a compact length prefix followed by the actual data.

from uwacomm import BaseMessage, BoundedInt, encode, decode
from uwacomm.models.fields import VarBytes, VarStr, VarList

class MissionUpdate(BaseMessage):
    vehicle_id: int = BoundedInt(ge=0, le=255)
    callsign: str = VarStr(max_length=8)              # 4-bit prefix + ≤64 bits
    waypoint_data: bytes = VarBytes(max_length=32)    # 6-bit prefix + ≤256 bits
    depths: list[int] = VarList(max_length=4, item_ge=0, item_le=5000)

small = MissionUpdate(vehicle_id=1, callsign="A", waypoint_data=b"\x00", depths=[0])
large = MissionUpdate(vehicle_id=1, callsign="ORCA-001", waypoint_data=bytes(32), depths=[0,1000,2500,5000])

print(len(encode(small)))  # 7 bytes
print(len(encode(large)))  # 50 bytes

See examples/varlen_fields.py for VarList[bool], VarList[float], and mixed messages.

CLI Tools

Message Analysis

Analyze message schemas and see field-by-field bit usage (inspired by dccl --analyze):

uwacomm --analyze message.py

Example output:

||||||| uwacomm: Underwater Communications Codec |||||||
2 messages loaded.
Field sizes are in bits unless otherwise noted.

=================== 10: StatusReport ===================
Actual maximum size of message: 4 bytes / 32 bits
        uwacomm.id head........................8 (if present)
        body..................................29
        padding to full byte...................3
Allowed maximum size of message: 32 bytes / 256 bits

--------------------------- Header ---------------------------
uwacomm.id............................................8 bits

---------------------------- Body ----------------------------
StatusReport..........................................29 bits
        1. vehicle_id...........................8 bits [0-255]
        2. depth_cm..........................14 bits [0-10000]
        3. battery_pct..........................7 bits [0-100]

======================== Summary ========================
Compression vs JSON: 22.2x smaller
Estimated transmission time @ 80 bps: 0.4 seconds

CLI commands:

uwacomm --analyze FILE    # Analyze message schema
uwacomm --version         # Show version
uwacomm --help            # Show help

Why uwacomm?

Bandwidth Matters Underwater

Underwater acoustic modems typically operate at 80-5000 bits per second—orders of magnitude slower than terrestrial networks. For comparison:

With limited transmission windows and high per-byte costs, every bit counts.

Multi-Mode Encoding

Choose the mode that fits your mission:

Efficient Float Encoding

Traditional IEEE 754 floats waste bandwidth underwater:

Example:

# Depth: -5.00 to 100.00 m (centimeter precision)
depth: float = BoundedFloat(min=-5.0, max=100.0, precision=2)
# 14 bits vs 64 bits for double → 78% bandwidth savings!

DCCL-Style Bounded Field Optimization

Unlike generic binary formats, uwacomm uses field constraints to minimize encoding size:

# Standard int: 32 bits
value: int

# Bounded int (0-255): only 8 bits!
value: int = Field(ge=0, le=255)

# Bounded int (0-15): only 4 bits!
value: int = Field(ge=0, le=15)

Pythonic and Type-Safe

Built on Pydantic v2, uwacomm provides:

• Automatic validation
• IDE autocomplete
• Type checking with mypy
• Clear error messages

Documentation

• User Guide: In-depth tutorials and concepts
• API Reference: Complete API documentation
• Examples: Runnable example scripts

Examples

See the examples/ directory for complete, runnable examples:

NEW in v0.4.0:

• nested_messages.py – Inline nested BaseMessage fields (GPS position inside vehicle status, two-level nesting)
• varlen_fields.py – VarBytes, VarStr, VarList — on-wire size scales with actual content

NEW in v0.3.0:

• hitl_simulation.py - Hardware-in-the-Loop simulation with MockModemDriver
• fragmentation_demo.py - Message fragmentation for size-limited acoustic modems

NEW in v0.2.0:

• generic_uw_messages.py - Generic underwater vehicle message definitions
• demo_multi_mode.py - All three encoding modes + broadcast patterns
• bandwidth_comparison.py - uwacomm vs JSON vs DCCL comparison

Core Examples:

• basic_usage.py - Message definition, encoding, decoding
• framing_example.py - Message framing with CRC
• protobuf_schema.py - Generate .proto schemas

Supported Features

Field Types

• ✅ Booleans (1 bit)
• ✅ Bounded unsigned integers (minimal bits)
• ✅ Bounded signed integers (minimal bits)
• ✅ Enums (minimal bits for value count)
• ✅ Fixed-length bytes
• ✅ Fixed-length strings (UTF-8)
• ✅ NEW: Floats with precision (DCCL-style bounded floats) - v0.2.0
• ✅ NEW: Nested BaseMessage fields (inline, zero overhead) – v0.4.0
• ✅ NEW: Variable-length bytes, strings, and lists (VarBytes, VarStr, VarList) – v0.4.0

Encoding Modes

• ✅ Mode 1: Point-to-point (8.2% smaller than DCCL)
• ✅ Mode 2: Self-describing messages (ties DCCL, enables auto-decode)
• ✅ Mode 3: Multi-vehicle routing (source/dest/priority/ack) - v0.2.0

Multi-Vehicle Features (Mode 3)

• ✅ Source/destination addressing (0-255 vehicle IDs)
• ✅ Priority levels (0=low, 3=high)
• ✅ ACK request flag
• ✅ Broadcast support (dest_id=255)
• ✅ MESSAGE_REGISTRY for auto-decode

Utilities

• ✅ CRC-16 and CRC-32 checksums
• ✅ Length-prefixed framing
• ✅ Message ID multiplexing
• ✅ Encoded size calculation
• ✅ Protobuf schema generation
• ✅ NEW: Message fragmentation/reassembly - v0.3.0
• ✅ NEW: Performance benchmarking suite (pytest-benchmark) - v0.3.0

Hardware-in-the-Loop (HITL) Simulation

• ✅ NEW: MockModemDriver for testing without hardware - v0.3.0
• ✅ Configurable acoustic channel simulation (delay, loss, bit errors)
• ✅ Loopback testing (echo sent frames back)
• ✅ Vendor-agnostic ModemDriver abstraction
• ✅ Multiple RX callback support
• ⏸️ Real modem drivers (WHOI, EvoLogics, Sonardyne) - planned for v0.4.0+

Design Principles

1. Explicit over implicit: All constraints must be declared
2. Deterministic: Same message → same bytes, always
3. Security-minded: Bounds checking, no unbounded recursion
4. Fail-fast: Clear exceptions, not silent corruption

Comparison to Alternatives

Summary:

• vs DCCL: 8.2% smaller (Mode 1), adds multi-mode encoding and routing
• vs Protobuf: 50-60% smaller, Python-native
• vs JSON: 88-90% smaller, type-safe

Development

Setup

git clone https://github.com/patel999jay/uwacomm.git
cd uwacomm
pip install -e ".[dev]"

Run Tests

pytest

Run Benchmarks

# Run all benchmarks (sorted by mean time)
pytest tests/benchmarks/ --benchmark-only --benchmark-sort=mean

# Save results to JSON for tracking over time
pytest tests/benchmarks/ --benchmark-only --benchmark-json=benchmark_results.json

Benchmarks cover encode/decode throughput, float precision overhead, routing header cost, and fragmentation/reassembly speed. They are excluded from the default pytest run.

Linting

black src tests examples
ruff check src tests examples
mypy src

Contributing

Contributions are welcome! Please see CONTRIBUTING.md for guidelines.

License

MIT License - see LICENSE for details.

Acknowledgments

• Inspired by DCCL (GobySoft)
• Built on Pydantic
• Influenced by arlpy usability principles
• Extends the author's prior work: ProtocolDataUnits

Citation

If you use uwacomm in your research, please cite:

@software{uwacomm2026,
  author = {Patel, Jay},
  title = {uwacomm: Python DCCL-inspired compact binary encoding for underwater communications},
  year = {2026},
  url = {https://github.com/patel999jay/uwacomm}
}

Related Projects

Predecessor: ProtocolDataUnits

• ProtocolDataUnits: https://github.com/patel999jay/ProtocolDataUnits
• PyPI: https://pypi.org/project/ProtocolDataUnits/
• Blog: https://patel999jay.github.io/post/protocoldataunits-python-package/

uwacomm is the evolution of ProtocolDataUnits, adding:

• Pydantic v2 integration
• DCCL-inspired bounded field optimization
• Better type safety and modern Python practices
• CLI analysis tools

For new projects, use uwacomm. ProtocolDataUnits remains available for existing users.

Official DCCL Project

• DCCL (C++): https://github.com/GobySoft/dccl
• Documentation: https://libdccl.org/
• Goby Framework: https://github.com/GobySoft/goby3

Python Underwater Acoustics

• arlpy: https://github.com/org-arl/arlpy - Underwater acoustics toolbox
• UnetStack: https://unetstack.net/ - Underwater network simulator

Other Python Binary Encodings

• Protobuf: https://protobuf.dev/ - Google's binary format
• MessagePack: https://msgpack.org/ - Efficient binary serialization
• construct: https://construct.readthedocs.io/ - Binary parsing library

uwacomm complements these tools by providing DCCL-inspired compact encoding specifically optimized for bandwidth-constrained underwater communications.