mirror of
https://github.com/liquidraver/ZephCore.git
synced 2026-09-02 00:38:50 +00:00
Implement MQTT Observer role
This commit is contained in:
liquidraver
+18
-2
@@ -411,6 +411,22 @@ if(CONFIG_ZEPHCORE_ROLE_REPEATER)
|
||||
${CMAKE_CURRENT_SOURCE_DIR}/adapters/usb
|
||||
)
|
||||
target_compile_definitions(app PRIVATE ZEPHCORE_REPEATER=1)
|
||||
elseif(CONFIG_ZEPHCORE_ROLE_OBSERVER)
|
||||
message(STATUS "ZephCore Role: OBSERVER")
|
||||
target_sources(app PRIVATE
|
||||
app/main_observer.cpp
|
||||
app/ObserverMesh.cpp
|
||||
app/RepeaterDataStore.cpp
|
||||
adapters/wifi/ZephyrWiFiStation.c
|
||||
adapters/mqtt/ZephyrMQTTPublisher.c
|
||||
)
|
||||
target_include_directories(app PRIVATE
|
||||
${CMAKE_CURRENT_SOURCE_DIR}/adapters/wifi
|
||||
${CMAKE_CURRENT_SOURCE_DIR}/adapters/mqtt
|
||||
${CMAKE_CURRENT_SOURCE_DIR}/adapters/usb
|
||||
${CMAKE_CURRENT_SOURCE_DIR}/app
|
||||
)
|
||||
target_compile_definitions(app PRIVATE ZEPHCORE_OBSERVER=1)
|
||||
else()
|
||||
message(STATUS "ZephCore Role: COMPANION")
|
||||
target_sources(app PRIVATE
|
||||
@@ -462,8 +478,8 @@ if(CONFIG_ZEPHCORE_UI_BUTTONS OR CONFIG_ZEPHCORE_UI_BUZZER OR CONFIG_ZEPHCORE_UI
|
||||
target_include_directories(app PRIVATE
|
||||
${CMAKE_CURRENT_SOURCE_DIR}/helpers/ui
|
||||
)
|
||||
# Repeater builds need weak stubs for companion-only UI mesh actions
|
||||
if(CONFIG_ZEPHCORE_ROLE_REPEATER)
|
||||
# Repeater and Observer builds need weak stubs for companion-only UI mesh actions
|
||||
if(CONFIG_ZEPHCORE_ROLE_REPEATER OR CONFIG_ZEPHCORE_ROLE_OBSERVER)
|
||||
target_sources(app PRIVATE
|
||||
helpers/ui/ui_mesh_actions_stubs.c
|
||||
)
|
||||
|
||||
@@ -42,6 +42,10 @@ module = ZEPHCORE_WIFI_OTA
|
||||
module-str = zephcore_wifi_ota
|
||||
source "$(ZEPHYR_BASE)/subsys/logging/Kconfig.template.log_config"
|
||||
|
||||
module = ZEPHCORE_OBSERVER
|
||||
module-str = zephcore_observer
|
||||
source "$(ZEPHYR_BASE)/subsys/logging/Kconfig.template.log_config"
|
||||
|
||||
menu "ZephCore"
|
||||
|
||||
menu "Device Role"
|
||||
@@ -66,6 +70,16 @@ config ZEPHCORE_ROLE_REPEATER
|
||||
No BLE stack. Has ACL authentication, region filtering,
|
||||
neighbor tracking, and full CLI command set.
|
||||
|
||||
config ZEPHCORE_ROLE_OBSERVER
|
||||
bool "Observer (listen-only, WiFi+MQTT)"
|
||||
help
|
||||
ESP32-only listen-only node. Receives LoRa packets without
|
||||
forwarding. Connects to WiFi (STA mode) and publishes received
|
||||
packets to an MQTT broker in meshcoretomqtt-compatible format.
|
||||
All parameters (WiFi SSID/PSK, MQTT host/port/TLS/user/pass,
|
||||
IATA location code) are configured at runtime via serial CLI
|
||||
and stored in LittleFS. No BLE, no advertising, no routing.
|
||||
|
||||
endchoice
|
||||
|
||||
if ZEPHCORE_ROLE_REPEATER
|
||||
|
||||
@@ -0,0 +1,469 @@
|
||||
/*
|
||||
* SPDX-License-Identifier: Apache-2.0
|
||||
* ZephyrMQTTPublisher — MQTT client thread for the Observer role.
|
||||
*
|
||||
* Thread flow:
|
||||
* 1. Wait for WIFI_READY_BIT (WiFi + DHCP + SNTP done)
|
||||
* 2. Resolve broker hostname via zsock_getaddrinfo()
|
||||
* 3. Connect MQTT (TLS with TLS_PEER_VERIFY_NONE, or plaintext)
|
||||
* 4. Publish retained "online" status (LWT handles "offline" on disconnect)
|
||||
* 5. Poll loop: drain publish queue + mqtt_input() keepalive
|
||||
* 6. On error/disconnect: publish "offline", back off 5s, goto 1
|
||||
*/
|
||||
|
||||
#include "ZephyrMQTTPublisher.h"
|
||||
#include "ZephyrWiFiStation.h" /* g_wifi_events, WIFI_READY_BIT, zc_wifi_station_* */
|
||||
#include "observer_creds.h"
|
||||
|
||||
#include <zephyr/kernel.h>
|
||||
#include <zephyr/net/mqtt.h>
|
||||
#include <zephyr/net/socket.h>
|
||||
#include <zephyr/net/tls_credentials.h>
|
||||
#include <zephyr/logging/log.h>
|
||||
|
||||
#include <errno.h>
|
||||
#include <string.h>
|
||||
#include <stdio.h>
|
||||
|
||||
LOG_MODULE_REGISTER(mqtt_pub, CONFIG_LOG_DEFAULT_LEVEL);
|
||||
|
||||
/* ========== Publish queue ========== */
|
||||
|
||||
/* Pre-serialized message: topic + JSON payload (both copied at enqueue). */
|
||||
#define PUB_TOPIC_MAX 160
|
||||
#define PUB_PAYLOAD_MAX 1024
|
||||
#define PUB_QUEUE_LEN 8
|
||||
|
||||
struct pub_msg {
|
||||
char topic[PUB_TOPIC_MAX];
|
||||
char payload[PUB_PAYLOAD_MAX];
|
||||
uint16_t payload_len;
|
||||
};
|
||||
|
||||
K_MSGQ_DEFINE(s_pub_queue, sizeof(struct pub_msg), PUB_QUEUE_LEN, 4);
|
||||
|
||||
/* ========== Module state ========== */
|
||||
|
||||
static const struct ObserverCreds *s_creds;
|
||||
static char s_client_id[64];
|
||||
static char s_status_topic[PUB_TOPIC_MAX];
|
||||
static char s_packets_topic[PUB_TOPIC_MAX];
|
||||
|
||||
static volatile bool s_connected;
|
||||
|
||||
/* Event bit to trigger reconnect from external callers */
|
||||
#define PUB_RECONNECT_BIT BIT(0)
|
||||
static K_EVENT_DEFINE(s_pub_events);
|
||||
|
||||
/* ========== MQTT buffers ========== */
|
||||
|
||||
#define MQTT_RX_BUF_SIZE 256
|
||||
#define MQTT_TX_BUF_SIZE 512
|
||||
|
||||
static uint8_t s_rx_buf[MQTT_RX_BUF_SIZE];
|
||||
static uint8_t s_tx_buf[MQTT_TX_BUF_SIZE];
|
||||
|
||||
/* ========== MQTT event callback ========== */
|
||||
|
||||
static void mqtt_evt_handler(struct mqtt_client *client,
|
||||
const struct mqtt_evt *evt)
|
||||
{
|
||||
switch (evt->type) {
|
||||
case MQTT_EVT_CONNACK:
|
||||
if (evt->result == 0) {
|
||||
LOG_INF("MQTT connected");
|
||||
s_connected = true;
|
||||
} else {
|
||||
/* Broker return codes: 1=bad proto, 2=id rejected, 3=server unavail,
|
||||
* 4=bad user/pass, 5=not authorized */
|
||||
LOG_WRN("MQTT CONNACK rejected: return_code=%d",
|
||||
evt->param.connack.return_code);
|
||||
}
|
||||
break;
|
||||
case MQTT_EVT_DISCONNECT:
|
||||
LOG_INF("MQTT disconnected");
|
||||
s_connected = false;
|
||||
break;
|
||||
case MQTT_EVT_PUBLISH:
|
||||
/* Observer never subscribes — ignore incoming publishes */
|
||||
break;
|
||||
case MQTT_EVT_PUBACK:
|
||||
case MQTT_EVT_PINGRESP:
|
||||
break;
|
||||
default:
|
||||
LOG_DBG("MQTT evt %d", evt->type);
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
/* ========== Publish helpers ========== */
|
||||
|
||||
static uint16_t s_pkt_id;
|
||||
|
||||
static int do_publish(struct mqtt_client *client,
|
||||
const char *topic, const char *payload, int payload_len,
|
||||
bool retain)
|
||||
{
|
||||
struct mqtt_publish_param p = {
|
||||
.message = {
|
||||
.topic = {
|
||||
.topic = {
|
||||
.utf8 = (const uint8_t *)topic,
|
||||
.size = strlen(topic),
|
||||
},
|
||||
.qos = MQTT_QOS_0_AT_MOST_ONCE,
|
||||
},
|
||||
.payload = {
|
||||
.data = (uint8_t *)payload,
|
||||
.len = (uint32_t)payload_len,
|
||||
},
|
||||
},
|
||||
.message_id = ++s_pkt_id,
|
||||
.dup_flag = 0,
|
||||
.retain_flag = retain ? 1 : 0,
|
||||
};
|
||||
return mqtt_publish(client, &p);
|
||||
}
|
||||
|
||||
static void publish_status(struct mqtt_client *client, bool online)
|
||||
{
|
||||
static const char online_json[] = "{\"status\":\"online\"}";
|
||||
static const char offline_json[] = "{\"status\":\"offline\"}";
|
||||
const char *msg = online ? online_json : offline_json;
|
||||
do_publish(client, s_status_topic, msg, strlen(msg), true);
|
||||
}
|
||||
|
||||
/* ========== DNS resolution ========== */
|
||||
|
||||
static int resolve_host(const char *host, uint16_t port,
|
||||
struct sockaddr_in *out_addr)
|
||||
{
|
||||
struct zsock_addrinfo hints = {
|
||||
.ai_family = AF_INET,
|
||||
.ai_socktype = SOCK_STREAM,
|
||||
};
|
||||
struct zsock_addrinfo *result;
|
||||
|
||||
char port_str[8];
|
||||
snprintf(port_str, sizeof(port_str), "%u", port);
|
||||
|
||||
int rc = zsock_getaddrinfo(host, port_str, &hints, &result);
|
||||
if (rc != 0) {
|
||||
LOG_ERR("DNS lookup failed for %s: %d", host, rc);
|
||||
return -EHOSTUNREACH;
|
||||
}
|
||||
|
||||
memcpy(out_addr, result->ai_addr, sizeof(*out_addr));
|
||||
zsock_freeaddrinfo(result);
|
||||
LOG_INF("Resolved %s → port %u", host, port);
|
||||
return 0;
|
||||
}
|
||||
|
||||
/* ========== Client setup ========== */
|
||||
|
||||
static struct mqtt_client s_client;
|
||||
static struct sockaddr_in s_broker_addr;
|
||||
|
||||
/* LWT strings (must outlive mqtt_connect()) */
|
||||
static const char s_lwt_offline[] = "{\"status\":\"offline\"}";
|
||||
static struct mqtt_topic s_lwt_topic;
|
||||
static struct mqtt_utf8 s_lwt_msg;
|
||||
static struct mqtt_utf8 s_mqtt_user;
|
||||
static struct mqtt_utf8 s_mqtt_pass;
|
||||
static struct mqtt_utf8 s_mqtt_client_id;
|
||||
|
||||
static int setup_client(void)
|
||||
{
|
||||
mqtt_client_init(&s_client);
|
||||
|
||||
/* Client ID */
|
||||
s_mqtt_client_id.utf8 = (const uint8_t *)s_client_id;
|
||||
s_mqtt_client_id.size = strlen(s_client_id);
|
||||
s_client.client_id = s_mqtt_client_id;
|
||||
|
||||
/* Broker address (already resolved) */
|
||||
s_client.broker = &s_broker_addr;
|
||||
|
||||
/* RX/TX buffers */
|
||||
s_client.rx_buf = s_rx_buf;
|
||||
s_client.rx_buf_size = sizeof(s_rx_buf);
|
||||
s_client.tx_buf = s_tx_buf;
|
||||
s_client.tx_buf_size = sizeof(s_tx_buf);
|
||||
|
||||
/* Keepalive (matches CONFIG_MQTT_KEEPALIVE=60) */
|
||||
s_client.keepalive = 60;
|
||||
s_client.clean_session = 1;
|
||||
|
||||
/* Credentials */
|
||||
if (s_creds->mqtt_user[0] != '\0') {
|
||||
s_mqtt_user.utf8 = (const uint8_t *)s_creds->mqtt_user;
|
||||
s_mqtt_user.size = strlen(s_creds->mqtt_user);
|
||||
s_client.user_name = &s_mqtt_user;
|
||||
}
|
||||
if (s_creds->mqtt_password[0] != '\0') {
|
||||
s_mqtt_pass.utf8 = (const uint8_t *)s_creds->mqtt_password;
|
||||
s_mqtt_pass.size = strlen(s_creds->mqtt_password);
|
||||
s_client.password = &s_mqtt_pass;
|
||||
}
|
||||
|
||||
/* LWT — retained "offline" published if connection drops */
|
||||
s_lwt_topic.topic.utf8 = (const uint8_t *)s_status_topic;
|
||||
s_lwt_topic.topic.size = strlen(s_status_topic);
|
||||
s_lwt_topic.qos = MQTT_QOS_0_AT_MOST_ONCE;
|
||||
s_lwt_msg.utf8 = (const uint8_t *)s_lwt_offline;
|
||||
s_lwt_msg.size = strlen(s_lwt_offline);
|
||||
s_client.will_topic = &s_lwt_topic;
|
||||
s_client.will_message = &s_lwt_msg;
|
||||
s_client.will_retain = 1;
|
||||
|
||||
/* Transport */
|
||||
if (s_creds->mqtt_tls) {
|
||||
s_client.transport.type = MQTT_TRANSPORT_SECURE;
|
||||
struct mqtt_sec_config *tls = &s_client.transport.tls.config;
|
||||
/* No certificate pinning — encrypted but no CA verification.
|
||||
* This avoids cert rotation issues (e.g. Let's Encrypt renewals). */
|
||||
tls->peer_verify = TLS_PEER_VERIFY_NONE;
|
||||
tls->cipher_count = 0;
|
||||
tls->cipher_list = NULL;
|
||||
tls->sec_tag_count = 0;
|
||||
tls->sec_tag_list = NULL;
|
||||
tls->hostname = s_creds->mqtt_host; /* SNI */
|
||||
} else {
|
||||
s_client.transport.type = MQTT_TRANSPORT_NON_SECURE;
|
||||
}
|
||||
|
||||
s_client.evt_cb = mqtt_evt_handler;
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
/* ========== MQTT poll loop ========== */
|
||||
|
||||
static int get_sock(void)
|
||||
{
|
||||
if (s_creds->mqtt_tls) {
|
||||
return s_client.transport.tls.sock;
|
||||
}
|
||||
return s_client.transport.tcp.sock;
|
||||
}
|
||||
|
||||
/* Wait for socket to be readable, with a timeout_ms cap.
|
||||
* Returns > 0 if data available, 0 if timeout, < 0 on error/hangup. */
|
||||
static int poll_socket(int sock, int timeout_ms)
|
||||
{
|
||||
struct zsock_pollfd pfd = {
|
||||
.fd = sock,
|
||||
.events = ZSOCK_POLLIN,
|
||||
};
|
||||
int rc = zsock_poll(&pfd, 1, timeout_ms);
|
||||
if (rc < 0) {
|
||||
return -errno;
|
||||
}
|
||||
if (pfd.revents & (ZSOCK_POLLHUP | ZSOCK_POLLERR)) {
|
||||
return -ECONNRESET;
|
||||
}
|
||||
return rc; /* 0 = timeout, 1 = data ready */
|
||||
}
|
||||
|
||||
static void run_poll_loop(void)
|
||||
{
|
||||
int sock = get_sock();
|
||||
|
||||
for (;;) {
|
||||
/* Check for external reconnect request */
|
||||
if (k_event_test(&s_pub_events, PUB_RECONNECT_BIT)) {
|
||||
k_event_clear(&s_pub_events, PUB_RECONNECT_BIT);
|
||||
LOG_INF("Reconnect requested");
|
||||
break;
|
||||
}
|
||||
|
||||
/* Drain publish queue */
|
||||
struct pub_msg msg;
|
||||
while (k_msgq_get(&s_pub_queue, &msg, K_NO_WAIT) == 0) {
|
||||
int rc = do_publish(&s_client, msg.topic,
|
||||
msg.payload, msg.payload_len, false);
|
||||
if (rc < 0) {
|
||||
LOG_WRN("Publish failed: %d", rc);
|
||||
s_connected = false;
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
/* Wait up to 100ms for incoming data (keepalive PINGRESP, etc.) */
|
||||
int pr = poll_socket(sock, 100);
|
||||
if (pr < 0) {
|
||||
LOG_WRN("Socket error in poll loop: %d", pr);
|
||||
s_connected = false;
|
||||
return;
|
||||
}
|
||||
if (pr > 0) {
|
||||
int rc = mqtt_input(&s_client);
|
||||
if (rc < 0 && rc != -EAGAIN && rc != -EWOULDBLOCK) {
|
||||
LOG_WRN("mqtt_input error: %d", rc);
|
||||
s_connected = false;
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
/* Send keepalive ping if needed */
|
||||
int rc = mqtt_live(&s_client);
|
||||
if (rc < 0 && rc != -EAGAIN) {
|
||||
LOG_WRN("mqtt_live error: %d", rc);
|
||||
s_connected = false;
|
||||
return;
|
||||
}
|
||||
|
||||
if (!s_connected) {
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/* ========== Publisher thread ========== */
|
||||
|
||||
#define MQTT_THREAD_STACK_SIZE 8192
|
||||
#define MQTT_THREAD_PRIORITY 5
|
||||
|
||||
static void mqtt_thread_fn(void *p1, void *p2, void *p3);
|
||||
K_THREAD_DEFINE(mqtt_pub_thread, MQTT_THREAD_STACK_SIZE,
|
||||
mqtt_thread_fn, NULL, NULL, NULL,
|
||||
MQTT_THREAD_PRIORITY, 0, 0);
|
||||
|
||||
static void mqtt_thread_fn(void *p1, void *p2, void *p3)
|
||||
{
|
||||
ARG_UNUSED(p1); ARG_UNUSED(p2); ARG_UNUSED(p3);
|
||||
|
||||
LOG_INF("MQTT publisher thread started");
|
||||
|
||||
for (;;) {
|
||||
s_connected = false;
|
||||
|
||||
/* Wait until WiFi is ready (blocks indefinitely) */
|
||||
LOG_INF("Waiting for WiFi ready...");
|
||||
k_event_wait(&g_wifi_events, WIFI_READY_BIT, false, K_FOREVER);
|
||||
|
||||
if (!s_creds || s_creds->mqtt_host[0] == '\0') {
|
||||
LOG_WRN("No MQTT host configured — set with: set mqtt.host <hostname>");
|
||||
k_sleep(K_SECONDS(10));
|
||||
continue;
|
||||
}
|
||||
|
||||
/* Clear reconnect flag before connecting */
|
||||
k_event_clear(&s_pub_events, PUB_RECONNECT_BIT);
|
||||
|
||||
/* Resolve broker hostname */
|
||||
int rc = resolve_host(s_creds->mqtt_host, s_creds->mqtt_port,
|
||||
&s_broker_addr);
|
||||
if (rc < 0) {
|
||||
LOG_WRN("DNS failed — retry in 10s");
|
||||
k_sleep(K_SECONDS(10));
|
||||
continue;
|
||||
}
|
||||
|
||||
/* Configure MQTT client */
|
||||
setup_client();
|
||||
|
||||
/* Connect */
|
||||
LOG_INF("Connecting to MQTT broker %s:%u (TLS=%u)",
|
||||
s_creds->mqtt_host, s_creds->mqtt_port,
|
||||
(unsigned)s_creds->mqtt_tls);
|
||||
|
||||
rc = mqtt_connect(&s_client);
|
||||
if (rc < 0) {
|
||||
LOG_WRN("mqtt_connect failed: %d — retry in 10s", rc);
|
||||
k_sleep(K_SECONDS(10));
|
||||
continue;
|
||||
}
|
||||
|
||||
/* Wait for CONNACK — poll up to 5s total, 500ms per iteration */
|
||||
{
|
||||
int sock = get_sock();
|
||||
for (int i = 0; i < 10 && !s_connected; i++) {
|
||||
int pr = poll_socket(sock, 500);
|
||||
if (pr < 0) {
|
||||
LOG_WRN("CONNACK wait: socket error %d", pr);
|
||||
break;
|
||||
}
|
||||
if (pr > 0) {
|
||||
int rc2 = mqtt_input(&s_client);
|
||||
if (rc2 < 0 && rc2 != -EAGAIN &&
|
||||
rc2 != -EWOULDBLOCK) {
|
||||
LOG_WRN("CONNACK wait: mqtt_input error %d", rc2);
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (!s_connected) {
|
||||
LOG_WRN("CONNACK not accepted (broker rejected or no response) — retry in 10s");
|
||||
mqtt_disconnect(&s_client, NULL);
|
||||
k_sleep(K_SECONDS(10));
|
||||
continue;
|
||||
}
|
||||
|
||||
/* Publish retained "online" status */
|
||||
publish_status(&s_client, true);
|
||||
|
||||
/* Run poll loop until disconnect or reconnect request */
|
||||
run_poll_loop();
|
||||
|
||||
/* Publish "offline" before closing (best-effort) */
|
||||
if (s_connected) {
|
||||
publish_status(&s_client, false);
|
||||
}
|
||||
mqtt_disconnect(&s_client, NULL);
|
||||
|
||||
LOG_INF("MQTT session ended — retry in 5s");
|
||||
k_sleep(K_SECONDS(5));
|
||||
}
|
||||
}
|
||||
|
||||
/* ========== Public API ========== */
|
||||
|
||||
void mqtt_publisher_start(const struct ObserverCreds *creds,
|
||||
const char *client_id,
|
||||
const char *status_topic,
|
||||
const char *packets_topic)
|
||||
{
|
||||
s_creds = creds;
|
||||
strncpy(s_client_id, client_id, sizeof(s_client_id) - 1);
|
||||
strncpy(s_status_topic, status_topic, sizeof(s_status_topic) - 1);
|
||||
strncpy(s_packets_topic, packets_topic, sizeof(s_packets_topic) - 1);
|
||||
s_client_id[sizeof(s_client_id) - 1] = '\0';
|
||||
s_status_topic[sizeof(s_status_topic) - 1] = '\0';
|
||||
s_packets_topic[sizeof(s_packets_topic) - 1] = '\0';
|
||||
|
||||
/* Thread is already created by K_THREAD_DEFINE — nothing more to do. */
|
||||
LOG_INF("MQTT publisher ready (client_id=%s)", s_client_id);
|
||||
}
|
||||
|
||||
void mqtt_publisher_enqueue(const char *topic, const char *payload, int payload_len)
|
||||
{
|
||||
struct pub_msg msg;
|
||||
|
||||
strncpy(msg.topic, topic, sizeof(msg.topic) - 1);
|
||||
msg.topic[sizeof(msg.topic) - 1] = '\0';
|
||||
|
||||
int copy_len = payload_len;
|
||||
if (copy_len >= (int)sizeof(msg.payload)) {
|
||||
copy_len = (int)sizeof(msg.payload) - 1;
|
||||
}
|
||||
memcpy(msg.payload, payload, copy_len);
|
||||
msg.payload[copy_len] = '\0';
|
||||
msg.payload_len = (uint16_t)copy_len;
|
||||
|
||||
if (k_msgq_put(&s_pub_queue, &msg, K_NO_WAIT) != 0) {
|
||||
LOG_WRN("Publish queue full — packet dropped");
|
||||
}
|
||||
}
|
||||
|
||||
bool mqtt_publisher_is_connected(void)
|
||||
{
|
||||
return s_connected;
|
||||
}
|
||||
|
||||
void mqtt_publisher_reconnect(void)
|
||||
{
|
||||
k_event_post(&s_pub_events, PUB_RECONNECT_BIT);
|
||||
}
|
||||
@@ -0,0 +1,53 @@
|
||||
/*
|
||||
* SPDX-License-Identifier: Apache-2.0
|
||||
* ZephyrMQTTPublisher — MQTT client thread for the Observer role.
|
||||
*
|
||||
* Dedicated thread: waits for WiFi ready, resolves broker hostname,
|
||||
* connects with optional TLS (TLS_PEER_VERIFY_NONE), publishes LWT and
|
||||
* status, then drains a pre-serialized JSON publish queue.
|
||||
*/
|
||||
|
||||
#pragma once
|
||||
|
||||
#include <stdbool.h>
|
||||
|
||||
#ifdef __cplusplus
|
||||
extern "C" {
|
||||
#endif
|
||||
|
||||
struct ObserverCreds;
|
||||
|
||||
/*
|
||||
* Start the MQTT publisher thread.
|
||||
* creds — runtime credentials; pointer kept, must remain valid.
|
||||
* client_id — MQTT client identifier string (e.g. "Observer-AABBCCDD").
|
||||
* status_topic — topic for retained online/offline status messages.
|
||||
* packets_topic — topic for received LoRa packet JSON messages.
|
||||
*
|
||||
* Must be called once from main() after wifi_station_start().
|
||||
*/
|
||||
void mqtt_publisher_start(const struct ObserverCreds *creds,
|
||||
const char *client_id,
|
||||
const char *status_topic,
|
||||
const char *packets_topic);
|
||||
|
||||
/*
|
||||
* Enqueue a pre-serialized JSON payload for publishing.
|
||||
* topic and payload are copied — safe to pass stack buffers.
|
||||
* Drops the message silently if the queue is full.
|
||||
* Safe to call from any thread.
|
||||
*/
|
||||
void mqtt_publisher_enqueue(const char *topic, const char *payload, int payload_len);
|
||||
|
||||
/* Returns true when the MQTT session is active. */
|
||||
bool mqtt_publisher_is_connected(void);
|
||||
|
||||
/*
|
||||
* Trigger an MQTT reconnect (e.g. after credential change via CLI).
|
||||
* Safe to call from any thread.
|
||||
*/
|
||||
void mqtt_publisher_reconnect(void);
|
||||
|
||||
#ifdef __cplusplus
|
||||
}
|
||||
#endif
|
||||
@@ -0,0 +1,210 @@
|
||||
/*
|
||||
* SPDX-License-Identifier: Apache-2.0
|
||||
* ZephyrWiFiStation — WiFi STA connection manager for the Observer role.
|
||||
*
|
||||
* Flow:
|
||||
* wifi_station_start() → issues NET_REQUEST_WIFI_CONNECT
|
||||
* NET_EVENT_WIFI_CONNECT_RESULT → link up, wait for DHCP
|
||||
* NET_EVENT_IPV4_DHCP_BOUND → IP assigned → run SNTP → signal WIFI_READY_BIT
|
||||
* NET_EVENT_WIFI_DISCONNECT_RESULT → clear WIFI_READY_BIT, schedule reconnect in 5s
|
||||
*/
|
||||
|
||||
#include "ZephyrWiFiStation.h"
|
||||
#include "observer_creds.h"
|
||||
|
||||
#include <zephyr/kernel.h>
|
||||
#include <zephyr/net/net_if.h>
|
||||
#include <zephyr/net/net_event.h>
|
||||
#include <zephyr/net/net_mgmt.h>
|
||||
#include <zephyr/net/wifi_mgmt.h>
|
||||
#include <zephyr/net/sntp.h>
|
||||
#include <zephyr/logging/log.h>
|
||||
|
||||
#include <string.h>
|
||||
#include <stdio.h>
|
||||
|
||||
LOG_MODULE_REGISTER(wifi_station, CONFIG_LOG_DEFAULT_LEVEL);
|
||||
|
||||
/* ========== Shared event object ========== */
|
||||
|
||||
K_EVENT_DEFINE(g_wifi_events);
|
||||
|
||||
/* ========== Module state ========== */
|
||||
|
||||
static const struct ObserverCreds *s_creds;
|
||||
static void (*s_time_sync_cb)(uint32_t unix_ts);
|
||||
|
||||
/* Protect link-up vs. SNTP state */
|
||||
static volatile bool s_wifi_link_up; /* WiFi associate event received */
|
||||
static volatile bool s_wifi_ready; /* DHCP + SNTP done */
|
||||
|
||||
/* ========== Reconnect work ========== */
|
||||
|
||||
static void connect_work_fn(struct k_work *work);
|
||||
static K_WORK_DELAYABLE_DEFINE(connect_work, connect_work_fn);
|
||||
|
||||
/* ========== net_mgmt callbacks ========== */
|
||||
|
||||
static struct net_mgmt_event_callback wifi_cb;
|
||||
static struct net_mgmt_event_callback ipv4_cb;
|
||||
|
||||
static void do_sntp_and_signal(void)
|
||||
{
|
||||
struct sntp_time ts;
|
||||
int rc = sntp_simple("pool.ntp.org", 8000, &ts);
|
||||
if (rc == 0) {
|
||||
LOG_INF("SNTP synced: %llu", (unsigned long long)ts.seconds);
|
||||
if (s_time_sync_cb) {
|
||||
s_time_sync_cb((uint32_t)ts.seconds);
|
||||
}
|
||||
} else {
|
||||
LOG_WRN("SNTP failed (rc=%d) — continuing without time sync", rc);
|
||||
}
|
||||
|
||||
s_wifi_ready = true;
|
||||
k_event_post(&g_wifi_events, WIFI_READY_BIT);
|
||||
LOG_INF("WiFi ready (DHCP+SNTP done)");
|
||||
}
|
||||
|
||||
/* Called when DHCP has bound an address */
|
||||
static void ipv4_event_handler(struct net_mgmt_event_callback *cb,
|
||||
uint64_t event, struct net_if *iface)
|
||||
{
|
||||
ARG_UNUSED(iface);
|
||||
if (event == NET_EVENT_IPV4_DHCP_BOUND) {
|
||||
LOG_INF("DHCP bound — starting SNTP sync");
|
||||
/* Run SNTP — this is called from the net_mgmt work queue.
|
||||
* sntp_simple() may block for up to 8s; net_mgmt stack must be
|
||||
* large enough (CONFIG_NET_MGMT_EVENT_STACK_SIZE >= 3072). */
|
||||
do_sntp_and_signal();
|
||||
}
|
||||
}
|
||||
|
||||
static void wifi_event_handler(struct net_mgmt_event_callback *cb,
|
||||
uint64_t event, struct net_if *iface)
|
||||
{
|
||||
switch (event) {
|
||||
case NET_EVENT_WIFI_CONNECT_RESULT: {
|
||||
#ifdef CONFIG_NET_MGMT_EVENT_INFO
|
||||
const struct wifi_status *status =
|
||||
(const struct wifi_status *)cb->info;
|
||||
bool success = (status == NULL ||
|
||||
status->conn_status == WIFI_STATUS_CONN_SUCCESS);
|
||||
int reason = status ? (int)status->conn_status : 0;
|
||||
#else
|
||||
bool success = true;
|
||||
int reason = 0;
|
||||
#endif
|
||||
if (success) {
|
||||
LOG_INF("WiFi link up (SSID: %s)", s_creds ? s_creds->wifi_ssid : "?");
|
||||
s_wifi_link_up = true;
|
||||
/* DHCP will fire ipv4_event_handler when lease is obtained */
|
||||
} else {
|
||||
LOG_WRN("WiFi connect failed (reason=%d) — retry in 10s", reason);
|
||||
k_work_reschedule(&connect_work, K_SECONDS(10));
|
||||
}
|
||||
break;
|
||||
}
|
||||
case NET_EVENT_WIFI_DISCONNECT_RESULT:
|
||||
LOG_INF("WiFi disconnected");
|
||||
s_wifi_link_up = false;
|
||||
s_wifi_ready = false;
|
||||
k_event_clear(&g_wifi_events, WIFI_READY_BIT);
|
||||
/* Reconnect after 5s backoff */
|
||||
k_work_reschedule(&connect_work, K_SECONDS(5));
|
||||
break;
|
||||
default:
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
/* ========== Connect work ========== */
|
||||
|
||||
static void connect_work_fn(struct k_work *work)
|
||||
{
|
||||
ARG_UNUSED(work);
|
||||
|
||||
if (!s_creds || s_creds->wifi_ssid[0] == '\0') {
|
||||
LOG_WRN("No WiFi SSID configured — set with: set wifi.ssid <name>");
|
||||
return;
|
||||
}
|
||||
|
||||
struct net_if *iface = net_if_get_default();
|
||||
if (!iface) {
|
||||
LOG_ERR("No network interface");
|
||||
return;
|
||||
}
|
||||
|
||||
struct wifi_connect_req_params params = {
|
||||
.ssid = (const uint8_t *)s_creds->wifi_ssid,
|
||||
.ssid_length = (uint8_t)strlen(s_creds->wifi_ssid),
|
||||
.channel = WIFI_CHANNEL_ANY,
|
||||
.band = WIFI_FREQ_BAND_UNKNOWN,
|
||||
.mfp = WIFI_MFP_OPTIONAL,
|
||||
};
|
||||
|
||||
if (s_creds->wifi_psk[0] != '\0') {
|
||||
params.psk = (const uint8_t *)s_creds->wifi_psk;
|
||||
params.psk_length = (uint8_t)strlen(s_creds->wifi_psk);
|
||||
params.security = WIFI_SECURITY_TYPE_PSK;
|
||||
} else {
|
||||
params.security = WIFI_SECURITY_TYPE_NONE;
|
||||
}
|
||||
|
||||
LOG_INF("Connecting to WiFi: %s", s_creds->wifi_ssid);
|
||||
int rc = net_mgmt(NET_REQUEST_WIFI_CONNECT, iface, ¶ms, sizeof(params));
|
||||
if (rc < 0 && rc != -EALREADY) {
|
||||
LOG_ERR("WiFi connect request failed: %d — retry in 10s", rc);
|
||||
k_work_reschedule(&connect_work, K_SECONDS(10));
|
||||
}
|
||||
}
|
||||
|
||||
/* ========== Public API ========== */
|
||||
|
||||
void zc_wifi_station_start(const struct ObserverCreds *creds,
|
||||
void (*time_sync_cb)(uint32_t unix_ts))
|
||||
{
|
||||
s_creds = creds;
|
||||
s_time_sync_cb = time_sync_cb;
|
||||
s_wifi_link_up = false;
|
||||
s_wifi_ready = false;
|
||||
|
||||
/* Register WiFi event callback */
|
||||
net_mgmt_init_event_callback(&wifi_cb, wifi_event_handler,
|
||||
NET_EVENT_WIFI_CONNECT_RESULT |
|
||||
NET_EVENT_WIFI_DISCONNECT_RESULT);
|
||||
net_mgmt_add_event_callback(&wifi_cb);
|
||||
|
||||
/* Register IPv4 DHCP callback */
|
||||
net_mgmt_init_event_callback(&ipv4_cb, ipv4_event_handler,
|
||||
NET_EVENT_IPV4_DHCP_BOUND);
|
||||
net_mgmt_add_event_callback(&ipv4_cb);
|
||||
|
||||
/* Trigger first connect attempt */
|
||||
k_work_schedule(&connect_work, K_MSEC(500));
|
||||
}
|
||||
|
||||
void zc_wifi_station_reconnect(void)
|
||||
{
|
||||
/* Attempt to disconnect first; ignore errors (may already be disconnected) */
|
||||
struct net_if *iface = net_if_get_default();
|
||||
if (iface) {
|
||||
net_mgmt(NET_REQUEST_WIFI_DISCONNECT, iface, NULL, 0);
|
||||
}
|
||||
|
||||
/* Clear state and immediately reschedule connect */
|
||||
s_wifi_link_up = false;
|
||||
s_wifi_ready = false;
|
||||
k_event_clear(&g_wifi_events, WIFI_READY_BIT);
|
||||
k_work_reschedule(&connect_work, K_MSEC(200));
|
||||
}
|
||||
|
||||
bool zc_wifi_station_is_connected(void)
|
||||
{
|
||||
return s_wifi_ready;
|
||||
}
|
||||
|
||||
const char *zc_wifi_station_ssid(void)
|
||||
{
|
||||
return (s_creds && s_creds->wifi_ssid[0]) ? s_creds->wifi_ssid : "";
|
||||
}
|
||||
@@ -0,0 +1,54 @@
|
||||
/*
|
||||
* SPDX-License-Identifier: Apache-2.0
|
||||
* ZephyrWiFiStation — WiFi STA connection manager for the Observer role.
|
||||
*
|
||||
* Connects to a WPA2-PSK (or open) network, obtains a DHCP lease, syncs
|
||||
* time via SNTP, then signals g_wifi_events / WIFI_READY_BIT so the MQTT
|
||||
* publisher thread can start. Auto-reconnects on link loss.
|
||||
*/
|
||||
|
||||
#pragma once
|
||||
|
||||
#include <stdbool.h>
|
||||
#include <stdint.h>
|
||||
#include <zephyr/kernel.h>
|
||||
|
||||
#ifdef __cplusplus
|
||||
extern "C" {
|
||||
#endif
|
||||
|
||||
struct ObserverCreds;
|
||||
|
||||
/*
|
||||
* Event object and bit shared with ZephyrMQTTPublisher so it can wait
|
||||
* for WiFi+SNTP to be ready before connecting to the broker.
|
||||
*/
|
||||
extern struct k_event g_wifi_events;
|
||||
#define WIFI_READY_BIT BIT(0) /* WiFi connected + DHCP + SNTP done */
|
||||
#define WIFI_RECONNECT_BIT BIT(1) /* Internal: trigger reconnect */
|
||||
|
||||
/*
|
||||
* Start the WiFi STA subsystem.
|
||||
* creds — runtime credentials (SSID, PSK); pointer kept, must remain valid.
|
||||
* time_sync_cb — called once after SNTP with Unix timestamp; may be NULL.
|
||||
*
|
||||
* Must be called once from main() before any other wifi_station_* calls.
|
||||
*/
|
||||
void zc_wifi_station_start(const struct ObserverCreds *creds,
|
||||
void (*time_sync_cb)(uint32_t unix_ts));
|
||||
|
||||
/*
|
||||
* Trigger an immediate reconnect (e.g. after credential change via CLI).
|
||||
* Safe to call from any thread.
|
||||
*/
|
||||
void zc_wifi_station_reconnect(void);
|
||||
|
||||
/* Returns true when WiFi is connected AND DHCP+SNTP have completed. */
|
||||
bool zc_wifi_station_is_connected(void);
|
||||
|
||||
/* Returns the SSID currently connected to, or an empty string. */
|
||||
const char *zc_wifi_station_ssid(void);
|
||||
|
||||
#ifdef __cplusplus
|
||||
}
|
||||
#endif
|
||||
@@ -0,0 +1,447 @@
|
||||
/*
|
||||
* SPDX-License-Identifier: Apache-2.0
|
||||
* ObserverMesh — listen-only LoRa mesh node implementation.
|
||||
*/
|
||||
|
||||
#include "ObserverMesh.h"
|
||||
#include "observer_creds.h"
|
||||
|
||||
#include <mesh/Utils.h>
|
||||
#include <mesh/LoRaConfig.h>
|
||||
#include <adapters/radio/LoRaRadioBase.h>
|
||||
|
||||
#include <zephyr/logging/log.h>
|
||||
LOG_MODULE_REGISTER(zephcore_observer, CONFIG_ZEPHCORE_OBSERVER_LOG_LEVEL);
|
||||
|
||||
#include <stdio.h>
|
||||
#include <string.h>
|
||||
#include <stdlib.h>
|
||||
#include <time.h>
|
||||
|
||||
/* Forward declaration — implemented in ZephyrMQTTPublisher.c */
|
||||
extern "C" {
|
||||
void mqtt_publisher_enqueue(const char *topic, const char *payload, int payload_len);
|
||||
bool mqtt_publisher_is_connected(void);
|
||||
void mqtt_publisher_reconnect(void);
|
||||
}
|
||||
|
||||
/* Forward declaration — implemented in ZephyrWiFiStation.c */
|
||||
extern "C" {
|
||||
bool zc_wifi_station_is_connected(void);
|
||||
void zc_wifi_station_reconnect(void);
|
||||
const char *zc_wifi_station_ssid(void);
|
||||
}
|
||||
|
||||
namespace mesh {
|
||||
|
||||
/* ========== Construction ========== */
|
||||
|
||||
ObserverMesh::ObserverMesh(Radio &radio, MillisecondClock &ms, RNG &rng, RTCClock &rtc)
|
||||
: Dispatcher(radio, ms, _pkt_mgr),
|
||||
_last_rssi(0.0f), _last_score(0.0f), _last_raw_len(0),
|
||||
_store(nullptr), _creds(nullptr), _rng(&rng), _rtc(&rtc)
|
||||
{
|
||||
memset(_pubkey_hex, 0, sizeof(_pubkey_hex));
|
||||
memset(_packets_topic, 0, sizeof(_packets_topic));
|
||||
memset(_status_topic, 0, sizeof(_status_topic));
|
||||
}
|
||||
|
||||
/* ========== begin() ========== */
|
||||
|
||||
void ObserverMesh::begin(RepeaterDataStore *store, struct ObserverCreds *creds)
|
||||
{
|
||||
_store = store;
|
||||
_creds = creds;
|
||||
|
||||
/* Initialize prefs with observer-specific defaults */
|
||||
initNodePrefs(&_prefs);
|
||||
_prefs.cr = 5; /* CR 4/5 */
|
||||
_prefs.tx_power_dbm = 0; /* observer never TXes anyway */
|
||||
/* freq=869.618, bw=62.5, sf=8 already set by initNodePrefs */
|
||||
|
||||
/* Load persisted prefs (overrides defaults with saved values) */
|
||||
if (!_store->loadPrefs(_prefs)) {
|
||||
/* First boot — save observer defaults */
|
||||
_store->savePrefs(_prefs);
|
||||
}
|
||||
|
||||
/* Load or generate node identity */
|
||||
if (!_store->loadIdentity(_self_id)) {
|
||||
LOG_INF("No identity found — generating new keypair");
|
||||
int attempts = 0;
|
||||
do {
|
||||
_self_id = LocalIdentity(_rng);
|
||||
attempts++;
|
||||
} while (attempts < 10 &&
|
||||
(_self_id.pub_key[0] == 0x00 || _self_id.pub_key[0] == 0xFF));
|
||||
_store->saveIdentity(_self_id);
|
||||
LOG_INF("New observer identity saved");
|
||||
}
|
||||
|
||||
/* Build hex pubkey string */
|
||||
Utils::toHex(_pubkey_hex, _self_id.pub_key, PUB_KEY_SIZE);
|
||||
_pubkey_hex[PUB_KEY_SIZE * 2] = '\0';
|
||||
|
||||
/* Build MQTT topic strings */
|
||||
buildTopics();
|
||||
|
||||
/* Log identity */
|
||||
LOG_INF("Observer ID: %.16s...", _pubkey_hex);
|
||||
|
||||
/* Start radio in continuous RX mode */
|
||||
Dispatcher::begin();
|
||||
}
|
||||
|
||||
void ObserverMesh::buildTopics()
|
||||
{
|
||||
const char *iata = (_creds && _creds->mqtt_iata[0] != '\0')
|
||||
? _creds->mqtt_iata : "XXX";
|
||||
|
||||
snprintf(_packets_topic, sizeof(_packets_topic),
|
||||
"meshcore/%s/%s/packets", iata, _pubkey_hex);
|
||||
snprintf(_status_topic, sizeof(_status_topic),
|
||||
"meshcore/%s/%s/status", iata, _pubkey_hex);
|
||||
}
|
||||
|
||||
/* ========== RX hooks ========== */
|
||||
|
||||
void ObserverMesh::logRxRaw(float snr, float rssi, const uint8_t raw[], int len)
|
||||
{
|
||||
_last_rssi = rssi;
|
||||
_last_raw_len = (len <= (int)sizeof(_last_raw)) ? len : (int)sizeof(_last_raw);
|
||||
memcpy(_last_raw, raw, _last_raw_len);
|
||||
}
|
||||
|
||||
void ObserverMesh::logRx(Packet *packet, int len, float score)
|
||||
{
|
||||
(void)packet; (void)len;
|
||||
_last_score = score;
|
||||
}
|
||||
|
||||
void ObserverMesh::enqueuePacket(Packet *pkt)
|
||||
{
|
||||
/* Compute packet hash (8 bytes → 16 hex chars) */
|
||||
uint8_t hash_bytes[MAX_HASH_SIZE];
|
||||
char hash_hex[MAX_HASH_SIZE * 2 + 1];
|
||||
pkt->calculatePacketHash(hash_bytes);
|
||||
Utils::toHex(hash_hex, hash_bytes, MAX_HASH_SIZE);
|
||||
hash_hex[MAX_HASH_SIZE * 2] = '\0';
|
||||
|
||||
/* Encode raw wire bytes as hex */
|
||||
/* Each byte → 2 hex chars; max MAX_TRANS_UNIT=255 bytes → 510 chars + NUL */
|
||||
static char raw_hex[MAX_TRANS_UNIT * 2 + 1];
|
||||
Utils::toHex(raw_hex, _last_raw, _last_raw_len);
|
||||
raw_hex[_last_raw_len * 2] = '\0';
|
||||
|
||||
/* Get current timestamp from RTC */
|
||||
uint32_t now_epoch = _rtc ? _rtc->getCurrentTime() : 0;
|
||||
struct tm tm_now;
|
||||
time_t t = (time_t)now_epoch;
|
||||
gmtime_r(&t, &tm_now);
|
||||
|
||||
/* Format ISO 8601 timestamp (microseconds always 0 — RTC has 1s resolution) */
|
||||
char ts_buf[48];
|
||||
snprintf(ts_buf, sizeof(ts_buf),
|
||||
"%04d-%02d-%02dT%02d:%02d:%02d.000000",
|
||||
tm_now.tm_year + 1900, tm_now.tm_mon + 1, tm_now.tm_mday,
|
||||
tm_now.tm_hour, tm_now.tm_min, tm_now.tm_sec);
|
||||
|
||||
/* Format time and date fields matching meshcoretomqtt */
|
||||
char time_buf[12], date_buf[32];
|
||||
snprintf(time_buf, sizeof(time_buf), "%02d:%02d:%02d",
|
||||
tm_now.tm_hour, tm_now.tm_min, tm_now.tm_sec);
|
||||
snprintf(date_buf, sizeof(date_buf), "%d/%d/%04d",
|
||||
tm_now.tm_mday, tm_now.tm_mon + 1, tm_now.tm_year + 1900);
|
||||
|
||||
/* Route letter: "F" = flood/transport-flood, "D" = direct/transport-direct */
|
||||
const char *route_str = pkt->isRouteDirect() ? "D" : "F";
|
||||
|
||||
/* score in meshcoretomqtt format: integer score * 1000 */
|
||||
int score_int = (int)(_last_score * 1000.0f);
|
||||
|
||||
/* Build JSON payload matching meshcoretomqtt packet format */
|
||||
static char json_buf[1024];
|
||||
int json_len = snprintf(json_buf, sizeof(json_buf),
|
||||
"{"
|
||||
"\"type\":\"PACKET\","
|
||||
"\"origin\":\"%s\","
|
||||
"\"origin_id\":\"%s\","
|
||||
"\"timestamp\":\"%s\","
|
||||
"\"direction\":\"rx\","
|
||||
"\"time\":\"%s\","
|
||||
"\"date\":\"%s\","
|
||||
"\"len\":\"%d\","
|
||||
"\"packet_type\":\"%u\","
|
||||
"\"route\":\"%s\","
|
||||
"\"payload_len\":\"%u\","
|
||||
"\"raw\":\"%s\","
|
||||
"\"SNR\":\"%d\","
|
||||
"\"RSSI\":\"%d\","
|
||||
"\"score\":\"%d\","
|
||||
"\"hash\":\"%s\""
|
||||
"}",
|
||||
_prefs.node_name,
|
||||
_pubkey_hex,
|
||||
ts_buf,
|
||||
time_buf,
|
||||
date_buf,
|
||||
_last_raw_len,
|
||||
(unsigned)pkt->getPayloadType(),
|
||||
route_str,
|
||||
(unsigned)pkt->payload_len,
|
||||
raw_hex,
|
||||
(int)pkt->getSNR(),
|
||||
(int)_last_rssi,
|
||||
score_int,
|
||||
hash_hex);
|
||||
|
||||
if (json_len < 0 || json_len >= (int)sizeof(json_buf)) {
|
||||
LOG_WRN("Packet JSON truncated (len=%d)", json_len);
|
||||
json_len = (int)sizeof(json_buf) - 1;
|
||||
}
|
||||
|
||||
mqtt_publisher_enqueue(_packets_topic, json_buf, json_len);
|
||||
}
|
||||
|
||||
DispatcherAction ObserverMesh::onRecvPacket(Packet *pkt)
|
||||
{
|
||||
/* Publish every reception — no deduplication.
|
||||
* The same flood packet heard from different repeaters is published
|
||||
* separately, each with its own SNR/RSSI (propagation data). */
|
||||
enqueuePacket(pkt);
|
||||
return ACTION_RELEASE; /* never retransmit */
|
||||
}
|
||||
|
||||
/* ========== Serial CLI ========== */
|
||||
|
||||
#define CLI_REPLY_SIZE 256
|
||||
|
||||
bool ObserverMesh::handleCLI(const char *command, char *reply, int reply_size)
|
||||
{
|
||||
reply[0] = '\0';
|
||||
|
||||
/* ---- help ---- */
|
||||
if (strcmp(command, "help") == 0 || command[0] == '\0') {
|
||||
return true; /* caller prints the banner */
|
||||
}
|
||||
|
||||
/* ---- get commands ---- */
|
||||
if (memcmp(command, "get ", 4) == 0) {
|
||||
const char *key = command + 4;
|
||||
|
||||
if (strcmp(key, "role") == 0) {
|
||||
snprintf(reply, reply_size, "observer");
|
||||
|
||||
} else if (strcmp(key, "name") == 0) {
|
||||
snprintf(reply, reply_size, "%s", _prefs.node_name);
|
||||
|
||||
} else if (strcmp(key, "public.key") == 0) {
|
||||
snprintf(reply, reply_size, "%s", _pubkey_hex);
|
||||
|
||||
} else if (strcmp(key, "board") == 0) {
|
||||
#ifdef CONFIG_ZEPHCORE_BOARD_NAME
|
||||
snprintf(reply, reply_size, "%s", CONFIG_ZEPHCORE_BOARD_NAME);
|
||||
#else
|
||||
snprintf(reply, reply_size, "unknown");
|
||||
#endif
|
||||
} else if (strcmp(key, "version") == 0) {
|
||||
snprintf(reply, reply_size, "%s (%s)", FIRMWARE_VERSION, FIRMWARE_BUILD_DATE);
|
||||
|
||||
} else if (strcmp(key, "radio") == 0) {
|
||||
snprintf(reply, reply_size,
|
||||
"freq=%.3f bw=%.1f sf=%u cr=%u tx=%ddBm",
|
||||
(double)_prefs.freq, (double)_prefs.bw,
|
||||
_prefs.sf, _prefs.cr, _prefs.tx_power_dbm);
|
||||
|
||||
} else if (strcmp(key, "wifi.ssid") == 0) {
|
||||
snprintf(reply, reply_size, "%s",
|
||||
(_creds && _creds->wifi_ssid[0]) ? _creds->wifi_ssid : "(not set)");
|
||||
|
||||
} else if (strcmp(key, "wifi.status") == 0) {
|
||||
snprintf(reply, reply_size, "%s",
|
||||
zc_wifi_station_is_connected() ? "connected" : "disconnected");
|
||||
|
||||
} else if (strcmp(key, "mqtt.status") == 0) {
|
||||
snprintf(reply, reply_size, "%s",
|
||||
mqtt_publisher_is_connected() ? "connected" : "disconnected");
|
||||
|
||||
} else if (strcmp(key, "mqtt.host") == 0) {
|
||||
snprintf(reply, reply_size, "%s",
|
||||
(_creds && _creds->mqtt_host[0]) ? _creds->mqtt_host : "(not set)");
|
||||
|
||||
} else if (strcmp(key, "mqtt.user") == 0) {
|
||||
snprintf(reply, reply_size, "%s",
|
||||
(_creds && _creds->mqtt_user[0]) ? _creds->mqtt_user : "(not set)");
|
||||
|
||||
} else if (strcmp(key, "mqtt.iata") == 0) {
|
||||
snprintf(reply, reply_size, "%s",
|
||||
(_creds && _creds->mqtt_iata[0]) ? _creds->mqtt_iata : "(not set)");
|
||||
|
||||
} else if (strcmp(key, "mqtt.port") == 0) {
|
||||
snprintf(reply, reply_size, "%u",
|
||||
(_creds) ? (unsigned)_creds->mqtt_port : 8883u);
|
||||
|
||||
} else if (strcmp(key, "mqtt.tls") == 0) {
|
||||
snprintf(reply, reply_size, "%u",
|
||||
(_creds) ? (unsigned)_creds->mqtt_tls : 1u);
|
||||
|
||||
} else {
|
||||
snprintf(reply, reply_size, "ERR unknown key: %s", key);
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
/* ---- set commands ---- */
|
||||
if (memcmp(command, "set ", 4) == 0) {
|
||||
const char *rest = command + 4;
|
||||
|
||||
/* Helper: find value after "key " */
|
||||
auto find_val = [](const char *s, const char *prefix) -> const char * {
|
||||
size_t n = strlen(prefix);
|
||||
if (memcmp(s, prefix, n) == 0 && s[n] == ' ')
|
||||
return s + n + 1;
|
||||
return nullptr;
|
||||
};
|
||||
|
||||
const char *val;
|
||||
|
||||
if ((val = find_val(rest, "name")) != nullptr) {
|
||||
strncpy(_prefs.node_name, val, sizeof(_prefs.node_name) - 1);
|
||||
_prefs.node_name[sizeof(_prefs.node_name) - 1] = '\0';
|
||||
_store->savePrefs(_prefs);
|
||||
snprintf(reply, reply_size, "name=%s", _prefs.node_name);
|
||||
|
||||
} else if ((val = find_val(rest, "freq")) != nullptr) {
|
||||
float f = (float)atof(val);
|
||||
/* Accept Hz (e.g. 869618000) or MHz (e.g. 869.618) */
|
||||
if (f > 1000000.0f) f /= 1000000.0f;
|
||||
if (f >= 150.0f && f <= 2500.0f) {
|
||||
_prefs.freq = f;
|
||||
_store->savePrefs(_prefs);
|
||||
((LoRaRadioBase *)_radio)->reconfigureWithParams(
|
||||
_prefs.freq, _prefs.bw, _prefs.sf, _prefs.cr);
|
||||
snprintf(reply, reply_size, "freq=%.3f MHz", (double)_prefs.freq);
|
||||
} else {
|
||||
snprintf(reply, reply_size, "ERR freq out of range");
|
||||
}
|
||||
|
||||
} else if ((val = find_val(rest, "sf")) != nullptr) {
|
||||
int sf = atoi(val);
|
||||
if (sf >= 7 && sf <= 12) {
|
||||
_prefs.sf = (uint8_t)sf;
|
||||
_store->savePrefs(_prefs);
|
||||
((LoRaRadioBase *)_radio)->reconfigureWithParams(
|
||||
_prefs.freq, _prefs.bw, _prefs.sf, _prefs.cr);
|
||||
snprintf(reply, reply_size, "sf=%u", _prefs.sf);
|
||||
} else {
|
||||
snprintf(reply, reply_size, "ERR sf must be 7-12");
|
||||
}
|
||||
|
||||
} else if ((val = find_val(rest, "bw")) != nullptr) {
|
||||
/* Accept index (0=125, 1=250, 2=500, 3=62.5, 4=41.7, 5=31.25)
|
||||
* or kHz value directly */
|
||||
float bw;
|
||||
int idx = atoi(val);
|
||||
const float bw_table[] = { 125.0f, 250.0f, 500.0f, 62.5f, 41.7f, 31.25f };
|
||||
if (idx >= 0 && idx <= 5) {
|
||||
bw = bw_table[idx];
|
||||
} else {
|
||||
bw = (float)atof(val);
|
||||
}
|
||||
if (bw > 0.0f) {
|
||||
_prefs.bw = bw;
|
||||
_store->savePrefs(_prefs);
|
||||
((LoRaRadioBase *)_radio)->reconfigureWithParams(
|
||||
_prefs.freq, _prefs.bw, _prefs.sf, _prefs.cr);
|
||||
snprintf(reply, reply_size, "bw=%.2f kHz", (double)_prefs.bw);
|
||||
} else {
|
||||
snprintf(reply, reply_size, "ERR invalid bw");
|
||||
}
|
||||
|
||||
} else if ((val = find_val(rest, "cr")) != nullptr) {
|
||||
int cr = atoi(val);
|
||||
if (cr >= 5 && cr <= 8) {
|
||||
_prefs.cr = (uint8_t)cr;
|
||||
_store->savePrefs(_prefs);
|
||||
((LoRaRadioBase *)_radio)->reconfigureWithParams(
|
||||
_prefs.freq, _prefs.bw, _prefs.sf, _prefs.cr);
|
||||
snprintf(reply, reply_size, "cr=%u", _prefs.cr);
|
||||
} else {
|
||||
snprintf(reply, reply_size, "ERR cr must be 5-8");
|
||||
}
|
||||
|
||||
} else if (!_creds) {
|
||||
snprintf(reply, reply_size, "ERR creds not initialized");
|
||||
|
||||
} else if ((val = find_val(rest, "wifi.ssid")) != nullptr) {
|
||||
strncpy(_creds->wifi_ssid, val, sizeof(_creds->wifi_ssid) - 1);
|
||||
_creds->wifi_ssid[sizeof(_creds->wifi_ssid) - 1] = '\0';
|
||||
observer_creds_save(_creds, _store->getBasePath());
|
||||
snprintf(reply, reply_size, "wifi.ssid=%s (reconnecting)", _creds->wifi_ssid);
|
||||
zc_wifi_station_reconnect();
|
||||
|
||||
} else if ((val = find_val(rest, "wifi.psk")) != nullptr) {
|
||||
strncpy(_creds->wifi_psk, val, sizeof(_creds->wifi_psk) - 1);
|
||||
_creds->wifi_psk[sizeof(_creds->wifi_psk) - 1] = '\0';
|
||||
observer_creds_save(_creds, _store->getBasePath());
|
||||
snprintf(reply, reply_size, "wifi.psk=*** (saved, reconnecting)");
|
||||
zc_wifi_station_reconnect();
|
||||
|
||||
} else if ((val = find_val(rest, "mqtt.host")) != nullptr) {
|
||||
strncpy(_creds->mqtt_host, val, sizeof(_creds->mqtt_host) - 1);
|
||||
_creds->mqtt_host[sizeof(_creds->mqtt_host) - 1] = '\0';
|
||||
observer_creds_save(_creds, _store->getBasePath());
|
||||
snprintf(reply, reply_size, "mqtt.host=%s (reconnecting)", _creds->mqtt_host);
|
||||
mqtt_publisher_reconnect();
|
||||
|
||||
} else if ((val = find_val(rest, "mqtt.port")) != nullptr) {
|
||||
int port = atoi(val);
|
||||
if (port > 0 && port <= 65535) {
|
||||
_creds->mqtt_port = (uint16_t)port;
|
||||
observer_creds_save(_creds, _store->getBasePath());
|
||||
snprintf(reply, reply_size, "mqtt.port=%u (reconnecting)", _creds->mqtt_port);
|
||||
mqtt_publisher_reconnect();
|
||||
} else {
|
||||
snprintf(reply, reply_size, "ERR port must be 1-65535");
|
||||
}
|
||||
|
||||
} else if ((val = find_val(rest, "mqtt.tls")) != nullptr) {
|
||||
int tls = atoi(val);
|
||||
_creds->mqtt_tls = (tls != 0) ? 1 : 0;
|
||||
observer_creds_save(_creds, _store->getBasePath());
|
||||
snprintf(reply, reply_size, "mqtt.tls=%u (reconnecting)", _creds->mqtt_tls);
|
||||
mqtt_publisher_reconnect();
|
||||
|
||||
} else if ((val = find_val(rest, "mqtt.user")) != nullptr) {
|
||||
strncpy(_creds->mqtt_user, val, sizeof(_creds->mqtt_user) - 1);
|
||||
_creds->mqtt_user[sizeof(_creds->mqtt_user) - 1] = '\0';
|
||||
observer_creds_save(_creds, _store->getBasePath());
|
||||
snprintf(reply, reply_size, "mqtt.user=%s (reconnecting)", _creds->mqtt_user);
|
||||
mqtt_publisher_reconnect();
|
||||
|
||||
} else if ((val = find_val(rest, "mqtt.password")) != nullptr) {
|
||||
strncpy(_creds->mqtt_password, val, sizeof(_creds->mqtt_password) - 1);
|
||||
_creds->mqtt_password[sizeof(_creds->mqtt_password) - 1] = '\0';
|
||||
observer_creds_save(_creds, _store->getBasePath());
|
||||
snprintf(reply, reply_size, "mqtt.password=*** (saved, reconnecting)");
|
||||
mqtt_publisher_reconnect();
|
||||
|
||||
} else if ((val = find_val(rest, "mqtt.iata")) != nullptr) {
|
||||
strncpy(_creds->mqtt_iata, val, sizeof(_creds->mqtt_iata) - 1);
|
||||
_creds->mqtt_iata[sizeof(_creds->mqtt_iata) - 1] = '\0';
|
||||
observer_creds_save(_creds, _store->getBasePath());
|
||||
buildTopics(); /* rebuild topic strings with new IATA */
|
||||
snprintf(reply, reply_size, "mqtt.iata=%s (topics updated, reconnecting)", _creds->mqtt_iata);
|
||||
mqtt_publisher_reconnect();
|
||||
|
||||
} else {
|
||||
snprintf(reply, reply_size, "ERR unknown key");
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
snprintf(reply, reply_size, "ERR unknown command (type 'help')");
|
||||
return false;
|
||||
}
|
||||
|
||||
} /* namespace mesh */
|
||||
@@ -0,0 +1,85 @@
|
||||
/*
|
||||
* SPDX-License-Identifier: Apache-2.0
|
||||
* ObserverMesh — listen-only LoRa mesh node.
|
||||
*
|
||||
* Extends mesh::Dispatcher directly (no routing, no flooding, no ACL).
|
||||
* Every received packet is forwarded to the MQTT publisher queue.
|
||||
* CLI handles WiFi/MQTT/radio configuration.
|
||||
*/
|
||||
|
||||
#pragma once
|
||||
|
||||
#include <mesh/Dispatcher.h>
|
||||
#include <mesh/StaticPoolPacketManager.h>
|
||||
#include <mesh/Identity.h>
|
||||
#include <mesh/RNG.h>
|
||||
#include <mesh/RTC.h>
|
||||
#include <helpers/NodePrefs.h>
|
||||
#include "RepeaterDataStore.h"
|
||||
#include "observer_creds.h"
|
||||
|
||||
#ifndef FIRMWARE_VERSION
|
||||
#define FIRMWARE_VERSION "v1.14.1-zephyr"
|
||||
#endif
|
||||
|
||||
#ifndef FIRMWARE_BUILD_DATE
|
||||
#define FIRMWARE_BUILD_DATE __DATE__
|
||||
#endif
|
||||
|
||||
namespace mesh {
|
||||
|
||||
class ObserverMesh : public Dispatcher {
|
||||
StaticPoolPacketManager _pkt_mgr;
|
||||
|
||||
/* Cached values set by logRxRaw / logRx before onRecvPacket */
|
||||
float _last_rssi;
|
||||
float _last_score;
|
||||
uint8_t _last_raw[MAX_TRANS_UNIT + 1];
|
||||
int _last_raw_len;
|
||||
|
||||
/* Identity and config */
|
||||
LocalIdentity _self_id;
|
||||
NodePrefs _prefs;
|
||||
RepeaterDataStore *_store;
|
||||
struct ObserverCreds *_creds;
|
||||
RNG *_rng;
|
||||
RTCClock *_rtc;
|
||||
|
||||
/* Pre-built MQTT topic strings (set in begin()) */
|
||||
char _pubkey_hex[PUB_KEY_SIZE * 2 + 1]; /* 64 hex chars + NUL */
|
||||
char _packets_topic[160];
|
||||
char _status_topic[160];
|
||||
|
||||
/* Private helpers */
|
||||
void buildTopics();
|
||||
void enqueuePacket(Packet *pkt);
|
||||
|
||||
protected:
|
||||
/* Capture RSSI + raw bytes before packet is parsed */
|
||||
void logRxRaw(float snr, float rssi, const uint8_t raw[], int len) override;
|
||||
/* Capture score (called between logRxRaw and onRecvPacket) */
|
||||
void logRx(Packet *packet, int len, float score) override;
|
||||
/* Build JSON and enqueue to MQTT publisher */
|
||||
DispatcherAction onRecvPacket(Packet *pkt) override;
|
||||
|
||||
public:
|
||||
ObserverMesh(Radio &radio, MillisecondClock &ms, RNG &rng, RTCClock &rtc);
|
||||
|
||||
/* Initialize: load/generate identity, load/init prefs, start radio RX. */
|
||||
void begin(RepeaterDataStore *store, struct ObserverCreds *creds);
|
||||
|
||||
/* Handle a single serial CLI command.
|
||||
* reply is filled with the response string (CLI_REPLY_SIZE bytes).
|
||||
* Returns true if the command was 'help' and the caller should print
|
||||
* the full banner (too long for the reply buffer). */
|
||||
bool handleCLI(const char *command, char *reply, int reply_size);
|
||||
|
||||
/* Accessors used by main_observer.cpp */
|
||||
NodePrefs *getNodePrefs() { return &_prefs; }
|
||||
const LocalIdentity &getSelfId() const { return _self_id; }
|
||||
const char *getPacketsTopic() const { return _packets_topic; }
|
||||
const char *getStatusTopic() const { return _status_topic; }
|
||||
const char *getPubkeyHex() const { return _pubkey_hex; }
|
||||
};
|
||||
|
||||
} /* namespace mesh */
|
||||
@@ -0,0 +1,400 @@
|
||||
/*
|
||||
* SPDX-License-Identifier: Apache-2.0
|
||||
* ZephCore Observer — listen-only LoRa node with WiFi+MQTT forwarding.
|
||||
*
|
||||
* Every received LoRa packet is published to an MQTT broker in
|
||||
* meshcoretomqtt-compatible JSON format. All parameters (WiFi, MQTT,
|
||||
* IATA, radio) are runtime-configurable via serial CLI and stored in
|
||||
* LittleFS. Nothing is hardcoded.
|
||||
*
|
||||
* Event loop:
|
||||
* LORA_RX → ObserverMesh::loop() → enqueuePacket() → mqtt_publisher_enqueue()
|
||||
* CLI_RX → ObserverMesh::handleCLI() (set/get commands)
|
||||
*/
|
||||
|
||||
#include <stdio.h>
|
||||
#include <string.h>
|
||||
#include <zephyr/kernel.h>
|
||||
#include <zephyr/drivers/uart.h>
|
||||
#include <zephyr/sys/ring_buffer.h>
|
||||
#include <zephyr/fs/fs.h>
|
||||
#include <zephyr/logging/log.h>
|
||||
|
||||
#define CLI_REPLY_SIZE 256
|
||||
|
||||
LOG_MODULE_REGISTER(zephcore_observer_main, CONFIG_ZEPHCORE_MAIN_LOG_LEVEL);
|
||||
|
||||
#include <app/RepeaterDataStore.h>
|
||||
#include <app/ObserverMesh.h>
|
||||
#include <adapters/clock/ZephyrRTCClock.h>
|
||||
#include <mesh/RadioIncludes.h>
|
||||
#include <ZephyrWiFiStation.h>
|
||||
#include <ZephyrMQTTPublisher.h>
|
||||
#include "observer_creds.h"
|
||||
|
||||
/* ========== observer_creds_load / observer_creds_save ========== */
|
||||
|
||||
extern "C" bool observer_creds_load(struct ObserverCreds *creds,
|
||||
const char *base_path)
|
||||
{
|
||||
char path[96];
|
||||
snprintf(path, sizeof(path), "%s/obs_creds", base_path);
|
||||
|
||||
struct fs_file_t f;
|
||||
fs_file_t_init(&f);
|
||||
|
||||
int rc = fs_open(&f, path, FS_O_READ);
|
||||
if (rc < 0) {
|
||||
LOG_DBG("obs_creds not found (%d) — using defaults", rc);
|
||||
return false;
|
||||
}
|
||||
|
||||
ssize_t n = fs_read(&f, creds, sizeof(*creds));
|
||||
fs_close(&f);
|
||||
|
||||
if (n != (ssize_t)sizeof(*creds)) {
|
||||
LOG_WRN("obs_creds truncated (%d/%d) — resetting", (int)n,
|
||||
(int)sizeof(*creds));
|
||||
memset(creds, 0, sizeof(*creds));
|
||||
observer_creds_init(creds);
|
||||
return false;
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
extern "C" bool observer_creds_save(const struct ObserverCreds *creds,
|
||||
const char *base_path)
|
||||
{
|
||||
char path[96];
|
||||
snprintf(path, sizeof(path), "%s/obs_creds", base_path);
|
||||
|
||||
struct fs_file_t f;
|
||||
fs_file_t_init(&f);
|
||||
|
||||
int rc = fs_open(&f, path, FS_O_WRITE | FS_O_CREATE | FS_O_TRUNC);
|
||||
if (rc < 0) {
|
||||
LOG_ERR("Cannot open %s for write: %d", path, rc);
|
||||
return false;
|
||||
}
|
||||
|
||||
ssize_t n = fs_write(&f, creds, sizeof(*creds));
|
||||
fs_close(&f);
|
||||
|
||||
if (n != (ssize_t)sizeof(*creds)) {
|
||||
LOG_ERR("obs_creds write failed (%d/%d)", (int)n,
|
||||
(int)sizeof(*creds));
|
||||
return false;
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
/* ========== LED (optional) ========== */
|
||||
|
||||
#if DT_NODE_HAS_PROP(DT_ALIAS(led0), gpios)
|
||||
#include <zephyr/drivers/gpio.h>
|
||||
#define LED0_NODE DT_ALIAS(led0)
|
||||
static const struct gpio_dt_spec led0 = GPIO_DT_SPEC_GET(LED0_NODE, gpios);
|
||||
#endif
|
||||
|
||||
/* ========== Event loop bits ========== */
|
||||
|
||||
#define MESH_EVENT_LORA_RX BIT(0)
|
||||
#define MESH_EVENT_CLI_RX BIT(1)
|
||||
#define MESH_EVENT_ALL (MESH_EVENT_LORA_RX | MESH_EVENT_CLI_RX)
|
||||
|
||||
static struct k_event mesh_events;
|
||||
|
||||
/* ========== USB serial CLI ========== */
|
||||
|
||||
#define USB_RING_BUF_SIZE 512
|
||||
#define CLI_LINE_BUF_SIZE 256
|
||||
|
||||
static const struct device *usb_dev;
|
||||
static uint8_t usb_ring_data[USB_RING_BUF_SIZE];
|
||||
static struct ring_buf usb_ring_buf;
|
||||
static char cli_line[CLI_LINE_BUF_SIZE];
|
||||
static char cli_reply[CLI_REPLY_SIZE];
|
||||
static uint16_t cli_line_idx;
|
||||
|
||||
static void cli_print(const char *s)
|
||||
{
|
||||
if (!usb_dev) return;
|
||||
while (*s) uart_poll_out(usb_dev, *s++);
|
||||
}
|
||||
|
||||
static void cli_println(const char *s)
|
||||
{
|
||||
cli_print(s);
|
||||
cli_print("\r\n");
|
||||
}
|
||||
|
||||
static void cli_uart_isr(const struct device *dev, void *user_data)
|
||||
{
|
||||
ARG_UNUSED(user_data);
|
||||
while (uart_irq_update(dev) && uart_irq_is_pending(dev)) {
|
||||
if (uart_irq_rx_ready(dev)) {
|
||||
uint8_t buf[64];
|
||||
int n = uart_fifo_read(dev, buf, sizeof(buf));
|
||||
if (n > 0) {
|
||||
ring_buf_put(&usb_ring_buf, buf, n);
|
||||
k_event_post(&mesh_events, MESH_EVENT_CLI_RX);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/* ========== LoRa callbacks ========== */
|
||||
|
||||
static void lora_rx_callback(void *user_data)
|
||||
{
|
||||
ARG_UNUSED(user_data);
|
||||
k_event_post(&mesh_events, MESH_EVENT_LORA_RX);
|
||||
}
|
||||
|
||||
/* Observer never transmits — TX done callback not needed */
|
||||
|
||||
/* ========== Help banner ========== */
|
||||
|
||||
/* Forward references */
|
||||
static mesh::ObserverMesh *s_mesh_ptr;
|
||||
static struct ObserverCreds s_creds;
|
||||
|
||||
static void print_banner(void)
|
||||
{
|
||||
char line[192];
|
||||
|
||||
cli_println("\r\n=== ZephCore Observer ===");
|
||||
|
||||
if (s_mesh_ptr) {
|
||||
const char *name = s_mesh_ptr->getNodePrefs()->node_name;
|
||||
const char *key = s_mesh_ptr->getPubkeyHex();
|
||||
snprintf(line, sizeof(line), "Node: %s", name);
|
||||
cli_println(line);
|
||||
snprintf(line, sizeof(line), "Key: %.32s...", key);
|
||||
cli_println(line);
|
||||
|
||||
NodePrefs *p = s_mesh_ptr->getNodePrefs();
|
||||
snprintf(line, sizeof(line),
|
||||
"Radio: %.3f MHz BW%.1f SF%u CR%u TX%ddBm",
|
||||
(double)p->freq, (double)p->bw,
|
||||
p->sf, p->cr, p->tx_power_dbm);
|
||||
cli_println(line);
|
||||
}
|
||||
|
||||
cli_println("");
|
||||
|
||||
snprintf(line, sizeof(line), "WiFi: %-14s (SSID: %s)",
|
||||
zc_wifi_station_is_connected() ? "CONNECTED" : "DISCONNECTED",
|
||||
s_creds.wifi_ssid[0] ? s_creds.wifi_ssid : "not set");
|
||||
cli_println(line);
|
||||
|
||||
snprintf(line, sizeof(line), "MQTT: %-14s (host: %s)",
|
||||
mqtt_publisher_is_connected() ? "CONNECTED" : "DISCONNECTED",
|
||||
s_creds.mqtt_host[0] ? s_creds.mqtt_host : "not set");
|
||||
cli_println(line);
|
||||
|
||||
snprintf(line, sizeof(line), "IATA: %s",
|
||||
s_creds.mqtt_iata[0] ? s_creds.mqtt_iata : "not set");
|
||||
cli_println(line);
|
||||
|
||||
cli_println("");
|
||||
cli_println("--- Configure ---");
|
||||
cli_println("set wifi.ssid <name> WiFi network name");
|
||||
cli_println("set wifi.psk <password> WiFi password (empty = open)");
|
||||
cli_println("set mqtt.host <hostname> MQTT broker hostname");
|
||||
cli_println("set mqtt.port <port> MQTT broker port (default 8883)");
|
||||
cli_println("set mqtt.tls <0|1> TLS on/off (default 1)");
|
||||
cli_println("set mqtt.user <username> MQTT username");
|
||||
cli_println("set mqtt.password <pass> MQTT password");
|
||||
cli_println("set mqtt.iata <code> Location code (e.g. BUD BTS VIE SEA)");
|
||||
cli_println("set name <name> Node display name");
|
||||
cli_println("set freq <MHz|Hz> LoRa frequency (e.g. 869.618 or 869618000)");
|
||||
cli_println("set sf <7-12> Spreading factor");
|
||||
cli_println("set bw <idx> Bandwidth: 3=62.5 0=125 1=250 2=500 kHz");
|
||||
cli_println("set cr <5-8> Coding rate");
|
||||
cli_println("");
|
||||
cli_println("--- Query ---");
|
||||
cli_println("get wifi.status WiFi connection state");
|
||||
cli_println("get mqtt.status MQTT connection state");
|
||||
cli_println("get radio LoRa radio parameters");
|
||||
cli_println("help Show this screen");
|
||||
cli_println("=========================");
|
||||
}
|
||||
|
||||
/* ========== CLI RX processing ========== */
|
||||
|
||||
static void process_cli_rx(void)
|
||||
{
|
||||
uint8_t byte;
|
||||
while (ring_buf_get(&usb_ring_buf, &byte, 1) == 1) {
|
||||
if (byte == '\r' || byte == '\n') {
|
||||
if (cli_line_idx > 0) {
|
||||
cli_line[cli_line_idx] = '\0';
|
||||
LOG_DBG("CLI: %s", cli_line);
|
||||
|
||||
cli_reply[0] = '\0';
|
||||
bool want_banner = false;
|
||||
if (s_mesh_ptr) {
|
||||
want_banner = s_mesh_ptr->handleCLI(
|
||||
cli_line, cli_reply,
|
||||
sizeof(cli_reply));
|
||||
}
|
||||
|
||||
if (want_banner) {
|
||||
print_banner();
|
||||
} else if (cli_reply[0] != '\0') {
|
||||
cli_print("\r\n -> ");
|
||||
cli_println(cli_reply);
|
||||
}
|
||||
cli_line_idx = 0;
|
||||
}
|
||||
cli_print("\r\n");
|
||||
} else if (byte == 0x7F || byte == 0x08) {
|
||||
if (cli_line_idx > 0) {
|
||||
cli_line_idx--;
|
||||
uart_poll_out(usb_dev, '\b');
|
||||
uart_poll_out(usb_dev, ' ');
|
||||
uart_poll_out(usb_dev, '\b');
|
||||
}
|
||||
} else if (cli_line_idx < sizeof(cli_line) - 1) {
|
||||
uart_poll_out(usb_dev, byte);
|
||||
cli_line[cli_line_idx++] = (char)byte;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/* ========== Time sync callback ========== */
|
||||
|
||||
static mesh::ZephyrRTCClock s_rtc_clock;
|
||||
|
||||
static void time_sync_cb(uint32_t unix_ts)
|
||||
{
|
||||
s_rtc_clock.setCurrentTime(unix_ts);
|
||||
LOG_INF("RTC synced from SNTP: %u", unix_ts);
|
||||
}
|
||||
|
||||
/* ========== Global instances ========== */
|
||||
|
||||
static mesh::ZephyrBoard s_board;
|
||||
static mesh::ZephyrMillisecondClock s_ms_clock;
|
||||
static mesh::ZephyrRNG s_rng;
|
||||
|
||||
static const struct device *const lora_dev = DEVICE_DT_GET(DT_ALIAS(lora0));
|
||||
|
||||
/* Radio prefs — observer-specific defaults set in main() */
|
||||
static NodePrefs s_radio_prefs;
|
||||
|
||||
#if IS_ENABLED(CONFIG_ZEPHCORE_RADIO_LR1110)
|
||||
static mesh::LR1110Radio lora_radio(lora_dev, s_board, &s_radio_prefs);
|
||||
#else
|
||||
static mesh::SX126xRadio lora_radio(lora_dev, s_board, &s_radio_prefs);
|
||||
#endif
|
||||
|
||||
static mesh::ObserverMesh observer_mesh(lora_radio, s_ms_clock, s_rng, s_rtc_clock);
|
||||
static RepeaterDataStore data_store;
|
||||
|
||||
/* ========== main() ========== */
|
||||
|
||||
int main(void)
|
||||
{
|
||||
/* Initialize radio prefs with observer-specific defaults */
|
||||
initNodePrefs(&s_radio_prefs);
|
||||
s_radio_prefs.cr = 5; /* CR 4/5 (initNodePrefs sets 8) */
|
||||
s_radio_prefs.tx_power_dbm = 0; /* observer never TXes */
|
||||
strncpy(s_radio_prefs.node_name, "Observer",
|
||||
sizeof(s_radio_prefs.node_name) - 1);
|
||||
|
||||
/* Brief boot delay — lets USB enumerate before first log */
|
||||
k_sleep(K_MSEC(1500));
|
||||
LOG_INF("=== ZephCore Observer starting ===");
|
||||
|
||||
/* Configure LED */
|
||||
#if DT_NODE_HAS_PROP(DT_ALIAS(led0), gpios)
|
||||
if (gpio_is_ready_dt(&led0)) {
|
||||
gpio_pin_configure_dt(&led0, GPIO_OUTPUT_INACTIVE);
|
||||
}
|
||||
#endif
|
||||
|
||||
/* Initialize LittleFS data store */
|
||||
if (!data_store.begin()) {
|
||||
LOG_ERR("RepeaterDataStore init failed");
|
||||
}
|
||||
|
||||
/* Load observer credentials (WiFi, MQTT, IATA) */
|
||||
memset(&s_creds, 0, sizeof(s_creds));
|
||||
observer_creds_init(&s_creds);
|
||||
observer_creds_load(&s_creds, data_store.getBasePath());
|
||||
|
||||
/* Init event object BEFORE any callbacks */
|
||||
k_event_init(&mesh_events);
|
||||
|
||||
/* LoRa RX callback — observer never needs TX done */
|
||||
lora_radio.setRxCallback(lora_rx_callback, nullptr);
|
||||
|
||||
/* Initialize and start mesh (loads prefs + identity from flash) */
|
||||
s_mesh_ptr = &observer_mesh;
|
||||
observer_mesh.begin(&data_store, &s_creds);
|
||||
|
||||
/* Generate a default node name based on pubkey if still generic */
|
||||
NodePrefs *prefs = observer_mesh.getNodePrefs();
|
||||
if (strlen(prefs->node_name) == 0 ||
|
||||
strcmp(prefs->node_name, "Observer") == 0 ||
|
||||
strcmp(prefs->node_name, "Repeater") == 0) {
|
||||
/* Use first 4 bytes of pubkey for uniqueness */
|
||||
const char *hex = observer_mesh.getPubkeyHex();
|
||||
snprintf(prefs->node_name, sizeof(prefs->node_name),
|
||||
"Observer-%.8s", hex);
|
||||
data_store.savePrefs(*prefs);
|
||||
}
|
||||
|
||||
/* Initialize USB serial for CLI */
|
||||
#if DT_HAS_COMPAT_STATUS_OKAY(zephyr_cdc_acm_uart)
|
||||
usb_dev = DEVICE_DT_GET_ONE(zephyr_cdc_acm_uart);
|
||||
#else
|
||||
/* ESP32 native USB or other console UART */
|
||||
usb_dev = DEVICE_DT_GET(DT_CHOSEN(zephyr_console));
|
||||
#endif
|
||||
if (device_is_ready(usb_dev)) {
|
||||
ring_buf_init(&usb_ring_buf, sizeof(usb_ring_data), usb_ring_data);
|
||||
uart_irq_callback_set(usb_dev, cli_uart_isr);
|
||||
uart_irq_rx_enable(usb_dev);
|
||||
LOG_INF("Serial CLI ready: %s", usb_dev->name);
|
||||
} else {
|
||||
LOG_WRN("Serial device not ready — CLI unavailable");
|
||||
usb_dev = nullptr;
|
||||
}
|
||||
|
||||
/* Print welcome banner */
|
||||
print_banner();
|
||||
|
||||
/* Start WiFi (non-blocking — MQTT thread waits for WIFI_READY_BIT) */
|
||||
zc_wifi_station_start(&s_creds, time_sync_cb);
|
||||
|
||||
/* Start MQTT publisher thread */
|
||||
char client_id[64];
|
||||
snprintf(client_id, sizeof(client_id), "%s", prefs->node_name);
|
||||
mqtt_publisher_start(&s_creds, client_id,
|
||||
observer_mesh.getStatusTopic(),
|
||||
observer_mesh.getPacketsTopic());
|
||||
|
||||
LOG_INF("Observer event loop running");
|
||||
|
||||
/* ========== Event loop ========== */
|
||||
for (;;) {
|
||||
uint32_t ev = k_event_wait(&mesh_events, MESH_EVENT_ALL,
|
||||
false, K_FOREVER);
|
||||
k_event_clear(&mesh_events, ev);
|
||||
|
||||
if (ev & MESH_EVENT_LORA_RX) {
|
||||
observer_mesh.loop();
|
||||
}
|
||||
|
||||
if (ev & MESH_EVENT_CLI_RX) {
|
||||
process_cli_rx();
|
||||
}
|
||||
}
|
||||
|
||||
return 0;
|
||||
}
|
||||
@@ -0,0 +1,46 @@
|
||||
/*
|
||||
* SPDX-License-Identifier: Apache-2.0
|
||||
* Observer runtime credentials — stored in LittleFS, configured via serial CLI.
|
||||
*
|
||||
* All connection parameters are runtime-configurable so no credentials ever
|
||||
* appear in the codebase. File path: /lfs/repeater/obs_creds
|
||||
*/
|
||||
|
||||
#pragma once
|
||||
|
||||
#include <stdint.h>
|
||||
#include <stdbool.h>
|
||||
|
||||
#ifdef __cplusplus
|
||||
extern "C" {
|
||||
#endif
|
||||
|
||||
struct ObserverCreds {
|
||||
char wifi_ssid[64]; /* WiFi network name */
|
||||
char wifi_psk[64]; /* WiFi password (empty = open network) */
|
||||
char mqtt_host[128]; /* MQTT broker hostname, e.g. "abydos.hu" */
|
||||
uint16_t mqtt_port; /* MQTT broker port, e.g. 8883 */
|
||||
uint8_t mqtt_tls; /* 1 = TLS (no cert verify), 0 = plaintext */
|
||||
char mqtt_user[64]; /* MQTT username */
|
||||
char mqtt_password[64]; /* MQTT password */
|
||||
char mqtt_iata[8]; /* IATA location code, e.g. "BUD", "BTS", "VIE" */
|
||||
uint8_t _reserved[5]; /* alignment / future use */
|
||||
};
|
||||
|
||||
/* Load creds from /lfs/repeater/obs_creds.
|
||||
* Returns true on success; on failure fills struct with safe zero defaults. */
|
||||
bool observer_creds_load(struct ObserverCreds *creds, const char *base_path);
|
||||
|
||||
/* Save creds to /lfs/repeater/obs_creds. Returns true on success. */
|
||||
bool observer_creds_save(const struct ObserverCreds *creds, const char *base_path);
|
||||
|
||||
/* Apply sensible defaults to a freshly-zeroed creds struct. */
|
||||
static inline void observer_creds_init(struct ObserverCreds *creds)
|
||||
{
|
||||
creds->mqtt_port = 8883;
|
||||
creds->mqtt_tls = 1;
|
||||
}
|
||||
|
||||
#ifdef __cplusplus
|
||||
}
|
||||
#endif
|
||||
@@ -0,0 +1,71 @@
|
||||
# SPDX-License-Identifier: Apache-2.0
|
||||
# Observer role configuration — listen-only LoRa node with WiFi+MQTT forwarding.
|
||||
# All parameters (WiFi SSID/PSK, MQTT host/port/TLS/user/pass, IATA location code)
|
||||
# are configured at runtime via serial CLI and stored in LittleFS.
|
||||
#
|
||||
# Build example:
|
||||
# west build -b xiao_esp32c3 zephcore --pristine -- \
|
||||
# -DEXTRA_CONF_FILE="boards/common/observer.conf"
|
||||
|
||||
CONFIG_ZEPHCORE_ROLE_OBSERVER=y
|
||||
|
||||
# No BLE — observer is serial-CLI-only
|
||||
CONFIG_BT=n
|
||||
CONFIG_ENTROPY_GENERATOR=y
|
||||
|
||||
# =========================================================
|
||||
# IPv4 networking stack (WiFi STA client mode)
|
||||
# =========================================================
|
||||
CONFIG_NETWORKING=y
|
||||
CONFIG_NET_IPV4=y
|
||||
CONFIG_NET_IPV6=n
|
||||
CONFIG_NET_TCP=y
|
||||
CONFIG_NET_UDP=y
|
||||
CONFIG_NET_SOCKETS=y
|
||||
CONFIG_NET_SOCKETS_SOCKOPT_TLS=y
|
||||
CONFIG_NET_MAX_CONTEXTS=8
|
||||
CONFIG_NET_MGMT=y
|
||||
CONFIG_NET_MGMT_EVENT=y
|
||||
CONFIG_NET_MGMT_EVENT_STACK_SIZE=4096
|
||||
CONFIG_DNS_RESOLVER=y
|
||||
CONFIG_DNS_SERVER_IP_ADDRESSES=y
|
||||
|
||||
# WiFi STA (client — NOT access point)
|
||||
CONFIG_WIFI=y
|
||||
CONFIG_NET_L2_WIFI_MGMT=y
|
||||
CONFIG_NET_L2_ETHERNET=y
|
||||
CONFIG_NET_DHCPV4=y
|
||||
|
||||
# =========================================================
|
||||
# MQTT with TLS (no cert verification — Zephyr has no CA bundle)
|
||||
# =========================================================
|
||||
CONFIG_MQTT_LIB=y
|
||||
CONFIG_MQTT_LIB_TLS=y
|
||||
CONFIG_MQTT_KEEPALIVE=60
|
||||
|
||||
# mbedTLS — TLS transport encryption without cert pinning
|
||||
CONFIG_MBEDTLS=y
|
||||
CONFIG_MBEDTLS_BUILTIN=y
|
||||
CONFIG_MBEDTLS_ENABLE_HEAP=y
|
||||
CONFIG_MBEDTLS_HEAP_SIZE=50000
|
||||
CONFIG_MBEDTLS_SSL_MAX_CONTENT_LEN=4096
|
||||
CONFIG_MBEDTLS_SSL_SERVER_NAME_INDICATION=y
|
||||
# TLS 1.2 ciphersuite (auto-selects KEY_EXCHANGE + PSA_WANT_* dependencies)
|
||||
CONFIG_MBEDTLS_CIPHERSUITE_TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256=y
|
||||
CONFIG_MBEDTLS_CIPHERSUITE_TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256=y
|
||||
# TLS 1.2 only (TLS 1.3 causes handshake_failure on most MQTT brokers)
|
||||
|
||||
# =========================================================
|
||||
# SNTP time synchronization after WiFi connect
|
||||
# =========================================================
|
||||
CONFIG_SNTP=y
|
||||
|
||||
# =========================================================
|
||||
# Stack / heap sizing for networking
|
||||
# =========================================================
|
||||
CONFIG_MAIN_STACK_SIZE=4096
|
||||
CONFIG_SYSTEM_WORKQUEUE_STACK_SIZE=4096
|
||||
CONFIG_NET_RX_STACK_SIZE=2048
|
||||
|
||||
# USB serial for CLI (ESP32 uses native usb_serial, not CDC ACM)
|
||||
CONFIG_CDC_ACM_SERIAL_INITIALIZE_AT_BOOT=n
|
||||
Reference in New Issue
Block a user