实验需求
在偶尔的实验中，会有如下需求：
- 主工程环境比较老；
- 有一套新的库/代码/工具环境比较新 要想在旧的工程继续扩展，调用新的工具，这时候依赖库会面临不可调和的问题。
理论上可以选择更新一下旧的库，但是如果旧的库包含很多历史工程，那更新环境就不太明智了。
解决方案
用 server-client 调用的方法，即把需要跑的攻击，写成一个 server，可以处理请求，内部处理就是根据输入的数据返回输出结果；再写一个 client 来反复和 server 通信调用，这样就可以在不动环境的 前提下，使得 A 代码可以调用 B 代码的结果。
server.py
from multiprocessing.connection import Listener
import argparse
import numpy as np
import cv2
from mesh_engine import MeshEngine
import pickle
import numpy as np
def convert_for_old_numpy(obj):
"""将对象转换为旧版numpy能识别的格式"""
if isinstance(obj, np.ndarray):
# 方法1：转换为列表（最简单）
return obj.tolist()
elif isinstance(obj, np.generic): # numpy标量
return obj.item()
elif isinstance(obj, list):
# 递归处理列表中的元素
return [convert_for_old_numpy(x) for x in obj]
elif isinstance(obj, dict):
return {k: convert_for_old_numpy(v) for k, v in obj.items()}
else:
return obj
if __name__ == "__main__":
args = parse_args()
print("[MeshServer] Initializing engine...")
engine = MeshEngine(args) # 主体的推理部分
print("[MeshServer] Ready.")
listener = Listener(('localhost', 6000), authkey=b'mesh-secret')
while True:
conn = listener.accept()
print("[MeshServer] Client connected")
try:
while True:
msg = conn.recv()
if msg == "close":
break
# msg: dict
volumes, depth, center_idx = engine.infer_single_image(image)
conn.send(convert_for_old_numpy({
"volumes": volumes,
"depth": depth,
"center_idx": center_idx,
"num_person": len(volumes)
}))
except EOFError:
print("[MeshServer] Client disconnected")
conn.close()
mesh_engine.py
import os, sys
import numpy as np
project_root = ""
sys.path.append(project_root)
import torch
from sam_3d_body import load_sam_3d_body, SAM3DBodyEstimator
class MeshEngine:
def __init__(self, args):
device = torch.device("cuda") if torch.cuda.is_available() else torch.device("cpu")
model, model_cfg = load_sam_3d_body(
args.checkpoint_path,
device=device,
mhr_path=args.mhr_path
)
self.estimator = SAM3DBodyEstimator(
sam_3d_body_model=model,
model_cfg=model_cfg,
human_detector=human_detector,
human_segmentor=human_segmentor,
fov_estimator=fov_estimator,
)
self.img_h = 256
self.img_w = 192
@torch.no_grad()
def infer_single_image(self, image: np.ndarray):
"""
image: np.ndarray (H, W, 3), uint8
return: List[np.ndarray]
"""
outputs = self.estimator.process_one_image(
image,
bbox_thr=0.8,
use_mask=False,
)
volumes = []
depth = []
bbox = []
for o in outputs:
volume, d = vertices_to_volume(
vertices=o["pred_vertices"],
cam_t=o["pred_cam_t"],
focal_length=o["focal_length"],
img_w=self.img_w,
img_h=self.img_h
)
bbox.append(o['bbox'])
binary_volume = (volume > 0).astype(np.uint8)
volumes.append(binary_volume)
depth.append(d)
center_idx = select_center_bbox(bbox, self.img_w, self.img_h)
return volumes, depth, center_idx以上两个文件，server 可以经过一次初始化，就能反复调用。
最后就是具体第三方调用的文件：client.py
from multiprocessing.connection import Client
class MeshClient:
def __init__(self):
self.conn = Client(('localhost', 6000), authkey=b'mesh-secret')
def infer(self, image_np):
self.conn.send({"image": image_np})
return self.conn.recv()
这样，在需要使用新工具的地方，直接初始化 MeshClient() 类即可，就会在约定的 6000 端口和服务端通信。