e1608: the settings, the DataSource state machine, and the connect form

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

src/source/e1608/e1608_kind.cpp

@@ -0,0 +1,288 @@
+// SPDX-License-Identifier: MIT
+// SPDX-FileCopyrightText: 2026 Martino Ferrari
+
+#include "source/e1608/e1608_source.hpp"
+#include "source/registry.hpp"
+
+#include <adwaita.h>
+
+#include <cstdio>
+#include <functional>
+#include <memory>
+#include <string>
+
+namespace udpscope {
+namespace {
+
+using namespace e1608;
+using ChangeCb = std::shared_ptr<std::function<void()>>;
+
+constexpr const char* kTriggerNames[] = {"none", "rising", "falling", "high", "low"};
+constexpr Trigger kTriggers[] = {Trigger::None, Trigger::Rising, Trigger::Falling, Trigger::High,
+                                 Trigger::Low};
+
+std::string num(double v) {
+  char b[48];
+  std::snprintf(b, sizeof b, "%.10g", v);
+  return b;
+}
+
+GtkWidget* entry_row(const char* title) {
+  GtkWidget* r = adw_entry_row_new();
+  adw_preferences_row_set_title(ADW_PREFERENCES_ROW(r), title);
+  adw_preferences_row_set_use_underline(ADW_PREFERENCES_ROW(r), TRUE);
+  return r;
+}
+
+GtkWidget* spin_row(const char* title, double lo, double hi, double step, double def, int digits) {
+  GtkWidget* r = GTK_WIDGET(adw_spin_row_new_with_range(lo, hi, step));
+  adw_preferences_row_set_title(ADW_PREFERENCES_ROW(r), title);
+  adw_preferences_row_set_use_underline(ADW_PREFERENCES_ROW(r), TRUE);
+  adw_spin_row_set_digits(ADW_SPIN_ROW(r), static_cast<guint>(digits));
+  adw_spin_row_set_value(ADW_SPIN_ROW(r), def);
+  return r;
+}
+
+GtkWidget* switch_row(const char* title, bool def) {
+  GtkWidget* r = adw_switch_row_new();
+  adw_preferences_row_set_title(ADW_PREFERENCES_ROW(r), title);
+  adw_preferences_row_set_use_underline(ADW_PREFERENCES_ROW(r), TRUE);
+  adw_switch_row_set_active(ADW_SWITCH_ROW(r), def);
+  return r;
+}
+
+GtkWidget* combo_row(const char* title, const char* const* items) {
+  GtkWidget* r = adw_combo_row_new();
+  adw_preferences_row_set_title(ADW_PREFERENCES_ROW(r), title);
+  adw_preferences_row_set_use_underline(ADW_PREFERENCES_ROW(r), TRUE);
+  adw_combo_row_set_model(ADW_COMBO_ROW(r), G_LIST_MODEL(gtk_string_list_new(items)));
+  return r;
+}
+
+GtkWidget* expander(const char* title, std::initializer_list<GtkWidget*> rows) {
+  GtkWidget* e = adw_expander_row_new();
+  adw_preferences_row_set_title(ADW_PREFERENCES_ROW(e), title);
+  for (GtkWidget* r : rows) adw_expander_row_add_row(ADW_EXPANDER_ROW(e), r);
+  return e;
+}
+
+std::string text_of(GtkWidget* w) { return gtk_editable_get_text(GTK_EDITABLE(w)); }
+double value_of(GtkWidget* w) { return adw_spin_row_get_value(ADW_SPIN_ROW(w)); }
+guint sel_of(GtkWidget* w) { return adw_combo_row_get_selected(ADW_COMBO_ROW(w)); }
+bool on_of(GtkWidget* w) { return adw_switch_row_get_active(ADW_SWITCH_ROW(w)); }
+
+/// The form's widgets. Held by a `shared_ptr` that every signal closure and every `ConfigForm`
+/// closure shares; the raw widget pointers are valid while the chooser keeps the rows.
+struct Form {
+  GtkWidget *address, *mac, *subnet, *port, *code, *timeout;
+  GtkWidget *rate, *ext_clock, *trigger, *auto_restart, *calibrate, *poll_rate;
+  GtkWidget *mode, *ai[8], *ai_group;
+  GtkWidget *dir[8], *do_init, *plot_counter, *ao_init[2];
+  ChangeCb changed = std::make_shared<std::function<void()>>();
+  bool updating = false;
+
+  bool diff() const { return sel_of(mode) == 1; }
+  int enabled_ai() const {
+    int n = 0;
+    for (int i = 0; i < (diff() ? 4 : 8); ++i) n += sel_of(ai[i]) > 0 ? 1 : 0;
+    return n;
+  }
+
+  /// In DIFF mode AI rows 4 to 7 are insensitive, not hidden; the group's subtitle shows the load.
+  void refresh() {
+    const bool d = diff();
+    for (int i = 0; i < 8; ++i) gtk_widget_set_sensitive(ai[i], !(d && i >= 4));
+    const int n = enabled_ai();
+    char b[96];
+    std::snprintf(b, sizeof b, "%d channel%s, %.3g of 250 kS/s", n, n == 1 ? "" : "s",
+                  n * value_of(rate) / 1000.0);
+    adw_expander_row_set_subtitle(ADW_EXPANDER_ROW(ai_group), b);
+  }
+
+  SourceConfig read() const {
+    SourceConfig c;
+    if (!text_of(address).empty()) c.set("address", text_of(address));
+    if (!text_of(mac).empty()) c.set("mac", text_of(mac));
+    if (!text_of(subnet).empty()) c.set("subnet", text_of(subnet));
+    c.set("port", num(value_of(port)));
+    c.set("code", num(value_of(code)));
+    c.set("timeout_ms", num(value_of(timeout)));
+    c.set("scan_rate", num(value_of(rate)));
+    c.set("ext_clock", on_of(ext_clock) ? "1" : "0");
+    c.set("trigger", kTriggerNames[std::min<guint>(sel_of(trigger), 4)]);
+    c.set("auto_restart", on_of(auto_restart) ? "1" : "0");
+    c.set("calibrate", on_of(calibrate) ? "1" : "0");
+    c.set("poll_rate", num(value_of(poll_rate)));
+    c.set("mode", diff() ? "diff" : "se");
+    static const char* const kRange[] = {"", "BIP10V", "BIP5V", "BIP2V", "BIP1V"};
+    for (int i = 0; i < 8; ++i) {
+      const guint s = sel_of(ai[i]);
+      if (s > 0 && s <= 4 && !(diff() && i >= 4)) c.set("ai" + std::to_string(i), kRange[s]);
+    }
+    for (int i = 0; i < 8; ++i) {
+      const guint s = sel_of(dir[i]);  // 0 input, 1 input plotted, 2 output
+      c.set("dir" + std::to_string(i), s == 2 ? "o" : "i");
+      c.set("plot" + std::to_string(i), s == 1 ? "1" : "0");
+    }
+    c.set("plot_counter", on_of(plot_counter) ? "1" : "0");
+    c.set("do_init", num(value_of(do_init)));
+    c.set("ao0_init", num(value_of(ao_init[0])));
+    c.set("ao1_init", num(value_of(ao_init[1])));
+    return c;
+  }
+
+  void write(const SourceConfig& cfg) {
+    const E1608Settings s = E1608Settings::from(cfg);
+    updating = true;
+    gtk_editable_set_text(GTK_EDITABLE(address), s.address.c_str());
+    gtk_editable_set_text(GTK_EDITABLE(mac), s.mac_text.c_str());
+    gtk_editable_set_text(GTK_EDITABLE(subnet), s.subnet.c_str());
+    adw_spin_row_set_value(ADW_SPIN_ROW(port), s.port);
+    adw_spin_row_set_value(ADW_SPIN_ROW(code), static_cast<double>(s.code));
+    adw_spin_row_set_value(ADW_SPIN_ROW(timeout), s.timeout_ms);
+    adw_spin_row_set_value(ADW_SPIN_ROW(rate), s.scan_rate);
+    adw_switch_row_set_active(ADW_SWITCH_ROW(ext_clock), s.ext_clock);
+    for (guint i = 0; i < 5; ++i) {
+      if (kTriggers[i] == s.trigger) adw_combo_row_set_selected(ADW_COMBO_ROW(trigger), i);
+    }
+    adw_switch_row_set_active(ADW_SWITCH_ROW(auto_restart), s.auto_restart);
+    adw_switch_row_set_active(ADW_SWITCH_ROW(calibrate), s.calibrate);
+    adw_spin_row_set_value(ADW_SPIN_ROW(poll_rate), s.poll_rate);
+    adw_combo_row_set_selected(ADW_COMBO_ROW(mode), s.diff ? 1 : 0);
+    for (int i = 0; i < 8; ++i) {
+      adw_combo_row_set_selected(
+          ADW_COMBO_ROW(ai[i]), s.ai_on[i] ? static_cast<guint>(s.ai_range[i]) + 1 : 0);
+      adw_combo_row_set_selected(ADW_COMBO_ROW(dir[i]), s.dir_out[i] ? 2 : (s.plot[i] ? 1 : 0));
+    }
+    adw_switch_row_set_active(ADW_SWITCH_ROW(plot_counter), s.plot_counter);
+    adw_spin_row_set_value(ADW_SPIN_ROW(do_init), s.do_init);
+    adw_spin_row_set_value(ADW_SPIN_ROW(ao_init[0]), s.ao_init[0]);
+    adw_spin_row_set_value(ADW_SPIN_ROW(ao_init[1]), s.ao_init[1]);
+    updating = false;
+    refresh();
+  }
+
+  bool valid() const { return enabled_ai() > 0 && E1608Settings::from(read()).problem().empty(); }
+};
+
+using FormPtr = std::shared_ptr<Form>;
+
+void on_notify(GObject*, GParamSpec*, gpointer user) {
+  Form& f = **static_cast<FormPtr*>(user);
+  f.refresh();
+  if (!f.updating && f.changed && *f.changed) (*f.changed)();
+}
+
+void on_edited(GtkEditable*, gpointer user) {
+  Form& f = **static_cast<FormPtr*>(user);
+  if (!f.updating && f.changed && *f.changed) (*f.changed)();
+}
+
+void free_form(gpointer d, GClosure*) { delete static_cast<FormPtr*>(d); }
+
+void hook_notify(GtkWidget* w, const char* signal, const FormPtr& f) {
+  g_signal_connect_data(w, signal, G_CALLBACK(on_notify), new FormPtr(f), free_form,
+                        static_cast<GConnectFlags>(0));
+}
+
+void hook_entry(GtkWidget* w, const FormPtr& f) {
+  g_signal_connect_data(w, "changed", G_CALLBACK(on_edited), new FormPtr(f), free_form,
+                        static_cast<GConnectFlags>(0));
+}
+
+class E1608Kind final : public SourceKind {
+ public:
+  std::string id() const override { return "e1608"; }
+  std::string display_name() const override { return "MCC E-1608"; }
+  bool has_datagrams() const override { return false; }
+
+  std::string identity(const SourceConfig& cfg) const override {
+    return E1608Settings::from(cfg).identity();
+  }
+
+  std::unique_ptr<DataSource> make(const SourceConfig& cfg) const override {
+    return std::make_unique<E1608Source>(cfg);
+  }
+
+  ConfigForm make_config_form() const override {
+    auto f = std::make_shared<Form>();
+    static const char* const kTrig[] = {"None", "Rising edge", "Falling edge", "High level",
+                                        "Low level", nullptr};
+    static const char* const kMode[] = {"Single-ended (8 channels)", "Differential (4 channels)",
+                                        nullptr};
+    static const char* const kAi[] = {"Off", "\u00b110 V", "\u00b15 V", "\u00b12 V", "\u00b11 V",
+                                      nullptr};
+    static const char* const kDir[] = {"Input", "Input, plotted", "Output", nullptr};
+
+    f->address = entry_row("_Address");
+    f->mac = entry_row("_MAC address");
+    f->subnet = entry_row("_Discovery subnet");
+    f->port = spin_row("_Port", 1, 65534, 1, kDefaultPort, 0);
+    f->code = spin_row("Connection _code", 0, 4294967295.0, 1, 0, 0);
+    f->timeout = spin_row("_Timeout (ms)", 100, 60000, 100, 2000, 0);
+    f->rate = spin_row("Scan _rate (S/s per channel)", 0.02, 250000, 1, 1000, 2);
+    f->ext_clock = switch_row("_External clock", false);
+    f->trigger = combo_row("T_rigger", kTrig);
+    f->auto_restart = switch_row("_Restart after an overrun", true);
+    f->calibrate = switch_row("Apply _factory calibration", true);
+    f->poll_rate = spin_row("_Input poll rate (Hz)", 0.1, 100, 1, 10, 1);
+    f->mode = combo_row("Input _mode", kMode);
+    for (int i = 0; i < 8; ++i) {
+      const std::string title = "Channel " + std::to_string(i);
+      f->ai[i] = combo_row(title.c_str(), kAi);
+      adw_combo_row_set_selected(ADW_COMBO_ROW(f->ai[i]), i == 0 ? 1 : 0);
+    }
+    for (int i = 0; i < 8; ++i) {
+      const std::string title = "Bit " + std::to_string(i);
+      f->dir[i] = combo_row(title.c_str(), kDir);
+    }
+    f->plot_counter = switch_row("Plot the _counter", false);
+    f->do_init = spin_row("Initial output value", 0, 255, 1, 0, 0);
+    f->ao_init[0] = spin_row("Initial DAC 0 (V)", -10, 10, 0.1, 0, 3);
+    f->ao_init[1] = spin_row("Initial DAC 1 (V)", -10, 10, 0.1, 0, 3);
+
+    GtkWidget* conn = expander("Connection", {f->address, f->mac, f->subnet, f->port, f->code,
+                                              f->timeout});
+    GtkWidget* acq = expander("Acquisition", {f->rate, f->ext_clock, f->trigger, f->auto_restart,
+                                              f->calibrate, f->poll_rate});
+    GtkWidget* ai = expander("Analog inputs", {f->mode, f->ai[0], f->ai[1], f->ai[2], f->ai[3],
+                                               f->ai[4], f->ai[5], f->ai[6], f->ai[7]});
+    GtkWidget* dio = expander("Digital I/O", {f->dir[0], f->dir[1], f->dir[2], f->dir[3], f->dir[4],
+                                              f->dir[5], f->dir[6], f->dir[7], f->plot_counter,
+                                              f->do_init, f->ao_init[0], f->ao_init[1]});
+    f->ai_group = ai;
+    adw_expander_row_set_expanded(ADW_EXPANDER_ROW(conn), TRUE);
+
+    for (GtkWidget* w : {f->address, f->mac, f->subnet}) hook_entry(w, f);
+    for (GtkWidget* w : {f->port, f->code, f->timeout, f->rate, f->poll_rate, f->do_init,
+                         f->ao_init[0], f->ao_init[1]})
+      hook_notify(w, "notify::value", f);
+    for (GtkWidget* w : {f->ext_clock, f->auto_restart, f->calibrate, f->plot_counter})
+      hook_notify(w, "notify::active", f);
+    hook_notify(f->trigger, "notify::selected", f);
+    hook_notify(f->mode, "notify::selected", f);
+    for (int i = 0; i < 8; ++i) {
+      hook_notify(f->ai[i], "notify::selected", f);
+      hook_notify(f->dir[i], "notify::selected", f);
+    }
+    f->refresh();
+
+    ConfigForm out;
+    out.rows = {conn, acq, ai, dio};
+    out.description =
+        "Give an address, or a MAC address to look up (optionally with a subnet such as 192.168.1)";
+    out.focus = f->address;
+    out.write = [f](const SourceConfig& c) { f->write(c); };
+    out.read = [f]() { return f->read(); };
+    out.valid = [f]() { return f->valid(); };
+    ChangeCb cb = f->changed;
+    out.on_change = [cb](std::function<void()> fn) { *cb = std::move(fn); };
+    return out;
+  }
+};
+
+}  // namespace
+
+std::unique_ptr<SourceKind> make_e1608_kind() { return std::make_unique<E1608Kind>(); }
+
+}  // namespace udpscope

