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()
|