MARTe-Integrated-Components/Docs/Protocol.md

# UDPStreamer Wire Protocol

This document specifies the binary protocol used between UDPStreamer (server) and any
compatible client (the included Go WebUI, a Python script, etc.).

All multi-byte integers are **little-endian**.

---

## Packet Header (17 bytes, packed)

Every datagram begins with a 17-byte header:

```
Offset  Size  Type    Field
──────  ────  ──────  ────────────────────────────────────────────────────
0       4     uint32  magic          = 0x53504455  ('UDPS' LE)
4       1     uint8   type           see Packet Types below
5       4     uint32  counter        per-update sequence number
                                     (same across all fragments of one update)
9       2     uint16  fragmentIdx    0-based index of this fragment
11      2     uint16  totalFragments number of fragments for this update
13      4     uint32  payloadBytes   bytes of payload following this header
```

**Total header size:** 17 bytes  
**Magic:** `0x55 0x44 0x50 0x53` (`UDPS`)

---

## Packet Types

| Value | Direction | Name | Description |
|-------|-----------|------|-------------|
| 0 | Server → Client | DATA | Signal data (may be fragmented) |
| 1 | Server → Client | CONFIG | Signal metadata sent on connect |
| 2 | Client → Server | ACK | Acknowledge a data counter (reserved) |
| 3 | Client → Server | CONNECT | Request a session |
| 4 | Client → Server | DISCONNECT | End the session |

---

## Session Flow

```
Client                                Server
──────                                ──────
                CONNECT (type=3) →
              ← CONFIG  (type=1)
              ← DATA    (type=0)  ┐
              ← DATA    (type=0)  │  repeated every RT cycle
              ← DATA    (type=0)  ┘
                DISCONNECT (type=4) →
```

1. Client sends a 17-byte CONNECT packet (`payloadBytes = 0`).
2. Server responds immediately with one or more CONFIG fragments describing all signals.
3. Server sends DATA fragments on every `Synchronise()` call while a client is connected.
4. Client sends DISCONNECT to terminate cleanly. A new CONNECT replaces an existing session.

---

## CONFIG Payload

The CONFIG payload is sent as one or more fragmented packets (`type = 1`).
After reassembly the layout is:

```
Offset  Size  Type     Field
──────  ────  ───────  ────────────────────────────────────
0       4     uint32   numSignals
── for each signal (136 bytes) ──────────────────────────────
0       64    char[64] name           null-terminated
64      1     uint8    typeCode       see Type Codes
65      1     uint8    quantType      see Quantization Types
66      1     uint8    numDimensions  0 = scalar, 1 = 1-D array, 2 = matrix
67      4     uint32   numRows        0 or 1 for scalar/1-D
71      4     uint32   numCols        number of elements along fastest axis
75      8     float64  rangeMin
83      8     float64  rangeMax
91      1     uint8    timeMode       see Time Modes
92      8     float64  samplingRate   Hz (0 if PacketTime)
100     4     uint32   timeSignalIdx  index of the time-reference signal;
                                      0xFFFFFFFF = PacketTime (no reference)
104     32    char[32] unit           null-terminated physical unit string
── (total per signal: 136 bytes) ────────────────────────────
```

### Type Codes

| Code | C type | Bytes/element |
|------|--------|---------------|
| 0 | uint8 | 1 |
| 1 | int8 | 1 |
| 2 | uint16 | 2 |
| 3 | int16 | 2 |
| 4 | uint32 | 4 |
| 5 | int32 | 4 |
| 6 | uint64 | 8 |
| 7 | int64 | 8 |
| 8 | float32 | 4 |
| 9 | float64 | 8 |

### Quantization Type Codes (wire side)

| Code | Wire type | Description |
|------|-----------|-------------|
| 0 | — | No quantization; raw type as above |
| 1 | uint8 | Linear map `[rangeMin, rangeMax]` → `[0, 255]` |
| 2 | int8 | Linear map `[rangeMin, rangeMax]` → `[-127, 127]` |
| 3 | uint16 | Linear map `[rangeMin, rangeMax]` → `[0, 65535]` |
| 4 | int16 | Linear map `[rangeMin, rangeMax]` → `[-32767, 32767]` |

### Time Mode Codes

