LingBot-Depth-ViT-L14实战教程：Python调用REST API实现自动化深度分析流水线

LingBot-Depth-ViT-L14实战教程：Python调用REST API实现自动化深度分析流水线

1. 引言

你有没有遇到过这样的场景？手头有一堆图片或视频，想快速知道里面物体的远近、场景的深度，或者想把手机拍的照片变成3D模型？过去，这需要昂贵的激光雷达设备或者复杂的立体视觉算法。现在，有了LingBot-Depth-ViT-L14模型，事情变得简单多了。

LingBot-Depth-ViT-L14是一个基于DINOv2视觉大模型的深度估计工具。简单来说，它能“看懂”图片，然后告诉你图片里每个像素点离摄像头有多远。更厉害的是，它还能把不完整的深度信息（比如激光雷达扫描的稀疏点）补全成完整的深度图。

本文不是简单的功能介绍，而是一个实战指南。我将带你一步步搭建一个完整的自动化深度分析流水线，让你能用Python脚本批量处理图片、调用模型、分析结果，并把整个过程自动化。无论你是做机器人导航、3D重建，还是AR/VR应用，这套方法都能直接拿来用。

2. 环境准备与快速部署

2.1 镜像部署

首先，我们需要把模型跑起来。这里用的是已经打包好的镜像，省去了安装各种依赖的麻烦。

部署成功后，你会看到一个实例列表，里面有刚创建的实例。记下它的IP地址，后面调用API要用到。

2.2 服务验证

部署完成后，模型提供了两个访问方式：

• Web界面：访问 http://<你的实例IP>:7860，这是一个可视化操作界面，适合手动测试和演示
• REST API：访问 http://<你的实例IP>:8000/docs，这是程序调用的接口，我们主要用这个

先验证一下服务是否正常。打开浏览器，访问Web界面，上传一张图片测试。系统自带了测试图片，路径是 /root/assets/lingbot-depth-main/examples/0/rgb.png。

选择“单目深度估计”模式，点击生成按钮。如果一切正常，2-3秒后你会看到右侧出现一张彩色热力图——这就是深度图。红色表示近处，蓝色表示远处。

3. 理解REST API接口

3.1 API端点详解

模型的核心功能通过一个简单的API提供。打开 http://<实例IP>:8000/docs，你会看到Swagger文档界面。主要接口是 /predict，支持两种调用方式：

POST /predict – 这是主要的预测接口，支持两种模式：

• 单目深度估计：只传RGB图片，模型自己估算深度
• 深度补全：传RGB图片+稀疏深度图，模型融合信息生成完整深度

接口返回的是JSON格式的数据，包含：

• depth_image_base64：深度图的base64编码（PNG格式，伪彩色显示）
• depth_data_base64：原始深度数据的base64编码（numpy数组，单位是米）
• depth_range：深度范围，比如“0.523m ~ 8.145m”
• status：处理状态，成功就是“success”

3.2 请求参数说明

调用API时，需要准备以下参数：

重要提示：

• 单目模式不需要传深度图和相机参数（传了也会被忽略）
• 深度补全模式强烈建议提供准确的相机内参，否则3D重建会有误差
• 图片格式支持常见的PNG、JPG等

4. Python调用实战：基础篇

4.1 最简单的调用示例

我们先从最简单的开始。假设你只想对单张图片做深度估计，下面是完整的Python代码：

import requests
import base64
import json
from PIL import Image
import io

# 配置API地址
API_URL = "http://你的实例IP:8000/predict"

def estimate_depth_single_image(image_path, save_depth_path=None):
"""
对单张图片进行深度估计

参数:
image_path: 输入图片路径
save_depth_path: 深度图保存路径（可选）
"""
# 1. 准备请求数据
files = {
'image': open(image_path, 'rb')
}
data = {
'mode': 'monocular' # 单目深度估计模式
}

# 2. 发送请求
print(f"正在处理图片: {image_path}")
response = requests.post(API_URL, files=files, data=data)

if response.status_code != 200:
print(f"请求失败: {response.status_code}")
print(response.text)
return None

# 3. 解析响应
result = response.json()

if result['status'] != 'success':
print(f"处理失败: {result.get('message', '未知错误')}")
return None

# 4. 保存深度图
if save_depth_path:
# 解码base64图片数据
depth_image_data = base64.b64decode(result['depth_image_base64'])
depth_image = Image.open(io.BytesIO(depth_image_data))
depth_image.save(save_depth_path)
print(f"深度图已保存到: {save_depth_path}")

# 5. 输出深度信息
print(f"深度范围: {result['depth_range']}")
print(f"输入尺寸: {result['input_size']}")
print(f"处理模式: {result['mode']}")
print(f"使用设备: {result['device']}")

return result

# 使用示例
if __name__ == "__main__":
# 替换为你的图片路径
input_image = "test_image.jpg"
output_depth = "depth_result.png"

