diff --git a/scripts/__pycache__/simulate_node.cpython-314.pyc b/scripts/__pycache__/simulate_node.cpython-314.pyc new file mode 100644 index 0000000..372e734 Binary files /dev/null and b/scripts/__pycache__/simulate_node.cpython-314.pyc differ diff --git a/scripts/simulate_node.py b/scripts/simulate_node.py new file mode 100755 index 0000000..c706942 --- /dev/null +++ b/scripts/simulate_node.py @@ -0,0 +1,309 @@ +#!/usr/bin/env python3 +""" +Simulate one audio Node for the ESP32 Hub over the node UDP protocol. + +The Hub (see src/network/node_protocol.h) runs: + * control plane on UDP 4210 (PROBE / START / STOP / POLL) + * data plane on UDP 4211 (DATA packets from Nodes) + +This script behaves like a Node with 4 microphones: + * answers PROBE with a PROBE_RESP (so the Hub registers it) + * streams 4 interleaved 48 kHz / 16-bit sine tones (one per mic) as + DATA packets while recording, paced to real time so the WAV plays back + at the correct pitch and duration + * sends HEARTBEATs so the Hub keeps it marked online + +Usage: + # PC connected to the Hub's AP (192.168.4.x), or any host on the subnet + python3 scripts/simulate_node.py + + # multiple virtual nodes (distinct tones): + python3 scripts/simulate_node.py --node-id 2 --frequencies 330,415,495,660 + +The four default tones are an A-major chord, one per mic, so each channel is +clearly distinguishable when the 40-channel WAV is played back. + +Requires only the Python standard library. +""" + +import argparse +import math +import select +import signal +import socket +import struct +import sys +import time + +# --- Protocol constants (mirror src/network/network_config.h + node_protocol.h) + +PROTOCOL_VERSION = 1 +CONTROL_PORT = 4210 +DATA_PORT = 4211 + +MSG_PROBE = 0x01 +MSG_PROBE_RESP = 0x02 +MSG_START = 0x03 +MSG_STOP = 0x04 +MSG_HEARTBEAT = 0x05 +MSG_POLL = 0x06 +MSG_POLL_ACK = 0x07 +MSG_DATA = 0x10 + +STATUS_IDLE = 0 +STATUS_RECORDING = 1 + +MICS_PER_NODE = 1 +SAMPLE_RATE = 16000 +BITS_PER_SAMPLE = 16 +FRAME_BYTES = MICS_PER_NODE * 2 # 8 bytes per frame (4 ch x 2 B) + +# Frames per DATA packet: (header 8 B) + n * 8 B must stay below 1472. +FRAMES_PER_PACKET = 100 + +AMPLITUDE = 8000.0 # ~0.25 x full scale +HEARTBEAT_MS = 2000 + + +def u16(x): + return struct.pack('= self._two_pi: + self.phase[ch] -= self._two_pi + sample = int(self.amplitude * math.sin(self.phase[ch])) + off = (f * MICS_PER_NODE + ch) * 2 + pcm[off] = sample & 0xFF + pcm[off + 1] = (sample >> 8) & 0xFF + + return bytes(pcm) + + +class SimulatedNode: + def __init__(self, args): + self.node_id = args.node_id + self.mac = bytes([0x02, 0x00, 0x00, 0x00, 0x00, 0x10 + (args.node_id & 0xEF)]) + self.firmware = "py-node-v1" + self.buffer_bytes = 64 * 1024 + + self.hub_ip = args.hub_ip + self.recording = False + self.status = STATUS_IDLE + self.uptime_ms = 0 + self.started_at = 0.0 + self.produced = 0 + self.last_heartbeat = 0.0 + self.running = True + + self.sine = SineGenerator(args.frequencies, args.amplitude) + self.stop_after = args.duration + + self.ctrl = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + self.ctrl.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + try: + self.ctrl.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) + except OSError: + pass + self.ctrl.bind(('', CONTROL_PORT)) + + self.data = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + try: + self.data.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) + except OSError: + pass + + # --- helpers ---------------------------------------------------------- + + def send_probe_resp(self): + pkt = build_probe_resp(self.node_id, self.mac, self.firmware, + MICS_PER_NODE, SAMPLE_RATE, BITS_PER_SAMPLE, + self.buffer_bytes, self.status) + self.ctrl.sendto(pkt, (self.hub_ip, CONTROL_PORT)) + print(f"[node {self.node_id}] PROBE_RESP sent to {self.hub_ip}") + + def send_heartbeat(self): + pkt = build_heartbeat(self.node_id, self.uptime_ms, + self.buffered_frames() * FRAME_BYTES, self.status) + self.ctrl.sendto(pkt, (self.hub_ip, CONTROL_PORT)) + + def buffered_frames(self): + if not self.recording: + return 0 + return max(0, int(time.monotonic() - self.started_at) * SAMPLE_RATE + - self.produced) + + def stream_pacing(self): + """Send the frames that real time says are due, in DATA packets.""" + if not self.recording or self.hub_ip is None: + return + + target = int((time.monotonic() - self.started_at) * SAMPLE_RATE) + while self.produced < target: + n = min(target - self.produced, FRAMES_PER_PACKET) + pcm = self.sine.frames(n) + pkt = build_data(self.node_id, MICS_PER_NODE, n, pcm) + self.data.sendto(pkt, (self.hub_ip, DATA_PORT)) + self.produced += n + + # --- control packet handling ------------------------------------------ + + def handle(self, packet, src): + if len(packet) < 2: + return + msg, ver = packet[0], packet[1] + if ver != PROTOCOL_VERSION: + return + + if src[0] != self.hub_ip: + self.hub_ip = src[0] # learn the Hub from any control packet + print(f"[node {self.node_id}] learned Hub at {self.hub_ip}") + + if msg == MSG_PROBE: + self.send_probe_resp() + + elif msg == MSG_START: + self.recording = True + self.status = STATUS_RECORDING + self.started_at = time.monotonic() + self.produced = 0 + print(f"[node {self.node_id}] START - streaming 4 tones at " + f"{SAMPLE_RATE} Hz, {MICS_PER_NODE} ch") + + elif msg == MSG_STOP: + if self.recording: + print(f"[node {self.node_id}] STOP - sent " + f"{self.produced} frames") + self.recording = False + self.status = STATUS_IDLE + + elif msg == MSG_POLL: + # We stream continuously; nothing extra to send. + pass + + # --- main loop -------------------------------------------------------- + + def run(self): + print(f"[node {self.node_id}] listening on UDP {CONTROL_PORT}...") + print(f"[node {self.node_id}] 4 tones: {self.sine.frequencies} Hz") + if self.hub_ip: + print(f"[node {self.node_id}] Hub set to {self.hub_ip}") + else: + print("[node %d] waiting for a PROBE to learn the Hub IP " + "(PC must be on the Hub's AP network)" % self.node_id) + + while self.running: + ready, _, _ = select.select([self.ctrl], [], [], 0.002) + if ready: + try: + packet, src = self.ctrl.recvfrom(2048) + self.handle(packet, src) + except OSError: + pass + + if self.recording: + self.stream_pacing() + if self.stop_after is not None: + if time.monotonic() - self.started_at >= self.stop_after: + print(f"[node {self.node_id}] --duration reached, " + "waiting for STOP") + self.recording = False + self.status = STATUS_IDLE + + now_ms = int(time.monotonic() * 1000) + self.uptime_ms = now_ms + if self.hub_ip and now_ms - self.last_heartbeat >= HEARTBEAT_MS: + self.last_heartbeat = now_ms + self.send_heartbeat() + + self.ctrl.close() + self.data.close() + + +def parse_args(argv): + parser = argparse.ArgumentParser( + description='Simulate a 4-mic audio Node for the ESP32 Hub.') + parser.add_argument('--node-id', type=int, default=1, + help='Node ID (default 1). Overrides the matching ' + 'built-in simulator node.') + parser.add_argument('--hub-ip', default=None, + help='Hub IP, e.g. 192.168.4.1 (auto-learned from ' + 'the first control packet if omitted).') + parser.add_argument('--frequencies', default='440,554.37,659.25,880', + help='Comma-separated sine frequencies per mic ' + '(default an A-major chord).') + parser.add_argument('--amplitude', type=float, default=AMPLITUDE, + help='Sine amplitude 0..32767 (default 8000).') + parser.add_argument('--duration', type=float, default=None, + help='Auto-stop after N seconds of recording ' + '(default: until the Hub sends STOP).') + return parser.parse_args(argv) + + +def main(argv=None): + args = parse_args(argv) + args.frequencies = [float(f) for f in args.frequencies.split(',')] + if len(args.frequencies) < MICS_PER_NODE: + print('error: provide at least %d frequencies' % MICS_PER_NODE, + file=sys.stderr) + return 1 + + node = SimulatedNode(args) + + def on_sigint(_sig, _frame): + node.running = False + print('\n[node %d] shutting down' % args.node_id) + + signal.signal(signal.SIGINT, on_sigint) + node.run() + return 0 + + +if __name__ == '__main__': + sys.exit(main()) diff --git a/src/core/node_registry.cpp b/src/core/node_registry.cpp new file mode 100644 index 0000000..9e74384 --- /dev/null +++ b/src/core/node_registry.cpp @@ -0,0 +1,268 @@ +#include "node_registry.h" + +#include + + +// --- RecordingController ----------------------------------------------------- + +void RecordingController::begin() +{ + recording = false; + startRequested = false; + stopRequested = false; + startedAt = 0; + stoppedAt = 0; +} + + +void RecordingController::requestStart() +{ + if (!recording) + { + startRequested = true; + } +} + + +void RecordingController::requestStop() +{ + if (recording) + { + stopRequested = true; + } +} + + +bool RecordingController::consumeStartRequest() +{ + if (startRequested) + { + startRequested = false; + return true; + } + return false; +} + + +bool RecordingController::consumeStopRequest() +{ + if (stopRequested) + { + stopRequested = false; + return true; + } + return false; +} + + +void RecordingController::onStartBroadcastSent() +{ + recording = true; + startedAt = millis(); +} + + +void RecordingController::onStopBroadcastSent() +{ + recording = false; + stoppedAt = millis(); +} + + +bool RecordingController::isRecording() const +{ + return recording; +} + + +uint32_t RecordingController::startedAtMs() const +{ + return startedAt; +} + + +uint32_t RecordingController::stoppedAtMs() const +{ + return stoppedAt; +} + + +// --- NodeRegistry ------------------------------------------------------------ + +void NodeRegistry::begin() +{ + for (int i = 0; i < NET_MAX_NODES; i++) + { + NodeInfo& n = const_cast(nodes[i]); + n.present = false; + n.online = false; + n.nodeId = 0xFF; + n.micCount = 0; + n.lastSeenMs = 0; + } +} + + +int NodeRegistry::findByMac(const uint8_t* mac) const +{ + for (int i = 0; i < NET_MAX_NODES; i++) + { + if (!nodes[i].present) + { + continue; + } + + bool same = true; + for (int b = 0; b < 6; b++) + { + if (nodes[i].mac[b] != mac[b]) + { + same = false; + break; + } + } + + if (same) + { + return i; + } + } + return -1; +} + + +int NodeRegistry::findById(uint8_t nodeId) const +{ + for (int i = 0; i < NET_MAX_NODES; i++) + { + if (nodes[i].present && nodes[i].nodeId == nodeId) + { + return i; + } + } + return -1; +} + + +int NodeRegistry::upsert(const nprot::ProbeResponse& resp, const IPAddress& ip, uint32_t nowMs) +{ + int idx = findByMac(resp.mac); + + if (idx < 0) + { + idx = findById(resp.nodeId); + } + + if (idx < 0) + { + // New node: take the first free slot. + for (int i = 0; i < NET_MAX_NODES; i++) + { + if (!nodes[i].present) + { + idx = i; + break; + } + } + + if (idx < 0) + { + return -1; // table full + } + } + + NodeInfo& n = const_cast(nodes[idx]); + n.present = true; + n.online = true; + n.nodeId = resp.nodeId; + memcpy(n.mac, resp.mac, 6); + n.ip = ip; + memset(n.firmware, 0, sizeof(n.firmware)); + memcpy(n.firmware, resp.firmware, 15); + n.micCount = resp.micCount; + n.sampleRate = resp.sampleRate; + n.bitsPerSample = resp.bitsPerSample; + n.bufferBytes = resp.bufferBytes; + n.status = resp.status; + n.lastSeenMs = nowMs; + + return idx; +} + + +void NodeRegistry::touch(uint8_t nodeId, uint32_t nowMs) +{ + int idx = findById(nodeId); + if (idx >= 0) + { + NodeInfo& n = const_cast(nodes[idx]); + n.lastSeenMs = nowMs; + n.online = true; + } +} + + +void NodeRegistry::updateOnline(uint32_t nowMs) +{ + for (int i = 0; i < NET_MAX_NODES; i++) + { + if (nodes[i].present) + { + NodeInfo& n = const_cast(nodes[i]); + n.online = (nowMs - n.lastSeenMs) < NET_OFFLINE_MS; + } + } +} + + +int NodeRegistry::count() const +{ + int c = 0; + for (int i = 0; i < NET_MAX_NODES; i++) + { + if (nodes[i].present) c++; + } + return c; +} + + +int NodeRegistry::onlineCount() const +{ + int c = 0; + for (int i = 0; i < NET_MAX_NODES; i++) + { + if (nodes[i].present && nodes[i].online) c++; + } + return c; +} + + +NodeInfo NodeRegistry::get(int index) const +{ + NodeInfo result = {}; + + if (index < 0 || index >= NET_MAX_NODES) + { + return result; // zeroed, present == false + } + + result.present = nodes[index].present; + result.online = nodes[index].online; + result.nodeId = nodes[index].nodeId; + for (int b = 0; b < 6; b++) result.mac[b] = nodes[index].mac[b]; + { + // IPAddress is a class, so copy its 4 payload bytes directly. + const volatile uint8_t* srcIp = + reinterpret_cast(&nodes[index].ip); + for (int b = 0; b < 4; b++) result.ip[b] = srcIp[b]; + } + for (int b = 0; b < 16; b++) result.firmware[b] = nodes[index].firmware[b]; + result.micCount = nodes[index].micCount; + result.sampleRate = nodes[index].sampleRate; + result.bitsPerSample= nodes[index].bitsPerSample; + result.bufferBytes = nodes[index].bufferBytes; + result.status = nodes[index].status; + result.lastSeenMs = nodes[index].lastSeenMs; + + return result; +} diff --git a/src/core/node_registry.h b/src/core/node_registry.h new file mode 100644 index 0000000..804ff61 --- /dev/null +++ b/src/core/node_registry.h @@ -0,0 +1,97 @@ +#pragma once + +// ============================================================================ +// Shared node registry + recording state. +// +// Written by the DiscoveryService / CollectionService (core 1) and read by +// the WebService (core 1) and StorageTask (core 0). Like the other *State +// classes, values live in a volatile snapshot so cross-core reads are sane. +// ============================================================================ + +#include + +#include "../network/network_config.h" +#include "../network/node_protocol.h" + + +// One discovered Node. `present` marks an occupied slot; `online` means it was +// heard from within the offline timeout. +struct NodeInfo +{ + bool present; + bool online; + uint8_t nodeId; + uint8_t mac[6]; + IPAddress ip; + char firmware[16]; + uint8_t micCount; + uint32_t sampleRate; + uint16_t bitsPerSample; + uint32_t bufferBytes; + uint8_t status; + uint32_t lastSeenMs; +}; + + +// Simple start/stop state machine shared between the FrontPanelService +// (requestor), DiscoveryService (broadcast sender) and StorageTask (watcher). +class RecordingController +{ +public: + void begin(); + + // Called by FrontPanelService on button press. + void requestStart(); + void requestStop(); + + // Consumed by DiscoveryService: true once per pending start/stop. + bool consumeStartRequest(); + bool consumeStopRequest(); + + // Called by DiscoveryService once the broadcast has been sent. + void onStartBroadcastSent(); + void onStopBroadcastSent(); + + bool isRecording() const; + uint32_t startedAtMs() const; + uint32_t stoppedAtMs() const; + +private: + volatile bool recording; + volatile bool startRequested; + volatile bool stopRequested; + volatile uint32_t startedAt; + volatile uint32_t stoppedAt; +}; + + +// Fixed-size table of discovered nodes (NET_MAX_NODES). Index == slot in the +// 40-channel interleave order is NOT assumed; the collector keys on nodeId. +class NodeRegistry +{ +public: + void begin(); + + // Index of a node by MAC, or -1. Used to detect new clients. + int findByMac(const uint8_t* mac) const; + + // Index of a node by nodeId, or -1. + int findById(uint8_t nodeId) const; + + // Creates or updates a node from a probe response. Returns the slot index + // or -1 when the table is full (no free slot and MAC unknown). + int upsert(const nprot::ProbeResponse& resp, const IPAddress& ip, uint32_t nowMs); + + // Refreshes lastSeenMs for a node (heartbeat / probe response). + void touch(uint8_t nodeId, uint32_t nowMs); + + // Recomputes `online` from lastSeenMs vs NET_OFFLINE_MS. + void updateOnline(uint32_t nowMs); + + int count() const; // occupied slots + int onlineCount() const; + NodeInfo get(int index) const; + +private: + volatile NodeInfo nodes[NET_MAX_NODES]; +}; diff --git a/src/core/storage_state.cpp b/src/core/storage_state.cpp index b1da4f6..4c64aa3 100644 --- a/src/core/storage_state.cpp +++ b/src/core/storage_state.cpp @@ -9,6 +9,8 @@ void StorageState::begin() current.usedMB = 0; current.freeMB = 0; current.writeSpeedBps = 0; + current.droppedChunks = 0; + current.bytesWritten = 0; current.cardType[0] = '\0'; } @@ -47,6 +49,13 @@ void StorageState::setWriteSpeedBps(uint32_t bytesPerSecond) } +void StorageState::setLoggingStats(uint32_t droppedChunks, uint32_t bytesWritten) +{ + current.droppedChunks = droppedChunks; + current.bytesWritten = bytesWritten; +} + + StorageSnapshot StorageState::getCurrent() { StorageSnapshot snapshot; @@ -57,6 +66,8 @@ StorageSnapshot StorageState::getCurrent() snapshot.usedMB = current.usedMB; snapshot.freeMB = current.freeMB; snapshot.writeSpeedBps = current.writeSpeedBps; + snapshot.droppedChunks = current.droppedChunks; + snapshot.bytesWritten = current.bytesWritten; for (size_t i = 0; i < sizeof(snapshot.cardType); i++) { diff --git a/src/core/storage_state.h b/src/core/storage_state.h index 326ad78..c42333d 100644 --- a/src/core/storage_state.h +++ b/src/core/storage_state.h @@ -14,6 +14,8 @@ struct StorageSnapshot uint32_t usedMB; uint32_t freeMB; uint32_t writeSpeedBps; + uint32_t droppedChunks; + uint32_t bytesWritten; char cardType[16]; }; @@ -27,6 +29,7 @@ public: void setCardType(const char* type); void setCapacity(uint64_t totalBytes, uint64_t usedBytes); void setWriteSpeedBps(uint32_t bytesPerSecond); + void setLoggingStats(uint32_t droppedChunks, uint32_t bytesWritten); StorageSnapshot getCurrent(); diff --git a/src/main.cpp b/src/main.cpp index a9e2f4e..340b200 100644 --- a/src/main.cpp +++ b/src/main.cpp @@ -6,13 +6,24 @@ #include "core/dashboard_state.h" #include "core/diagnostics_state.h" #include "core/storage_state.h" +#include "core/node_registry.h" #include "services/service_manager.h" #include "services/wifi_service.h" #include "services/ota_service.h" #include "services/web_service.h" +#include "services/front_panel_service.h" + +#include "network/network_config.h" +#include "network/discovery_service.h" +#include "network/collection_service.h" +#undef NET_SIM_ENABLED +#if NET_SIM_ENABLED +#include "network/node_simulator.h" +#endif #include "storage/sd_manager.h" +#include "storage/data_logger.h" #include "tasks/system_task.h" #include "tasks/diagnostics_task.h" @@ -22,14 +33,26 @@ DashboardState dashboardState; DiagnosticsState diagnosticsState; StorageState storageState; +NodeRegistry nodeRegistry; +RecordingController recordingController; // Simple service scheduler nice for grouping tasks ServiceManager services; WiFiService wifi; OTAService ota; -WebService web(dashboardState, diagnosticsState, storageState); +WebService web(dashboardState, diagnosticsState, storageState, + nodeRegistry, recordingController); +DiscoveryService discovery(nodeRegistry, recordingController); +DataLogger dataLogger; +CollectionService collection(nodeRegistry, recordingController, dataLogger); + +FrontPanelService panel(recordingController); + +#if NET_SIM_ENABLED +NodeSimulator simulator(nodeRegistry, recordingController); +#endif // Actual FreeRTOS tasks that are scheduled SystemTask systemTask(services); @@ -37,16 +60,26 @@ SystemTask systemTask(services); DiagnosticsTask diagnosticsTask(diagnosticsState); SDManager sdManager; -StorageTask storageTask(sdManager, storageState); +StorageTask storageTask(sdManager, storageState, dataLogger, recordingController); void setup() { Serial.begin(115200); + nodeRegistry.begin(); + recordingController.begin(); + services.add(&wifi); services.add(&ota); services.add(&web); + services.add(&discovery); + services.add(&collection); + services.add(&panel); +#if NET_SIM_ENABLED + services.add(&simulator); + collection.setSource(&simulator); +#endif systemTask.start(); diagnosticsTask.start(); diff --git a/src/network/collection_service.cpp b/src/network/collection_service.cpp new file mode 100644 index 0000000..dc27262 --- /dev/null +++ b/src/network/collection_service.cpp @@ -0,0 +1,360 @@ +#include "collection_service.h" + +#include + +#include "node_protocol.h" + + +namespace +{ +constexpr size_t FRAME_BYTES = NET_MICS_PER_NODE * 2; // 8 +constexpr size_t ROUND_BYTES = NET_ROUND_FRAMES * FRAME_BYTES; // 768 +constexpr size_t OUTPUT_FRAME = NET_MAX_NODES * NET_MICS_PER_NODE * 2; // 80 + +// One interleave round is written into exactly one chunk. If the round ever +// exceeds the chunk capacity, produceRound() overflows the DMA pool and +// corrupts the heap, so the two sizes are bound together at build time. +static_assert((size_t)NET_ROUND_FRAMES * OUTPUT_FRAME + <= (size_t)STORAGE_LOG_CHUNK_SIZE, + "Interleave round exceeds the DataLogger chunk: reduce " + "NET_ROUND_FRAMES or enlarge STORAGE_LOG_CHUNK_SIZE"); +} + + +CollectionService::CollectionService(NodeRegistry& registry, + RecordingController& recorder, + DataLogger& logger, + NodeSource* source) +: +Service("Collect", 10), +nodes(registry), +recorder(recorder), +logger(logger), +source(source), +lastPollIndex(-1), +lastPollMs(0), +blockedSinceMs(0), +chunkSequence(0), +started(false) +{ + memset(stages, 0, sizeof(stages)); +} + + +void CollectionService::setSource(NodeSource* src) +{ + source = src; +} + + +void CollectionService::begin() +{ + dataUdp.begin(NET_DATA_PORT); + + Serial.printf("[Collector] Data UDP on port %d\n", NET_DATA_PORT); +} + + +void CollectionService::update() +{ + uint32_t now = millis(); + + if (recorder.isRecording()) + { + if (!started) + { + started = true; + Serial.println("[Collector] Recording started, polling nodes"); + } + + drainDataSocket(now); + pullSimulator(now); + pollNextNode(now); + interleaveAvailable(now); + } + else + { + if (started) + { + started = false; + Serial.println("[Collector] Recording stopped"); + } + + for (int i = 0; i < NET_MAX_NODES; i++) + { + stages[i].len = 0; + stages[i].present = false; + } + + blockedSinceMs = 0; + } +} + + +bool CollectionService::isOnline(const NodeInfo& n) +{ + return n.present && n.online; +} + + +void CollectionService::drainDataSocket(uint32_t nowMs) +{ + uint8_t buf[NET_MAX_UDP_PAYLOAD]; + + int len; + while ((len = dataUdp.parsePacket()) > 0) + { + int n = dataUdp.read(buf, sizeof(buf)); + if (n < 8) + { + continue; + } + + nprot::DataHeader h; + if (!nprot::parseDataHeader(buf, n, h)) + { + continue; + } + + if (h.micCount != NET_MICS_PER_NODE) + { + continue; + } + + int idx = nodes.findById(h.nodeId); + if (idx < 0) + { + continue; + } + + appendPcmToStage(idx, buf + 8, (size_t)(n - 8)); + } +} + + +void CollectionService::pullSimulator(uint32_t nowMs) +{ + if (source == nullptr) + { + return; + } + + for (int idx = 0; idx < NET_MAX_NODES; idx++) + { + NodeInfo ni = nodes.get(idx); + if (!isOnline(ni)) + { + continue; + } + + NodeStage& s = stages[idx]; + if (!s.present || s.nodeId != ni.nodeId) + { + s.present = true; + s.nodeId = ni.nodeId; + s.len = 0; + } + + size_t free = sizeof(s.data) - s.len; + size_t want = free > 2 * ROUND_BYTES ? 2 * ROUND_BYTES : free; + if (want < FRAME_BYTES) + { + continue; + } + + uint8_t tmp[2 * ROUND_BYTES]; + size_t got = source->readAudio(ni.nodeId, tmp, want); + if (got > 0) + { + appendPcmToStage(idx, tmp, got); + } + } +} + + +void CollectionService::pollNextNode(uint32_t nowMs) +{ + if (nodes.onlineCount() == 0) + { + return; + } + + if (nowMs - lastPollMs < NET_POLL_PERIOD_MS) + { + return; + } + + lastPollMs = nowMs; + + int start = (lastPollIndex + 1) % NET_MAX_NODES; + + for (int i = 0; i < NET_MAX_NODES; i++) + { + int idx = (start + i) % NET_MAX_NODES; + NodeInfo ni = nodes.get(idx); + + if (isOnline(ni)) + { + lastPollIndex = idx; + sendPoll(ni); + return; + } + } +} + + +void CollectionService::sendPoll(const NodeInfo& node) +{ + uint8_t buf[8]; + size_t n = nprot::buildPoll(buf, sizeof(buf), node.nodeId); + + dataUdp.beginPacket(node.ip, NET_CONTROL_PORT); + dataUdp.write(buf, n); + dataUdp.endPacket(); +} + + +void CollectionService::interleaveAvailable(uint32_t nowMs) +{ + bool allReady = true; + + for (int idx = 0; idx < NET_MAX_NODES; idx++) + { + NodeInfo ni = nodes.get(idx); + if (isOnline(ni) && stages[idx].len < ROUND_BYTES) + { + allReady = false; + break; + } + } + + if (!allReady) + { + // A node fell behind (e.g. its DATA packets were lost). Wait up to a + // full poll cycle, then force a round with silence for stragglers so + // the pipeline never stalls. + if (blockedSinceMs == 0) + { + blockedSinceMs = nowMs; + } + + if (nowMs - blockedSinceMs >= NET_POLL_PERIOD_MS) + { + blockedSinceMs = 0; + produceRound(); + } + return; + } + + blockedSinceMs = 0; + + while (produceRound()) + { + allReady = true; + for (int idx = 0; idx < NET_MAX_NODES; idx++) + { + NodeInfo ni = nodes.get(idx); + if (isOnline(ni) && stages[idx].len < ROUND_BYTES) + { + allReady = false; + break; + } + } + if (!allReady) + { + break; + } + } +} + + +bool CollectionService::produceRound() +{ + AudioChunk* c = logger.acquireChunk(0); + if (c == nullptr) + { + return false; // pool drained; SD is catching up + } + + uint8_t* out = c->data; + uint8_t silence[FRAME_BYTES] = {0}; + + for (uint32_t f = 0; f < NET_ROUND_FRAMES; f++) + { + for (int idx = 0; idx < NET_MAX_NODES; idx++) + { + NodeInfo ni = nodes.get(idx); + + if (isOnline(ni) && stages[idx].present && stages[idx].len >= ROUND_BYTES) + { + memcpy(out, stages[idx].data + f * FRAME_BYTES, FRAME_BYTES); + } + else + { + memcpy(out, silence, FRAME_BYTES); + } + + out += FRAME_BYTES; + } + } + + for (int idx = 0; idx < NET_MAX_NODES; idx++) + { + NodeInfo ni = nodes.get(idx); + if (!isOnline(ni)) + { + continue; + } + + NodeStage& s = stages[idx]; + if (s.present && s.len >= ROUND_BYTES) + { + size_t remainder = s.len - ROUND_BYTES; + memmove(s.data, s.data + ROUND_BYTES, remainder); + s.len = (uint16_t)remainder; + } + } + + c->length = NET_ROUND_FRAMES * OUTPUT_FRAME; + c->sequence = chunkSequence++; + + logger.submitChunk(c); + + return true; +} + + +void CollectionService::appendPcmToStage(int idx, const uint8_t* p, size_t n) +{ + if (idx < 0 || idx >= NET_MAX_NODES) + { + return; + } + + NodeStage& s = stages[idx]; + NodeInfo ni = nodes.get(idx); + + if (!s.present || s.nodeId != ni.nodeId) + { + s.present = true; + s.nodeId = ni.nodeId; + s.len = 0; + } + + if (n >= sizeof(s.data)) + { + memcpy(s.data, p + (n - sizeof(s.data)), sizeof(s.data)); + s.len = sizeof(s.data); + return; + } + + if (s.len + n > sizeof(s.data)) + { + size_t drop = s.len + n - sizeof(s.data); + memmove(s.data, s.data + drop, s.len - drop); + s.len -= (uint16_t)drop; + } + + memcpy(s.data + s.len, p, n); + s.len += (uint16_t)n; +} diff --git a/src/network/collection_service.h b/src/network/collection_service.h new file mode 100644 index 0000000..4f91f41 --- /dev/null +++ b/src/network/collection_service.h @@ -0,0 +1,80 @@ +#pragma once + +// ============================================================================ +// CollectionService - data plane of the node protocol. +// +// Round-robin POLLs online nodes, reassembles their DATA packets into +// per-node staging buffers, interleaves 40 channels into frame order, and +// feeds the resulting PCM to the DataLogger as chunks. Missing/offline nodes +// contribute silence (zero fill), per design.md. +// +// Audio sources plug in behind the NodeSource interface so the built-in +// NodeSimulator and real UDP nodes share the exact same path. +// ============================================================================ + +#include + +#include "../services/service.h" +#include "../core/node_registry.h" +#include "../storage/data_logger.h" +#include "network_config.h" + + +// Anything that produces node-interleaved (4 ch) PCM for a node. Real nodes +// arrive via the data UDP socket; the simulator generates PCM in software. +class NodeSource +{ +public: + virtual ~NodeSource() {} + + // Writes up to maxBytes of PCM for `nodeId` into `out`. Returns bytes + // written (a multiple of 2 * NET_MICS_PER_NODE). Called on core 1. + virtual size_t readAudio(uint8_t nodeId, uint8_t* out, size_t maxBytes) = 0; +}; + + +class CollectionService : public Service +{ +public: + CollectionService(NodeRegistry& registry, + RecordingController& recorder, + DataLogger& logger, + NodeSource* source = nullptr); + + void begin() override; + void update() override; + + // Registered by main.cpp so the simulator feeds this collector. + void setSource(NodeSource* source); + +private: + struct NodeStage + { + bool present; + uint8_t nodeId; + uint8_t data[NET_STAGE_BYTES]; + uint16_t len; + }; + + void drainDataSocket(uint32_t nowMs); + void pullSimulator(uint32_t nowMs); + void pollNextNode(uint32_t nowMs); + void interleaveAvailable(uint32_t nowMs); + bool produceRound(); + void sendPoll(const NodeInfo& node); + void appendPcmToStage(int idx, const uint8_t* p, size_t n); + static bool isOnline(const NodeInfo& n); + + NodeRegistry& nodes; + RecordingController& recorder; + DataLogger& logger; + NodeSource* source; + WiFiUDP dataUdp; + + NodeStage stages[NET_MAX_NODES]; + int lastPollIndex; + uint32_t lastPollMs; + uint32_t blockedSinceMs; + uint32_t chunkSequence; + bool started; +}; diff --git a/src/network/discovery_service.cpp b/src/network/discovery_service.cpp new file mode 100644 index 0000000..3e21df1 --- /dev/null +++ b/src/network/discovery_service.cpp @@ -0,0 +1,196 @@ +#include "discovery_service.h" + +#include +#include + + +DiscoveryService::DiscoveryService(NodeRegistry& registry, RecordingController& recorder) +: +Service("Discovery", 100), +nodes(registry), +recorder(recorder), +lastScanMs(0), +lastBeaconMs(0) +{ +} + + +void DiscoveryService::begin() +{ + udp.begin(NET_CONTROL_PORT); + + Serial.printf("[Discovery] Control UDP on port %d\n", NET_CONTROL_PORT); +} + + +void DiscoveryService::update() +{ + uint32_t now = millis(); + + handlePacket(now); + + // Front panel asked to start/stop: broadcast it. + if (recorder.consumeStartRequest()) + { + sendStart(); + recorder.onStartBroadcastSent(); + Serial.println("[Discovery] START broadcast sent"); + } + + if (recorder.consumeStopRequest()) + { + sendStop(); + recorder.onStopBroadcastSent(); + Serial.println("[Discovery] STOP broadcast sent"); + } + + // Probe any station that just associated. + if (now - lastScanMs >= NET_PROBE_SCAN_MS) + { + lastScanMs = now; + scanStations(now); + } + + // Safety-net beacon so nodes that missed a unicast probe still answer. + if (now - lastBeaconMs >= NET_BEACON_MS) + { + lastBeaconMs = now; + broadcastBeacon(now); + } + + nodes.updateOnline(now); +} + + +void DiscoveryService::scanStations(uint32_t nowMs) +{ + wifi_sta_list_t staList; + esp_netif_sta_list_t netifList; + + if (esp_wifi_ap_get_sta_list(&staList) != ESP_OK) + { + return; + } + + if (esp_netif_get_sta_list(&staList, &netifList) != ESP_OK) + { + return; + } + + for (int i = 0; i < netifList.num; i++) + { + uint32_t ipRaw = netifList.sta[i].ip.addr; + if (ipRaw == 0) + { + continue; // not yet assigned by DHCP + } + + if (nodes.findByMac(netifList.sta[i].mac) < 0) + { + IPAddress ip(ipRaw); + sendProbeTo(ip); + Serial.printf("[Discovery] New client %s, sending probe\n", + ip.toString().c_str()); + } + } +} + + +void DiscoveryService::sendProbeTo(const IPAddress& ip) +{ + uint8_t buf[8]; + size_t n = nprot::buildProbe(buf, sizeof(buf), recorder.isRecording()); + + udp.beginPacket(ip, NET_CONTROL_PORT); + udp.write(buf, n); + udp.endPacket(); +} + + +void DiscoveryService::broadcastBeacon(uint32_t nowMs) +{ + IPAddress bcast; + bcast.fromString(NET_AP_BCAST_IP); + sendProbeTo(bcast); +} + + +void DiscoveryService::handlePacket(uint32_t nowMs) +{ + int len = udp.parsePacket(); + if (len <= 0) + { + return; + } + + uint8_t buf[256]; + int n = udp.read(buf, sizeof(buf)); + if (n <= 1) + { + return; + } + + IPAddress remoteIp = udp.remoteIP(); + + switch (buf[0]) + { + case nprot::MSG_PROBE_RESP: + { + nprot::ProbeResponse resp; + if (nprot::parseProbeResponse(buf, n, resp)) + { + int idx = nodes.upsert(resp, remoteIp, nowMs); + if (idx >= 0) + { + Serial.printf("[Discovery] Node %u online (%u mics, %u Hz)\n", + resp.nodeId, resp.micCount, resp.sampleRate); + } + else + { + Serial.println("[Discovery] Node table full, node ignored"); + } + } + break; + } + + case nprot::MSG_HEARTBEAT: + { + nprot::Heartbeat hb; + if (nprot::parseHeartbeat(buf, n, hb)) + { + nodes.touch(hb.nodeId, nowMs); + } + break; + } + + default: + break; + } +} + + +void DiscoveryService::broadcast(const uint8_t* buf, size_t len) +{ + IPAddress bcast; + bcast.fromString(NET_AP_BCAST_IP); + + udp.beginPacket(bcast, NET_CONTROL_PORT); + udp.write(buf, len); + udp.endPacket(); +} + + +void DiscoveryService::sendStart() +{ + uint8_t buf[8]; + size_t n = nprot::buildStart(buf, sizeof(buf)); + broadcast(buf, n); +} + + +void DiscoveryService::sendStop() +{ + uint8_t buf[8]; + size_t n = nprot::buildStop(buf, sizeof(buf)); + broadcast(buf, n); +} diff --git a/src/network/discovery_service.h b/src/network/discovery_service.h new file mode 100644 index 0000000..9814f13 --- /dev/null +++ b/src/network/discovery_service.h @@ -0,0 +1,42 @@ +#pragma once + +// ============================================================================ +// DiscoveryService - control plane of the node protocol. +// +// Detects new WiFi clients (AP station list -> unicast PROBE), discovers +// nodes that missed a probe via periodic broadcast beacons, registers +// PROBE_RESP / HEARTBEAT packets in the NodeRegistry, tracks offline state, +// and broadcasts START / STOP when the RecordingController requests it. +// ============================================================================ + +#include + +#include "../services/service.h" +#include "../core/node_registry.h" +#include "network_config.h" + + +class DiscoveryService : public Service +{ +public: + DiscoveryService(NodeRegistry& registry, RecordingController& recorder); + + void begin() override; + void update() override; + +private: + void scanStations(uint32_t nowMs); + void sendProbeTo(const IPAddress& ip); + void broadcastBeacon(uint32_t nowMs); + void handlePacket(uint32_t nowMs); + void broadcast(const uint8_t* buf, size_t len); + void sendStart(); + void sendStop(); + + NodeRegistry& nodes; + RecordingController& recorder; + WiFiUDP udp; + + uint32_t lastScanMs; + uint32_t lastBeaconMs; +}; diff --git a/src/network/network_config.h b/src/network/network_config.h new file mode 100644 index 0000000..21d2182 --- /dev/null +++ b/src/network/network_config.h @@ -0,0 +1,82 @@ +#pragma once + +// ============================================================================ +// Network configuration for the Hub <-> Node UDP protocol. +// See docs/network_protocol.md for the wire format. +// ============================================================================ + +// --- UDP ports -------------------------------------------------------------- + +// Control plane: discovery, status, start/stop, polling. +#define NET_CONTROL_PORT 4210 + +// Data plane: audio DATA packets. +#define NET_DATA_PORT 4211 + +// Subnet broadcast address for the soft-AP (192.168.4.1/24). +#define NET_AP_BCAST_IP "192.168.4.255" + +// --- Node registry ---------------------------------------------------------- + +#define NET_MAX_NODES 10 // 10 nodes x 4 mics = 40 channels + +// --- Discovery -------------------------------------------------------------- + +// How often to scan the AP station list for newly-connected clients and send +// them a unicast PROBE (ms). +#define NET_PROBE_SCAN_MS 1000 + +// How often to broadcast a PROBE to the whole subnet as a fallback so nodes +// that missed a unicast probe are still discovered (ms). +#define NET_BEACON_MS 10000 + +// A node that has not been heard from in this long is marked offline (ms). +#define NET_OFFLINE_MS 5000 + +// --- Data collection -------------------------------------------------------- + +// Round-robin: interval between sending POLL to consecutive nodes (ms). +#define NET_POLL_PERIOD_MS 200 + +// How long to wait for a polled node's DATA packets before moving on (ms). +#define NET_DATA_RX_TIMEOUT_MS 30 + +// Maximum UDP payload accepted for DATA packets (1472 keeps below the +// Ethernet MTU so no IP fragmentation occurs). +#define NET_MAX_UDP_PAYLOAD 1472 + +// Microphones per node. Fixed by the hardware design (4 mics x 10 nodes). +#define NET_MICS_PER_NODE 4 + +// Frames per interleave round. One round is ROUND_FRAMES x 40 ch x 2 B +// = 7680 bytes, sized to fit an 8 KB DataLogger chunk (STORAGE_LOG_CHUNK_SIZE). +// Enforced at compile time by a static_assert in the collector. +#define NET_ROUND_FRAMES 96 + +// Per-node staging buffer (bytes) used to reassemble interleave rounds. +// A round is only 1536 B/node, so 4 KB absorbs network burstiness without +// blowing the DRAM budget (10 nodes x 4 KB = 40 KB static). +#define NET_STAGE_BYTES 4096 + +// --- Node simulator --------------------------------------------------------- + +// 1 = build the built-in NodeSimulator (virtual nodes generating sample +// audio). No node hardware or PC required to produce sample data. +#define NET_SIM_ENABLED 1 + +#define NET_SIM_NODES 10 + +#define NET_SIM_SAMPLE_RATE_HZ 48000 + +// --- Front panel (LED / button) -------------------------------------------- + +// Recording status LED. GPIO 12 is a strapping pin: wire it ACTIVE-HIGH +// (GPIO12 -> LED -> 220R -> GND) and keep it low/floating at boot. +#define NET_LED_GPIO 22 + +// Start/stop button. GPIO 34 is input-only with NO internal pull-up: wire a +// 10k pull-up to 3.3V and the button between the pin and GND (active-low). +#define NET_BUTTON_GPIO 34 + +// Button debounce window (ms). +#define NET_BUTTON_DEBOUNCE_MS 50 diff --git a/src/network/node_protocol.h b/src/network/node_protocol.h new file mode 100644 index 0000000..0ccb0b0 --- /dev/null +++ b/src/network/node_protocol.h @@ -0,0 +1,237 @@ +#pragma once + +// ============================================================================ +// Hub <-> Node wire protocol (see docs/network_protocol.md). +// +// Self-contained: the node firmware project should be able to include this +// header as-is. All multi-byte values are little-endian. Packets are: +// +// [0] message type (1 byte) +// [1] protocol version (1 byte) +// [2..] payload +// +// Control plane lives on NET_CONTROL_PORT, audio data on NET_DATA_PORT. +// ============================================================================ + +#include +#include + + +namespace nprot +{ + +constexpr uint8_t PROTOCOL_VERSION = 1; + +// --- Message types ---------------------------------------------------------- + +enum MsgType : uint8_t +{ + MSG_PROBE = 0x01, // Hub -> Node (unicast on connect, or beacon) + MSG_PROBE_RESP = 0x02, // Node -> Hub (status + capabilities) + MSG_START = 0x03, // Hub -> Node (broadcast: begin recording) + MSG_STOP = 0x04, // Hub -> Node (broadcast: stop recording) + MSG_HEARTBEAT = 0x05, // Node -> Hub (keep-alive + status) + MSG_POLL = 0x06, // Hub -> Node (unicast: send buffered audio) + MSG_POLL_ACK = 0x07, // Node -> Hub (response sizing, optional) + MSG_DATA = 0x10 // Node -> Hub (audio payload, data port) +}; + +enum NodeStatus : uint8_t +{ + STATUS_IDLE = 0, + STATUS_RECORDING = 1, + STATUS_ERROR = 2 +}; + +// --- Packet sizes ----------------------------------------------------------- + +constexpr size_t PROBE_RESP_PAYLOAD = 1 + 6 + 16 + 1 + 4 + 2 + 4 + 1; // 35 +constexpr size_t HEARTBEAT_PAYLOAD = 1 + 4 + 4 + 1; // 10 +constexpr size_t POLL_ACK_PAYLOAD = 1 + 4 + 4; // 9 +constexpr size_t DATA_HEADER_PAYLOAD = 1 + 1 + 4; // 6 + +// --- Control plane builders (Hub side) -------------------------------------- + +// PROBE: [ver][flags] flags bit0 = recording +inline size_t buildProbe(uint8_t* buf, size_t cap, bool recording) +{ + if (cap < 3) return 0; + buf[0] = MSG_PROBE; + buf[1] = PROTOCOL_VERSION; + buf[2] = recording ? 0x01 : 0x00; + return 3; +} + +// START / STOP: [ver] +inline size_t buildStart(uint8_t* buf, size_t cap) +{ + if (cap < 2) return 0; + buf[0] = MSG_START; + buf[1] = PROTOCOL_VERSION; + return 2; +} + +inline size_t buildStop(uint8_t* buf, size_t cap) +{ + if (cap < 2) return 0; + buf[0] = MSG_STOP; + buf[1] = PROTOCOL_VERSION; + return 2; +} + +// POLL: [ver][nodeId] +inline size_t buildPoll(uint8_t* buf, size_t cap, uint8_t nodeId) +{ + if (cap < 3) return 0; + buf[0] = MSG_POLL; + buf[1] = PROTOCOL_VERSION; + buf[2] = nodeId; + return 3; +} + +// --- Node side builders ----------------------------------------------------- + +struct ProbeResponse +{ + uint8_t nodeId; + uint8_t mac[6]; + char firmware[16]; + uint8_t micCount; + uint32_t sampleRate; + uint16_t bitsPerSample; + uint32_t bufferBytes; + uint8_t status; +}; + +// PROBE_RESP: [ver][nodeId][mac(6)][firmware(16)][mics][rate(4)][bits(2)][buffer(4)][status] +inline size_t buildProbeResponse(uint8_t* buf, size_t cap, const ProbeResponse& r) +{ + if (cap < PROBE_RESP_PAYLOAD + 1) return 0; + size_t o = 0; + buf[o++] = MSG_PROBE_RESP; + buf[o++] = PROTOCOL_VERSION; + buf[o++] = r.nodeId; + memcpy(buf + o, r.mac, 6); o += 6; + memcpy(buf + o, r.firmware, 16); o += 16; + buf[o++] = r.micCount; + buf[o++] = r.sampleRate & 0xFF; buf[o++] = (r.sampleRate >> 8) & 0xFF; + buf[o++] = (r.sampleRate >> 16) & 0xFF; buf[o++] = (r.sampleRate >> 24) & 0xFF; + buf[o++] = r.bitsPerSample & 0xFF; buf[o++] = (r.bitsPerSample >> 8) & 0xFF; + buf[o++] = r.bufferBytes & 0xFF; buf[o++] = (r.bufferBytes >> 8) & 0xFF; + buf[o++] = (r.bufferBytes >> 16) & 0xFF; buf[o++] = (r.bufferBytes >> 24) & 0xFF; + buf[o++] = r.status; + return o; +} + +struct Heartbeat +{ + uint8_t nodeId; + uint32_t uptimeMs; + uint32_t bufferFillBytes; + uint8_t status; +}; + +// HEARTBEAT: [ver][nodeId][uptime(4)][bufferFill(4)][status] +inline size_t buildHeartbeat(uint8_t* buf, size_t cap, const Heartbeat& h) +{ + if (cap < HEARTBEAT_PAYLOAD + 1) return 0; + size_t o = 0; + buf[o++] = MSG_HEARTBEAT; + buf[o++] = PROTOCOL_VERSION; + buf[o++] = h.nodeId; + buf[o++] = h.uptimeMs & 0xFF; buf[o++] = (h.uptimeMs >> 8) & 0xFF; + buf[o++] = (h.uptimeMs >> 16) & 0xFF; buf[o++] = (h.uptimeMs >> 24) & 0xFF; + buf[o++] = h.bufferFillBytes & 0xFF; buf[o++] = (h.bufferFillBytes >> 8) & 0xFF; + buf[o++] = (h.bufferFillBytes >> 16) & 0xFF; buf[o++] = (h.bufferFillBytes >> 24) & 0xFF; + buf[o++] = h.status; + return o; +} + +// POLL_ACK: [ver][nodeId][seq(4)][framesReady(4)] +inline size_t buildPollAck(uint8_t* buf, size_t cap, uint8_t nodeId, uint32_t seq, uint32_t framesReady) +{ + if (cap < POLL_ACK_PAYLOAD + 1) return 0; + size_t o = 0; + buf[o++] = MSG_POLL_ACK; + buf[o++] = PROTOCOL_VERSION; + buf[o++] = nodeId; + for (int i = 0; i < 4; i++) buf[o++] = (seq >> (8 * i)) & 0xFF; + for (int i = 0; i < 4; i++) buf[o++] = (framesReady >> (8 * i)) & 0xFF; + return o; +} + +// --- Data plane ------------------------------------------------------------- + +struct DataHeader +{ + uint8_t nodeId; + uint8_t micCount; + uint32_t frameCount; // number of frames per mic in this packet +}; + +inline uint16_t dataFrameSize(uint8_t micCount) +{ + return (uint16_t)micCount * 2; // 16-bit samples +} + +// DATA: [ver][nodeId][mics][frameCount(4)] then PCM (frameCount * mics * 2 bytes) +inline size_t buildData(uint8_t* buf, size_t cap, const DataHeader& h, const void* pcm) +{ + size_t payload = DATA_HEADER_PAYLOAD + (size_t)h.frameCount * dataFrameSize(h.micCount); + if (cap < payload + 1) return 0; + size_t o = 0; + buf[o++] = MSG_DATA; + buf[o++] = PROTOCOL_VERSION; + buf[o++] = h.nodeId; + buf[o++] = h.micCount; + for (int i = 0; i < 4; i++) buf[o++] = (h.frameCount >> (8 * i)) & 0xFF; + memcpy(buf + o, pcm, (size_t)h.frameCount * dataFrameSize(h.micCount)); + return payload + 1; +} + +// --- Parsers (Hub side) ----------------------------------------------------- + +inline uint32_t readU32(const uint8_t* b) { return b[0] | (b[1] << 8) | (b[2] << 16) | ((uint32_t)b[3] << 24); } +inline uint16_t readU16(const uint8_t* b) { return b[0] | (b[1] << 8); } + +inline bool parseProbeResponse(const uint8_t* buf, size_t len, ProbeResponse& out) +{ + if (len < PROBE_RESP_PAYLOAD + 1) return false; + if (buf[0] != MSG_PROBE_RESP || buf[1] != PROTOCOL_VERSION) return false; + size_t o = 2; + out.nodeId = buf[o++]; + memcpy(out.mac, buf + o, 6); o += 6; + memcpy(out.firmware, buf + o, 16); o += 16; + out.micCount = buf[o++]; + out.sampleRate = readU32(buf + o); o += 4; + out.bitsPerSample = readU16(buf + o); o += 2; + out.bufferBytes = readU32(buf + o); o += 4; + out.status = buf[o]; + out.firmware[15] = '\0'; + return true; +} + +inline bool parseHeartbeat(const uint8_t* buf, size_t len, Heartbeat& out) +{ + if (len < HEARTBEAT_PAYLOAD + 1) return false; + if (buf[0] != MSG_HEARTBEAT || buf[1] != PROTOCOL_VERSION) return false; + size_t o = 2; + out.nodeId = buf[o++]; + out.uptimeMs = readU32(buf + o); o += 4; + out.bufferFillBytes = readU32(buf + o); o += 4; + out.status = buf[o]; + return true; +} + +inline bool parseDataHeader(const uint8_t* buf, size_t len, DataHeader& out) +{ + if (len < DATA_HEADER_PAYLOAD + 1) return false; + if (buf[0] != MSG_DATA || buf[1] != PROTOCOL_VERSION) return false; + size_t o = 2; + out.nodeId = buf[o++]; + out.micCount = buf[o++]; + out.frameCount = readU32(buf + o); + return true; +} + +} // namespace nprot diff --git a/src/network/node_simulator.cpp b/src/network/node_simulator.cpp new file mode 100644 index 0000000..7e2e6e0 --- /dev/null +++ b/src/network/node_simulator.cpp @@ -0,0 +1,139 @@ +#include "node_simulator.h" + +#include +#include + + +namespace +{ +constexpr float SIM_TWO_PI = 6.28318530718f; + +// PCM frame size for one node (4 ch x 2 B), matches CollectionService. +constexpr size_t FRAME_BYTES = NET_MICS_PER_NODE * 2; + +// Mic frequencies step by 30 Hz so each mic of a node is distinct. +constexpr float MIC_STEP_HZ = 30.0f; + +// Amplitude ~0.25 FS; the DC marker per node is small and inaudible but easy +// to spot in a waveform editor when verifying channel mapping. +constexpr float AMPLITUDE = 8000.0f; +constexpr float DC_MARKER_STEP = 50.0f; +} + + +NodeSimulator::NodeSimulator(NodeRegistry& registry, RecordingController& recorder) +: +Service("Sim", 500), +nodes(registry), +recorder(recorder) +{ + memset(lastCallMs, 0, sizeof(lastCallMs)); + memset(phase, 0, sizeof(phase)); +} + + +void NodeSimulator::begin() +{ + uint32_t now = millis(); + + for (uint8_t i = 0; i < NET_SIM_NODES; i++) + { + nprot::ProbeResponse resp; + memset(&resp, 0, sizeof(resp)); + + resp.nodeId = i; + resp.mac[0] = 0x02; + resp.mac[5] = i; + memcpy(resp.firmware, "node-sim", 9); + resp.micCount = NET_MICS_PER_NODE; + resp.sampleRate = NET_SIM_SAMPLE_RATE_HZ; + resp.bitsPerSample = 16; + resp.bufferBytes = 64 * 1024; + resp.status = nprot::STATUS_IDLE; + + IPAddress ip(192, 168, 4, 200 + i); + nodes.upsert(resp, ip, now); + lastCallMs[i] = now; + } + + Serial.printf("[Sim] %d virtual nodes registered\n", NET_SIM_NODES); +} + + +void NodeSimulator::update() +{ + uint32_t now = millis(); + + // Keep the virtual nodes "online" so discovery/dashboard show them. + for (uint8_t i = 0; i < NET_SIM_NODES; i++) + { + nodes.touch(i, now); + } +} + + +float NodeSimulator::baseFrequency(uint8_t nodeId) +{ + return 440.0f + nodeId * 40.0f; +} + + +size_t NodeSimulator::readAudio(uint8_t nodeId, uint8_t* out, size_t maxBytes) +{ + if (!recorder.isRecording() || nodeId >= NET_SIM_NODES) + { + return 0; + } + + uint32_t now = millis(); + + if (lastCallMs[nodeId] == 0) + { + lastCallMs[nodeId] = now; + return 0; + } + + uint32_t elapsed = now - lastCallMs[nodeId]; + lastCallMs[nodeId] = now; + + size_t maxFrames = maxBytes / FRAME_BYTES; + uint32_t frames = (uint32_t)(((uint64_t)elapsed * NET_SIM_SAMPLE_RATE_HZ) / 1000ULL); + + if (frames > maxFrames) + { + frames = (uint32_t)maxFrames; + } + + if (frames == 0) + { + return 0; + } + + float base = baseFrequency(nodeId); + float dc = DC_MARKER_STEP * nodeId; + + uint8_t* p = out; + + for (uint32_t f = 0; f < frames; f++) + { + for (uint8_t mic = 0; mic < NET_MICS_PER_NODE; mic++) + { + float& ph = phase[nodeId][mic]; + float freq = base + MIC_STEP_HZ * mic; + + ph += SIM_TWO_PI * freq / (float)NET_SIM_SAMPLE_RATE_HZ; + if (ph >= SIM_TWO_PI) + { + ph -= SIM_TWO_PI; + } + + int16_t sample = (int16_t)(AMPLITUDE * sinf(ph) + dc); + + p[0] = (uint8_t)(sample & 0xFF); + p[1] = (uint8_t)((sample >> 8) & 0xFF); + p += 2; + } + } + + return frames * FRAME_BYTES; +} diff --git a/src/network/node_simulator.h b/src/network/node_simulator.h new file mode 100644 index 0000000..48ed2c1 --- /dev/null +++ b/src/network/node_simulator.h @@ -0,0 +1,40 @@ +#pragma once + +// ============================================================================ +// NodeSimulator - virtual nodes that generate sample audio in software. +// +// Implements NodeSource so the CollectionService feeds simulated PCM through +// the exact same path as real UDP nodes. Each virtual node emits a distinct +// sine tone per mic plus a small DC marker per node, so channel -> track +// mapping is easy to verify in the WAV on a PC. +// +// Enable/disable with NET_SIM_ENABLED (network_config.h). Use the simulator +// OR real nodes, not both (node IDs would collide). +// ============================================================================ + +#include "../services/service.h" +#include "../core/node_registry.h" +#include "collection_service.h" +#include "network_config.h" + + +class NodeSimulator : public Service, public NodeSource +{ +public: + NodeSimulator(NodeRegistry& registry, RecordingController& recorder); + + void begin() override; + void update() override; + + // NodeSource + size_t readAudio(uint8_t nodeId, uint8_t* out, size_t maxBytes) override; + +private: + static float baseFrequency(uint8_t nodeId); + + NodeRegistry& nodes; + RecordingController& recorder; + + uint32_t lastCallMs[NET_SIM_NODES]; + float phase[NET_SIM_NODES][NET_MICS_PER_NODE]; +}; diff --git a/src/services/front_panel_service.cpp b/src/services/front_panel_service.cpp new file mode 100644 index 0000000..6002d46 --- /dev/null +++ b/src/services/front_panel_service.cpp @@ -0,0 +1,95 @@ +#include "front_panel_service.h" + + +FrontPanelService::FrontPanelService(RecordingController& recorder) +: +Service("Panel", 20), +recorder(recorder), +lastBlinkMs(0), +ledOn(false), +buttonRaw(false), +buttonDebounced(false), +buttonChangeMs(0) +{ +} + + +void FrontPanelService::begin() +{ + pinMode(NET_LED_GPIO, OUTPUT); + digitalWrite(NET_LED_GPIO, LOW); + + // GPIO34 is input-only; the external 10k pull-up is required. + pinMode(NET_BUTTON_GPIO, INPUT); + + buttonRaw = (digitalRead(NET_BUTTON_GPIO) == LOW); + buttonDebounced = buttonRaw; + + Serial.printf("[Panel] LED on GPIO%d, button on GPIO%d\n", + NET_LED_GPIO, NET_BUTTON_GPIO); +} + + +void FrontPanelService::update() +{ + uint32_t now = millis(); + + updateLed(now); + updateButton(now); +} + + +void FrontPanelService::updateLed(uint32_t nowMs) +{ + if (recorder.isRecording()) + { + digitalWrite(NET_LED_GPIO, HIGH); // solid + return; + } + + // 1 Hz blink while idle (500 ms on / 500 ms off). + if (nowMs - lastBlinkMs >= 500) + { + lastBlinkMs = nowMs; + ledOn = !ledOn; + digitalWrite(NET_LED_GPIO, ledOn ? HIGH : LOW); + } +} + + +void FrontPanelService::updateButton(uint32_t nowMs) +{ + bool raw = (digitalRead(NET_BUTTON_GPIO) == LOW); + + if (raw != buttonRaw) + { + buttonRaw = raw; + buttonChangeMs = nowMs; + return; + } + + if (nowMs - buttonChangeMs < NET_BUTTON_DEBOUNCE_MS) + { + return; + } + + if (raw != buttonDebounced) + { + buttonDebounced = raw; + + if (buttonDebounced) + { + // Pressed edge: toggle recording. + if (recorder.isRecording()) + { + recorder.requestStop(); + Serial.println("[Panel] Button: STOP requested"); + } + else + { + recorder.requestStart(); + Serial.println("[Panel] Button: START requested"); + } + } + } +} diff --git a/src/services/front_panel_service.h b/src/services/front_panel_service.h new file mode 100644 index 0000000..af9d8f0 --- /dev/null +++ b/src/services/front_panel_service.h @@ -0,0 +1,38 @@ +#pragma once + +// ============================================================================ +// FrontPanelService - user-facing start/stop + status LED. +// +// GPIO12 LED: blinks at 1 Hz while idle, solid while recording. Active-high +// (GPIO12 -> LED -> 220R -> GND; the pin is a strapping pin, keep it low). +// +// GPIO34 button: debounced, active-low (10k pull-up to 3.3V, button to GND). +// Toggles recording start/stop via the RecordingController. +// ============================================================================ + +#include "../services/service.h" +#include "../core/node_registry.h" +#include "../network/network_config.h" + + +class FrontPanelService : public Service +{ +public: + FrontPanelService(RecordingController& recorder); + + void begin() override; + void update() override; + +private: + void updateLed(uint32_t nowMs); + void updateButton(uint32_t nowMs); + + RecordingController& recorder; + + uint32_t lastBlinkMs; + bool ledOn; + + bool buttonRaw; + bool buttonDebounced; + uint32_t buttonChangeMs; +}; diff --git a/src/services/web_service.cpp b/src/services/web_service.cpp index 7eea953..3462d28 100644 --- a/src/services/web_service.cpp +++ b/src/services/web_service.cpp @@ -1,11 +1,14 @@ #include "web_service.h" -WebService::WebService(DashboardState& state, DiagnosticsState& diag_state, StorageState& storage_state) +WebService::WebService(DashboardState& state, DiagnosticsState& diag_state, StorageState& storage_state, + NodeRegistry& nodes, RecordingController& recorder) : Service("Web", 10), dashboardState(state), diagnosticsState(diag_state), storageState(storage_state), + nodes(nodes), + recorder(recorder), server(80), webSocket(81) { @@ -95,6 +98,7 @@ function connectWebSocket() { document.getElementById("cpu_frequency").innerHTML = "CPU Frequency: " + data.diagnostics.cpu_frequency + "MHz"; updateStorage(data.storage); + updateNetwork(data.network); }; socket.onclose = function() { @@ -136,6 +140,27 @@ function updateStorage(storage) { window.onload = connectWebSocket; +function updateNetwork(network) { + if (!network) return; + + document.getElementById("rec_status").innerHTML = + network.recording === "true" ? "Recording: ACTIVE" : "Recording: idle"; + document.getElementById("rec_status").style.color = + network.recording === "true" ? "#00ff99" : "#ffcc00"; + + document.getElementById("node_count").innerHTML = + "Connected: " + network.online_count + " / " + network.node_count; + + let html = ""; + for (const n of network.nodes) { + const color = n.online === "true" ? "#00ff99" : "#ff5555"; + html += '
Node ' + n.id + + " @ " + n.ip + " (" + n.mics + " mics)" + + (n.online === "true" ? " online" : " OFFLINE") + "
"; + } + document.getElementById("node_list").innerHTML = html || "(no nodes)"; +} + @@ -161,6 +186,13 @@ window.onload = connectWebSocket;
Estimated write speed: Loading...
+
+

Network:

+
Recording: Loading...
+
Connected: Loading...
+
(no nodes)
+
+