src/source/e1608/e1608_source.cpp

@@ -0,0 +1,503 @@
+// SPDX-License-Identifier: MIT
+// SPDX-FileCopyrightText: 2026 Martino Ferrari
+
+#include "source/e1608/e1608_source.hpp"
+
+#include <algorithm>
+#include <cctype>
+#include <cstdio>
+#include <stdexcept>
+#include <thread>
+
+namespace udpscope::e1608 {
+
+namespace {
+
+std::string lower(std::string s) {
+  for (char& c : s) c = static_cast<char>(std::tolower(static_cast<unsigned char>(c)));
+  return s;
+}
+
+std::string trim(const std::string& s) {
+  std::size_t a = 0, b = s.size();
+  while (a < b && std::isspace(static_cast<unsigned char>(s[a]))) ++a;
+  while (b > a && std::isspace(static_cast<unsigned char>(s[b - 1]))) --b;
+  return s.substr(a, b - a);
+}
+
+double clamp(double v, double lo, double hi) { return std::min(hi, std::max(lo, v)); }
+
+double wave(AoMode m, double f) {
+  switch (m) {
+    case AoMode::Sine: return std::sin(2.0 * M_PI * f);
+    case AoMode::Square: return f < 0.5 ? 1.0 : -1.0;
+    case AoMode::Triangle: return f < 0.5 ? 4.0 * f - 1.0 : 3.0 - 4.0 * f;
+    case AoMode::Ramp: return 2.0 * f - 1.0;
+    case AoMode::Level: break;
+  }
+  return 0.0;
+}
+
+double epoch_seconds() {
+  return std::chrono::duration<double>(std::chrono::system_clock::now().time_since_epoch()).count();
+}
+
+}  // namespace
+
+// --- E1608Settings --------------------------------------------------------------------------
+
+E1608Settings E1608Settings::from(const SourceConfig& c) {
+  E1608Settings s;
+  s.address = trim(c.get("address"));
+  s.mac_text = trim(c.get("mac"));
+  s.subnet = trim(c.get("subnet"));
+  s.port = static_cast<int>(c.get_int("port", kDefaultPort));
+  const long code = c.get_int("code", 0);
+  s.code = static_cast<std::uint32_t>(std::min(4294967295L, std::max(0L, code)));
+  s.timeout_ms = static_cast<int>(c.get_int("timeout_ms", 2000));
+  s.scan_rate = c.get_double("scan_rate", 1000.0);
+  s.ext_clock = c.get_bool("ext_clock", false);
+  parse_trigger(c.get("trigger", "none"), s.trigger);
+  s.auto_restart = c.get_bool("auto_restart", true);
+  s.calibrate = c.get_bool("calibrate", true);
+  s.poll_rate = c.get_double("poll_rate", 10.0);
+  s.diff = lower(c.get("mode", "se")) == "diff";
+
+  bool any_key = false;
+  for (int i = 0; i < 8; ++i) {
+    const std::string key = "ai" + std::to_string(i);
+    if (!c.has(key)) continue;
+    any_key = true;
+    Range r;
+    if (parse_range(c.get(key), r)) {
+      s.ai_on[i] = true;
+      s.ai_range[i] = r;
+    }
+  }
+  if (!any_key) {  // a bare `--source-opt address=...` is connectable
+    s.ai_on[0] = true;
+    s.ai_range[0] = Range::Bip10V;
+  }
+  if (s.diff) {
+    for (int i = 4; i < 8; ++i) s.ai_on[i] = false;
+  }
+
+  for (int i = 0; i < 8; ++i) {
+    const std::string d = lower(c.get("dir" + std::to_string(i), "i"));
+    s.dir_out[i] = d == "o" || d == "out" || d == "output";
+    s.plot[i] = c.get_bool("plot" + std::to_string(i), false) && !s.dir_out[i];
+  }
+  s.plot_counter = c.get_bool("plot_counter", false);
+  s.do_init = static_cast<int>(c.get_int("do_init", 0) & 0xFF);
+  for (int i = 0; i < 2; ++i)
+    s.ao_init[i] = clamp(c.get_double("ao" + std::to_string(i) + "_init", 0.0), -10.0, 10.0);
+  return s;
+}
+
+int E1608Settings::active_channels() const {
+  int n = 0;
+  for (int i = 0; i < 8; ++i) n += ai_on[i] ? 1 : 0;
+  return n;
+}
+
+std::vector<QueueEntry> E1608Settings::queue() const {
+  std::vector<QueueEntry> q;
+  for (int i = 0; i < 8; ++i) {
+    if (ai_on[i]) q.push_back({static_cast<std::uint8_t>(diff ? i + 8 : i), ai_range[i]});
+  }
+  return q;
+}
+
+std::string E1608Settings::identity() const {
+  if (!address.empty()) return address;
+  Mac m;
+  return parse_mac(mac_text, m) ? format_mac(m) : mac_text;
+}
+
+std::string E1608Settings::problem() const {
+  Mac m;
+  if (address.empty() && mac_text.empty()) return "Enter an address, or a MAC address to look up";
+  if (!mac_text.empty() && !parse_mac(mac_text, m))
+    return "The MAC address must be six hex bytes, such as 00:80:2f:00:16:08";
+  if (address.empty() && !subnet.empty() && sweep_targets(subnet).empty())
+    return "The subnet must look like 192.168.1";
+  if (port < 1 || port > 65534) return "The port must be between 1 and 65534";
+  if (timeout_ms < 100 || timeout_ms > 60000) return "The timeout must be 100 to 60000 ms";
+  if (!(scan_rate > 0.0)) return "The scan rate must be positive";
+  if (active_channels() < 1) return "Enable at least one analog input";
+  if (aggregate() > kMaxAggregateHz) {
+    char b[96];
+    std::snprintf(b, sizeof b, "%d channels at %.6g S/s is %.6g kS/s; the board takes 250 kS/s",
+                  active_channels(), scan_rate, aggregate() / 1000.0);
+    return b;
+  }
+  if (aggregate() < kMinAggregateHz) return "The aggregate rate must be at least 0.02 S/s";
+  if (!(poll_rate >= 0.1 && poll_rate <= 100.0)) return "The input poll rate must be 0.1 to 100 Hz";
+  return {};
+}
+
+// --- E1608Source ----------------------------------------------------------------------------
+
+E1608Source::E1608Source(const SourceConfig& cfg) : cfg_(E1608Settings::from(cfg)) {
+  queue_ = cfg_.queue();
+  block_ = static_cast<std::size_t>(clamp(std::round(cfg_.scan_rate / 100.0), 1.0, 4096.0));
+  rx_.resize(65536);
+  ready_.resize(queue_.size());
+
+  auto base = [&](const std::string& name, const char* unit, double lo, double hi) {
+    SignalInfo s;
+    s.name = name;
+    s.unit = unit;
+    s.type_code = 9;  // float64
+    s.rows = 1;
+    s.cols = 1;
+    s.range_min = lo;
+    s.range_max = hi;
+    s.time_mode = kTimePacket;
+    s.sampling_rate = cfg_.scan_rate;
+    return s;
+  };
+  for (int i = 0; i < 8; ++i) {
+    if (!cfg_.ai_on[i]) continue;
+    const double fs = full_scale(cfg_.ai_range[i]);
+    signals_.push_back(base("AI" + std::to_string(i), "V", -fs, fs));
+  }
+  for (int b = 0; b < 8; ++b) {
+    if (!cfg_.plot[b]) continue;
+    plotted_bits_.push_back(b);
+    signals_.push_back(base("DI" + std::to_string(b), "", 0.0, 1.0));
+  }
+  if (cfg_.plot_counter) signals_.push_back(base("Counter", "counts", 0.0, 1000.0));
+  held_di_.assign(plotted_bits_.size(), 0.0);
+}
+
+std::string E1608Source::last_message() const {
+  std::lock_guard<std::mutex> lk(msg_mu_);
+  return last_message_;
+}
+
+void E1608Source::open(SourceSink& sink) {
+  const std::string p = cfg_.problem();
+  if (!p.empty()) throw std::runtime_error(p);
+  sink_ = &sink;
+  state_ = State::Down;
+  retry_at_ = Clock::now();
+}
+
+void E1608Source::close() {
+  if (state_ == State::Scanning) dev_.scan_stop();
+  link_.close();
+  connected_.store(false, std::memory_order_relaxed);
+}
+
+SourceStats E1608Source::stats() const {
+  SourceStats s;
+  s.packets = packets_;
+  s.frames = frames_;
+  s.reconnects = reconnects_;
+  s.connected = connected_.load(std::memory_order_relaxed);
+  return s;
+}
+
+void E1608Source::note(const std::string& msg) {
+  {
+    std::lock_guard<std::mutex> lk(msg_mu_);
+    last_message_ = msg;
+  }
+  if (msg != last_logged_) {  // a retry every 2 s must not flood stderr
+    last_logged_ = msg;
+    std::fprintf(stderr, "[e1608] %s\n", msg.c_str());
+    std::fflush(stderr);
+  }
+}
+
+void E1608Source::fail(const std::string& msg, bool fatal) {
+  link_.close();
+  const bool was_up = connected_.exchange(false, std::memory_order_relaxed);
+  note(msg);
+  if (was_up) sink_->on_event(SourceEvent::Disconnected, msg.c_str());
+  sink_->on_event(SourceEvent::Error, msg.c_str());
+  any_gen_ = false;
+  readouts_->gen_on[0].store(false);
+  readouts_->gen_on[1].store(false);
+  if (fatal) {
+    state_ = State::Halted;
+  } else {
+    state_ = State::Down;
+    retry_at_ = Clock::now() + std::chrono::seconds(2);
+  }
+}
+
+void E1608Source::poll(int timeout_ms) {
+  if (sink_ == nullptr) return;
+  try {
+    drain_commands();
+    switch (state_) {
+      case State::Down: {
+        const auto now = Clock::now();
+        if (now < retry_at_) {
+          const auto left = std::chrono::duration_cast<std::chrono::milliseconds>(retry_at_ - now);
+          std::this_thread::sleep_for(std::min(left, std::chrono::milliseconds(timeout_ms)));
+          return;
+        }
+        connect_and_start();
+        break;
+      }
+      case State::Scanning:
+        scan_step(timeout_ms);
+        break;
+      case State::Halted:
+        std::this_thread::sleep_for(std::chrono::milliseconds(timeout_ms));
+        break;
+    }
+  } catch (const LinkError& e) {
+    fail(e.what(), e.kind() == LinkError::Kind::Rejected);
+  } catch (const std::exception& e) {
+    fail(e.what(), false);
+  }
+}
+
+void E1608Source::connect_and_start() {
+  std::string host = cfg_.address;
+  Mac want{};
+  const bool have_mac = !cfg_.mac_text.empty() && parse_mac(cfg_.mac_text, want);
+  if (host.empty() || have_mac) {
+    const std::vector<std::string> targets =
+        host.empty() ? sweep_targets(cfg_.subnet) : std::vector<std::string>{host};
+    const std::vector<DeviceInfo> found =
+        discover(targets, cfg_.port, std::min(cfg_.timeout_ms, 1500));
+    if (host.empty()) {
+      for (const DeviceInfo& d : found) {
+        if (d.mac == want) host = d.ip;
+      }
+      if (host.empty())
+        throw LinkError(LinkError::Kind::Io,
+                        "no E-1608 with MAC " + format_mac(want) + " answered");
+    } else {
+      if (found.empty()) throw LinkError(LinkError::Kind::Io, "no E-1608 answered at " + host);
+      if (found[0].mac != want)
+        throw LinkError(LinkError::Kind::Protocol, "the board at " + host + " has MAC " +
+                                                       format_mac(found[0].mac) + ", not " +
+                                                       format_mac(want));
+    }
+  }
+
+  link_.connect(host, cfg_.port, cfg_.code, cfg_.timeout_ms);
+  cal_ = Calibration{};
+  if (cfg_.calibrate && !dev_.read_calibration(cal_))
+    note("calibration memory refused; using nominal values");
+
+  std::uint8_t inputs = 0;
+  for (int b = 0; b < 8; ++b) {
+    if (!cfg_.dir_out[b]) inputs = static_cast<std::uint8_t>(inputs | (1u << b));
+  }
+  dev_.dconf(inputs);
+  do_latch_ = static_cast<std::uint8_t>(cfg_.do_init);
+  dev_.dout(do_latch_);
+  readouts_->do_latch.store(do_latch_);
+  for (int ch = 0; ch < 2; ++ch) {
+    ao_[ch] = AoSettings{};
+    ao_[ch].level = cfg_.ao_init[ch];
+    write_ao(ch, cfg_.ao_init[ch]);
+  }
+  any_gen_ = false;
+  update_gen_flags();
+
+  start_scan();
+
+  pending_.clear();
+  for (auto& r : ready_) r.clear();
+  pending_lost_ = 0;
+  const auto now = Clock::now();
+  next_slow_ = now;
+  next_status_ = now + std::chrono::milliseconds(500);
+  next_gen_ = now + std::chrono::milliseconds(20);
+  slow_read();
+
+  state_ = State::Scanning;
+  connected_.store(true, std::memory_order_relaxed);
+  if (connects_.fetch_add(1) > 0) ++reconnects_;
+  readouts_->epoch.fetch_add(1);
+  last_logged_.clear();
+  sink_->on_config(signals_, 1);
+  sink_->on_event(SourceEvent::Connected, "");
+}
+
+int E1608Source::wait_ms(int timeout_ms, Clock::time_point now) const {
+  Clock::time_point next = std::min(next_slow_, next_status_);
+  if (any_gen_) next = std::min(next, next_gen_);
+  const long long d = std::chrono::duration_cast<std::chrono::milliseconds>(next - now).count();
+  return static_cast<int>(std::max(0LL, std::min<long long>(d, timeout_ms)));
+}
+
+void E1608Source::scan_step(int timeout_ms) {
+  const std::size_t n = link_.read_scan(rx_.data(), rx_.size(), wait_ms(timeout_ms, Clock::now()));
+  if (n > 0) {
+    ++packets_;
+    pending_.insert(pending_.end(), rx_.begin(), rx_.begin() + static_cast<std::ptrdiff_t>(n));
+    const std::size_t nq = queue_.size();
+    const std::size_t scans = pending_.size() / (2 * nq);
+    if (scans > 0) {
+      std::vector<std::uint16_t> raw(scans * nq);
+      for (std::size_t i = 0; i < raw.size(); ++i)
+        raw[i] = static_cast<std::uint16_t>(pending_[2 * i] | (pending_[2 * i + 1] << 8));
+      pending_.erase(pending_.begin(),
+                     pending_.begin() + static_cast<std::ptrdiff_t>(scans * nq * 2));
+      decode_scans(raw.data(), scans, queue_, cal_, dec_);
+      for (std::size_t q = 0; q < nq; ++q)
+        ready_[q].insert(ready_[q].end(), dec_[q].begin(), dec_[q].end());
+    }
+  }
+  while (!ready_.empty() && ready_[0].size() >= block_) emit_frame(block_);
+  run_timers();
+}
+
+void E1608Source::emit_frame(std::size_t count) {
+  const std::size_t nq = queue_.size();
+  const std::size_t total = signals_.size();
+  vals_.resize(total);
+  views_.resize(total);
+  for (std::size_t q = 0; q < nq; ++q) {
+    vals_[q].assign(ready_[q].begin(), ready_[q].begin() + static_cast<std::ptrdiff_t>(count));
+    ready_[q].erase(ready_[q].begin(), ready_[q].begin() + static_cast<std::ptrdiff_t>(count));
+  }
+  for (std::size_t k = 0; k < plotted_bits_.size(); ++k) vals_[nq + k].assign(count, held_di_[k]);
+  if (cfg_.plot_counter) vals_[total - 1].assign(count, held_counter_);
+  for (std::size_t i = 0; i < total; ++i)
+    views_[i] = SignalValues{vals_[i].data(), static_cast<std::uint32_t>(count)};
+
+  Frame f;
+  f.counter = frame_counter_++;
+  f.lost = pending_lost_;
+  pending_lost_ = 0;
+  f.late = false;
+  f.recv_time = epoch_seconds();
+  f.publish_mode = 1;
+  f.num_samples = static_cast<std::uint32_t>(count);
+  f.num_signals = static_cast<std::uint32_t>(total);
+  f.signals = signals_.data();
+  f.values = views_.data();
+  ++frames_;
+  sink_->on_data(f);
+}
+
+void E1608Source::run_timers() {
+  const auto now = Clock::now();
+  if (now >= next_slow_) {
+    next_slow_ = now + std::chrono::duration_cast<Clock::duration>(
+                           std::chrono::duration<double>(1.0 / cfg_.poll_rate));
+    slow_read();
+  }
+  if (now >= next_status_) {
+    next_status_ = now + std::chrono::milliseconds(500);
+    check_status();
+  }
+  if (any_gen_ && now >= next_gen_) {
+    next_gen_ = now + std::chrono::milliseconds(20);
+    generator_tick(now);
+  }
+}
+
+void E1608Source::slow_read() {
+  const std::uint8_t d = dev_.din();
+  const std::uint32_t c = dev_.counter();
+  readouts_->di.store(d);
+  readouts_->counter.store(c);
+  for (std::size_t k = 0; k < plotted_bits_.size(); ++k)
+    held_di_[k] = (d >> plotted_bits_[k]) & 1 ? 1.0 : 0.0;
+  held_counter_ = static_cast<double>(c);
+}
+
+void E1608Source::check_status() {
+  const std::uint16_t st = dev_.status();
+  if (st & 4) {
+    if (!cfg_.auto_restart) {
+      fail("scan overrun; the scan is stopped (auto-restart is off)", true);
+      return;
+    }
+    note("scan overrun; restarting");
+    sink_->on_event(SourceEvent::Error, "scan overrun; restarting");
+    restart_scan();
+    pending_lost_ = 1;
+  } else if (!(st & 2) && cfg_.auto_restart) {
+    restart_scan();
+  }
+}
+
+void E1608Source::restart_scan() {
+  dev_.scan_stop();
+  pending_.clear();
+  for (auto& r : ready_) r.clear();
+  start_scan();
+}
+
+void E1608Source::start_scan() {
+  // The device layer does not validate the rate; the board's cap is enforced here.
+  if (cfg_.aggregate() > kMaxAggregateHz)
+    throw LinkError(LinkError::Kind::Command, "the aggregate rate exceeds 250 kS/s");
+  try {
+    dev_.scan_start(queue_, cfg_.aggregate(), cfg_.ext_clock, cfg_.trigger);
+  } catch (...) {
+    // The board may have accepted the start (0x11) before a later step failed (the filler byte):
+    // tell it to stop, so it is not left scanning with nobody reading.
+    try {
+      dev_.scan_stop();
+    } catch (...) {
+    }
+    throw;
+  }
+}
+
+void E1608Source::write_ao(int ch, double volts) {
+  readouts_->ao[ch].store(dev_.aout(ch, volts, cal_));
+}
+
+void E1608Source::update_gen_flags() {
+  any_gen_ = false;
+  for (int ch = 0; ch < 2; ++ch) {
+    const bool on = ao_[ch].mode != AoMode::Level && ao_[ch].run;
+    readouts_->gen_on[ch].store(on);
+    any_gen_ = any_gen_ || on;
+  }
+}
+
+void E1608Source::generator_tick(Clock::time_point now) {
+  for (int ch = 0; ch < 2; ++ch) {
+    const AoSettings& s = ao_[ch];
+    if (s.mode == AoMode::Level || !s.run) continue;
+    const double t = std::chrono::duration<double>(now - gen_t0_[ch]).count();
+    const double phase = std::fmod(t * s.freq_hz, 1.0);
+    write_ao(ch, clamp(s.offset + s.amplitude * wave(s.mode, phase), -10.0, 10.0));
+  }
+}
+
+void E1608Source::set_output_bit(int bit, bool on) {
+  if (state_ != State::Scanning || bit < 0 || bit > 7 || !cfg_.dir_out[bit]) return;
+  do_latch_ = static_cast<std::uint8_t>(on ? (do_latch_ | (1u << bit)) : (do_latch_ & ~(1u << bit)));
+  dev_.dout(do_latch_);
+  readouts_->do_latch.store(do_latch_);
+}
+
+void E1608Source::reset_counter() {
+  if (state_ != State::Scanning) return;
+  dev_.counter_reset();
+  held_counter_ = 0.0;
+  readouts_->counter.store(0);
+}
+
+void E1608Source::set_analog_out(int ch, const AoSettings& s) {
+  if (state_ != State::Scanning || ch < 0 || ch > 1) return;
+  ao_[ch] = s;
+  ao_[ch].freq_hz = clamp(s.freq_hz, 0.1, 10.0);
+  ao_[ch].amplitude = clamp(s.amplitude, 0.0, 10.0);
+  ao_[ch].offset = clamp(s.offset, -10.0, 10.0);
+  if (s.mode == AoMode::Level) {
+    ao_[ch].run = false;
+    write_ao(ch, clamp(s.level, -10.0, 10.0));
+  } else if (s.run) {
+    gen_t0_[ch] = Clock::now();
+  }
+  update_gen_flags();
+}
+
+}  // namespace udpscope::e1608

