wangzhibo
2026-10-08 220e4fab27b7f419b3e89e2f2d28ae789aac09d7
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
from pathlib import Path
 
ROOT = Path(__file__).resolve().parents[1]
 
 
def read_text(path: str) -> str:
    return (ROOT / path).read_text(encoding="utf-8")
 
 
def require(condition: bool, message: str) -> None:
    if not condition:
        raise SystemExit(message)
 
 
def main() -> None:
    system_h = read_text("main/boards/sening-xiaozhi-s3/system.h")
    mqtt_h = read_text("main/boards/sening-xiaozhi-s3/mqtt.h")
    mqtt_c = read_text("main/boards/sening-xiaozhi-s3/mqtt.c")
    app = read_text("main/application.cc")
 
    require(
        'BOARD_MQTT_RPT_EVENT_TOPIC_FORMAT "snlp/v1/%s/rpt/event"' in system_h,
        "system config must define the MQTT rpt/event topic format",
    )
    require(
        "void BoardMqtt_PublishSttEvent(const char *text);" in mqtt_h,
        "MQTT header must expose STT event reporting API",
    )
    require(
        "static char s_rpt_event_topic[BOARD_MQTT_TOPIC_MAX_LEN];" in mqtt_c,
        "MQTT client must keep a rpt/event topic buffer",
    )
    require(
        "BoardMqtt_FormatTopic(s_rpt_event_topic" in mqtt_c,
        "MQTT startup must format the rpt/event topic",
    )
    require(
        "void BoardMqtt_PublishSttEvent(const char *text)" in mqtt_c,
        "MQTT client must implement STT event reporting",
    )
 
    publish_body = mqtt_c.split("void BoardMqtt_PublishSttEvent(const char *text)", 1)[1].split(
        "static void", 1
    )[0]
    require('cJSON_AddStringToObject(payload, "event", "asr_result");' in publish_body, "STT event must use asr_result event type")
    require('cJSON_AddStringToObject(payload, "text", text);' in publish_body, "STT event must include recognized text")
    require("s_rpt_event_topic" in publish_body, "STT event must publish to rpt/event topic")
    require('cJSON_AddStringToObject(payload, "ts"' not in publish_body, "STT event payload must not add its own ts field")
    require('cJSON_AddNumberToObject(payload, "timestamp"' not in publish_body, "STT event payload must not add its own timestamp field")
 
    stt_body = app.split('} else if (strcmp(type->valuestring, "stt") == 0)', 1)[1].split(
        '} else if (strcmp(type->valuestring, "llm") == 0)', 1
    )[0]
    require(
        "BoardMqtt_PublishSttEvent(text->valuestring);" in stt_body,
        "application STT handler must report recognized text to business MQTT",
    )
 
 
if __name__ == "__main__":
    main()