result = estimate_depth_single_image(input_image, output_depth)

if result:
print("处理成功！")
# 你还可以获取原始深度数据用于进一步分析
depth_data = base64.b64decode(result['depth_data_base64'])
# 这里可以添加你自己的处理逻辑

这段代码做了几件事：

4.2 处理深度补全任务

如果你有RGB图片和对应的稀疏深度图（比如来自激光雷达或ToF传感器），可以用深度补全模式：

def complete_depth(rgb_path, sparse_depth_path, camera_params=None, save_path=None):
"""
深度补全：结合RGB和稀疏深度生成完整深度图

参数:
rgb_path: RGB图片路径
sparse_depth_path: 稀疏深度图路径
camera_params: 相机参数字典（可选）
save_path: 结果保存路径（可选）
"""
# 准备文件
files = {
'image': open(rgb_path, 'rb'),
'depth_image': open(sparse_depth_path, 'rb')
}

# 准备数据
data = {
'mode': 'completion'
}

# 如果有相机参数，一起传过去
if camera_params:
data.update(camera_params)

# 发送请求
response = requests.post(API_URL, files=files, data=data)

if response.status_code == 200:
result = response.json()

if save_path and result['status'] == 'success':
# 保存结果
depth_data = base64.b64decode(result['depth_image_base64'])
with open(save_path, 'wb') as f:
f.write(depth_data)
print(f"补全深度图已保存: {save_path}")

return result
else:
print(f"请求失败: {response.status_code}")
return None

# 使用示例
camera_config = {
'fx': 460.14,
'fy': 460.20,
'cx': 319.66,
'cy': 237.40
}

result = complete_depth(
rgb_path="scene_rgb.jpg",
sparse_depth_path="sparse_depth.png",
camera_params=camera_config,
save_path="completed_depth.png"
)

关键点：

• 深度补全需要同时提供RGB图和稀疏深度图
• 相机参数不是必须的，但有了会更准确
• 稀疏深度图可以是PNG格式，值代表深度（单位米或毫米）

5. 构建自动化流水线

5.1 批量处理图片

实际项目中，我们往往需要处理大量图片。下面是一个批量处理的完整示例：

import os
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
import pandas as pd

class DepthProcessingPipeline:
"""深度处理流水线"""

def __init__(self, api_url, output_dir="output"):
self.api_url = api_url
self.output_dir = output_dir
os.makedirs(output_dir, exist_ok=True)

# 创建结果记录文件
self.results_file = os.path.join(output_dir, "processing_results.csv")
self.results = []

def process_single_image(self, image_path, mode='monocular', **kwargs):
"""处理单张图片"""
try:
files = {'image': open(image_path, 'rb')}
data = {'mode': mode}

# 添加额外参数
if kwargs:
data.update(kwargs)

start_time = time.time()
response = requests.post(self.api_url, files=files, data=data)
processing_time = time.time() – start_time

if response.status_code == 200:
result = response.json()

if result['status'] == 'success':
# 生成输出文件名
basename = os.path.basename(image_path)
name_without_ext = os.path.splitext(basename)[0]
output_path = os.path.join(
self.output_dir,
f"{name_without_ext}_depth.png"
)

# 保存深度图
depth_data = base64.b64decode(result['depth_image_base64'])
with open(output_path, 'wb') as f:
f.write(depth_data)

# 记录结果
record = {
'image': basename,
'status': 'success',
'processing_time': round(processing_time, 3),
'depth_range': result['depth_range'],
'output_file': output_path,
'timestamp': time.strftime("%Y-%m-%d %H:%M:%S")
}

return record
else:
return {
'image': os.path.basename(image_path),
'status': 'failed',
'error': result.get('message', '未知错误'),
'timestamp': time.strftime("%Y-%m-%d %H:%M:%S")
}
else:
return {
'image': os.path.basename(image_path),
'status': 'failed',
'error': f"HTTP {response.status_code}",
'timestamp': time.strftime("%Y-%m-%d %H:%M:%S")
}

except Exception as e:
return {
'image': os.path.basename(image_path),
'status': 'failed',
'error': str(e),
'timestamp': time.strftime("%Y-%m-%d %H:%M:%S")
}

def batch_process(self, image_dir, max_workers=4):
"""批量处理目录中的所有图片"""
# 获取所有图片文件
image_extensions = ['.jpg', '.jpeg', '.png', '.bmp']
image_files = []

for file in os.listdir(image_dir):
if any(file.lower().endswith(ext) for ext in image_extensions):
image_files.append(os.path.join(image_dir, file))

print(f"找到 {len(image_files)} 张待处理图片")

# 使用线程池并行处理
with ThreadPoolExecutor(max_workers=max_workers) as executor:
# 提交所有任务
future_to_image = {
executor.submit(self.process_single_image, img): img
for img in image_files
}

# 收集结果
completed = 0
for future in as_completed(future_to_image):
result = future.result()
self.results.append(result)

completed += 1
if result['status'] == 'success':
print(f"[{completed}/{len(image_files)}] 成功处理: {result['image']} "
f"(耗时: {result.get('processing_time', 0):.2f}s)")
else:
print(f"[{completed}/{len(image_files)}] 处理失败: {result['image']} "
f"(错误: {result.get('error', '未知')})")

# 保存结果到CSV
self.save_results()

# 生成统计报告
self.generate_report()

def save_results(self):
"""保存处理结果到CSV"""
df = pd.DataFrame(self.results)
df.to_csv(self.results_file, index=False, encoding='utf-8')
print(f"结果已保存到: {self.results_file}")

def generate_report(self):
"""生成处理报告"""
success_count = sum(1 for r in self.results if r['status'] == 'success')
fail_count = len(self.results) – success_count

if success_count > 0:
avg_time = sum(
r.get('processing_time', 0) for r in self.results
if r['status'] == 'success'
) / success_count

print("\\n" + "="*50)
print("处理报告")
print("="*50)
print(f"总图片数: {len(self.results)}")
print(f"成功: {success_count}")
print(f"失败: {fail_count}")
print(f"成功率: {success_count/len(self.results)*100:.1f}%")
print(f"平均处理时间: {avg_time:.2f}秒")
print(f"输出目录: {self.output_dir}")
print("="*50)

# 使用示例
if __name__ == "__main__":
# 初始化流水线
pipeline = DepthProcessingPipeline(
api_url="http://你的实例IP:8000/predict",
output_dir="depth_results"
)

# 批量处理图片
pipeline.batch_process(
image_dir="input_images", # 你的图片目录
max_workers=2 # 并发数，根据你的实例配置调整
)

这个流水线提供了：

• 批量处理：自动扫描目录中的所有图片
• 并行处理：使用线程池提高效率
• 错误处理：记录失败原因，不影响其他图片
• 结果记录：保存处理日志和统计信息
• 进度显示：实时显示处理进度

5.2 与3D处理流程集成

深度图最常见的用途之一是生成3D点云。下面是如何将深度图转换为点云的示例：

import numpy as np
import open3d as o3d
from scipy import ndimage

class DepthToPointCloud:
"""深度图转点云工具"""

@staticmethod
def depth_to_pointcloud(depth_map, camera_matrix, max_points=50000):
"""
将深度图转换为点云

参数:
depth_map: 深度图数组 (H, W)，单位米
camera_matrix: 相机内参矩阵 [[fx, 0, cx], [0, fy, cy], [0, 0, 1]]
max_points: 最大点数（太多点会降低处理速度）

返回:
open3d点云对象
"""
height, width = depth_map.shape

# 创建网格坐标
u, v = np.meshgrid(np.arange(width), np.arange(height))

# 转换为相机坐标系
z = depth_map
x = (u – camera_matrix[0, 2]) * z / camera_matrix[0, 0]
y = (v – camera_matrix[1, 2]) * z / camera_matrix[1, 1]

# 过滤无效点（深度为0或NaN）
valid_mask = (z > 0) & ~np.isnan(z) & ~np.isinf(z)
x_valid = x[valid_mask]
y_valid = y[valid_mask]
z_valid = z[valid_mask]

# 如果点太多，进行下采样
if len(x_valid) > max_points:
step = len(x_valid) // max_points
indices = np.arange(0, len(x_valid), step)
x_valid = x_valid[indices]
y_valid = y_valid[indices]
z_valid = z_valid[indices]

# 创建点云
points = np.stack([x_valid, y_valid, z_valid], axis=-1)

# 计算法向量（可选）
# 这里使用简单的邻域法向量估计
if len(points) > 100:
pcd = o3d.geometry.PointCloud()
pcd.points = o3d.utility.Vector3dVector(points)

# 估计法向量
pcd.estimate_normals(
search_param=o3d.geometry.KDTreeSearchParamHybrid(
radius=0.1, max_nn=30
)
)

# 统一法向量方向（朝向相机）
pcd.orient_normals_towards_camera_location(
camera_location=np.array([0, 0, 0])
)
else:
pcd = o3d.geometry.PointCloud()
pcd.points = o3d.utility.Vector3dVector(points)

return pcd

@staticmethod
def process_depth_api_result(api_result, camera_matrix, output_ply=None):
"""
直接处理API返回的结果

参数:
api_result: API返回的JSON结果
camera_matrix: 相机内参矩阵
output_ply: PLY文件保存路径（可选）

返回:
点云对象和统计信息
"""
# 解码深度数据
depth_data_base64 = api_result['depth_data_base64']
depth_bytes = base64.b64decode(depth_data_base64)

# 注意：这里需要根据实际编码方式调整
# 假设是numpy数组的bytes表示
depth_array = np.frombuffer(depth_bytes, dtype=np.float32)

# 从metadata获取尺寸信息
# 这里需要根据API返回的实际数据结构调整
# 假设有shape信息
if 'input_size' in api_result:
# 解析 "640×480" 这样的字符串
h, w = map(int, api_result['input_size'].split('x'))
depth_array = depth_array.reshape(h, w)
else:
# 尝试自动推断
size = int(np.sqrt(len(depth_array)))
depth_array = depth_array.reshape(size, size)

# 转换为点云
pcd = DepthToPointCloud.depth_to_pointcloud(depth_array, camera_matrix)

# 保存为PLY文件
if output_ply:
o3d.io.write_point_cloud(output_ply, pcd)
print(f"点云已保存到: {output_ply}")

# 统计信息
stats = {
'num_points': len(pcd.points),
'depth_range': api_result['depth_range'],
'bounds': pcd.get_axis_aligned_bounding_box()
}

return pcd, stats

# 使用示例
def create_3d_from_depth():
"""从深度图创建3D模型"""

# 假设我们已经通过API获取了深度结果
api_result = estimate_depth_single_image("input.jpg")

if api_result:
# 定义相机参数（需要根据你的相机调整）
camera_matrix = np.array([
[460.14, 0, 319.66],
[0, 460.20, 237.40],
[0, 0, 1]
])

# 转换为点云
pcd, stats = DepthToPointCloud.process_depth_api_result(
api_result=api_result,
camera_matrix=camera_matrix,
output_ply="output.ply"
)

print(f"生成点云点数: {stats['num_points']}")
print(f"深度范围: {stats['depth_range']}")
print(f"包围盒: {stats['bounds']}")

# 可视化（需要GUI环境）
# o3d.visualization.draw_geometries([pcd])

return pcd

# 如果没有open3d，可以使用简单的numpy版本
def simple_depth_to_points(depth_map, camera_matrix):
"""简化版的深度转点云，只返回点坐标"""
h, w = depth_map.shape
fx, fy = camera_matrix[0, 0], camera_matrix[1, 1]
cx, cy = camera_matrix[0, 2], camera_matrix[1, 2]

# 生成网格
u, v = np.meshgrid(np.arange(w), np.arange(h))

# 转换
z = depth_map
x = (u – cx) * z / fx
y = (v – cy) * z / fy

# 展平并过滤
points = np.stack([x.flatten(), y.flatten(), z.flatten()], axis=1)
valid = points[:, 2] > 0
points = points[valid]

return points

6. 高级应用与优化技巧

6.1 实时视频流处理

对于需要实时处理的应用，比如机器人导航或AR应用，我们可以优化处理流程：

import cv2
import threading
import queue
import time

class RealTimeDepthProcessor:
"""实时深度处理器"""

def __init__(self, api_url, camera_id=0, process_interval=0.1):
self.api_url = api_url
self.camera_id = camera_id
self.process_interval = process_interval # 处理间隔（秒）

self.cap = None
self.processing_queue = queue.Queue(maxsize=2) # 限制队列大小
self.result_queue = queue.Queue(maxsize=2)
self.running = False

# 统计信息
self.frame_count = 0
self.process_count = 0
self.avg_latency = 0

def start(self):
"""启动处理流水线"""
self.running = True

# 打开摄像头
self.cap = cv2.VideoCapture(self.camera_id)
if not self.cap.isOpened():
print("无法打开摄像头")
return False

print(f"摄像头已打开，分辨率: {self.cap.get(cv2.CAP_PROP_FRAME_WIDTH)}x"
f"{self.cap.get(cv2.CAP_PROP_FRAME_HEIGHT)}")

# 启动处理线程
self.capture_thread = threading.Thread(target=self._capture_frames)
self.process_thread = threading.Thread(target=self._process_frames)
self.display_thread = threading.Thread(target=self._display_results)

self.capture_thread.start()
self.process_thread.start()
self.display_thread.start()

return True

def _capture_frames(self):
"""捕获帧线程"""
last_process_time = 0

while self.running:
ret, frame = self.cap.read()
if not ret:
print("无法读取帧")
time.sleep(0.1)
continue

self.frame_count += 1

# 控制处理频率
current_time = time.time()
if current_time – last_process_time >= self.process_interval:
# 缩小尺寸以提高处理速度
small_frame = cv2.resize(frame, (320, 240))

# 放入处理队列（非阻塞）
try:
self.processing_queue.put(small_frame, block=False)
last_process_time = current_time
except queue.Full:
pass # 队列已满，跳过这一帧

def _process_frames(self):
"""处理帧线程"""
while self.running:
try:
# 从队列获取帧（带超时）
frame = self.processing_queue.get(timeout=0.5)

# 转换为RGB（OpenCV是BGR）
rgb_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)

# 编码为JPEG
_, img_encoded = cv2.imencode('.jpg', rgb_frame)
img_bytes = img_encoded.tobytes()

# 准备API请求
files = {'image': ('frame.jpg', img_bytes, 'image/jpeg')}
data = {'mode': 'monocular'}

# 发送请求
start_time = time.time()
try:
response = requests.post(
self.api_url,
files=files,
data=data,
timeout=2.0 # 设置超时
)

if response.status_code == 200:
result = response.json()
latency = time.time() – start_time

if result['status'] == 'success':
# 解码深度图
depth_data = base64.b64decode(result['depth_image_base64'])
depth_array = np.frombuffer(depth_data, np.uint8)
depth_image = cv2.imdecode(depth_array, cv2.IMREAD_COLOR)

# 放入结果队列
self.result_queue.put({
'frame': frame,
'depth': depth_image,
'latency': latency,
'depth_range': result.get('depth_range', 'N/A')
})

self.process_count += 1
self.avg_latency = (self.avg_latency * (self.process_count – 1) + latency) / self.process_count

except requests.exceptions.Timeout:
print("请求超时，跳过这一帧")
except Exception as e:
print(f"处理错误: {e}")

except queue.Empty:
continue # 队列为空，继续等待

def _display_results(self):
"""显示结果线程"""
cv2.namedWindow('Real-time Depth', cv2.WINDOW_NORMAL)

while self.running:
try:
result = self.result_queue.get(timeout=0.5)

frame = result['frame']
depth = result['depth']
latency = result['latency']

# 调整深度图尺寸以匹配原图
depth_resized = cv2.resize(depth, (frame.shape[1], frame.shape[0]))

# 并排显示
combined = np.hstack([frame, depth_resized])

# 添加信息叠加
info_text = f"Latency: {latency*1000:.0f}ms | FPS: {1/self.process_interval:.1f}"
cv2.putText(combined, info_text, (10, 30),
cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 255, 0), 2)

cv2.imshow('Real-time Depth', combined)

# 按'q'退出
if cv2.waitKey(1) & 0xFF == ord('q'):
self.stop()
break

except queue.Empty:
continue

def stop(self):
"""停止处理"""
self.running = False

if self.cap:
self.cap.release()

cv2.destroyAllWindows()

# 等待线程结束
if hasattr(self, 'capture_thread'):
self.capture_thread.join(timeout=1.0)
if hasattr(self, 'process_thread'):
self.process_thread.join(timeout=1.0)
if hasattr(self, 'display_thread'):
self.display_thread.join(timeout=1.0)

print(f"\\n处理统计:")
print(f"总帧数: {self.frame_count}")
print(f"处理帧数: {self.process_count}")
print(f"平均延迟: {self.avg_latency*1000:.1f}ms")

# 使用示例
if __name__ == "__main__":
processor = RealTimeDepthProcessor(
api_url="http://你的实例IP:8000/predict",
camera_id=0, # 0通常是默认摄像头
process_interval=0.3 # 每0.3秒处理一帧
)

if processor.start():
try:
# 主线程等待
while processor.running:
time.sleep(0.1)
except KeyboardInterrupt:
processor.stop()

这个实时处理器提供了：

• 多线程架构：捕获、处理、显示分离，提高效率
• 队列缓冲：防止帧丢失
• 自适应频率：控制处理速度，避免过载
• 实时显示：并排显示原图和深度图
• 性能统计：显示处理延迟和帧率

6.2 性能优化建议

在实际部署中，你可能需要进一步优化性能：

def optimize_image_for_depth(image_path, target_size=(448, 448)):
"""优化图片尺寸以提高处理速度"""
img = cv2.imread(image_path)

# 保持长宽比调整尺寸（建议使用14的倍数）
h, w = img.shape[:2]

# 计算调整后的尺寸
scale = min(target_size[0]/h, target_size[1]/w)
new_h, new_w = int(h * scale), int(w * scale)

# 确保是14的倍数（模型要求）
new_h = (new_h // 14) * 14
new_w = (new_w // 14) * 14

resized = cv2.resize(img, (new_w, new_h))
return resized

import asyncio
import aiohttp

async def batch_process_async(image_paths, api_url, batch_size=4):
"""异步批量处理"""
async with aiohttp.ClientSession() as session:
tasks = []

for i, image_path in enumerate(image_paths):
if i % batch_size == 0 and tasks:
# 处理一批任务
await asyncio.gather(*tasks)
tasks = []
print(f"已处理 {i+1}/{len(image_paths)} 张图片")

task = process_single_async(session, image_path, api_url)
tasks.append(task)

# 处理剩余任务
if tasks:
await asyncio.gather(*tasks)

print("所有图片处理完成")

async def process_single_async(session, image_path, api_url):
"""异步处理单张图片"""
with open(image_path, 'rb') as f:
files = {'image': f}
data = {'mode': 'monocular'}

async with session.post(api_url, data=data, files=files) as response:
if response.status == 200:
result = await response.json()
return result
else:
print(f"处理失败: {image_path}")
return None

from functools import lru_cache
import hashlib

class CachedDepthProcessor:
"""带缓存的深度处理器"""

def __init__(self, api_url):
self.api_url = api_url
self.cache = {}

def get_image_hash(self, image_path):
"""计算图片哈希值作为缓存键"""
with open(image_path, 'rb') as f:
return hashlib.md5(f.read()).hexdigest()

@lru_cache(maxsize=100)
def process_with_cache(self, image_hash, image_path):
"""带缓存的处理"""
# 检查缓存
if image_hash in self.cache:
print(f"缓存命中: {image_path}")
return self.cache[image_hash]

# 实际处理
print(f"处理新图片: {image_path}")
result = estimate_depth_single_image(image_path, api_url=self.api_url)

if result:
self.cache[image_hash] = result

return result

def process_image(self, image_path):
"""处理图片（自动使用缓存）"""
image_hash = self.get_image_hash(image_path)
return self.process_with_cache(image_hash, image_path)

7. 实际应用案例

7.1 机器人导航避障

假设我们有一个移动机器人，需要实时避障。我们可以用深度图来检测障碍物：

class ObstacleDetector:
"""基于深度图的障碍物检测器"""

def __init__(self, safe_distance=1.0, danger_distance=0.5):
"""
参数:
safe_distance: 安全距离（米）
danger_distance: 危险距离（米）
"""
self.safe_distance = safe_distance
self.danger_distance = danger_distance

def detect_obstacles(self, depth_map, camera_matrix, robot_height=0.3):
"""
检测障碍物

参数:
depth_map: 深度图数组
camera_matrix: 相机内参
robot_height: 机器人高度（米），用于过滤地面

返回:
障碍物位置列表和警告级别
"""
h, w = depth_map.shape
fx, fy = camera_matrix[0, 0], camera_matrix[1, 1]
cx, cy = camera_matrix[0, 2], camera_matrix[1, 2]

# 转换为3D点
u, v = np.meshgrid(np.arange(w), np.arange(h))
z = depth_map
x = (u – cx) * z / fx
y = (v – cy) * z / fy

# 过滤无效点和地面
valid_mask = (z > 0) & (z < 10) # 只考虑10米内的点
ground_mask = np.abs(y + robot_height) < 0.1 # 假设地面在相机下方robot_height处

obstacle_mask = valid_mask & ~ground_mask

# 分析障碍物
obstacles = []

# 按距离分区
close_mask = obstacle_mask & (z < self.danger_distance)
warning_mask = obstacle_mask & (z >= self.danger_distance) & (z < self.safe_distance)
safe_mask = obstacle_mask & (z >= self.safe_distance)

# 检测最近的障碍物
if np.any(close_mask):
min_depth = np.min(z[close_mask])
min_idx = np.argmin(z[close_mask])
min_y, min_x = np.unravel_index(min_idx, z.shape)

obstacles.append({
'type': 'danger',
'distance': float(min_depth),
'position': (float(x[min_y, min_x]), float(y[min_y, min_x]), float(z[min_y, min_x])),
'pixel_position': (int(min_x), int(min_y)),
'message': f"危险！前方{min_depth:.2f}米有障碍物"
})

# 统计各区域障碍物
stats = {
'danger_count': np.sum(close_mask),
'warning_count': np.sum(warning_mask),
'safe_count': np.sum(safe_mask),
'closest_obstacle': obstacles[0] if obstacles else None,
'depth_map_shape': depth_map.shape
}

return obstacles, stats

def visualize_detection(self, rgb_image, depth_map, obstacles, stats):
"""可视化检测结果"""
# 创建可视化图像
vis_image = rgb_image.copy()

# 在深度图上标记障碍物
depth_colored = cv2.applyColorMap(
(depth_map * 255 / depth_map.max()).astype(np.uint8),
cv2.COLORMAP_JET
)

# 标记危险区域
danger_mask = depth_map < self.danger_distance
warning_mask = (depth_map >= self.danger_distance) & (depth_map < self.safe_distance)

# 在原图上标记
vis_image[danger_mask] = [0, 0, 255] # 红色标记危险
vis_image[warning_mask] = [0, 165, 255] # 橙色标记警告

# 标记最近的障碍物
if obstacles and obstacles[0]['type'] == 'danger':
closest = obstacles[0]
px, py = closest['pixel_position']
cv2.circle(vis_image, (px, py), 10, (0, 0, 255), 2)
cv2.putText(vis_image, f"{closest['distance']:.2f}m",
(px+15, py), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 0, 255), 2)

# 添加统计信息
cv2.putText(vis_image, f"危险区域: {stats['danger_count']}",
(10, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 0, 255), 2)
cv2.putText(vis_image, f"警告区域: {stats['warning_count']}",
(10, 60), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 165, 255), 2)
cv2.putText(vis_image, f"安全区域: {stats['safe_count']}",
(10, 90), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 255, 0), 2)

return vis_image, depth_colored

# 使用示例
def robot_navigation_demo():
"""机器人导航演示"""
# 初始化检测器
detector = ObstacleDetector(safe_distance=2.0, danger_distance=1.0)

# 相机参数（需要根据实际相机校准）
camera_matrix = np.array([
[320, 0, 320],
[0, 320, 240],
[0, 0, 1]
])

# 模拟处理流程
while True:
# 1. 获取当前帧（这里用静态图片模拟）
rgb_image = cv2.imread("current_scene.jpg")

# 2. 获取深度图（通过API）
depth_result = estimate_depth_single_image("current_scene.jpg")

if depth_result:
# 3. 解码深度数据
depth_data = base64.b64decode(depth_result['depth_data_base64'])
depth_array = np.frombuffer(depth_data, dtype=np.float32)

# 假设是224×224的深度图
depth_map = depth_array.reshape(224, 224)

# 4. 检测障碍物
obstacles, stats = detector.detect_obstacles(
depth_map, camera_matrix, robot_height=0.3
)

# 5. 可视化
vis_image, depth_colored = detector.visualize_detection(
rgb_image, depth_map, obstacles, stats
)

# 6. 显示结果
combined = np.hstack([vis_image, depth_colored])
cv2.imshow('Obstacle Detection', combined)

# 7. 输出导航建议
if obstacles:
closest = obstacles[0]
print(f"导航建议: {closest['message']}")

# 根据障碍物位置决定行动
x_pos = closest['position'][0]
if abs(x_pos) < 0.2: # 障碍物在正前方
print("建议: 停止或后退")
elif x_pos > 0: # 障碍物在右侧
print("建议: 向左转")
else: # 障碍物在左侧
print("建议: 向右转")
else:
print("路径清晰，可以前进")

# 按'q'退出
if cv2.waitKey(1) & 0xFF == ord('q'):
break

cv2.destroyAllWindows()

7.2 3D场景重建

另一个常见应用是从视频序列重建3D场景：

class SceneReconstructor:
"""3D场景重建器"""

def __init__(self, api_url, output_dir="reconstruction"):
self.api_url = api_url
self.output_dir = output_dir
os.makedirs(output_dir, exist_ok=True)

self.point_clouds = []
self.camera_poses = [] # 相机位姿列表
self.reference_points = []

def process_video_sequence(self, video_path, frame_interval=10):
"""
处理视频序列进行3D重建

参数:
video_path: 视频文件路径
frame_interval: 帧间隔（每隔多少帧处理一帧）
"""
cap = cv2.VideoCapture(video_path)
frame_count = 0
processed_count = 0

print(f"开始处理视频: {video_path}")
print(f"帧间隔: {frame_interval}")

while True:
ret, frame = cap.read()
if not ret:
break

frame_count += 1

# 按间隔处理帧
if frame_count % frame_interval == 0:
print(f"处理第 {frame_count} 帧…")

# 保存当前帧
frame_path = os.path.join(self.output_dir, f"frame_{frame_count:06d}.jpg")
cv2.imwrite(frame_path, frame)

# 获取深度图
depth_result = estimate_depth_single_image(frame_path, api_url=self.api_url)

if depth_result:
# 解码深度数据
depth_data = base64.b64decode(depth_result['depth_data_base64'])
depth_array = np.frombuffer(depth_data, dtype=np.float32)

# 假设是正方形深度图
size = int(np.sqrt(len(depth_array)))
depth_map = depth_array.reshape(size, size)

# 简化：使用固定相机参数
# 实际应用中应该通过SLAM或SfM估计相机位姿
camera_matrix = np.array([
[500, 0, size/2],
[0, 500, size/2],
[0, 0, 1]
])

# 生成点云
points = self.depth_to_points_simple(depth_map, camera_matrix)

# 估计相机位姿（简化版，实际需要特征匹配）
# 这里假设相机沿直线移动
camera_pose = np.eye(4)
camera_pose[0, 3] = processed_count * 0.1 # 每帧移动0.1米

# 变换点云到世界坐标系
points_world = self.transform_points(points, camera_pose)

self.point_clouds.append(points_world)
self.camera_poses.append(camera_pose)

processed_count += 1
print(f"已处理 {processed_count} 个关键帧")

cap.release()
print(f"视频处理完成，共处理 {processed_count} 个关键帧")

# 合并所有点云
if self.point_clouds:
all_points = np.vstack(self.point_clouds)
self.save_point_cloud(all_points, "full_scene.ply")

# 生成简化版点云用于可视化
self.generate_simplified_cloud(all_points)

def depth_to_points_simple(self, depth_map, camera_matrix):
"""简化版深度转点云"""
h, w = depth_map.shape
fx, fy = camera_matrix[0, 0], camera_matrix[1, 1]
cx, cy = camera_matrix[0, 2], camera_matrix[1, 2]

