e1608: link and device layers (discovery, handshake, framed commands, calibration, decode)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

CMakeLists.txt

@@ -242,6 +242,8 @@ add_executable(udpscope
   src/source/udps_source.cpp
   src/source/udps_kind.cpp
   src/source/probe_source.cpp
+  src/source/e1608/e1608_link.cpp
+  src/source/e1608/e1608_device.cpp
   src/plot_host.cpp
   src/export.cpp
   src/search.cpp

src/source/e1608/e1608_device.cpp

@@ -0,0 +1,200 @@
+// SPDX-License-Identifier: MIT
+// SPDX-FileCopyrightText: 2026 Martino Ferrari
+
+#include "source/e1608/e1608_device.hpp"
+
+#include <algorithm>
+#include <cctype>
+#include <cmath>
+#include <cstring>
+
+namespace udpscope::e1608 {
+
+namespace {
+
+constexpr double kClockHz = 80e6;
+
+std::string lower(std::string s) {
+  for (char& c : s) c = static_cast<char>(std::tolower(static_cast<unsigned char>(c)));
+  return s;
+}
+
+double clamp_code(double v) { return std::min(65535.0, std::max(0.0, v)); }
+
+float le_f32(const std::uint8_t* p) {
+  const std::uint32_t u = static_cast<std::uint32_t>(p[0]) | (static_cast<std::uint32_t>(p[1]) << 8) |
+                          (static_cast<std::uint32_t>(p[2]) << 16) |
+                          (static_cast<std::uint32_t>(p[3]) << 24);
+  float f;
+  std::memcpy(&f, &u, sizeof f);
+  return f;
+}
+
+void put_le32(std::vector<std::uint8_t>& v, std::uint32_t x) {
+  for (int i = 0; i < 4; ++i) v.push_back(static_cast<std::uint8_t>((x >> (8 * i)) & 0xFF));
+}
+
+/// Reads `count` (slope, intercept) pairs at `addr`.
+void read_table(Link& link, std::uint16_t addr, int count, Cal* out) {
+  const std::uint16_t bytes = static_cast<std::uint16_t>(count * 8);
+  const std::vector<std::uint8_t> req = {static_cast<std::uint8_t>(addr & 0xFF),
+                                         static_cast<std::uint8_t>(addr >> 8),
+                                         static_cast<std::uint8_t>(bytes & 0xFF),
+                                         static_cast<std::uint8_t>(bytes >> 8)};
+  const std::vector<std::uint8_t> r = link.command(0x40, req, bytes);
+  for (int i = 0; i < count; ++i) {
+    out[i].slope = le_f32(&r[static_cast<std::size_t>(i) * 8]);
+    out[i].intercept = le_f32(&r[static_cast<std::size_t>(i) * 8 + 4]);
+  }
+}
+
+}  // namespace
+
+double full_scale(Range r) {
+  static const double fs[] = {10.0, 5.0, 2.0, 1.0};
+  return fs[static_cast<int>(r)];
+}
+
+const char* range_name(Range r) {
+  static const char* const n[] = {"BIP10V", "BIP5V", "BIP2V", "BIP1V"};
+  return n[static_cast<int>(r)];
+}
+
+bool parse_range(const std::string& text, Range& out) {
+  const std::string t = lower(text);
+  for (int i = 0; i < 4; ++i) {
+    if (t == lower(range_name(static_cast<Range>(i)))) {
+      out = static_cast<Range>(i);
+      return true;
+    }
+  }
+  return false;
+}
+
+const char* trigger_name(Trigger t) {
+  switch (t) {
+    case Trigger::None: return "none";
+    case Trigger::Rising: return "rising";
+    case Trigger::Falling: return "falling";
+    case Trigger::High: return "high";
+    case Trigger::Low: return "low";
+  }
+  return "none";
+}
+
+bool parse_trigger(const std::string& text, Trigger& out) {
+  const std::string t = lower(text);
+  for (Trigger c : {Trigger::None, Trigger::Rising, Trigger::Falling, Trigger::High, Trigger::Low}) {
+    if (t == trigger_name(c)) {
+      out = c;
+      return true;
+    }
+  }
+  return false;
+}
+
+double Calibration::volts(std::uint16_t raw, Range range, bool diff) const {
+  const Cal& c = (diff ? df : se)[static_cast<int>(range)];
+  const double code = clamp_code(std::round(static_cast<double>(raw) * c.slope + c.intercept));
+  return (code - 32768.0) * full_scale(range) / 32768.0;
+}
+
+std::uint16_t Calibration::dac_code(int ch, double v) const {
+  v = std::min(10.0, std::max(-10.0, v));
+  const double nominal = clamp_code(std::round(v * 32768.0 / 10.0 + 32768.0));
+  const Cal& c = ao[ch & 1];
+  return static_cast<std::uint16_t>(clamp_code(std::round(nominal * c.slope + c.intercept)));
+}
+
+void decode_scans(const std::uint16_t* raw, std::size_t scans, const std::vector<QueueEntry>& queue,
+                  const Calibration& cal, std::vector<std::vector<double>>& out) {
+  const std::size_t nq = queue.size();
+  out.resize(nq);
+  for (std::size_t q = 0; q < nq; ++q) {
+    out[q].resize(scans);
+    for (std::size_t s = 0; s < scans; ++s)
+      out[q][s] = cal.volts(raw[s * nq + q], queue[q].range, queue[q].diff());
+  }
+}
+
+bool Device::read_calibration(Calibration& cal) {
+  Calibration c;
+  try {
+    read_table(link_, 0x00, 4, c.df);
+    read_table(link_, 0x20, 4, c.se);
+    read_table(link_, 0x40, 2, c.ao);
+  } catch (const LinkError& e) {
+    if (e.kind() != LinkError::Kind::Command) throw;
+    cal = Calibration{};
+    return false;
+  }
+  c.loaded = true;
+  cal = c;
+  return true;
+}
+
+std::uint8_t Device::din() { return link_.command(0x00, {}, 1)[0]; }
+
+void Device::dout(std::uint8_t value) { link_.command(0x03, {value}); }
+
+void Device::dconf(std::uint8_t inputs_mask) { link_.command(0x05, {inputs_mask}); }
+
+double Device::aout(int ch, double volts, const Calibration& cal) {
+  const double v = std::min(10.0, std::max(-10.0, volts));
+  const std::uint16_t code = cal.dac_code(ch, v);
+  link_.command(0x21, {static_cast<std::uint8_t>(ch & 1), static_cast<std::uint8_t>(code & 0xFF),
+                       static_cast<std::uint8_t>(code >> 8)});
+  return v;
+}
+
+std::uint32_t Device::counter() {
+  const std::vector<std::uint8_t> r = link_.command(0x30, {}, 4);
+  return static_cast<std::uint32_t>(r[0]) | (static_cast<std::uint32_t>(r[1]) << 8) |
+         (static_cast<std::uint32_t>(r[2]) << 16) | (static_cast<std::uint32_t>(r[3]) << 24);
+}
+
+void Device::counter_reset() { link_.command(0x31); }
+
+std::uint16_t Device::status() {
+  const std::vector<std::uint8_t> r = link_.command(0x52, {}, 2);
+  return static_cast<std::uint16_t>(r[0] | (r[1] << 8));
+}
+
+void Device::scan_start(const std::vector<QueueEntry>& queue, double aggregate_hz, bool ext_clock,
+                        Trigger trigger) {
+  link_.open_scan();
+  try {
+    std::vector<std::uint8_t> q;
+    q.push_back(static_cast<std::uint8_t>(queue.size()));
+    for (const QueueEntry& e : queue) {
+      q.push_back(e.wire_channel);
+      q.push_back(static_cast<std::uint8_t>(e.range));
+    }
+    link_.command(0x15, q);
+
+    std::uint32_t pacer = 0;
+    if (!ext_clock) {
+      const double p = std::round(kClockHz / aggregate_hz - 1.0);
+      pacer = static_cast<std::uint32_t>(std::min(4294967295.0, std::max(1.0, p)));
+    }
+    std::vector<std::uint8_t> s;
+    put_le32(s, 0);  // count 0: scan until stopped
+    put_le32(s, pacer);
+    s.push_back(static_cast<std::uint8_t>(trigger));
+    link_.command(0x11, s);
+    link_.send_filler();
+  } catch (...) {
+    link_.close_scan();
+    throw;
+  }
+}
+
+void Device::scan_stop() {
+  link_.close_scan();
+  try {
+    link_.command(0x13, {1});
+  } catch (const LinkError&) {
+  }
+}
+
+}  // namespace udpscope::e1608

src/source/e1608/e1608_device.hpp

@@ -0,0 +1,91 @@
+// SPDX-License-Identifier: MIT
+// SPDX-FileCopyrightText: 2026 Martino Ferrari
+
+#pragma once
+
+#include "source/e1608/e1608_link.hpp"
+
+#include <cstddef>
+#include <cstdint>
+#include <string>
+#include <vector>
+
+namespace udpscope::e1608 {
+
+enum class Range : std::uint8_t { Bip10V = 0, Bip5V = 1, Bip2V = 2, Bip1V = 3 };
+
+double full_scale(Range r);          ///< 10, 5, 2, 1
+const char* range_name(Range r);     ///< "BIP10V" ...
+/// Case-insensitive: "BIP10V", "bip5v", ...
+bool parse_range(const std::string& text, Range& out);
+
+enum class Trigger : std::uint8_t { None = 0, Rising = 4, Falling = 8, High = 12, Low = 16 };
+
+const char* trigger_name(Trigger t);  ///< "none", "rising", "falling", "high", "low"
+bool parse_trigger(const std::string& text, Trigger& out);  ///< case-insensitive
+
+struct Cal {
+  double slope = 1.0;
+  double intercept = 0.0;
+};
+
+/// The factory calibration: one (slope, intercept) per range for single-ended and differential
+/// inputs, one per DAC. Defaults are the nominal (uncalibrated) values.
+struct Calibration {
+  Cal se[4];
+  Cal df[4];
+  Cal ao[2];
+  bool loaded = false;
+
+  /// A raw ADC count as calibrated volts.
+  double volts(std::uint16_t raw, Range range, bool diff) const;
+  /// The DAC code that makes `ch` output `volts` (clamped to +-10 V).
+  std::uint16_t dac_code(int ch, double volts) const;
+};
+
+/// One entry of the scan queue. DIFF channels are `channel + 8` on the wire.
+struct QueueEntry {
+  std::uint8_t wire_channel;
+  Range range;
+  bool diff() const noexcept { return wire_channel >= 8; }
+};
+
+/// Turns `scans` interleaved scans of `queue.size()` raw counts into calibrated volts:
+/// `out[queue index][scan]`.
+void decode_scans(const std::uint16_t* raw, std::size_t scans, const std::vector<QueueEntry>& queue,
+                  const Calibration& cal, std::vector<std::vector<double>>& out);
+
+/// Typed commands over a `Link`. Holds a reference: the link must outlive it.
+class Device {
+ public:
+  explicit Device(Link& link) : link_(link) {}
+
+  /// Fills `cal` from the board's calibration memory and sets `cal.loaded`. A refusal from the
+  /// board (`LinkError::Kind::Command`) leaves the nominal values and returns false; any other
+  /// error propagates.
+  bool read_calibration(Calibration& cal);
+
+  std::uint8_t din();
+  void dout(std::uint8_t value);
+  /// `inputs_mask`: a set bit makes that line an input.
+  void dconf(std::uint8_t inputs_mask);
+  /// Writes `volts` (clamped to +-10 V, calibrated) to DAC `ch`; returns the clamped volts.
+  double aout(int ch, double volts, const Calibration& cal);
+  std::uint32_t counter();
+  void counter_reset();
+  /// Bit 0 scan socket open, bit 1 scan running, bit 2 overrun.
+  std::uint16_t status();
+
+  /// Opens the scan socket, writes the queue, starts the scan and sends the filler byte.
+  /// `aggregate_hz` is the scan rate times the number of channels. On failure the scan socket
+  /// is closed again.
+  void scan_start(const std::vector<QueueEntry>& queue, double aggregate_hz, bool ext_clock,
+                  Trigger trigger);
+  /// Closes the scan socket and tells the board to stop; board errors are ignored.
+  void scan_stop();
+
+ private:
+  Link& link_;
+};
+
+}  // namespace udpscope::e1608

src/source/e1608/e1608_link.cpp

@@ -0,0 +1,360 @@
+// SPDX-License-Identifier: MIT
+// SPDX-FileCopyrightText: 2026 Martino Ferrari
+
+#include "source/e1608/e1608_link.hpp"
+
+#include <algorithm>
+#include <arpa/inet.h>
+#include <cerrno>
+#include <chrono>
+#include <cstdio>
+#include <cstring>
+#include <netdb.h>
+#include <netinet/in.h>
+#include <netinet/tcp.h>
+#include <poll.h>
+#include <sys/socket.h>
+#include <unistd.h>
+
+namespace udpscope::e1608 {
+
+void Fd::reset(int fd) noexcept {
+  if (fd_ >= 0) ::close(fd_);
+  fd_ = fd;
+}
+
+namespace {
+
+using Clock = std::chrono::steady_clock;
+using Kind = LinkError::Kind;
+
+constexpr std::size_t kMaxFrame = 1024;
+
+[[noreturn]] void sys_fail(const std::string& what) {
+  throw LinkError(Kind::Io, what + ": " + std::strerror(errno));
+}
+
+int ms_left(Clock::time_point deadline) {
+  const auto d = std::chrono::duration_cast<std::chrono::milliseconds>(deadline - Clock::now());
+  return d.count() < 0 ? 0 : static_cast<int>(d.count());
+}
+
+/// True when `fd` is ready for `events` (or has an error/hang-up to report); false on timeout.
+bool wait_fd(int fd, short events, int timeout_ms) {
+  pollfd p{fd, events, 0};
+  for (;;) {
+    const int r = ::poll(&p, 1, timeout_ms);
+    if (r > 0) return true;
+    if (r == 0) return false;
+    if (errno == EINTR) continue;
+    sys_fail("poll");
+  }
+}
+
+sockaddr_in make_addr(const std::string& host, int port) {
+  sockaddr_in a{};
+  a.sin_family = AF_INET;
+  a.sin_port = htons(static_cast<std::uint16_t>(port));
+  if (::inet_pton(AF_INET, host.c_str(), &a.sin_addr) == 1) return a;
+  addrinfo hints{};
+  hints.ai_family = AF_INET;
+  hints.ai_socktype = SOCK_STREAM;
+  addrinfo* res = nullptr;
+  if (::getaddrinfo(host.c_str(), nullptr, &hints, &res) != 0 || res == nullptr)
+    throw LinkError(Kind::Io, "cannot resolve '" + host + "'");
+  a.sin_addr = reinterpret_cast<const sockaddr_in*>(res->ai_addr)->sin_addr;
+  ::freeaddrinfo(res);
+  return a;
+}
+
+void send_all(int fd, const std::uint8_t* p, std::size_t n, Clock::time_point deadline) {
+  while (n > 0) {
+    if (!wait_fd(fd, POLLOUT, ms_left(deadline)))
+      throw LinkError(Kind::Timeout, "timed out sending to the board");
+    const ssize_t w = ::send(fd, p, n, MSG_NOSIGNAL | MSG_DONTWAIT);
+    if (w < 0) {
+      if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) continue;
+      sys_fail("send");
+    }
+    p += w;
+    n -= static_cast<std::size_t>(w);
+  }
+}
+
+void read_exact(int fd, std::uint8_t* p, std::size_t n, Clock::time_point deadline) {
+  while (n > 0) {
+    if (!wait_fd(fd, POLLIN, ms_left(deadline)))
+      throw LinkError(Kind::Timeout, "timed out waiting for the board");
+    const ssize_t r = ::recv(fd, p, n, MSG_DONTWAIT);
+    if (r == 0) throw LinkError(Kind::Io, "the board closed the connection");
+    if (r < 0) {
+      if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) continue;
+      sys_fail("recv");
+    }
+    p += r;
+    n -= static_cast<std::size_t>(r);
+  }
+}
+
+/// A connected, non-blocking TCP socket with TCP_NODELAY. `rcvbuf > 0` sets SO_RCVBUF first.
+Fd tcp_connect(const std::string& host, int port, int timeout_ms, int rcvbuf) {
+  const sockaddr_in a = make_addr(host, port);
+  Fd fd(::socket(AF_INET, SOCK_STREAM | SOCK_NONBLOCK, 0));
+  if (!fd.valid()) sys_fail("socket");
+  const int one = 1;
+  ::setsockopt(fd.get(), IPPROTO_TCP, TCP_NODELAY, &one, sizeof one);
+  if (rcvbuf > 0) ::setsockopt(fd.get(), SOL_SOCKET, SO_RCVBUF, &rcvbuf, sizeof rcvbuf);
+  const std::string where = host + ":" + std::to_string(port);
+  if (::connect(fd.get(), reinterpret_cast<const sockaddr*>(&a), sizeof a) < 0 &&
+      errno != EINPROGRESS)
+    sys_fail("cannot connect to " + where);
+  if (!wait_fd(fd.get(), POLLOUT, timeout_ms))
+    throw LinkError(Kind::Timeout, "timed out connecting to " + where);
+  int err = 0;
+  socklen_t len = sizeof err;
+  ::getsockopt(fd.get(), SOL_SOCKET, SO_ERROR, &err, &len);
+  if (err != 0) {
+    errno = err;
+    sys_fail("cannot connect to " + where);
+  }
+  return fd;
+}
+
+}  // namespace
+
+// --- addresses ------------------------------------------------------------------------------
+
+bool parse_mac(const std::string& text, Mac& out) {
+  Mac m{};
+  std::size_t pos = 0;
+  for (std::size_t i = 0; i < 6; ++i) {
+    if (pos + 2 > text.size()) return false;
+    unsigned v = 0;
+    for (std::size_t k = 0; k < 2; ++k) {
+      const char c = text[pos + k];
+      v <<= 4;
+      if (c >= '0' && c <= '9') v |= static_cast<unsigned>(c - '0');
+      else if (c >= 'a' && c <= 'f') v |= static_cast<unsigned>(c - 'a' + 10);
+      else if (c >= 'A' && c <= 'F') v |= static_cast<unsigned>(c - 'A' + 10);
+      else return false;
+    }
+    m[i] = static_cast<std::uint8_t>(v);
+    pos += 2;
+    if (i < 5) {
+      if (pos >= text.size() || (text[pos] != ':' && text[pos] != '-')) return false;
+      ++pos;
+    }
+  }
+  if (pos != text.size()) return false;
+  out = m;
+  return true;
+}
+
+std::string format_mac(const Mac& mac) {
+  char buf[24];
+  std::snprintf(buf, sizeof buf, "%02x:%02x:%02x:%02x:%02x:%02x", mac[0], mac[1], mac[2], mac[3],
+                mac[4], mac[5]);
+  return buf;
+}
+
+std::vector<std::string> sweep_targets(const std::string& subnet) {
+  std::string s = subnet;
+  while (!s.empty() && s.front() == ' ') s.erase(s.begin());
+  while (!s.empty() && (s.back() == ' ' || s.back() == '.')) s.pop_back();
+  if (s.empty()) return {"255.255.255.255"};
+  int parts = 0;
+  std::size_t i = 0;
+  while (i <= s.size()) {
+    const std::size_t dot = s.find('.', i);
+    const std::string part = s.substr(i, dot == std::string::npos ? std::string::npos : dot - i);
+    if (part.empty() || part.size() > 3) return {};
+    int v = 0;
+    for (char c : part) {
+      if (c < '0' || c > '9') return {};
+      v = v * 10 + (c - '0');
+    }
+    if (v > 255) return {};
+    ++parts;
+    if (dot == std::string::npos) break;
+    i = dot + 1;
+  }
+  if (parts != 3) return {};
+  std::vector<std::string> out;
+  out.reserve(254);
+  for (int h = 1; h <= 254; ++h) out.push_back(s + "." + std::to_string(h));
+  return out;
+}
+
+std::vector<DeviceInfo> discover(const std::vector<std::string>& targets, int port, int wait_ms) {
+  std::vector<DeviceInfo> found;
+  Fd u(::socket(AF_INET, SOCK_DGRAM | SOCK_NONBLOCK, 0));
+  if (!u.valid()) sys_fail("socket");
+  const int one = 1;
+  ::setsockopt(u.get(), SOL_SOCKET, SO_BROADCAST, &one, sizeof one);
+  const std::uint8_t probe = 'D';
+  for (const std::string& t : targets) {
+    sockaddr_in a{};
+    try {
+      a = make_addr(t, port);
+    } catch (const LinkError&) {
+      continue;
+    }
+    // A sweep may hit unreachable hosts; those sends fail and are simply skipped.
+    (void)::sendto(u.get(), &probe, 1, MSG_NOSIGNAL, reinterpret_cast<const sockaddr*>(&a),
+                   sizeof a);
+  }
+  const auto deadline = Clock::now() + std::chrono::milliseconds(wait_ms);
+  for (;;) {
+    if (!wait_fd(u.get(), POLLIN, ms_left(deadline))) break;
+    std::uint8_t buf[128];
+    sockaddr_in from{};
+    socklen_t flen = sizeof from;
+    const ssize_t n = ::recvfrom(u.get(), buf, sizeof buf, MSG_DONTWAIT,
+                                 reinterpret_cast<sockaddr*>(&from), &flen);
+    if (n < 0) {
+      if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) continue;
+      break;
+    }
+    if (n < 29 || buf[0] != 'D') continue;
+    DeviceInfo info;
+    info.product_id = static_cast<std::uint16_t>(buf[7] | (buf[8] << 8));
+    if (info.product_id != kProductId) continue;
+    std::copy(buf + 1, buf + 7, info.mac.begin());
+    bool dup = false;
+    for (const DeviceInfo& d : found) dup = dup || d.mac == info.mac;
+    if (dup) continue;
+    char ip[INET_ADDRSTRLEN] = {};
+    ::inet_ntop(AF_INET, &from.sin_addr, ip, sizeof ip);
+    info.ip = ip;
+    found.push_back(info);
+  }
+  return found;
+}
+
+// --- Link -----------------------------------------------------------------------------------
+
+void Link::connect(const std::string& host, int port, std::uint32_t code, int timeout_ms) {
+  close();
+  host_ = host;
+  port_ = port;
+  timeout_ms_ = timeout_ms;
+  next_id_ = 0;
+
+  // The connection-code handshake: a connected UDP socket, so an ICMP refusal surfaces.
+  {
+    const sockaddr_in a = make_addr(host, port);
+    Fd u(::socket(AF_INET, SOCK_DGRAM, 0));
+    if (!u.valid()) sys_fail("socket");
+    if (::connect(u.get(), reinterpret_cast<const sockaddr*>(&a), sizeof a) < 0)
+      sys_fail("handshake with " + host);
+    const std::uint8_t msg[5] = {'C', static_cast<std::uint8_t>(code & 0xFF),
+                                 static_cast<std::uint8_t>((code >> 8) & 0xFF),
+                                 static_cast<std::uint8_t>((code >> 16) & 0xFF),
+                                 static_cast<std::uint8_t>((code >> 24) & 0xFF)};
+    if (::send(u.get(), msg, sizeof msg, MSG_NOSIGNAL) < 0) sys_fail("handshake with " + host);
+    if (!wait_fd(u.get(), POLLIN, timeout_ms))
+      throw LinkError(Kind::Timeout, "no answer from the board at " + host);
+    std::uint8_t rep[16];
+    const ssize_t n = ::recv(u.get(), rep, sizeof rep, 0);
+    if (n < 0) sys_fail("handshake with " + host);
+    if (n < 2 || rep[0] != 'C') throw LinkError(Kind::Protocol, "unexpected handshake reply");
+    switch (rep[1]) {
+      case 0:
+      case 2:
+        break;
+      case 1:
+        throw LinkError(Kind::Rejected, "the board rejected the connection code");
+      case 3:
+        throw LinkError(Kind::Busy, "the board is busy (another client holds it)");
+      default:
+        throw LinkError(Kind::Protocol, "unexpected handshake status " + std::to_string(rep[1]));
+    }
+  }
+  cmd_ = tcp_connect(host, port, timeout_ms, 0);
+}
+
+void Link::close() {
+  scan_.reset();
+  cmd_.reset();
+}
+
+std::vector<std::uint8_t> Link::command(std::uint8_t cmd, const std::vector<std::uint8_t>& payload,
+                                        int expect) {
+  if (!cmd_.valid()) throw LinkError(Kind::Io, "not connected");
+  if (payload.size() + 7 > kMaxFrame) throw LinkError(Kind::Protocol, "command too long");
+  // Bytes left by an earlier exchange that timed out must not be read as this reply.
+  {
+    std::uint8_t junk[256];
+    while (::recv(cmd_.get(), junk, sizeof junk, MSG_DONTWAIT) > 0) {
+    }
+  }
+  const std::uint8_t id = ++next_id_;
+  std::vector<std::uint8_t> f;
+  f.reserve(7 + payload.size());
+  f.push_back(0xDB);
+  f.push_back(cmd);
+  f.push_back(id);
+  f.push_back(0);
+  f.push_back(static_cast<std::uint8_t>(payload.size() & 0xFF));
+  f.push_back(static_cast<std::uint8_t>((payload.size() >> 8) & 0xFF));
+  f.insert(f.end(), payload.begin(), payload.end());
+  unsigned sum = 0;
+  for (std::uint8_t b : f) sum += b;
+  f.push_back(static_cast<std::uint8_t>(0xFF - (sum & 0xFF)));
+
+  const auto deadline = Clock::now() + std::chrono::milliseconds(timeout_ms_);
+  send_all(cmd_.get(), f.data(), f.size(), deadline);
+
+  std::uint8_t h[6];
+  read_exact(cmd_.get(), h, sizeof h, deadline);
+  if (h[0] != 0xDB) throw LinkError(Kind::Protocol, "bad sync byte in a reply");
+  const std::size_t n = static_cast<std::size_t>(h[4] | (h[5] << 8));
+  if (n + 7 > kMaxFrame) throw LinkError(Kind::Protocol, "oversized reply");
+  std::vector<std::uint8_t> body(n + 1);
+  read_exact(cmd_.get(), body.data(), body.size(), deadline);
+  unsigned rsum = 0;
+  for (std::uint8_t b : h) rsum += b;
+  for (std::uint8_t b : body) rsum += b;
+  if ((rsum & 0xFF) != 0xFF) throw LinkError(Kind::Protocol, "bad checksum in a reply");
+  if (h[1] != (cmd | 0x80) || h[2] != id)
+    throw LinkError(Kind::Protocol, "reply does not match the command");
+  if (h[3] != 0) {
+    static const char* const kText[] = {"ok",        "protocol error", "parameter error",
+                                        "busy",      "not ready",      "timeout",
+                                        "other error"};
+    const char* t = h[3] < 7 ? kText[h[3]] : "unknown error";
+    char hex[8];
+    std::snprintf(hex, sizeof hex, "%02x", cmd);
+    throw LinkError(Kind::Command, std::string("the board refused command 0x") + hex + ": " + t);
+  }
+  body.pop_back();
+  if (expect >= 0 && body.size() != static_cast<std::size_t>(expect))
+    throw LinkError(Kind::Command, "reply of " + std::to_string(body.size()) +
+                                       " bytes, expected " + std::to_string(expect));
+  return body;
+}
+
+void Link::open_scan() {
+  close_scan();
+  scan_ = tcp_connect(host_, port_ + 1, timeout_ms_, 1 << 20);
+}
+
+void Link::close_scan() { scan_.reset(); }
+
+void Link::send_filler() {
+  if (!cmd_.valid()) throw LinkError(Kind::Io, "not connected");
+  const std::uint8_t z = 0;
+  send_all(cmd_.get(), &z, 1, Clock::now() + std::chrono::milliseconds(timeout_ms_));
+}
+
+std::size_t Link::read_scan(std::uint8_t* buf, std::size_t cap, int timeout_ms) {
+  if (!scan_.valid()) throw LinkError(Kind::Io, "the scan socket is not open");
+  if (!wait_fd(scan_.get(), POLLIN, timeout_ms)) return 0;
+  const ssize_t n = ::recv(scan_.get(), buf, cap, MSG_DONTWAIT);
+  if (n > 0) return static_cast<std::size_t>(n);
+  if (n == 0) throw LinkError(Kind::Io, "scan socket closed by the board");
+  if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) return 0;
+  sys_fail("scan recv");
+}
+
+}  // namespace udpscope::e1608

