AzureKinectViewer无法进行数据保存而且现在的开源代码基本已C++为主，同时缺少可视化页面的自动保存，本程序使用Python插件pyk4v进行相机控制，前端通过React+Ant Design进行页面控制，支持RGB和深度图的实时预览。以及提供了脚本将ply、pcd、npy文件转为JSON格式。 代码开源至Github:https://github.com/Lrunlin/py_azure_kinect_viewer

预览

前端页面讲解

服务器讲解

使用main.py启动服务器，采集数据通过fastapi接收，保存至 /data文件夹下，根据 日期/时间戳/文件 的规则路径进行保存

使用

本系统在 /script提供了npy、ply、pcd转为JSON格式的脚本，自动将指定文件夹下的全部xx文件转为json文件。 同时 merge.py文件可以将三种类型的文件结合除XYZ、RGB外可以生成faces以及nx、ny、nz，该文件功能就是将单次采集的全部数据进行计算和汇总。 preview_npy.py和 preview_npy.bat 文件为npy文件的预览程序将.npy文件指定为该bat打开可以直接查看npy文化。

启动

打开项目,启动前端，服务器直接运行main.py文件即可

cd admin
npm i
npm run dev

静态文件服务器使用Node.js(Python的性能懂的都懂)，因为该程序除了前端8001端口外，还要保证3000(服务器端口)和3001(文件存储端口)正常

实现原理

帧保存：因为每次采集的时候技能前摇时间较长，因为修改为实时采集将上一帧缓存，在fastapi接收到请求后直接使用缓存的最新帧来保存数据。

def background_capture():
    global latest_frame
    while True:
        try:
            frame = k4a.get_capture()
            if frame.color is not None and frame.depth is not None:
                with frame_lock:
                    latest_frame = frame
        except Exception as e:
            debug_log(f"后台采集失败: {e}")
        time.sleep(0.01)

点云预览使用websocket实现：

@app.websocket("/ws/pointcloud")
async def websocket_pointcloud(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            await asyncio.sleep(0.05)
            with frame_lock:
                frame = latest_frame
            if frame is None or frame.color is None or frame.depth is None:
                continue
            fx, fy = 600.0, 600.0
            cx, cy = frame.depth.shape[1] / 2.0, frame.depth.shape[0] / 2.0
            points, _ = generate_point_cloud(  # 只要 points
                frame.depth, fx, fy, cx, cy, frame.color)
            pc = points.astype(np.float32)  # (N,3)
            await websocket.send_bytes(pc.tobytes())
    except Exception as e:
        # 断开链接
        pass
    finally:
        await websocket.close()

RGB预览:因为 multipart/x-mixed-replace; boundary=frame无法主动断开所以前端每次建立都携带时间戳作为key，服务器内创建字典进行缓存，在前端每次进行切换是否预览，时会将发送/close_stream请求主动端口。

# 全局流断开控制(判断实时相机是否断开)
stream_stop_flags = {}  # {stream_id: bool}