Collector: UDP and TCP packets

UDP and TCP binary packets

Simulators, home-made data acquisition boxes and embedded devices often just send their readings as a fixed binary structure over UDP (broadcast) or a TCP socket. This source decodes those packets with a small layout file that says which bytes are which channel, or reads JSON and CSV text lines. It uses only Python's standard library: no extra to install. Commercial sims that already have a Kymoscope plugin (rFactor, Le Mans Ultimate, Assetto Corsa) do not need it.

Configuration

sources:
  - name: Bench DAQ
    type: udp
    bind: 0.0.0.0:5005               # inbound port: ask the customer first. Default 127.0.0.1:5005
    allow_inbound: true              # required for any bind that is not 127.0.0.1 / localhost / ::1
    source_filter: [192.168.1.50]    # only this sender (addresses or networks)
    layout: udp_daq_layout.yaml

  - name: Embedded controller (TCP)
    type: udp
    transport: tcp
    connect: 192.168.1.60:7000       # or  bind: 0.0.0.0:7000 + allow_inbound: true  to listen for the device
    framing: length                  # 2-byte length prefix, then the packet
    length_endian: big
    layout: controller.csv

  - name: Gateway JSON lines
    type: udp
    transport: tcp
    connect: 10.0.0.9:9000
    format: json                     # one JSON object per line
    values: {Temperature: temp, Pressure: p}
    time_path: ts
    time_format: epoch_s

The layout

A layout lists the fields of a packet in order. A field has a name, a type (u8 i8 u16 i16 u32 i32 u64 i64 f16 f32 f64 bool) and optionally an offset in bytes (default: right after the previous field), a count (an array: channels name[0], name[1]...), endian, scale and add (value = raw x scale + add), unit, and bit (an integer type only: that bit becomes a true / false channel). A packet may be longer than its fields (the rest is ignored); a shorter one is counted in decode_errors. A NaN or infinite float becomes a gap.

# A DAQ broadcasting <u32 seq><f64 t><f32 x 8>, little endian, 44 bytes per packet
endian: little
fields:
  - {name: seq, type: u32}                      # no offset: follows the previous field
  - {name: t, type: f64}
  - {name: x, type: f32, count: 8, unit: V}     # channels x[0] .. x[7]
seq: seq                                         # counts lost packets (seq_gaps in health)
timestamp: {field: t, unit: s, mode: epoch}      # the device clock; omit to use the arrival time

seq names an integer field that the device increments per packet; missing numbers are added to seq_gaps (a repeat, reorder or device restart is not a gap; the counter may wrap). timestamp names the time field: unit s, ms, us or ns; mode: epoch means Unix time, relative means a counter from the device start, anchored to the arrival time of the first packet (and again if the device restarts its counter, or if its clock drifts more than max_clock_drift_s from ours, so a slow or fast device clock does not walk away from real time; each re-anchor is counted in clock_resyncs and can step the timestamps by up to that amount). Without it, values are stamped with their arrival time. The seq and timestamp fields are not channels.

Several packet types

When the device sends different packets on one port, name the discriminator with packet_id and give each packet its fields under packets. Sequential offsets start after the id. An id with no packet is counted in decode_errors. seq and timestamp are set per packet.

endian: big
packet_id: {type: u8, offset: 0}       # the byte that says which packet this is
packets:
  1:                                    # the id; "0x02" works too
    fields:
      - {name: temp, type: i16, scale: 0.1, add: 5, unit: degC}   # value = raw x scale + add
      - {name: flags, type: u16, offset: 3, bit: 2}                # bit 2 of the u16 -> true / false
      - {name: alive, type: bool}
  2:
    prefix: "b."                        # channels become b.p, b.q
    fields:
      - {name: p, type: u32}
      - {name: q, type: f32, endian: little}   # a field can override the byte order

The same layout as a CSV file (one row per field; role is id, seq, time or time_relative):

packet,name,type,offset,count,endian,scale,add,unit,bit,role
,id,u8,0,,,,,,,id
1,temp,i16,,,,0.1,5,degC,,
1,flags,u16,3,,,,,,2,
2,seq,u32,,,,,,,,seq
2,t,f64,,,,,,ms,,time_relative
2,p,f32,,4,,,,bar,,

JSON and text lines

With format: json each datagram (or tcp line) is one JSON document, mapped with the same options as the REST source: items_path, values, tag_path, value_path, time_path, time_format (iso, epoch_s, epoch_ms), quality_path and good_quality. With format: csv-line, give columns (names without dots or brackets), delimiter (default a comma) and optionally time_path naming the time column; a datagram may hold several lines. Without a time, values get the arrival time. tags limits the channels; otherwise new ones are added up to max_tags (5000).

discover, backfill and loss

discover lists the layout's channels (text formats have no list until data arrives). backfill is not available: UDP and TCP streams have no history, so packets sent while the collector is down or the network drops them are gone. Use seq_gaps to see how many.

Health

packets, packets_per_s, values_per_s, decode_errors, dropped (oversized packets and packets that did not fit the bounded queue), seq_gaps, rejected_sources (wrong sender, or too many tcp clients), connected (tcp: connections; udp: 1 while listening), reconnects, idle_closed (idle tcp clients closed), clock_resyncs and queue. A NaN or infinite float is a gap with status BadDeviceFailure. Memory is bounded: at most 100,000 values wait to be sent.

Troubleshooting