Step 4.2: PTP-paced AES67 RTP sender with 1 kHz test tone

- aes67_tx: TX task woken by an esp_timer every packet time; sends every
  packet whose last sample is in the past (PTP time), RTP ts =
  (sample index from PTP + clk_offset) mod 2^32, packets aligned to
  multiples of the packet size. Sends only while PTP is locked; resyncs
  on lock, clock step or > 20 ms lag. L24/L16 big-endian, TTL/DSCP/
  multicast IF from config; a config save restarts TX.
- Built-in 1 kHz tone at -18 dBFS, phase from the PTP sample index;
  aes67_tx_set_source() pull callback for later sources (underruns
  counted, padded with silence). status: tx_packets, underruns.
- Verified with a receiver script on the PC: 1 ms L24 stereo 1000/s,
  0 gaps, ts step 48, tone -18.00 dBFS within 1.2 LSB of the ideal
  PTP-phased sine; also L16 mono 0.333 ms, 0.125 ms (8000/s) and 4 ms.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

components/aes67_tx/CMakeLists.txt

@@ -1,3 +1,3 @@
 idf_component_register(SRCS "aes67_tx.c"
                        INCLUDE_DIRS "include"
-                       PRIV_REQUIRES aes67_web lwip)
+                       PRIV_REQUIRES aes67_net aes67_ptp aes67_web esp_netif esp_timer lwip)

components/aes67_tx/aes67_tx.c

@@ -1,11 +1,34 @@
 #include "aes67_tx.h"
 
+#include <math.h>
 #include <stdio.h>
+#include <string.h>
 
 #include "aes67_cfg.h"
+#include "aes67_net.h"
+#include "aes67_ptp.h"
 #include "aes67_web.h"
+#include "esp_log.h"
+#include "esp_random.h"
+#include "esp_timer.h"
+#include "freertos/FreeRTOS.h"
+#include "freertos/task.h"
 #include "lwip/inet.h"
+#include "lwip/sockets.h"
 
+#define MAX_FRAMES      192         // 4 ms at 48 kHz
+#define MAX_CH          2
+#define RTP_HDR         12
+#define MAX_LAG_NS      20000000    // more than 20 ms behind: resync instead of bursting
+#define TONE_HZ         1000
+#define TONE_DBFS       -18.0
+
+static const char *TAG = "aes67_tx";
+
+static TaskHandle_t s_task;
+static volatile aes67_tx_read_cb_t s_read;
+static volatile bool s_reconfig = true;
+static volatile uint32_t s_packets, s_underruns;
 // Defaults follow the Riedel Director 4-wire AES67 output; channels 2 (core default).
 static const char AES67_DEFAULTS[] =
     "{\"name\":\"AES67\",\"enabled\":true,\"discovery\":\"sap\",\"mcast\":\"239.69.1.10\","
@@ -58,16 +81,224 @@ static bool aes67_validate(const cJSON *g, char *err, size_t n)
     return ok;
 }
 
