From 875e6554d4dd627494375a398d06ebf12b3591e8 Mon Sep 17 00:00:00 2001 From: liquidraver <504870+liquidraver@users.noreply.github.com> Date: Sun, 5 Apr 2026 14:33:09 +0200 Subject: [PATCH] Implement MQTT Observer role --- zephcore/CMakeLists.txt | 20 +- zephcore/Kconfig | 14 + zephcore/adapters/mqtt/ZephyrMQTTPublisher.c | 469 +++++++++++++++++++ zephcore/adapters/mqtt/ZephyrMQTTPublisher.h | 53 +++ zephcore/adapters/wifi/ZephyrWiFiStation.c | 210 +++++++++ zephcore/adapters/wifi/ZephyrWiFiStation.h | 54 +++ zephcore/app/ObserverMesh.cpp | 447 ++++++++++++++++++ zephcore/app/ObserverMesh.h | 85 ++++ zephcore/app/main_observer.cpp | 400 ++++++++++++++++ zephcore/app/observer_creds.h | 46 ++ zephcore/boards/common/observer.conf | 71 +++ 11 files changed, 1867 insertions(+), 2 deletions(-) create mode 100644 zephcore/adapters/mqtt/ZephyrMQTTPublisher.c create mode 100644 zephcore/adapters/mqtt/ZephyrMQTTPublisher.h create mode 100644 zephcore/adapters/wifi/ZephyrWiFiStation.c create mode 100644 zephcore/adapters/wifi/ZephyrWiFiStation.h create mode 100644 zephcore/app/ObserverMesh.cpp create mode 100644 zephcore/app/ObserverMesh.h create mode 100644 zephcore/app/main_observer.cpp create mode 100644 zephcore/app/observer_creds.h create mode 100644 zephcore/boards/common/observer.conf diff --git a/zephcore/CMakeLists.txt b/zephcore/CMakeLists.txt index 6b24e42..8e30ba3 100644 --- a/zephcore/CMakeLists.txt +++ b/zephcore/CMakeLists.txt @@ -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 ) diff --git a/zephcore/Kconfig b/zephcore/Kconfig index 2fcd174..58754ca 100644 --- a/zephcore/Kconfig +++ b/zephcore/Kconfig @@ -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 diff --git a/zephcore/adapters/mqtt/ZephyrMQTTPublisher.c b/zephcore/adapters/mqtt/ZephyrMQTTPublisher.c new file mode 100644 index 0000000..8b6fd38 --- /dev/null +++ b/zephcore/adapters/mqtt/ZephyrMQTTPublisher.c @@ -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 +#include +#include +#include +#include + +#include +#include +#include + +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 "); + 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); +} diff --git a/zephcore/adapters/mqtt/ZephyrMQTTPublisher.h b/zephcore/adapters/mqtt/ZephyrMQTTPublisher.h new file mode 100644 index 0000000..c635284 --- /dev/null +++ b/zephcore/adapters/mqtt/ZephyrMQTTPublisher.h @@ -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 + +#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 diff --git a/zephcore/adapters/wifi/ZephyrWiFiStation.c b/zephcore/adapters/wifi/ZephyrWiFiStation.c new file mode 100644 index 0000000..f0317e5 --- /dev/null +++ b/zephcore/adapters/wifi/ZephyrWiFiStation.c @@ -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 +#include +#include +#include +#include +#include +#include + +#include +#include + +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 "); + 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 : ""; +} diff --git a/zephcore/adapters/wifi/ZephyrWiFiStation.h b/zephcore/adapters/wifi/ZephyrWiFiStation.h new file mode 100644 index 0000000..a908200 --- /dev/null +++ b/zephcore/adapters/wifi/ZephyrWiFiStation.h @@ -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 +#include +#include + +#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 diff --git a/zephcore/app/ObserverMesh.cpp b/zephcore/app/ObserverMesh.cpp new file mode 100644 index 0000000..8ec1baa --- /dev/null +++ b/zephcore/app/ObserverMesh.cpp @@ -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 +#include +#include + +#include +LOG_MODULE_REGISTER(zephcore_observer, CONFIG_ZEPHCORE_OBSERVER_LOG_LEVEL); + +#include +#include +#include +#include + +/* 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 */ diff --git a/zephcore/app/ObserverMesh.h b/zephcore/app/ObserverMesh.h new file mode 100644 index 0000000..381c879 --- /dev/null +++ b/zephcore/app/ObserverMesh.h @@ -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 +#include +#include +#include +#include +#include +#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 */ diff --git a/zephcore/app/main_observer.cpp b/zephcore/app/main_observer.cpp new file mode 100644 index 0000000..d446163 --- /dev/null +++ b/zephcore/app/main_observer.cpp @@ -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 +#include +#include +#include +#include +#include +#include + +#define CLI_REPLY_SIZE 256 + +LOG_MODULE_REGISTER(zephcore_observer_main, CONFIG_ZEPHCORE_MAIN_LOG_LEVEL); + +#include +#include +#include +#include +#include +#include +#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 +#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 WiFi network name"); + cli_println("set wifi.psk WiFi password (empty = open)"); + cli_println("set mqtt.host MQTT broker hostname"); + cli_println("set mqtt.port MQTT broker port (default 8883)"); + cli_println("set mqtt.tls <0|1> TLS on/off (default 1)"); + cli_println("set mqtt.user MQTT username"); + cli_println("set mqtt.password MQTT password"); + cli_println("set mqtt.iata Location code (e.g. BUD BTS VIE SEA)"); + cli_println("set name Node display name"); + cli_println("set freq LoRa frequency (e.g. 869.618 or 869618000)"); + cli_println("set sf <7-12> Spreading factor"); + cli_println("set bw 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; +} diff --git a/zephcore/app/observer_creds.h b/zephcore/app/observer_creds.h new file mode 100644 index 0000000..a7574f1 --- /dev/null +++ b/zephcore/app/observer_creds.h @@ -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 +#include + +#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 diff --git a/zephcore/boards/common/observer.conf b/zephcore/boards/common/observer.conf new file mode 100644 index 0000000..df78ef6 --- /dev/null +++ b/zephcore/boards/common/observer.conf @@ -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