# 下采样以减少点数
step = 2
u, v = np.meshgrid(np.arange(0, w, step), np.arange(0, h, step))
z = depth_map[v, u]

# 过滤无效点
valid = z > 0
u_valid = u[valid]
v_valid = v[valid]
z_valid = z[valid]

# 转换为3D坐标
x = (u_valid – cx) * z_valid / fx
y = (v_valid – cy) * z_valid / fy

points = np.stack([x, y, z_valid], axis=1)
return points

def transform_points(self, points, transform):
"""变换点云坐标"""
# 添加齐次坐标
points_homo = np.hstack([points, np.ones((len(points), 1))])

# 应用变换
points_transformed = (transform @ points_homo.T).T

# 返回3D坐标
return points_transformed[:, :3]

def save_point_cloud(self, points, filename):
"""保存点云到PLY文件"""
# 创建open3d点云
pcd = o3d.geometry.PointCloud()
pcd.points = o3d.utility.Vector3dVector(points)

# 估计颜色（简化版，实际应该从RGB图像获取）
# 这里使用高度着色
colors = np.zeros_like(points)
z_min, z_max = points[:, 2].min(), points[:, 2].max()
colors[:, 0] = 1.0 # 红色
colors[:, 1] = (points[:, 2] – z_min) / (z_max – z_min) # 绿色随高度变化
colors[:, 2] = 0.0 # 蓝色

pcd.colors = o3d.utility.Vector3dVector(colors)

# 保存
output_path = os.path.join(self.output_dir, filename)
o3d.io.write_point_cloud(output_path, pcd)
print(f"点云已保存到: {output_path}")
print(f"点数: {len(points)}")

def generate_simplified_cloud(self, points, target_points=100000):
"""生成简化版点云"""
if len(points) > target_points:
# 随机下采样
indices = np.random.choice(len(points), target_points, replace=False)
simplified = points[indices]
else:
simplified = points

self.save_point_cloud(simplified, "simplified_scene.ply")

# 生成统计信息
stats = {
'total_points': len(points),
'simplified_points': len(simplified),
'bounding_box': {
'x_range': (points[:, 0].min(), points[:, 0].max()),
'y_range': (points[:, 1].min(), points[:, 1].max()),
'z_range': (points[:, 2].min(), points[:, 2].max())
}
}

print("\\n场景统计:")
print(f"总点数: {stats['total_points']}")
print(f"简化后点数: {stats['simplified_points']}")
print(f"X范围: {stats['bounding_box']['x_range'][0]:.2f} ~ {stats['bounding_box']['x_range'][1]:.2f}")
print(f"Y范围: {stats['bounding_box']['y_range'][0]:.2f} ~ {stats['bounding_box']['y_range'][1]:.2f}")
print(f"Z范围: {stats['bounding_box']['z_range'][0]:.2f} ~ {stats['bounding_box']['z_range'][1]:.2f}")

return simplified

# 使用示例
def reconstruct_from_video():
"""从视频重建3D场景"""
reconstructor = SceneReconstructor(
api_url="http://你的实例IP:8000/predict",
output_dir="scene_reconstruction"
)

# 处理视频
reconstructor.process_video_sequence(
video_path="walkthrough.mp4",
frame_interval=15 # 每15帧处理一帧
)

print("3D重建完成！")
print("查看 output_dir 目录中的PLY文件，可以用MeshLab或CloudCompare查看")

8. 总结

通过本文的实战教程，你应该已经掌握了如何使用LingBot-Depth-ViT-L14模型构建完整的深度分析流水线。我们来回顾一下关键要点：

8.1 核心收获

8.2 实用建议

在实际使用中，有几个经验值得分享：

性能优化方面：

• 对于实时应用，适当降低图片分辨率可以显著提高处理速度
• 使用异步请求可以更好地利用网络带宽
• 对于相似场景的图片，考虑使用缓存机制

精度提升方面：

• 深度补全模式需要准确的相机内参，建议先进行相机标定
• 对于室外大场景，可能需要分段处理然后拼接
• 考虑使用时间一致性约束来处理视频序列

错误处理方面：

• 总是检查API返回的状态码和错误信息
• 对于网络不稳定的环境，实现重试机制
• 记录详细的处理日志，便于调试和优化

8.3 下一步探索

掌握了基础用法后，你可以进一步探索：

深度估计技术正在快速发展，从机器人导航到AR/VR，从自动驾驶到工业检测，应用场景越来越广泛。LingBot-Depth-ViT-L14作为一个开箱即用的解决方案，为你提供了快速上手的机会。

希望这个教程能帮助你快速启动项目，把深度感知能力集成到自己的应用中。记住，最好的学习方式就是动手实践——选一个感兴趣的应用场景，用本文的代码作为起点，开始构建你自己的深度分析系统吧。

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。