Add heartbeat endpoint message schema

2026-07-08 18:35:44 +08:00
parent ebc7aaa11c
commit 041c830bfb
5 changed files with 234 additions and 7 deletions
@@ -88,6 +88,7 @@ void convert_a_frame(MppFrame frame)
    }
    base = (RK_U8 *)mpp_buffer_get_ptr(buffer);
    rga_client.convert(base, width, height, h_stride, v_stride, out);
+    send_inference_heartbeat_if_needed();
    // printf("convert a frame ok");
    if (pthread_mutex_trylock(&frame_mutex) == 0)
    {
@@ -179,16 +179,18 @@ h264 分段文件会写到：

 ## 上报逻辑

-程序向本地管理程序发送：
+推理结果业务包向本地管理程序发送：

 ```text
 POST http://127.0.0.1:<upload-port>/video/post
 ```

-JSON 结构保持兼容：
+JSON 结构：

 ```json
 {
+  "message_type": "inference_result",
+  "schema_version": 1,
  "serial": "camera001",
  "type": 19,
  "at": 1778826272,
@@ -202,7 +204,59 @@ JSON 结构保持兼容：
 }
 ```

-当前版本只主动上传 `number > 0` 的正例。目标未检出时不上传 `number=0`，只写日志。
+字段说明：
+
+- `message_type` 固定为 `inference_result`，用于和心跳包区分。
+- `schema_version` 当前为 `1`。
+- `serial` 来自启动参数 `-c`，在管理程序中可作为推理进程身份。
+- `type` 来自启动参数 `-t`。
+- `at` 为推理端当前时间戳。
+- `params[].class_idx` 来自模型目录下 `<model>_code.txt`。
+- `params[].name` 来自模型目录下 `<model>.txt`。
+- `params[].number` 为当前帧同一 `class_idx` 的检测数量；多个 label 映射到同一个 `class_idx` 时会合并数量。
+
+当前版本只主动上传 `number > 0` 的正例。目标未检出时不上传 `number=0`，只写日志。业务包不会承担心跳职责。
+
+## 心跳机制
+
+推理程序新增独立心跳接口：
+
+```text
+POST http://127.0.0.1:<upload-port>/heartbeat
+```
+
+只有启动参数指定了有效 `-p <upload-port>`，并且 `-c <processUUID>` 非空时，才会发送心跳。当前心跳周期为 `HEARTBEAT_SECS = 60` 秒。
+
+心跳包结构：
+
+```json
+{
+  "message_type": "heartbeat",
+  "schema_version": 1,
+  "serial": "camera001",
+  "process_uuid": "camera001",
+  "status": "running",
+  "timestamp": 1778826272
+}
+```
+
+字段说明：
+
+- `message_type` 固定为 `heartbeat`。
+- `schema_version` 当前为 `1`。
+- `serial` 来自启动参数 `-c`，必填且非空。
+- `process_uuid` 同样来自启动参数 `-c`，用于管理程序校验端口对应的推理进程。
+- `status` 当前固定为 `running`。
+- `timestamp` 为推理端当前时间戳。
+
+心跳发送点在解码帧完成 RGA 转换之后、`-P/--time-period` 时间段判断之前。因此即使当前时间段不执行识别，只要程序仍在正常读流解码，也会继续发送心跳。
+
+管理程序建议按以下规则区分：
+
+```text
+/heartbeat + message_type=heartbeat          -> 只刷新心跳，不进入业务判断
+/video/post + message_type=inference_result  -> 推理业务结果
+```

 ## G1000 截图通知客户端

@@ -0,0 +1,91 @@
+# 2026-07-06 修改记录
+
+## 推理程序心跳
+
+新增独立心跳上报：
+
+```text
+POST http://127.0.0.1:<upload-port>/heartbeat
+```
+
+心跳只在启动参数指定了有效 `-p <upload-port>`，且 `-c <processUUID>` 非空时发送。当前发送周期沿用 `HEARTBEAT_SECS = 60`。
+
+心跳包结构：
+
+```json
+{
+  "message_type": "heartbeat",
+  "schema_version": 1,
+  "serial": "<processUUID>",
+  "process_uuid": "<processUUID>",
+  "status": "running",
+  "timestamp": 1778826272
+}
+```
+
+其中 `serial` 和 `process_uuid` 都来自启动参数 `-c`。例如 `fire_check` 进程应传入 `deviceUUID_fire_check`。
+
+心跳发送点放在解码帧完成 RGA 转换之后、时间段判断之前。因此即使 `-P/--time-period` 当前不执行识别，只要推理程序仍在读流解码，也会继续刷新心跳。
+
+## 推理结果业务包
+
+业务结果仍然走原接口：
+
+```text
+POST http://127.0.0.1:<upload-port>/video/post
+```
+
+本次只新增顶层区分字段，不改变原有业务字段：
+
+```json
+{
+  "message_type": "inference_result",
+  "schema_version": 1,
+  "serial": "<processUUID>",
+  "type": 19,
+  "at": 1778826272,
+  "params": [
+    {
+      "class_idx": 1,
+      "name": "person",
+      "number": 1
+    }
+  ]
+}
+```
+
+管理程序可按以下规则区分：
+
+```text
+/heartbeat + message_type=heartbeat          -> 只刷新心跳
+/video/post + message_type=inference_result  -> 业务推理结果
+```
+
+## 未修改内容
+
+- 未修改识别阈值、workspace 过滤、类别判断、NMS、模型后处理逻辑。
+- 未修改 `/video/post` 中 `serial/type/at/params` 原有字段含义。
+- 未恢复旧的 `number=0` 心跳业务包。
+- 当前推理程序缓存文件仍是 `.h264`，没有改成 `.mp4`。
+
+## mp4-merge 代码确认
+
+已查看本地 `mp4-merge/` 源码。该项目是 MP4 box 级合并工具，依赖输入文件已有 `moov/mdat/stbl` 等 MP4 容器结构，当前不能直接合并裸 `.h264`。
+
+如果要支持 `.h264`，推荐在 `mp4-merge` 命令入口增加预处理：
+
+```text
+.h264 -> ffmpeg 无损封装为临时 .mp4 -> 复用现有 join_files 合并
+```
+
+不建议直接在现有 MP4 box 合并逻辑里解析裸 H264。
+
+## 构建验证
+
+已使用 WSL + Linaro 交叉编译通过，生成产物：
+
+```text
+C:\Users\Smos.DESKTOP-6MT98U8\Desktop\yolov5-inference\build-aarch64\yolov5
+```
+
+构建输出为 ARM aarch64 ELF。构建过程中出现 Windows/WSL 时间戳导致的 `Clock skew` warning，但目标文件已成功生成。
@@ -24,12 +24,14 @@
 #include "vertix.h"
 #include "snapshot_notify.h"
 #include "myrga.h"
+#include "utils.h"

 #ifdef mkdir
 #undef mkdir
 #endif

 void *upload_info(void *arg);
+static void detect_log_line(const char *fmt, ...);

 typedef struct _upload_target_info_t
 {
@@ -59,7 +61,9 @@ int TypeID = 0;
 int UploadPort = 0;

 static char is_uploading = 0;
+static char is_heartbeat_uploading = 0;
 static time_t last_upload = 0;
+static time_t last_heartbeat_post = 0;
 static time_t last_heartbeat_log = 0;
 static int last_object_count[OBJ_CLASS_MAX] = {0};
 static pthread_mutex_t upload_state_mutex = PTHREAD_MUTEX_INITIALIZER;
@@ -107,6 +111,80 @@ static void release_upload()
    pthread_mutex_unlock(&upload_state_mutex);
 }

+static int try_reserve_heartbeat_upload()
+{
+    int reserved = 0;
+    pthread_mutex_lock(&upload_state_mutex);
+    if (is_heartbeat_uploading == 0)
+    {
+        is_heartbeat_uploading = 1;
+        reserved = 1;
+    }
+    pthread_mutex_unlock(&upload_state_mutex);
+    return reserved;
+}
+
+static void release_heartbeat_upload()
+{
+    pthread_mutex_lock(&upload_state_mutex);
+    is_heartbeat_uploading = 0;
+    pthread_mutex_unlock(&upload_state_mutex);
+}
+
+static void *upload_heartbeat_info(void *arg)
+{
+    (void)arg;
+
+    cJSON *heartbeat = cJSON_CreateObject();
+    cJSON_AddStringToObject(heartbeat, "message_type", "heartbeat");
+    cJSON_AddNumberToObject(heartbeat, "schema_version", 1);
+    cJSON_AddStringToObject(heartbeat, "serial", CameraID);
+    cJSON_AddStringToObject(heartbeat, "process_uuid", CameraID);
+    cJSON_AddStringToObject(heartbeat, "status", "running");
+    cJSON_AddNumberToObject(heartbeat, "timestamp", current_time_stamp());
+
+    int ret = post_json("127.0.0.1", UploadPort, "localhost", "/heartbeat", heartbeat);
+    detect_log_line("heartbeat_post camera=%s port=%d result=%d",
+                    CameraID ? CameraID : "", UploadPort, ret);
+
+    cJSON_Delete(heartbeat);
+    release_heartbeat_upload();
+    return NULL;
+}
+
+void send_inference_heartbeat_if_needed()
+{
+    if (UploadPort <= 0 || CameraID == NULL || CameraID[0] == '\0')
+    {
+        return;
+    }
+
+    time_t now_timestamp;
+    time(&now_timestamp);
+    if (now_timestamp - last_heartbeat_post < HEARTBEAT_SECS)
+    {
+        return;
+    }
+
+    if (!try_reserve_heartbeat_upload())
+    {
+        return;
+    }
+
+    last_heartbeat_post = now_timestamp;
+    pthread_t uploadth;
+    if (pthread_create(&uploadth, NULL, upload_heartbeat_info, NULL) == 0)
+    {
+        pthread_detach(uploadth);
+    }
+    else
+    {
+        detect_log_line("heartbeat_thread_create_failed camera=%s port=%d",
+                        CameraID ? CameraID : "", UploadPort);
+        release_heartbeat_upload();
+    }
+}
+
 static int is_edge_touch(int left, int top, int right, int bottom)
 {
    return left <= 2 || top <= 2 || right >= DSTWidth - 3 || bottom >= DSTHeight - 3;
@@ -823,6 +901,8 @@ void *upload_info(void *arg)
        return NULL;
    }
    cJSON *uploadinfo = cJSON_CreateObject();
+    cJSON_AddStringToObject(uploadinfo, "message_type", "inference_result");
+    cJSON_AddNumberToObject(uploadinfo, "schema_version", 1);
    cJSON_AddStringToObject(uploadinfo, "serial", CameraID);
    // 内部事件编码，因为没法表示多个，所以，即将废弃
    cJSON_AddNumberToObject(uploadinfo, "type", TypeID);
@@ -69,6 +69,7 @@ rknn_t *do_init_model();
 rknn_t *reload_rknn(char *model_path);
 */
 int process_a_frame(char *buf, int sw, int sh);
+void send_inference_heartbeat_if_needed();

 /*
 int post_process_u8(uint8_t *input0, uint8_t *input1, uint8_t *input2, int model_in_h, int model_in_w,