Added an expanded connectivity view with MQTT message history
This commit is contained in:
parent
6c1941cc80
commit
bc4d2ec002
GPG Key ID:
D4A6C394DF13F1D7
|
|
@ -61,6 +61,8 @@ add_library(fgc_core STATIC
|
|||
src/core/CalibrationRoutine.cpp
|
||||
src/core/MtiProtocol.cpp
|
||||
src/core/CameraMode.cpp
|
||||
src/core/ChannelHistory.cpp
|
||||
src/core/ChannelTopics.cpp
|
||||
src/sensors/Sht41Protocol.cpp
|
||||
src/ui/UiSnapshot.cpp
|
||||
src/ui/Scroll.cpp
|
||||
|
|
|
|||
|
|
@ -0,0 +1,53 @@
|
|||
#pragma once
|
||||
|
||||
#include "fgc/IControlChannel.h"
|
||||
|
||||
#include <cstddef>
|
||||
#include <deque>
|
||||
#include <mutex>
|
||||
#include <string>
|
||||
#include <vector>
|
||||
|
||||
namespace fgc {
|
||||
|
||||
// A bounded, thread-safe record of recent control-channel traffic, plus the
|
||||
// cumulative counters behind the connectivity view.
|
||||
//
|
||||
// It lives here rather than inside MqttControlChannel for two reasons: the null
|
||||
// channel keeps the same history (so the view works with --no-mqtt), and holding
|
||||
// no MQTT types means the counter logic is unit-testable with WITH_MQTT=OFF.
|
||||
//
|
||||
// Its mutex is its own, deliberately not MqttControlChannel's: that one guards
|
||||
// sub_data_ and is taken by poll() on the control thread, while record() is
|
||||
// called from the Paho callback thread and from publish() on both the control
|
||||
// and ImagePipeline worker threads. Separate locks also keep the history lock
|
||||
// off the 3-second publish wait.
|
||||
class ChannelHistory {
|
||||
public:
|
||||
// Newest-first eviction bound. Enough to cover a scan's worth of waypoints
|
||||
// without making the per-tick snapshot copy expensive.
|
||||
static constexpr std::size_t kCapacity = 40;
|
||||
|
||||
// Anything but Delivery::Ok on an outbound message means it never reached the
|
||||
// broker. Those are the entries worth looking for.
|
||||
void record(bool outbound, Delivery delivery, std::string topic, std::string payload);
|
||||
|
||||
void noteConnect();
|
||||
void noteLoss();
|
||||
|
||||
// What actually went out on the StatusCode topic. The connectivity view used
|
||||
// to infer this from the live control code, which assumes the echo rather
|
||||
// than reporting the send.
|
||||
void noteStatus(std::string payload);
|
||||
|
||||
// Oldest first, so the view can print it top-to-bottom as it happened.
|
||||
std::vector<ChannelMessage> recent() const;
|
||||
ChannelStats stats() const;
|
||||
|
||||
private:
|
||||
mutable std::mutex mutex_;
|
||||
std::deque<ChannelMessage> ring_;
|
||||
ChannelStats stats_;
|
||||
};
|
||||
|
||||
} // namespace fgc
|
||||
|
|
@ -0,0 +1,29 @@
|
|||
#pragma once
|
||||
|
||||
#include "fgc/IControlChannel.h"
|
||||
|
||||
#include <string>
|
||||
|
||||
namespace fgc {
|
||||
|
||||
// Topic names and payload encoding for the control channel, kept here rather
|
||||
// than inside MqttControlChannel so the null channel produces byte-identical
|
||||
// output (its history has to show what WOULD have been sent, not an
|
||||
// approximation) and so the JSON is unit-testable without a broker.
|
||||
//
|
||||
// Scheme: GGS/FWT/<tower>/<leaf> — see docs/mqtt-api.md.
|
||||
|
||||
// Leaf names, so the two channels and the UI agree on spelling.
|
||||
inline constexpr const char* kTopicTargetHdg = "target_HDG";
|
||||
inline constexpr const char* kTopicControlCode = "ControlCode";
|
||||
inline constexpr const char* kTopicStatusCode = "StatusCode";
|
||||
inline constexpr const char* kTopicCamEvent = "CamEvent";
|
||||
inline constexpr const char* kTopicEnv = "Env";
|
||||
|
||||
std::string towerTopic(const std::string& tower, const std::string& leaf);
|
||||
|
||||
// Payload builders. One definition each, used by both channels.
|
||||
std::string camEventJson(const CamEvent& e);
|
||||
std::string envJson(const EnvEvent& e);
|
||||
|
||||
} // namespace fgc
|
||||
|
|
@ -1,6 +1,7 @@
|
|||
#pragma once
|
||||
|
||||
#include <string>
|
||||
#include <vector>
|
||||
|
||||
namespace fgc {
|
||||
|
||||
|
|
@ -30,6 +31,32 @@ struct EnvEvent {
|
|||
long long timestamp_ms = 0;
|
||||
};
|
||||
|
||||
// What became of an outbound message. Dropped and Failed are separated because
|
||||
// they mean different things operationally: Dropped is "the link was down and we
|
||||
// discarded it silently", Failed is "the broker or library rejected it".
|
||||
// Inbound messages are always Ok.
|
||||
enum class Delivery { Ok, Dropped, Failed };
|
||||
|
||||
// One message that crossed the control channel, for the connectivity view.
|
||||
struct ChannelMessage {
|
||||
long long timestamp_ms = 0;
|
||||
bool outbound = true; // PUB (ours) vs RX (broker's)
|
||||
Delivery delivery = Delivery::Ok;
|
||||
std::string topic;
|
||||
std::string payload;
|
||||
};
|
||||
|
||||
// Cumulative channel counters since start-up.
|
||||
struct ChannelStats {
|
||||
long long published_ok = 0;
|
||||
long long published_dropped = 0; // attempted while disconnected: silently discarded
|
||||
long long published_failed = 0; // the broker/library rejected it
|
||||
long long received = 0;
|
||||
long long connects = 0; // successful (re)connections
|
||||
long long last_connect_ms = 0, last_loss_ms = 0;
|
||||
std::string last_status_payload; // what actually went out on StatusCode
|
||||
};
|
||||
|
||||
// Abstraction over the remote control/telemetry channel. Implemented by
|
||||
// MqttControlChannel (Eclipse Paho) and NullControlChannel (no broker; used
|
||||
// for development - publishes are dropped and poll() yields a default).
|
||||
|
|
@ -49,6 +76,11 @@ public:
|
|||
// Latest control input; clears the *_available flags so each update is
|
||||
// acted on once.
|
||||
virtual ControlCommand poll() = 0;
|
||||
|
||||
// Recent traffic and counters for the connectivity ('m') view. Default: none,
|
||||
// so an implementation that keeps no history needs no changes.
|
||||
virtual std::vector<ChannelMessage> recentMessages() const { return {}; }
|
||||
virtual ChannelStats channelStats() const { return {}; }
|
||||
};
|
||||
|
||||
} // namespace fgc
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
#pragma once
|
||||
|
||||
#include "fgc/ChannelHistory.h"
|
||||
#include "fgc/IControlChannel.h"
|
||||
|
||||
#include <atomic>
|
||||
|
|
@ -37,6 +38,9 @@ public:
|
|||
void publishEnv(const EnvEvent& event) override;
|
||||
ControlCommand poll() override;
|
||||
|
||||
std::vector<ChannelMessage> recentMessages() const override { return history_.recent(); }
|
||||
ChannelStats channelStats() const override { return history_.stats(); }
|
||||
|
||||
private:
|
||||
// mqtt::callback
|
||||
void connected(const std::string& cause) override;
|
||||
|
|
@ -48,6 +52,7 @@ private:
|
|||
|
||||
void publish(const std::string& topic, const std::string& payload);
|
||||
|
||||
ChannelHistory history_;
|
||||
std::string broker_uri_;
|
||||
std::string tower_;
|
||||
std::string topic_target_hdg_;
|
||||
|
|
|
|||
|
|
@ -11,8 +11,18 @@ namespace fgc {
|
|||
// Simulated ambient sensor for development without hardware: synthesizes a
|
||||
// slowly varying, plausible temp/humidity reading so the Sensors panel
|
||||
// animates without an SHT41 attached.
|
||||
//
|
||||
// It holds each reading for `period`, exactly as the real driver does (which
|
||||
// only refreshes latest/timestamp_ms once per [Env] period_ms). Stamping a new
|
||||
// timestamp on every sample() call would make every ~10 ms control tick look
|
||||
// like a fresh reading, and Application's "publish once per fresh sample" check
|
||||
// keys on that timestamp - so the mock would publish Env hundreds of times a
|
||||
// second where the real sensor publishes once every two seconds.
|
||||
class MockEnvSensor : public IEnvSensor {
|
||||
public:
|
||||
explicit MockEnvSensor(std::chrono::milliseconds period = std::chrono::milliseconds(2000))
|
||||
: period_(period) {}
|
||||
|
||||
void start() override {
|
||||
start_ = clock::now();
|
||||
LOG_INFO << "[mock] env sensor started";
|
||||
|
|
@ -22,22 +32,28 @@ public:
|
|||
bool connected() const override { return true; }
|
||||
|
||||
std::optional<EnvSample> sample() override {
|
||||
const double t = std::chrono::duration<double>(clock::now() - start_).count();
|
||||
EnvSample s;
|
||||
s.valid = true;
|
||||
s.temp_c = static_cast<float>(22.0 + 3.0 * std::sin(t * 0.02));
|
||||
s.humidity_pct = static_cast<float>(45.0 + 10.0 * std::sin(t * 0.013 + 1.0));
|
||||
s.timestamp_ms = std::chrono::duration_cast<std::chrono::milliseconds>(
|
||||
std::chrono::system_clock::now().time_since_epoch())
|
||||
.count();
|
||||
return s;
|
||||
const auto now = clock::now();
|
||||
if (latest_.valid && (now - taken_) < period_) return latest_; // same reading as before
|
||||
|
||||
const double t = std::chrono::duration<double>(now - start_).count();
|
||||
latest_.valid = true;
|
||||
latest_.temp_c = static_cast<float>(22.0 + 3.0 * std::sin(t * 0.02));
|
||||
latest_.humidity_pct = static_cast<float>(45.0 + 10.0 * std::sin(t * 0.013 + 1.0));
|
||||
latest_.timestamp_ms = std::chrono::duration_cast<std::chrono::milliseconds>(
|
||||
std::chrono::system_clock::now().time_since_epoch())
|
||||
.count();
|
||||
taken_ = now;
|
||||
return latest_;
|
||||
}
|
||||
|
||||
std::string name() const override { return "SHT41 (mock)"; }
|
||||
|
||||
private:
|
||||
using clock = std::chrono::steady_clock;
|
||||
clock::time_point start_ = clock::now();
|
||||
clock::time_point start_ = clock::now();
|
||||
std::chrono::milliseconds period_;
|
||||
EnvSample latest_;
|
||||
clock::time_point taken_{};
|
||||
};
|
||||
|
||||
} // namespace fgc
|
||||
|
|
|
|||
|
|
@ -1,34 +1,63 @@
|
|||
#pragma once
|
||||
|
||||
#include "fgc/ChannelHistory.h"
|
||||
#include "fgc/ChannelTopics.h"
|
||||
#include "fgc/IControlChannel.h"
|
||||
#include "fgc/Logger.h"
|
||||
|
||||
#include <string>
|
||||
#include <utility>
|
||||
|
||||
namespace fgc {
|
||||
|
||||
// No-op control channel for running without an MQTT broker. Publishes are
|
||||
// dropped (logged at debug) and poll() never supplies a control command, so
|
||||
// the scheduler stays in automatic-sweep mode (ControlCode 0).
|
||||
// No-op control channel for running without an MQTT broker. Publishes go
|
||||
// nowhere and poll() never supplies a control command, so the scheduler stays
|
||||
// in automatic-sweep mode (ControlCode 0).
|
||||
//
|
||||
// It still records what it was asked to send, using the same topic and payload
|
||||
// builders as the real channel, so the connectivity view works under --no-mqtt
|
||||
// and shows exactly what would have gone out. Every entry is Delivery::Dropped,
|
||||
// which is the truth: nothing was sent.
|
||||
class NullControlChannel : public IControlChannel {
|
||||
public:
|
||||
explicit NullControlChannel(std::string tower = "") : tower_(std::move(tower)) {}
|
||||
|
||||
bool connect() override { return true; }
|
||||
void disconnect() override {}
|
||||
bool connected() const override { return true; }
|
||||
|
||||
void publishStatus(int code) override {
|
||||
const std::string payload = std::to_string(code);
|
||||
history_.noteStatus(payload);
|
||||
LOG_TRACE_CAT(LogCat::Mqtt) << "PUB StatusCode " << code << " (null channel)";
|
||||
drop(kTopicStatusCode, payload);
|
||||
}
|
||||
void publishCamEvent(const CamEvent& e) override {
|
||||
LOG_TRACE_CAT(LogCat::Mqtt) << "PUB CamEvent cam=" << e.camera
|
||||
<< " hdg=" << e.heading_decideg << " t=" << e.timestamp_ms
|
||||
<< " (null channel)";
|
||||
drop(kTopicCamEvent, camEventJson(e));
|
||||
}
|
||||
void publishEnv(const EnvEvent& e) override {
|
||||
LOG_TRACE_CAT(LogCat::Mqtt) << "PUB Env temp_c=" << e.temp_c
|
||||
<< " humid_pct=" << e.humidity_pct << " t=" << e.timestamp_ms
|
||||
<< " (null channel)";
|
||||
drop(kTopicEnv, envJson(e));
|
||||
}
|
||||
|
||||
ControlCommand poll() override { return {}; }
|
||||
|
||||
std::vector<ChannelMessage> recentMessages() const override { return history_.recent(); }
|
||||
ChannelStats channelStats() const override { return history_.stats(); }
|
||||
|
||||
private:
|
||||
void drop(const std::string& leaf, std::string payload) {
|
||||
history_.record(/*outbound=*/true, Delivery::Dropped, towerTopic(tower_, leaf),
|
||||
std::move(payload));
|
||||
}
|
||||
|
||||
std::string tower_;
|
||||
ChannelHistory history_;
|
||||
};
|
||||
|
||||
} // namespace fgc
|
||||
|
|
|
|||
|
|
@ -1,5 +1,7 @@
|
|||
#pragma once
|
||||
|
||||
#include "fgc/IControlChannel.h"
|
||||
|
||||
#include "fgc/IMotorController.h" // AxisState
|
||||
#include "fgc/Logger.h" // LogLevel
|
||||
|
||||
|
|
@ -149,6 +151,10 @@ struct ConnView {
|
|||
int control_code = 0; // 0 = auto-sweep, 1 = directed
|
||||
std::string target_heading;
|
||||
int last_status_code = 0;
|
||||
|
||||
// Recent traffic + counters for the expanded 'm' view.
|
||||
std::vector<ChannelMessage> recent;
|
||||
ChannelStats stats;
|
||||
};
|
||||
|
||||
struct HeaderView {
|
||||
|
|
@ -309,6 +315,16 @@ std::string formatDegrees(double deg);
|
|||
// Returns "—" when then_ms is 0 (never).
|
||||
std::string formatTimeAgo(long long now_ms, long long then_ms);
|
||||
|
||||
// Last path segment of an MQTT topic ("GGS/FWT/Tower/CamEvent" -> "CamEvent").
|
||||
// Every row in the connectivity view shares the same prefix, which only eats
|
||||
// width. Returns the input unchanged when there is no '/'.
|
||||
std::string topicLeaf(const std::string& topic);
|
||||
|
||||
// "11:56:14.201" (local wall clock) for an absolute ms timestamp; "—" when 0.
|
||||
// The message history needs the actual time, not "3s ago": operators correlate
|
||||
// it against the broker's own log.
|
||||
std::string formatClockMs(long long ms);
|
||||
|
||||
// MVP placeholder: the Sensors panel fields shown before the DHT11/MTi drivers
|
||||
// exist (all `present=false`). Centralised so the panel and tests agree.
|
||||
SensorsView pendingSensorsView();
|
||||
|
|
|
|||
|
|
@ -252,7 +252,7 @@ struct Application::Impl {
|
|||
#else
|
||||
if (want) LOG_WARN << "MQTT requested but binary built without MQTT; using null channel";
|
||||
#endif
|
||||
return std::make_unique<NullControlChannel>();
|
||||
return std::make_unique<NullControlChannel>(cfg.general.tower_name);
|
||||
}
|
||||
|
||||
std::unique_ptr<IMotorController> makeMotor() {
|
||||
|
|
@ -277,7 +277,8 @@ struct Application::Impl {
|
|||
const bool force_mock = opts.mock_env.value_or(false);
|
||||
if (!cfg.features.enable_env && !force_mock)
|
||||
return nullptr; // sensors panel stays "pending"
|
||||
if (force_mock || cfg.features.mock_env) return std::make_unique<MockEnvSensor>();
|
||||
if (force_mock || cfg.features.mock_env)
|
||||
return std::make_unique<MockEnvSensor>(std::chrono::milliseconds(cfg.env.period_ms));
|
||||
if (cfg.env.i2c_device.empty()) {
|
||||
LOG_WARN << "Env sensor enabled but [Env] i2c_device is empty; disabling";
|
||||
return nullptr;
|
||||
|
|
@ -724,6 +725,10 @@ struct Application::Impl {
|
|||
s.conn.control_code = scheduler ? scheduler->controlCode() : 0;
|
||||
s.conn.target_heading = scheduler ? scheduler->targetHeading() : "0";
|
||||
s.conn.last_status_code = s.conn.control_code; // echoed back as status
|
||||
if (channel) {
|
||||
s.conn.recent = channel->recentMessages();
|
||||
s.conn.stats = channel->channelStats();
|
||||
}
|
||||
|
||||
// --- Diagnostics (last firmware DUMP) ---
|
||||
s.dump.text = motor->lastDump();
|
||||
|
|
|
|||
|
|
@ -0,0 +1,68 @@
|
|||
#include "fgc/ChannelHistory.h"
|
||||
|
||||
#include <chrono>
|
||||
|
||||
namespace fgc {
|
||||
|
||||
namespace {
|
||||
|
||||
long long nowMs() {
|
||||
return std::chrono::duration_cast<std::chrono::milliseconds>(
|
||||
std::chrono::system_clock::now().time_since_epoch())
|
||||
.count();
|
||||
}
|
||||
|
||||
} // namespace
|
||||
|
||||
void ChannelHistory::record(bool outbound, Delivery delivery, std::string topic,
|
||||
std::string payload) {
|
||||
std::lock_guard<std::mutex> lock(mutex_);
|
||||
|
||||
if (!outbound) {
|
||||
++stats_.received;
|
||||
} else {
|
||||
switch (delivery) {
|
||||
case Delivery::Ok: ++stats_.published_ok; break;
|
||||
case Delivery::Dropped: ++stats_.published_dropped; break;
|
||||
case Delivery::Failed: ++stats_.published_failed; break;
|
||||
}
|
||||
}
|
||||
|
||||
ChannelMessage m;
|
||||
m.timestamp_ms = nowMs();
|
||||
m.outbound = outbound;
|
||||
m.delivery = delivery;
|
||||
m.topic = std::move(topic);
|
||||
m.payload = std::move(payload);
|
||||
|
||||
ring_.push_back(std::move(m));
|
||||
while (ring_.size() > kCapacity) ring_.pop_front();
|
||||
}
|
||||
|
||||
void ChannelHistory::noteConnect() {
|
||||
std::lock_guard<std::mutex> lock(mutex_);
|
||||
++stats_.connects;
|
||||
stats_.last_connect_ms = nowMs();
|
||||
}
|
||||
|
||||
void ChannelHistory::noteStatus(std::string payload) {
|
||||
std::lock_guard<std::mutex> lock(mutex_);
|
||||
stats_.last_status_payload = std::move(payload);
|
||||
}
|
||||
|
||||
void ChannelHistory::noteLoss() {
|
||||
std::lock_guard<std::mutex> lock(mutex_);
|
||||
stats_.last_loss_ms = nowMs();
|
||||
}
|
||||
|
||||
std::vector<ChannelMessage> ChannelHistory::recent() const {
|
||||
std::lock_guard<std::mutex> lock(mutex_);
|
||||
return {ring_.begin(), ring_.end()};
|
||||
}
|
||||
|
||||
ChannelStats ChannelHistory::stats() const {
|
||||
std::lock_guard<std::mutex> lock(mutex_);
|
||||
return stats_;
|
||||
}
|
||||
|
||||
} // namespace fgc
|
||||
|
|
@ -0,0 +1,22 @@
|
|||
#include "fgc/ChannelTopics.h"
|
||||
|
||||
namespace fgc {
|
||||
|
||||
std::string towerTopic(const std::string& tower, const std::string& leaf) {
|
||||
return "GGS/FWT/" + tower + "/" + leaf;
|
||||
}
|
||||
|
||||
std::string camEventJson(const CamEvent& e) {
|
||||
return "{ \"fwt\":\"" + e.tower + "\", \"cam\":\"" + e.camera +
|
||||
"\", \"hdg\":" + std::to_string(e.heading_decideg) +
|
||||
", \"pit\":" + std::to_string(e.pitch_decideg) +
|
||||
", \"time\":" + std::to_string(e.timestamp_ms) + " }";
|
||||
}
|
||||
|
||||
std::string envJson(const EnvEvent& e) {
|
||||
return "{ \"fwt\":\"" + e.tower + "\", \"temp_c\":" + std::to_string(e.temp_c) +
|
||||
", \"humid_pct\":" + std::to_string(e.humidity_pct) +
|
||||
", \"time\":" + std::to_string(e.timestamp_ms) + " }";
|
||||
}
|
||||
|
||||
} // namespace fgc
|
||||
|
|
@ -118,10 +118,11 @@ const std::vector<HelpSection>& helpCatalog() {
|
|||
"Example: trace serial on / trace off"}},
|
||||
}},
|
||||
{"Display", "Terminal dashboard keys (TUI only).", {
|
||||
{"? / g / i / c",
|
||||
"Toggle the Help, Gimbal, Sensors and Camera overlays.", {
|
||||
{"? / g / i / c / m",
|
||||
"Toggle the Help, Gimbal, Sensors, Camera and Connectivity overlays.", {
|
||||
"An overlay takes the full terminal height. Esc closes it; with none open,",
|
||||
"Esc cancels a running procedure."}},
|
||||
"Esc cancels a running procedure. 'm' shows broker state, per-topic traffic",
|
||||
"and the recent MQTT messages, including any dropped while offline."}},
|
||||
{"Up/Down, j/k",
|
||||
"Scroll the open overlay one line (Help: change section).", {
|
||||
"The scrollbar on the right shows how much is off-screen."}},
|
||||
|
|
|
|||
|
|
@ -1,5 +1,7 @@
|
|||
#include "fgc/MqttControlChannel.h"
|
||||
|
||||
#include "fgc/ChannelTopics.h"
|
||||
|
||||
#include "fgc/Logger.h"
|
||||
|
||||
#include <chrono>
|
||||
|
|
@ -16,11 +18,11 @@ MqttControlChannel::MqttControlChannel(const std::string& broker_uri, const std:
|
|||
const std::string& user, const std::string& password)
|
||||
: broker_uri_(broker_uri),
|
||||
tower_(tower),
|
||||
topic_target_hdg_("GGS/FWT/" + tower + "/target_HDG"),
|
||||
topic_control_code_("GGS/FWT/" + tower + "/ControlCode"),
|
||||
topic_status_("GGS/FWT/" + tower + "/StatusCode"),
|
||||
topic_cam_event_("GGS/FWT/" + tower + "/CamEvent"),
|
||||
topic_env_("GGS/FWT/" + tower + "/Env"),
|
||||
topic_target_hdg_(towerTopic(tower, kTopicTargetHdg)),
|
||||
topic_control_code_(towerTopic(tower, kTopicControlCode)),
|
||||
topic_status_(towerTopic(tower, kTopicStatusCode)),
|
||||
topic_cam_event_(towerTopic(tower, kTopicCamEvent)),
|
||||
topic_env_(towerTopic(tower, kTopicEnv)),
|
||||
client_(broker_uri, tower) {
|
||||
conn_opts_.set_keep_alive_interval(20);
|
||||
conn_opts_.set_clean_session(true);
|
||||
|
|
@ -59,6 +61,7 @@ bool MqttControlChannel::connected() const { return connected_; }
|
|||
|
||||
void MqttControlChannel::connected(const std::string& /*cause*/) {
|
||||
LOG_INFO << "MQTT (re)connected; subscribing";
|
||||
history_.noteConnect();
|
||||
client_.subscribe(topic_target_hdg_, kQos);
|
||||
client_.subscribe(topic_control_code_, kQos);
|
||||
connected_ = true;
|
||||
|
|
@ -66,6 +69,7 @@ void MqttControlChannel::connected(const std::string& /*cause*/) {
|
|||
|
||||
void MqttControlChannel::connection_lost(const std::string& cause) {
|
||||
LOG_WARN << "MQTT connection lost" << (cause.empty() ? "" : ": " + cause);
|
||||
history_.noteLoss();
|
||||
connected_ = false;
|
||||
// Paho will auto-reconnect when configured; the broker callback re-subscribes.
|
||||
}
|
||||
|
|
@ -78,6 +82,7 @@ void MqttControlChannel::message_arrived(mqtt::const_message_ptr msg) {
|
|||
const std::string topic = msg->get_topic();
|
||||
const std::string payload = msg->to_string();
|
||||
LOG_TRACE_CAT(LogCat::Mqtt) << "RX " << topic << ' ' << payload;
|
||||
history_.record(/*outbound=*/false, Delivery::Ok, topic, payload);
|
||||
std::lock_guard<std::mutex> lock(mutex_);
|
||||
try {
|
||||
if (topic == topic_target_hdg_) {
|
||||
|
|
@ -102,36 +107,37 @@ ControlCommand MqttControlChannel::poll() {
|
|||
}
|
||||
|
||||
void MqttControlChannel::publish(const std::string& topic, const std::string& payload) {
|
||||
if (!client_.is_connected()) return;
|
||||
if (!client_.is_connected()) {
|
||||
// Discarded: there is no queue and no retry, so this message is simply
|
||||
// gone. Recording it is the only trace it ever existed.
|
||||
history_.record(/*outbound=*/true, Delivery::Dropped, topic, payload);
|
||||
return;
|
||||
}
|
||||
LOG_TRACE_CAT(LogCat::Mqtt) << "PUB " << topic << ' ' << payload;
|
||||
try {
|
||||
auto msg = mqtt::make_message(topic, payload);
|
||||
msg->set_qos(kQos);
|
||||
msg->set_retained(true);
|
||||
client_.publish(msg)->wait_for(kPublishTimeout);
|
||||
history_.record(/*outbound=*/true, Delivery::Ok, topic, payload);
|
||||
} catch (const mqtt::exception& e) {
|
||||
LOG_WARN << "MQTT publish failed: " << e.what();
|
||||
history_.record(/*outbound=*/true, Delivery::Failed, topic, payload);
|
||||
}
|
||||
}
|
||||
|
||||
void MqttControlChannel::publishStatus(int code) {
|
||||
publish(topic_status_, std::to_string(code));
|
||||
const std::string payload = std::to_string(code);
|
||||
history_.noteStatus(payload);
|
||||
publish(topic_status_, payload);
|
||||
}
|
||||
|
||||
void MqttControlChannel::publishCamEvent(const CamEvent& e) {
|
||||
std::string payload = "{ \"fwt\":\"" + e.tower + "\" ,\"cam\":\"" + e.camera +
|
||||
"\", \"hdg\":" + std::to_string(e.heading_decideg) +
|
||||
", \"pit\":" + std::to_string(e.pitch_decideg) +
|
||||
", \"time\":" + std::to_string(e.timestamp_ms) + " }";
|
||||
publish(topic_cam_event_, payload);
|
||||
publish(topic_cam_event_, camEventJson(e));
|
||||
}
|
||||
|
||||
void MqttControlChannel::publishEnv(const EnvEvent& e) {
|
||||
std::string payload = "{ \"fwt\":\"" + e.tower +
|
||||
"\", \"temp_c\":" + std::to_string(e.temp_c) +
|
||||
", \"humid_pct\":" + std::to_string(e.humidity_pct) +
|
||||
", \"time\":" + std::to_string(e.timestamp_ms) + " }";
|
||||
publish(topic_env_, payload);
|
||||
publish(topic_env_, envJson(e));
|
||||
}
|
||||
|
||||
} // namespace fgc
|
||||
|
|
|
|||
115
src/ui/TuiUi.cpp
115
src/ui/TuiUi.cpp
|
|
@ -1,5 +1,7 @@
|
|||
#include "fgc/ui/TuiUi.h"
|
||||
|
||||
#include "fgc/ChannelTopics.h"
|
||||
|
||||
#include "fgc/DumpParser.h"
|
||||
#include "fgc/HelpText.h"
|
||||
#include "fgc/Logger.h"
|
||||
|
|
@ -375,6 +377,103 @@ Element connPanel(const ConnView& v) {
|
|||
}));
|
||||
}
|
||||
|
||||
// Expanded connectivity view ('m'): broker state, the five topics with their last
|
||||
// traffic, the counters, and a scrollable history of recent messages.
|
||||
//
|
||||
// The history is the point of the view. `dropped` in particular has no other
|
||||
// route to the operator: MqttControlChannel::publish() discards a message when
|
||||
// the link is down without queueing, retrying or logging, so a lost CamEvent -
|
||||
// and with it an image the ground station never fetches - is otherwise invisible.
|
||||
Element connDetailPanel(const ConnView& v, ScrollState& st) {
|
||||
auto row = [](const std::string& k, Element val) {
|
||||
return hbox({text(k) | dim | size(WIDTH, EQUAL, 12), std::move(val)});
|
||||
};
|
||||
const long long now = std::chrono::duration_cast<std::chrono::milliseconds>(
|
||||
std::chrono::system_clock::now().time_since_epoch()).count();
|
||||
|
||||
std::vector<Element> rows;
|
||||
|
||||
// --- Broker ---
|
||||
rows.push_back(text("BROKER") | bold | color(Color::Green));
|
||||
Element state = !v.mqtt_enabled
|
||||
? (text(" disabled - messages not sent ") | color(Color::Yellow) | bold)
|
||||
: v.mqtt_connected
|
||||
? (text(" connected ") | color(Color::Green) | bold)
|
||||
: (text(" disconnected - retrying ") | color(Color::Red) | bold);
|
||||
const bool really_up = v.mqtt_enabled && v.mqtt_connected && v.stats.last_connect_ms > 0;
|
||||
rows.push_back(hbox({text("state ") | dim, state, filler(),
|
||||
text(really_up ? "since " + formatTimeAgo(now, v.stats.last_connect_ms)
|
||||
: "") | dim}));
|
||||
rows.push_back(row("uri", text(v.broker.empty() ? "\xE2\x80\x94" : v.broker)));
|
||||
rows.push_back(row("client id", text(v.tower)));
|
||||
rows.push_back(row("settings", text("QoS 1 \xC2\xB7 retained \xC2\xB7 keep-alive 20 s "
|
||||
"\xC2\xB7 clean session") | dim));
|
||||
rows.push_back(row("link", text(std::to_string(v.stats.connects) + " connect(s), last loss " +
|
||||
formatTimeAgo(now, v.stats.last_loss_ms)) | dim));
|
||||
|
||||
// --- Counters ---
|
||||
rows.push_back(separator());
|
||||
rows.push_back(text("COUNTERS") | bold | color(Color::Green));
|
||||
rows.push_back(row("published", hbox({
|
||||
text(std::to_string(v.stats.published_ok) + " ok ") | color(Color::Green),
|
||||
text(std::to_string(v.stats.published_dropped) + " dropped ") |
|
||||
(v.stats.published_dropped ? color(Color::Red) : color(Color::Default)),
|
||||
text(std::to_string(v.stats.published_failed) + " failed") |
|
||||
(v.stats.published_failed ? color(Color::Red) : color(Color::Default)),
|
||||
})));
|
||||
rows.push_back(row("received", text(std::to_string(v.stats.received))));
|
||||
if (v.stats.published_dropped)
|
||||
rows.push_back(text("dropped = attempted while the link was down; not queued, not "
|
||||
"retried, gone.") | color(Color::Red) | dim);
|
||||
|
||||
// --- Topics: last traffic per topic, newest wins ---
|
||||
rows.push_back(separator());
|
||||
rows.push_back(text("TOPICS") | bold | color(Color::Green));
|
||||
struct TopicRow { const char* dir; const char* leaf; };
|
||||
static const TopicRow kTopics[] = {
|
||||
{"sub", kTopicTargetHdg}, {"sub", kTopicControlCode}, {"pub", kTopicStatusCode},
|
||||
{"pub", kTopicCamEvent}, {"pub", kTopicEnv},
|
||||
};
|
||||
for (const auto& t : kTopics) {
|
||||
const ChannelMessage* last = nullptr;
|
||||
for (const auto& m : v.recent)
|
||||
if (topicLeaf(m.topic) == t.leaf) last = &m; // recent is oldest-first
|
||||
std::string when = last ? formatTimeAgo(now, last->timestamp_ms) : "\xE2\x80\x94";
|
||||
rows.push_back(hbox({
|
||||
text(std::string(" ") + t.dir + " ") | dim,
|
||||
text(t.leaf) | size(WIDTH, EQUAL, 14),
|
||||
text(when) | dim | size(WIDTH, EQUAL, 10),
|
||||
text(last ? last->payload : "") | dim,
|
||||
}));
|
||||
}
|
||||
|
||||
// --- History ---
|
||||
rows.push_back(separator());
|
||||
rows.push_back(hbox({text("RECENT MESSAGES") | bold | color(Color::Green),
|
||||
text(" (oldest first)") | dim}));
|
||||
if (v.recent.empty()) {
|
||||
rows.push_back(text("(nothing sent or received yet)") | dim);
|
||||
} else {
|
||||
for (const auto& m : v.recent) {
|
||||
const bool bad = m.delivery != Delivery::Ok;
|
||||
Element tag = text(m.outbound ? "PUB" : "RX ") |
|
||||
(m.outbound ? color(Color::Yellow) : color(Color::Cyan));
|
||||
std::string note = m.delivery == Delivery::Dropped ? " DROPPED"
|
||||
: m.delivery == Delivery::Failed ? " FAILED"
|
||||
: "";
|
||||
rows.push_back(hbox({
|
||||
text(formatClockMs(m.timestamp_ms) + " ") | dim,
|
||||
tag,
|
||||
text(" " + topicLeaf(m.topic)) | size(WIDTH, EQUAL, 14),
|
||||
text(" " + m.payload + note) | (bad ? color(Color::Red) : color(Color::Default)),
|
||||
}));
|
||||
}
|
||||
}
|
||||
|
||||
return window(overlayTitle("CONNECTIVITY", "m/Esc:close", st, Color::Green),
|
||||
scrollableBody(vbox(std::move(rows)), st));
|
||||
}
|
||||
|
||||
Element logPanel(const std::vector<LogLine>& lines, ScrollState& st) {
|
||||
std::vector<Element> rows;
|
||||
for (const auto& l : lines) {
|
||||
|
|
@ -877,7 +976,7 @@ void TuiUi::refreshLoop() {
|
|||
void TuiUi::uiLoop() {
|
||||
std::string cmd_buffer;
|
||||
bool command_mode = false;
|
||||
enum class Overlay { None, Help, Gimbal, Sensors, Cameras };
|
||||
enum class Overlay { None, Help, Gimbal, Sensors, Cameras, Conn };
|
||||
Overlay overlay = Overlay::None; // which takeover panel owns the main area
|
||||
int help_sel = 0;
|
||||
bool gimbal_dump_requested = false; // auto-pull a dump the first time
|
||||
|
|
@ -886,7 +985,7 @@ void TuiUi::uiLoop() {
|
|||
// Per-pane scroll positions. UI-thread-only view state (the renderer lambda
|
||||
// and CatchEvent below both run on this thread), so unlike screen_/paused_
|
||||
// these need no synchronisation and stay locals rather than TuiUi members.
|
||||
ScrollState cam_scroll, sensors_scroll, help_scroll, log_scroll;
|
||||
ScrollState cam_scroll, sensors_scroll, help_scroll, log_scroll, conn_scroll;
|
||||
std::array<ScrollState, 2> gimbal_scroll{}; // YAW, PITCH
|
||||
log_scroll.follow = true; // the log tracks the tail until told otherwise
|
||||
size_t log_dropped_at_freeze = 0; // buffer evictions when scrollback was entered
|
||||
|
|
@ -944,7 +1043,7 @@ void TuiUi::uiLoop() {
|
|||
bottom = hbox({
|
||||
keyHint("s", "Start"), keyHint("x", "Stop"), keyHint("h", "Home"),
|
||||
keyHint("g", "Gimbal"), keyHint("i", "Sensors"), keyHint("c", "Scan"),
|
||||
keyHint("p", "Pause"), keyHint("r", "Refresh"),
|
||||
keyHint("m", "Comms"), keyHint("p", "Pause"), keyHint("r", "Refresh"),
|
||||
keyHint("\xE2\x86\x90\xE2\x86\x92\xE2\x86\x91\xE2\x86\x93", "Nudge"),
|
||||
keyHint(":", "Cmd"), keyHint("?", "Help"), filler(), keyHint("q", "Quit"),
|
||||
});
|
||||
|
|
@ -966,6 +1065,8 @@ void TuiUi::uiLoop() {
|
|||
sensorsDetailPanel(s.imu, s.env, sensors_scroll) | flex, bottom});
|
||||
case Overlay::Cameras:
|
||||
return vbox({header, separator(), cameraDetailPanel(s.capture, cam_scroll) | flex, bottom});
|
||||
case Overlay::Conn:
|
||||
return vbox({header, separator(), connDetailPanel(s.conn, conn_scroll) | flex, bottom});
|
||||
default: {
|
||||
// Activity strip sits between the log and the key bar; shown only
|
||||
// once a special op has run or is running, else it costs no space.
|
||||
|
|
@ -1027,6 +1128,9 @@ void TuiUi::uiLoop() {
|
|||
case Overlay::Cameras:
|
||||
if (scrollKeys(cam_scroll, e, arrows)) return true;
|
||||
break;
|
||||
case Overlay::Conn:
|
||||
if (scrollKeys(conn_scroll, e, arrows)) return true;
|
||||
break;
|
||||
case Overlay::Sensors:
|
||||
if (scrollKeys(sensors_scroll, e, arrows)) return true;
|
||||
break;
|
||||
|
|
@ -1116,6 +1220,11 @@ void TuiUi::uiLoop() {
|
|||
if (overlay == Overlay::Cameras) cam_scroll = {};
|
||||
return true;
|
||||
}
|
||||
if (c == "m") {
|
||||
overlay = (overlay == Overlay::Conn) ? Overlay::None : Overlay::Conn;
|
||||
if (overlay == Overlay::Conn) conn_scroll = {};
|
||||
return true;
|
||||
}
|
||||
if (c == "p") { // freeze/unfreeze the live refresh so text can be selected
|
||||
paused_ = !paused_;
|
||||
return true; // this keypress redraws once so the indicator updates
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@
|
|||
|
||||
#include <cmath>
|
||||
#include <cstdio>
|
||||
#include <ctime>
|
||||
|
||||
namespace fgc {
|
||||
|
||||
|
|
@ -47,6 +48,23 @@ std::string formatTimeAgo(long long now_ms, long long then_ms) {
|
|||
return buf;
|
||||
}
|
||||
|
||||
std::string topicLeaf(const std::string& topic) {
|
||||
const auto slash = topic.rfind('/');
|
||||
if (slash == std::string::npos) return topic;
|
||||
return topic.substr(slash + 1);
|
||||
}
|
||||
|
||||
std::string formatClockMs(long long ms) {
|
||||
if (ms <= 0) return "\xE2\x80\x94"; // em dash: never
|
||||
const std::time_t secs = static_cast<std::time_t>(ms / 1000);
|
||||
std::tm tm{};
|
||||
localtime_r(&secs, &tm);
|
||||
char buf[32];
|
||||
std::snprintf(buf, sizeof(buf), "%02d:%02d:%02d.%03d", tm.tm_hour, tm.tm_min, tm.tm_sec,
|
||||
static_cast<int>(ms % 1000));
|
||||
return buf;
|
||||
}
|
||||
|
||||
SensorsView pendingSensorsView() {
|
||||
SensorsView v;
|
||||
// Xsens MTi-28 AHRS: drift-free Euler orientation. The device's internal
|
||||
|
|
|
|||
|
|
@ -23,6 +23,7 @@ add_executable(fgc_tests
|
|||
test_singleinstance.cpp
|
||||
test_config.cpp
|
||||
test_cameramode.cpp
|
||||
test_channelhistory.cpp
|
||||
test_telemetry.cpp
|
||||
test_command.cpp
|
||||
test_scheduler.cpp
|
||||
|
|
|
|||
|
|
@ -0,0 +1,92 @@
|
|||
#include <doctest/doctest.h>
|
||||
|
||||
#include "fgc/ChannelHistory.h"
|
||||
#include "fgc/ChannelTopics.h"
|
||||
|
||||
#include <string>
|
||||
|
||||
using namespace fgc;
|
||||
|
||||
TEST_CASE("ChannelHistory counts each delivery outcome separately") {
|
||||
ChannelHistory h;
|
||||
h.record(true, Delivery::Ok, "t/a", "1");
|
||||
h.record(true, Delivery::Ok, "t/a", "2");
|
||||
h.record(true, Delivery::Dropped, "t/a", "3");
|
||||
h.record(true, Delivery::Failed, "t/a", "4");
|
||||
h.record(false, Delivery::Ok, "t/b", "in");
|
||||
|
||||
ChannelStats s = h.stats();
|
||||
CHECK(s.published_ok == 2);
|
||||
CHECK(s.published_dropped == 1); // the silent-drop path: the number that matters
|
||||
CHECK(s.published_failed == 1);
|
||||
CHECK(s.received == 1); // inbound never touches the publish counters
|
||||
}
|
||||
|
||||
TEST_CASE("ChannelHistory returns messages oldest first and keeps their fields") {
|
||||
ChannelHistory h;
|
||||
h.record(true, Delivery::Dropped, "t/first", "a");
|
||||
h.record(false, Delivery::Ok, "t/second", "b");
|
||||
|
||||
auto r = h.recent();
|
||||
REQUIRE(r.size() == 2);
|
||||
CHECK(r[0].topic == "t/first");
|
||||
CHECK(r[0].outbound);
|
||||
CHECK(r[0].delivery == Delivery::Dropped);
|
||||
CHECK(r[0].payload == "a");
|
||||
CHECK(r[1].topic == "t/second");
|
||||
CHECK_FALSE(r[1].outbound);
|
||||
CHECK(r[1].timestamp_ms > 0);
|
||||
}
|
||||
|
||||
TEST_CASE("ChannelHistory evicts the oldest beyond its capacity") {
|
||||
ChannelHistory h;
|
||||
const int n = static_cast<int>(ChannelHistory::kCapacity) + 10;
|
||||
for (int i = 0; i < n; ++i) h.record(true, Delivery::Ok, "t", std::to_string(i));
|
||||
|
||||
auto r = h.recent();
|
||||
REQUIRE(r.size() == ChannelHistory::kCapacity);
|
||||
CHECK(r.front().payload == std::to_string(n - static_cast<int>(ChannelHistory::kCapacity)));
|
||||
CHECK(r.back().payload == std::to_string(n - 1));
|
||||
// Counters are cumulative; they must not be bounded by the ring.
|
||||
CHECK(h.stats().published_ok == n);
|
||||
}
|
||||
|
||||
TEST_CASE("ChannelHistory tracks connects, losses and the last status sent") {
|
||||
ChannelHistory h;
|
||||
CHECK(h.stats().connects == 0);
|
||||
CHECK(h.stats().last_connect_ms == 0);
|
||||
|
||||
h.noteConnect();
|
||||
h.noteLoss();
|
||||
h.noteConnect();
|
||||
h.noteStatus("1");
|
||||
|
||||
ChannelStats s = h.stats();
|
||||
CHECK(s.connects == 2);
|
||||
CHECK(s.last_connect_ms > 0);
|
||||
CHECK(s.last_loss_ms > 0);
|
||||
CHECK(s.last_status_payload == "1");
|
||||
}
|
||||
|
||||
TEST_CASE("topic and payload builders match the documented wire format") {
|
||||
CHECK(towerTopic("Tower", kTopicCamEvent) == "GGS/FWT/Tower/CamEvent");
|
||||
|
||||
CamEvent c;
|
||||
c.tower = "Tower"; c.camera = "RGB";
|
||||
c.heading_decideg = 1373; c.pitch_decideg = 300; c.timestamp_ms = 1719312345678LL;
|
||||
const std::string j = camEventJson(c);
|
||||
CHECK(j.find("\"fwt\":\"Tower\"") != std::string::npos);
|
||||
CHECK(j.find("\"cam\":\"RGB\"") != std::string::npos);
|
||||
CHECK(j.find("\"hdg\":1373") != std::string::npos);
|
||||
CHECK(j.find("\"pit\":300") != std::string::npos);
|
||||
CHECK(j.find("\"time\":1719312345678") != std::string::npos);
|
||||
// The quality gate is gone; nothing should still be publishing a verdict.
|
||||
CHECK(j.find("degraded") == std::string::npos);
|
||||
|
||||
EnvEvent e;
|
||||
e.tower = "Tower"; e.temp_c = 21.5f; e.humidity_pct = 48.0f; e.timestamp_ms = 42;
|
||||
const std::string ej = envJson(e);
|
||||
CHECK(ej.find("\"fwt\":\"Tower\"") != std::string::npos);
|
||||
CHECK(ej.find("\"temp_c\":21.5") != std::string::npos);
|
||||
CHECK(ej.find("\"time\":42") != std::string::npos);
|
||||
}
|
||||
|
|
@ -101,3 +101,20 @@ TEST_CASE("EnvView defaults to disabled with no reading") {
|
|||
CHECK_FALSE(e.mock);
|
||||
CHECK(e.timestamp_ms == 0);
|
||||
}
|
||||
|
||||
TEST_CASE("topicLeaf strips the constant GGS/FWT/<tower>/ prefix") {
|
||||
CHECK(topicLeaf("GGS/FWT/Staeffelsberg/CamEvent") == "CamEvent");
|
||||
CHECK(topicLeaf("CamEvent") == "CamEvent"); // already a leaf
|
||||
CHECK(topicLeaf("a/b/") == ""); // trailing slash: empty leaf, not "b"
|
||||
CHECK(topicLeaf("") == "");
|
||||
}
|
||||
|
||||
TEST_CASE("formatClockMs renders wall-clock time, and never for 0") {
|
||||
CHECK(formatClockMs(0) == "\xE2\x80\x94");
|
||||
const std::string t = formatClockMs(1719312345678LL);
|
||||
REQUIRE(t.size() == 12); // HH:MM:SS.mmm
|
||||
CHECK(t[2] == ':');
|
||||
CHECK(t[5] == ':');
|
||||
CHECK(t[8] == '.');
|
||||
CHECK(t.substr(9) == "678"); // the ms field comes straight off the input
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue