Skip to content

feat(rtc-tcp-client): deliver received images via on_image callback #6

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
sedawwk merged 1 commit into from
Jul 13, 2026
Merged

feat(rtc-tcp-client): deliver received images via on_image callback #6

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 20 additions & 0 deletions modules/rtc-tcp-client/include/tuya_ai.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -181,6 +181,25 @@ typedef struct tai_text_msg {
uint8_t _reserved[8];
} tai_text_msg_t;

/* --- Image --------------------------------------------------------------- */
/* A received image (e.g. a cloud-generated picture) arrives as a stream of
* chunks: START (or ONE_SHOT) carries the first bytes and the image-params,
* MIDDLE continues, END terminates (len may be 0). The caller accumulates the
* chunks by stream_flag and decodes once the stream ends. format/width/height
* are populated from image-params on START/ONE_SHOT and 0 on MIDDLE/END. */
typedef struct tai_image_msg {
const uint8_t *data; /* encoded image bytes (JPEG/PNG); callback-lifetime */
size_t len;
uint8_t format; /* TAI_IMG_JPEG / TAI_IMG_PNG / 0=unknown */
uint16_t width; /* px, 0 if unknown / not on this chunk */
uint16_t height; /* px, 0 if unknown */
uint8_t stream_flag; /* TAI_STREAM_* */
uint16_t data_id; /* Data ID */
const char *event_id; /* turn id, borrowed; "" if none */
uint64_t timestamp_ms; /* stream-start ts (media header) */
uint8_t _reserved[8];
} tai_image_msg_t;