src/source/e1608/e1608_source.hpp

@@ -0,0 +1,163 @@
+// SPDX-License-Identifier: MIT
+// SPDX-FileCopyrightText: 2026 Martino Ferrari
+
+#pragma once
+
+#include "source/data_source.hpp"
+#include "source/e1608/e1608_device.hpp"
+#include "source/e1608/e1608_link.hpp"
+#include "source/source_config.hpp"
+
+#include <atomic>
+#include <chrono>
+#include <cmath>
+#include <cstdint>
+#include <memory>
+#include <mutex>
+#include <string>
+#include <vector>
+
+namespace udpscope::e1608 {
+
+constexpr double kMaxAggregateHz = 250000.0;
+constexpr double kMinAggregateHz = 0.02;
+
+/// Everything a configured E-1608 source needs, parsed once from the flat `SourceConfig`.
+struct E1608Settings {
+  std::string address;   ///< host or IP; empty means "find it by MAC"
+  std::string mac_text;  ///< as typed; empty means "any board at `address`"
+  std::string subnet;    ///< `a.b.c` to sweep when looking up a MAC; empty means broadcast
+  int port = kDefaultPort;
+  std::uint32_t code = 0;
+  int timeout_ms = 2000;
+  double scan_rate = 1000.0;  ///< per channel; with an external clock, that clock's rate
+  bool ext_clock = false;
+  Trigger trigger = Trigger::None;
+  bool auto_restart = true;
+  bool calibrate = true;
+  double poll_rate = 10.0;  ///< Hz, for the digital inputs and the counter
+  bool diff = false;
+  bool ai_on[8] = {};
+  Range ai_range[8] = {};
+  bool dir_out[8] = {};
+  bool plot[8] = {};
+  bool plot_counter = false;
+  int do_init = 0;
+  double ao_init[2] = {0.0, 0.0};
+
+  static E1608Settings from(const SourceConfig& cfg);
+  /// The first reason this cannot connect, in words for the user; empty when it can.
+  std::string problem() const;
+  int active_channels() const;
+  std::vector<QueueEntry> queue() const;
+  double aggregate() const { return scan_rate * active_channels(); }
+  /// The address, else the MAC; names the session file.
+  std::string identity() const;
+};
+
+/// What the panels read. Written by the producer thread, read by the GTK thread's timer.
+struct E1608Readouts {
+  std::atomic<unsigned> di{0};          ///< the DIN byte as last read
+  std::atomic<unsigned> do_latch{0};    ///< the DOUT byte as last written
+  std::atomic<std::uint32_t> counter{0};
+  std::atomic<double> ao[2];            ///< the last DAC voltage written; NaN before any
+  std::atomic<bool> gen_on[2];
+  std::atomic<unsigned> epoch{0};       ///< bumped at each (re)connect: panels re-sync from it
+  E1608Readouts() {
+    for (int i = 0; i < 2; ++i) {
+      ao[i].store(std::nan(""));
+      gen_on[i].store(false);
+    }
+  }
+};
+
+enum class AoMode { Level, Sine, Square, Triangle, Ramp };
+
+struct AoSettings {
+  AoMode mode = AoMode::Level;
+  double level = 0.0;      ///< Level mode, volts
+  double amplitude = 1.0;  ///< waveform modes, volts
+  double offset = 0.0;
+  double freq_hz = 1.0;    ///< clamped to 0.1 .. 10
+  bool run = false;        ///< waveform modes only
+};
+
+class E1608Source final : public DataSource {
+ public:
+  explicit E1608Source(const SourceConfig& cfg);
+
+  void open(SourceSink& sink) override;
+  void poll(int timeout_ms) override;
+  void close() override;
+  SourceStats stats() const override;
+  bool connected() const override { return connected_.load(std::memory_order_relaxed); }
+
+  // Producer thread only: reach these through `post()`. Each is ignored unless scanning.
+  void set_output_bit(int bit, bool on);
+  void reset_counter();
+  void set_analog_out(int ch, const AoSettings& s);
+
+  std::shared_ptr<E1608Readouts> readouts() const { return readouts_; }
+  const E1608Settings& settings() const { return cfg_; }
+  unsigned connects() const { return connects_.load(std::memory_order_relaxed); }
+  std::string last_message() const;
+
+ private:
+  using Clock = std::chrono::steady_clock;
+  enum class State { Down, Scanning, Halted };
+
+  void connect_and_start();
+  void scan_step(int timeout_ms);
+  void emit_frame(std::size_t count);
+  void run_timers();
+  void slow_read();
+  void check_status();
+  void restart_scan();
+  void start_scan();
+  void generator_tick(Clock::time_point now);
+  void write_ao(int ch, double volts);
+  void update_gen_flags();
+  void fail(const std::string& msg, bool fatal);
+  void note(const std::string& msg);
+  int wait_ms(int timeout_ms, Clock::time_point now) const;
+
+  E1608Settings cfg_;
+  SourceSink* sink_ = nullptr;
+  State state_ = State::Down;
+  Link link_;
+  Device dev_{link_};
+  Calibration cal_;
+  std::vector<QueueEntry> queue_;
+  std::vector<SignalInfo> signals_;
+  std::vector<int> plotted_bits_;  ///< the DI bits that are plot channels, ascending
+  std::size_t block_ = 10;
+
+  std::shared_ptr<E1608Readouts> readouts_ = std::make_shared<E1608Readouts>();
+  std::atomic<bool> connected_{false};
+  std::atomic<unsigned> connects_{0};
+  mutable std::mutex msg_mu_;
+  std::string last_message_;
+  std::string last_logged_;
+
+  std::vector<std::uint8_t> rx_;       ///< the read buffer
+  std::vector<std::uint8_t> pending_;  ///< bytes of an incomplete scan
+  std::vector<std::vector<double>> ready_;  ///< decoded, not yet framed: [queue index][scan]
+  std::vector<std::vector<double>> dec_;
+  std::vector<std::vector<double>> vals_;   ///< the frame being built: [signal][sample]
+  std::vector<SignalValues> views_;
+  std::vector<double> held_di_;
+  double held_counter_ = 0.0;
+  std::uint8_t do_latch_ = 0;
+
+  AoSettings ao_[2];
+  Clock::time_point gen_t0_[2];
+  bool any_gen_ = false;
+
+  Clock::time_point retry_at_;
+  Clock::time_point next_slow_, next_status_, next_gen_;
+  std::uint32_t frame_counter_ = 0;
+  std::uint32_t pending_lost_ = 0;
+  std::uint64_t packets_ = 0, frames_ = 0, reconnects_ = 0;
+};
+
+}  // namespace udpscope::e1608