-// Stub (build step 2): counters stay 0 until the RTP sender exists (step 4).
 static void tx_status(cJSON *st)
 {
-    cJSON_AddNumberToObject(st, "tx_packets", 0);
-    cJSON_AddNumberToObject(st, "underruns", 0);
+    cJSON_AddNumberToObject(st, "tx_packets", s_packets);
+    cJSON_AddNumberToObject(st, "underruns", s_underruns);
+}
+
+static void aes67_apply(const cJSON *g)
+{
+    s_reconfig = true;  // the TX task picks up the new settings
+}
+
+void aes67_tx_set_source(aes67_tx_read_cb_t read)
+{
+    s_read = read;
+}
+
+/* ----- sender ----- */
+
+typedef struct {
+    bool enabled;
+    struct sockaddr_in dst;
+    uint8_t ttl, dscp, pt;
+    uint32_t ssrc, clk_offset;
+    int channels, rate, bytes;      // bytes per sample: 3 (L24) or 2 (L16)
+    int frames;                     // samples per channel per packet
+} tx_cfg_t;
+
+static bool load(tx_cfg_t *c)
+{
+    cJSON *a = cfg_get("aes67");
+    if (!a) {
+        return false;
+    }
+#define NUM(k) cJSON_GetObjectItemCaseSensitive(a, k)->valuedouble
+    c->enabled = cJSON_IsTrue(cJSON_GetObjectItemCaseSensitive(a, "enabled"));
+    memset(&c->dst, 0, sizeof(c->dst));
+    c->dst.sin_family = AF_INET;
+    c->dst.sin_port = htons((uint16_t)NUM("port"));
+    inet_aton(cJSON_GetObjectItemCaseSensitive(a, "mcast")->valuestring, &c->dst.sin_addr);
+    c->ttl = (uint8_t)NUM("ttl");
+    c->dscp = (uint8_t)NUM("dscp");
+    c->pt = (uint8_t)NUM("pt");
+    c->ssrc = (uint32_t)NUM("ssrc");
+    c->clk_offset = (uint32_t)NUM("clk_offset");
+    c->channels = (int)NUM("channels");
+    c->rate = (int)NUM("rate");
+    c->bytes = strcmp(cJSON_GetObjectItemCaseSensitive(a, "encoding")->valuestring, "L16") == 0 ? 2 : 3;
+    c->frames = (int)lround(NUM("ptime") * c->rate / 1000.0);   // 0.333 ms -> 16 at 48 kHz
+#undef NUM
+    cJSON_Delete(a);
+    return c->frames > 0 && c->frames <= MAX_FRAMES && c->channels <= MAX_CH;
+}
+
+static int open_socket(const tx_cfg_t *c)
+{
+    int fd = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
+    if (fd < 0) {
+        return -1;
+    }
+    int tos = c->dscp << 2;
+    uint8_t ttl = c->ttl, loop = 0;
+    setsockopt(fd, IPPROTO_IP, IP_TOS, &tos, sizeof(tos));
+    setsockopt(fd, IPPROTO_IP, IP_MULTICAST_TTL, &ttl, sizeof(ttl));
+    setsockopt(fd, IPPROTO_IP, IP_MULTICAST_LOOP, &loop, sizeof(loop));
+    esp_netif_ip_info_t ip;
+    if (esp_netif_get_ip_info(aes67_net_netif(), &ip) == ESP_OK) {
+        struct in_addr ifa = { .s_addr = ip.ip.addr };
+        setsockopt(fd, IPPROTO_IP, IP_MULTICAST_IF, &ifa, sizeof(ifa));
+    }
+    return fd;
+}
+
+// Sample index since the PTP epoch (no overflow: seconds * rate first).
+static int64_t ns_to_samples(int64_t ns, int rate)
+{
+    return (ns / 1000000000LL) * rate + (ns % 1000000000LL) * rate / 1000000000LL;
+}
+
+// PTP time at which sample index s starts (rounded up).
+static int64_t samples_to_ns(int64_t s, int rate)
+{
+    return (s / rate) * 1000000000LL + ((s % rate) * 1000000000LL + rate - 1) / rate;
+}
+
+// 1 kHz tone, phase from the PTP sample index, so every sender's tone lines up.
+static void tone(int32_t *buf, int frames, int channels, int rate, int64_t s0)
+{
+    static float amp;
+    if (!amp) {
+        amp = (float)(pow(10.0, TONE_DBFS / 20.0) * 2147483647.0);
+    }
+    int period = rate / TONE_HZ;
+    for (int i = 0; i < frames; i++) {
+        int32_t v = (int32_t)(amp * sinf(2.0f * (float)M_PI * (float)((s0 + i) % period) / (float)period));
+        for (int ch = 0; ch < channels; ch++) {
+            buf[i * channels + ch] = v;
+        }
+    }
+}
+
+static void timer_cb(void *arg)
+{
+    xTaskNotifyGive(s_task);
+}
+
+static void tx_task(void *arg)
+{
+    tx_cfg_t c = { 0 };
+    int fd = -1;
+    esp_timer_handle_t timer = NULL;
+    const esp_timer_create_args_t targs = { .callback = timer_cb, .name = "aes67_tx" };
+    esp_timer_create(&targs, &timer);
+    static int32_t pcm[MAX_FRAMES * MAX_CH];
+    static uint8_t pkt[RTP_HDR + MAX_FRAMES * MAX_CH * 3];
+    uint16_t seq = (uint16_t)esp_random();
+    int64_t next = 0;           // sample index of the next packet; 0 = resync needed
+    const char *state = NULL, *new_state;
+
+    while (1) {
+        ulTaskNotifyTake(pdTRUE, pdMS_TO_TICKS(100));
+
+        if (s_reconfig) {
+            s_reconfig = false;
+            if (fd >= 0) {
+                close(fd);
+                fd = -1;
+            }
+            esp_timer_stop(timer);
+            if (!load(&c)) {
+                ESP_LOGE(TAG, "unsupported config (ptime/rate/channels)");
+                c.enabled = false;
+            }
+            next = 0;
+            if (c.enabled) {
+                fd = open_socket(&c);
+                // wake once per packet time; late wakes are caught up below
+                esp_timer_start_periodic(timer, (uint64_t)c.frames * 1000000ULL / c.rate);
+            }
+        }
+
+        int64_t now;
+        if (!c.enabled || fd < 0) {
+            new_state = "disabled";
+        } else if (!aes67_ptp_locked() || aes67_ptp_now_ns(&now) != ESP_OK) {
+            new_state = "waiting for PTP lock";
+            next = 0;
+        } else {
+            new_state = "sending";
+            int64_t due = ns_to_samples(now, c.rate);
+            // (Re)start aligned to a packet boundary, or after falling behind / a clock step back.
+            if (!next || due - next > (int64_t)c.rate * MAX_LAG_NS / 1000000000LL || next - due > 2 * c.frames) {
+                if (next) {
+                    ESP_LOGW(TAG, "resync (%lld samples off)", due - next);
+                }
+                next = (due / c.frames + 1) * c.frames;
+            }
+            // Send every packet whose last sample is in the past.
+            while (samples_to_ns(next + c.frames, c.rate) <= now) {
+                aes67_tx_read_cb_t read = s_read;
+                if (read) {
+                    size_t got = read(pcm, c.frames);
+                    if (got < (size_t)c.frames) {
+                        memset(pcm + got * c.channels, 0, (c.frames - got) * c.channels * sizeof(int32_t));
+                        s_underruns++;
+                    }
+                } else {
+                    tone(pcm, c.frames, c.channels, c.rate, next);
+                }
+                uint32_t ts = (uint32_t)next + c.clk_offset;
+                pkt[0] = 0x80;          // V=2
+                pkt[1] = c.pt & 0x7f;
+                pkt[2] = seq >> 8;
+                pkt[3] = seq & 0xff;
+                pkt[4] = ts >> 24;
+                pkt[5] = ts >> 16;
+                pkt[6] = ts >> 8;
+                pkt[7] = ts;
+                pkt[8] = c.ssrc >> 24;
+                pkt[9] = c.ssrc >> 16;
+                pkt[10] = c.ssrc >> 8;
+                pkt[11] = c.ssrc;
+                uint8_t *p = pkt + RTP_HDR;
+                for (int i = 0; i < c.frames * c.channels; i++) {
+                    uint32_t v = (uint32_t)pcm[i];      // big-endian, top bytes of the 32-bit sample
+                    *p++ = v >> 24;
+                    *p++ = v >> 16;
+                    if (c.bytes == 3) {
+                        *p++ = v >> 8;
+                    }
+                }
+                if (sendto(fd, pkt, p - pkt, 0, (struct sockaddr *)&c.dst, sizeof(c.dst)) > 0) {
+                    s_packets++;
+                }
+                seq++;
+                next += c.frames;
+            }
+        }
+        if (new_state != state) {
+            state = new_state;
+            if (c.enabled && state[0] == 's') {
+                ESP_LOGI(TAG, "%s: %s:%u, %s/%d/%d, %d samples per packet, %s", state,
+                         inet_ntoa(c.dst.sin_addr), ntohs(c.dst.sin_port), c.bytes == 3 ? "L24" : "L16",
+                         c.rate, c.channels, c.frames, s_read ? "source" : "1 kHz test tone");
+            } else {
+                ESP_LOGI(TAG, "%s", state);
+            }
+        }
+    }
+}
+
+esp_err_t aes67_tx_start(void)
+{
+    return xTaskCreate(tx_task, "aes67_tx", 4096, NULL, 16, &s_task) == pdPASS ? ESP_OK : ESP_ERR_NO_MEM;
 }
 
 esp_err_t aes67_tx_init(void)
 {
-    esp_err_t err = cfg_register("aes67", AES67_DEFAULTS, aes67_validate, NULL);
+    esp_err_t err = cfg_register("aes67", AES67_DEFAULTS, aes67_validate, aes67_apply);
     if (err != ESP_OK) {
         return err;
     }

components/aes67_tx/include/aes67_tx.h

@@ -2,7 +2,18 @@
 // Core component: must not depend on main/ (project code).
 #pragma once
 
+#include <stddef.h>
+#include <stdint.h>
+
 #include "esp_err.h"
 
-// Registers the "aes67" config group and the TX status fields. (Sender itself: step 4.)
+// Audio source: fill buf with frames x channels interleaved samples (full scale = INT32_MAX).
+// Return the number of frames delivered; fewer than asked counts as an underrun (padded with silence).
+typedef size_t (*aes67_tx_read_cb_t)(int32_t *buf, size_t frames);
+
+// Registers the "aes67" config group and the TX status fields.
 esp_err_t aes67_tx_init(void);
+// Start the sender task. Sends while PTP is locked and aes67.enabled is set.
+esp_err_t aes67_tx_start(void);
+// Set the audio source. NULL = built-in 1 kHz test tone at -18 dBFS, phase-locked to PTP time.
+void aes67_tx_set_source(aes67_tx_read_cb_t read);