src/source/e1608/e1608_link.hpp

@@ -0,0 +1,113 @@
+// SPDX-License-Identifier: MIT
+// SPDX-FileCopyrightText: 2026 Martino Ferrari
+
+#pragma once
+
+#include <array>
+#include <cstddef>
+#include <cstdint>
+#include <stdexcept>
+#include <string>
+#include <vector>
+
+namespace udpscope::e1608 {
+
+constexpr int kDefaultPort = 54211;
+constexpr std::uint16_t kProductId = 0x012F;
+
+/// Every failure of the link. `Rejected` (a bad connection code) is the only one a retry cannot
+/// fix. `Command` is a reply the board answered with a nonzero status, or a reply of the wrong
+/// length: the link itself is fine. `Protocol` is a framing error (sync, checksum, id).
+class LinkError : public std::runtime_error {
+ public:
+  enum class Kind { Io, Timeout, Rejected, Busy, Protocol, Command };
+  LinkError(Kind kind, const std::string& what) : std::runtime_error(what), kind_(kind) {}
+  Kind kind() const noexcept { return kind_; }
+
+ private:
+  Kind kind_;
+};
+
+/// An owned file descriptor.
+class Fd {
+ public:
+  Fd() = default;
+  explicit Fd(int fd) : fd_(fd) {}
+  Fd(const Fd&) = delete;
+  Fd& operator=(const Fd&) = delete;
+  Fd(Fd&& o) noexcept : fd_(o.release()) {}
+  Fd& operator=(Fd&& o) noexcept {
+    if (this != &o) reset(o.release());
+    return *this;
+  }
+  ~Fd() { reset(); }
+  int get() const noexcept { return fd_; }
+  bool valid() const noexcept { return fd_ >= 0; }
+  int release() noexcept {
+    const int f = fd_;
+    fd_ = -1;
+    return f;
+  }
+  /// Closes the current descriptor (if any) and takes `fd`.
+  void reset(int fd = -1) noexcept;
+
+ private:
+  int fd_ = -1;
+};
+
+using Mac = std::array<std::uint8_t, 6>;
+
+/// Six hex bytes separated by `:` or `-`. False (and `out` untouched) on anything else.
+bool parse_mac(const std::string& text, Mac& out);
+/// Lower-case `aa:bb:cc:dd:ee:ff`.
+std::string format_mac(const Mac& mac);
+
+struct DeviceInfo {
+  std::string ip;
+  Mac mac{};
+  std::uint16_t product_id = 0;
+};
+
+/// Where to send a discovery `D`: an empty subnet is the limited broadcast; `a.b.c` (an optional
+/// trailing dot) is a unicast to hosts .1 to .254; anything else is empty.
+std::vector<std::string> sweep_targets(const std::string& subnet);
+
+/// Sends `D` to every target, then collects replies for `wait_ms`. Keeps E-1608s only (product id
+/// `kProductId`) and one entry per MAC.
+std::vector<DeviceInfo> discover(const std::vector<std::string>& targets, int port, int wait_ms);
+
+/// One board: the UDP handshake, the TCP command channel, and the scan socket (`port + 1`).
+/// Not thread safe; the source's producer thread owns it.
+class Link {
+ public:
+  /// Handshake, then the command connection. Throws `LinkError` (`Rejected` for a bad code).
+  void connect(const std::string& host, int port, std::uint32_t code, int timeout_ms);
+  /// Closes the command and scan sockets. Safe to call at any time.
+  void close();
+  bool is_open() const noexcept { return cmd_.valid(); }
+
+  /// One framed exchange. `expect >= 0` demands a reply payload of exactly that length. Throws
+  /// `Command` for a nonzero status or a length mismatch, `Protocol` for a bad frame.
+  std::vector<std::uint8_t> command(std::uint8_t cmd, const std::vector<std::uint8_t>& payload = {},
+                                    int expect = -1);
+
+  /// Opens the scan stream socket (a 1 MiB receive buffer, set before connecting).
+  void open_scan();
+  void close_scan();
+  bool scan_open() const noexcept { return scan_.valid(); }
+  /// The single `0x00` byte on the command socket that starts the stream after scan start.
+  void send_filler();
+  /// Waits up to `timeout_ms` for scan bytes and reads what is there (at most `cap`). Returns 0 on
+  /// a timeout. Throws `Io` when the board closed the socket.
+  std::size_t read_scan(std::uint8_t* buf, std::size_t cap, int timeout_ms);
+
+ private:
+  Fd cmd_;
+  Fd scan_;
+  std::string host_;
+  int port_ = kDefaultPort;
+  int timeout_ms_ = 2000;
+  std::uint8_t next_id_ = 0;
+};
+
+}  // namespace udpscope::e1608