/* --- Event (generic) ----------------------------------------------------- */
typedef struct tai_event_msg {
uint16_t event_type; /* TAI_EVT_* */
Expand Down Expand Up @@ -293,6 +312,7 @@ typedef struct tai_config {
*/
void (*on_audio) (tai_ctx_t *ctx, const tai_audio_msg_t *msg, void *user_data);
void (*on_text) (tai_ctx_t *ctx, const tai_text_msg_t *msg, void *user_data);
void (*on_image) (tai_ctx_t *ctx, const tai_image_msg_t *msg, void *user_data);
void (*on_event) (tai_ctx_t *ctx, const tai_event_msg_t *msg, void *user_data);
void (*on_disconnect)(tai_ctx_t *ctx, const tai_disconnect_msg_t *msg, void *user_data);
void *user_data;
Expand Down
1 change: 1 addition & 0 deletions modules/rtc-tcp-client/src/tai_client.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -312,6 +312,7 @@ tai_ctx_t *tai_ctx_init(void *mem, const tai_config_t *cfg)
ctx->agent_token = cfg->agent_token;
ctx->on_audio = cfg->on_audio;
ctx->on_text = cfg->on_text;
ctx->on_image = cfg->on_image;
ctx->on_event = cfg->on_event;
ctx->on_disconnect = cfg->on_disconnect;
ctx->user_data = cfg->user_data;
Expand Down
1 change: 1 addition & 0 deletions modules/rtc-tcp-client/src/tai_internal.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -294,6 +294,7 @@ struct tai_ctx {
/* Receive callbacks (struct-based; see tuya_ai.h) */
void (*on_audio) (tai_ctx_t *, const tai_audio_msg_t *, void *);
void (*on_text) (tai_ctx_t *, const tai_text_msg_t *, void *);
void (*on_image) (tai_ctx_t *, const tai_image_msg_t *, void *);
void (*on_event) (tai_ctx_t *, const tai_event_msg_t *, void *);
void (*on_disconnect)(tai_ctx_t *, const tai_disconnect_msg_t *, void *);
void *user_data;
Expand Down
56 changes: 56 additions & 0 deletions modules/rtc-tcp-client/src/tai_protocol.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -518,6 +518,24 @@ static void emit_event(tai_ctx_t *ctx, uint16_t event_type,
ctx->on_event(ctx, &m, ctx->user_data);
}

static void emit_image(tai_ctx_t *ctx, const uint8_t *data, size_t len,
uint8_t format, uint16_t width, uint16_t height,
uint8_t stream_flag, uint16_t data_id, uint64_t ts_ms)
{
if (!ctx->on_image) return;
tai_image_msg_t m = {0};
m.data = data;
m.len = len;
m.format = format;
m.width = width;
m.height = height;
m.stream_flag = stream_flag;
m.data_id = data_id;
m.event_id = ctx->rx_event_id;
m.timestamp_ms = ts_ms;
ctx->on_image(ctx, &m, ctx->user_data);
}

/* Strict known-event set: an unknown event type is fail-fast (§3.4). */
static int is_known_event(uint16_t t)
{
Expand Down Expand Up @@ -651,6 +669,41 @@ static int media_text(tai_ctx_t *ctx,
return TAI_OK;
}

/* IMAGE packet payload: [data_id:2][48-bit stream_flag|ts_ms][image bytes…].
* A received image is delivered chunk-by-chunk (the caller reassembles by
* stream_flag). image-params (attr 90) on START/ONE_SHOT carries
* "payload_type format width height"; format/width/height are 0 otherwise. */
static int media_image(tai_ctx_t *ctx,
const tai_attr_t *attrs, int attr_count,
const uint8_t *payload, size_t payload_len)
{
if (payload_len < 8) {
TAI_LOGW(ctx->pal, TAG, "image media header truncated (%zu < 8)", payload_len);
return TAI_PROTO_ERR_MEDIA_HDR;
}
uint16_t data_id = 0; uint8_t stream_flag = 0; uint64_t ts_ms = 0;
tai_unpack_media_hdr(payload, 8, &data_id, &stream_flag, &ts_ms);

uint8_t format = 0; uint16_t width = 0, height = 0;
const tai_attr_t *ip = tai_attr_find(attrs, attr_count, TAI_ATTR_IMAGE_PARAMS);
if (ip && ip->len > 0) {
char tmp[64];
size_t cplen = ip->len < sizeof(tmp) - 1 ? ip->len : sizeof(tmp) - 1;
memcpy(tmp, ip->value, cplen);
tmp[cplen] = '\0';
unsigned f[4] = {0};
sscanf(tmp, "%u %u %u %u", &f[0], &f[1], &f[2], &f[3]);
format = (uint8_t)f[1]; /* f[0]=payload_type, f[1]=format, f[2]=w, f[3]=h */
width = (uint16_t)f[2];
height = (uint16_t)f[3];
}

latch_event_id(ctx, attrs, attr_count);
emit_image(ctx, payload + 8, payload_len - 8, format, width, height,
stream_flag, data_id, ts_ms);
return TAI_OK;
}

int tai_proto_dispatch(tai_ctx_t *ctx,
uint8_t pkt_type,
const tai_attr_t *attrs, int attr_count,
Expand All @@ -671,6 +724,9 @@ int tai_proto_dispatch(tai_ctx_t *ctx,
case TAI_PKT_TEXT:
return media_text(ctx, attrs, attr_count, payload, payload_len);

case TAI_PKT_IMAGE:
return media_image(ctx, attrs, attr_count, payload, payload_len);

case TAI_PKT_EVENT: {
uint16_t evt_type;
const uint8_t *evt_data;
Expand Down
94 changes: 92 additions & 2 deletions modules/rtc-tcp-client/test/test_integration.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,12 @@ typedef struct {
uint8_t audio_concat[8192]; /* concatenation of all delivered frames */
size_t audio_concat_len;

int image_calls;
size_t image_bytes;
uint8_t image_last_flag;
uint8_t image_format;
uint16_t image_w, image_h;

int event_calls;
uint16_t event_types[16];
int event_count;
Expand Down Expand Up @@ -108,6 +114,12 @@ static void st_reset(void)
g_st.audio_sample_rate = 0;
g_st.audio_frame_duration = 0;
g_st.audio_concat_len = 0;
g_st.image_calls = 0;
g_st.image_bytes = 0;
g_st.image_last_flag = 0xFF;
g_st.image_format = 0;
g_st.image_w = 0;
g_st.image_h = 0;
g_st.event_calls = 0;
g_st.event_count = 0;
g_st.event_payload_len = 0;
Expand Down Expand Up @@ -157,6 +169,19 @@ static void on_audio(tai_ctx_t *ctx, const tai_audio_msg_t *msg, void *ud)
pthread_mutex_unlock(&g_st.mtx);
}

static void on_image(tai_ctx_t *ctx, const tai_image_msg_t *msg, void *ud)
{
(void)ctx; (void)ud;
pthread_mutex_lock(&g_st.mtx);
g_st.image_calls++;
g_st.image_bytes += msg->len;
g_st.image_last_flag = msg->stream_flag;
if (msg->format) g_st.image_format = msg->format;
if (msg->width) g_st.image_w = msg->width;
if (msg->height) g_st.image_h = msg->height;
pthread_mutex_unlock(&g_st.mtx);
}

static void on_event(tai_ctx_t *ctx, const tai_event_msg_t *msg, void *ud)
{
(void)ctx; (void)ud;
Expand Down Expand Up @@ -377,6 +402,31 @@ static size_t server_send_audio(tai_ctx_t *ctx,
payload, (size_t)hlen + alen, frame_seq);
}

/* Send one IMAGE chunk. image_params_str (e.g. "0 1 320 480" = raw/JPEG/w/h)
* rides on START/ONE_SHOT only; pass NULL for MIDDLE/END. */
static size_t server_send_image(tai_ctx_t *ctx,
uint8_t stream_flag,
const char *image_params_str,
const uint8_t *img, size_t ilen,
uint16_t frame_seq)
{
uint8_t payload[4096];
int hlen = tai_pack_media_hdr(TAI_VER_21, TAI_DATA_ID_IMAGE_UP,
stream_flag, ctx->pal->time_ms(),
payload, sizeof(payload));
if (hlen <= 0) return 0;
if ((size_t)hlen + ilen > sizeof(payload)) return 0;
if (ilen) memcpy(payload + hlen, img, ilen);

tai_attr_t attrs[1];
int na = 0;
if (image_params_str)
attrs[na++] = tai_attr_strv(TAI_ATTR_IMAGE_PARAMS, image_params_str);

return server_send(ctx, TAI_PKT_IMAGE, na ? attrs : NULL, na,
payload, (size_t)hlen + ilen, frame_seq);
}

/* =========================================================================
* Helpers: transport fragmentation
*
Expand Down Expand Up @@ -481,6 +531,7 @@ static tai_ctx_t *setup_ctx(void *mem)
cfg.pal = tai_pal_loopback();
cfg.on_text = on_text;
cfg.on_audio = on_audio;
cfg.on_image = on_image;
cfg.on_event = on_event;
cfg.on_disconnect = on_disconnect;

Expand Down Expand Up @@ -1016,21 +1067,59 @@ static void test_fail_fast(void)
* by delivering a valid TEXT right after the unknown packet and seeing it
* arrive with no disconnect.
* ========================================================================= */
/* =========================================================================
* Test: received image (on_image, downlink). The server streams an image as
* START (carrying image-params) + END; the client must deliver each chunk to
* on_image with the parsed format/size and the raw bytes, with no disconnect.
* ========================================================================= */
static void test_image_recv(void)
{
SECTION("image_recv");
static uint8_t ctx_mem[sizeof(struct tai_ctx)];
uint8_t tx[2048];

tai_ctx_t *ctx = setup_ctx(ctx_mem);
CHECK(ctx != NULL);
CHECK_EQ_INT(tai_connect(ctx), TAI_OK);
(void)tai_loopback_pop_sent(tx, sizeof(tx));

/* on_image delivers raw encoded bytes verbatim (no decode); a byte pattern
* stands in for the JPEG body. */
uint8_t body[64];
for (int i = 0; i < (int)sizeof(body); i++) body[i] = (uint8_t)i;

/* "0 1 320 480" = payload_type RAW, format JPEG, 320x480. */
server_send_image(ctx, TAI_STREAM_START, "0 1 320 480", body, sizeof(body), 1);
server_send_image(ctx, TAI_STREAM_END, NULL, NULL, 0, 2);

CHECK(WAIT_FOR(g_st.image_calls >= 2, 1000));
CHECK_EQ_INT((int)g_st.image_bytes, (int)sizeof(body)); /* START body + END(0) */
CHECK_EQ_INT(g_st.image_format, TAI_IMG_JPEG);
CHECK_EQ_INT(g_st.image_w, 320);
CHECK_EQ_INT(g_st.image_h, 480);
CHECK_EQ_INT(g_st.image_last_flag, TAI_STREAM_END);
CHECK_EQ_INT(g_st.disconnect_calls, 0);

tai_disconnect(ctx); tai_ctx_deinit(ctx);
}

static void test_unknown_tolerated(void)
{
SECTION("unknown_tolerated");
static uint8_t ctx_mem[sizeof(struct tai_ctx)];
uint8_t tx[2048];

/* --- unknown packet type: an inbound IMAGE packet the device never expects. --- */
/* --- unknown packet type: a reserved type the device does not implement.
* (IMAGE, once an example here, is now a handled type — see on_image.) --- */
{
tai_ctx_t *ctx = setup_ctx(ctx_mem);
CHECK(ctx != NULL);
CHECK_EQ_INT(tai_connect(ctx), TAI_OK);
(void)tai_loopback_pop_sent(tx, sizeof(tx));

uint8_t pl[4] = {0};
server_send(ctx, TAI_PKT_IMAGE, NULL, 0, pl, sizeof(pl), 1);
const uint8_t TAI_PKT_RESERVED_UNKNOWN = 99; /* not enumerated / not dispatched */
server_send(ctx, TAI_PKT_RESERVED_UNKNOWN, NULL, 0, pl, sizeof(pl), 1);
/* A valid TEXT after the unknown packet: if the unknown frame was
* skipped cleanly (stream in sync) this arrives; if it desynced or tore
* down, text never comes. */
Expand Down Expand Up @@ -1823,6 +1912,7 @@ int main(void)
test_text_query();
test_audio_roundtrip();
test_image_query();
test_image_recv();
test_disconnect_eof();
test_session_close_then_transport();
test_liveness_during_stream();
Expand Down
Loading