| Code | Name | Meaning |
|------|------|---------|
| 0 | PacketTime | HRT timestamp at `Synchronise()` — see DATA payload |
| 1 | FullArray | `timeSignalIdx` signal has same `numElements`; element `[k]` time = `timeSignal[k]` |
| 2 | FirstSample | `timeSignalIdx` is scalar; `t[k] = t[0] + k / samplingRate` |
| 3 | LastSample | `timeSignalIdx` is scalar; `t[k] = t[N-1] - (N-1-k) / samplingRate` |

---

## DATA Payload

After reassembly, the DATA payload layout is:

```
Offset  Size  Type    Field
──────  ────  ──────  ────────────────────────────────────────────────────
0       8     uint64  hrtTimestamp   hardware reference timer count at Synchronise()
── for each signal (in config order) ────────────────────────────────────
varies  N×sz  —       signal data    N = numRows×numCols, sz = element size
                                     (wire size if quantized, raw size otherwise)
```

Signal data for quantized signals uses the wire element size (see Quantization Type Codes),
not the original MARTe2 type size.

### Dequantization

To recover physical values from quantized integers:

```
// uint16 → float
span      = rangeMax - rangeMin
physical  = rangeMin + (wire_uint16 / 65535.0) × span

// int16 → float
physical  = rangeMin + ((wire_int16 + 32767) / 65534.0) × span
```

---

## Fragmentation

When a payload exceeds `MaxPayloadSize` bytes, it is split into fragments:

```
chunkSize   = MaxPayloadSize - 17   // usable bytes per datagram
numFragments = ceil(payloadSize / chunkSize)
```

Fragment `i` carries bytes `[i × chunkSize .. min((i+1) × chunkSize, payloadSize))`.
All fragments share the same `counter`; `fragmentIdx` and `totalFragments` allow
the client to reassemble them in any order.

**Example:** `MaxPayloadSize = 1400`, payload = 8016 B  
→ `chunkSize = 1383`, `numFragments = ceil(8016/1383) = 6`

---

## Ordering DATA (required of every receiver)

DATA carries its own `counter` sequence, incremented once per sent packet
(CONFIG is numbered independently). Reassembly completes in arrival order, not
counter order, so a packet reordered or duplicated on the wire surfaces after a
newer one has already been consumed. Its values are well-formed but carry an
older time base: accepting it writes them over samples the consumer already
holds and leaves the span they should have filled empty — a collision on one
side and a hole on the other.

A receiver must therefore drop any DATA packet that does not advance the
counter, and must order it by the *signed* difference:

```c
int32_t delta = (int32_t)(counter - lastCounter);   /* survives the uint32 wrap */
if (delta <= 0) { /* stale or duplicate: drop */ }
lost = (uint32_t)delta - 1u;                        /* packets missing before this one */
```

Comparing the values directly would call the first packet after the wrap stale
and reject the stream from then on.

`lost` matters beyond diagnostics. A consumer that spaces batched samples from
the elapsed time since the previous packet must divide that gap by `lost + 1`
batches; dividing by one batch reports a period too long by exactly that factor
and walks the samples past their own end into the next packet's range. Reset
the sequence on (re)connect: the producer's counter restarts independently.

Implemented in `UDPSClient::AcceptDataCounter` (C++),
`udpsprotocol.SequenceGate` (Go) and `decode_data` (C).

---

## Minimal Python Client Example

```python
import socket, struct, time

MAGIC    = 0x53504455
HDR_FMT  = '<IBHHI'     # magic, type, counter, fragIdx, totalFrags, payloadBytes
HDR_SIZE = 17

def build_connect():
    return struct.pack(HDR_FMT, MAGIC, 3, 0, 0, 1, 0)

def parse_header(data):
    return struct.unpack_from(HDR_FMT, data)

sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.bind(('', 44900))
sock.sendto(build_connect(), ('127.0.0.1', 44500))
sock.settimeout(5.0)

fragments = {}
while True:
    data, _ = sock.recvfrom(65536)
    magic, ptype, counter, frag_idx, total_frags, payload_bytes = parse_header(data)
    payload = data[HDR_SIZE:]

    if ptype == 1:   # CONFIG
        print(f"CONFIG fragment {frag_idx+1}/{total_frags}")
    elif ptype == 0: # DATA
        fragments.setdefault(counter, {})[frag_idx] = payload
        if len(fragments[counter]) == total_frags:
            full = b''.join(fragments.pop(counter)[i] for i in range(total_frags))
            hrt = struct.unpack_from('<Q', full)[0]
            print(f"DATA counter={counter} hrt={hrt} payload={len(full)}B")
```