dongchangxi 1 month ago
parent
commit
e3ae862300
  1. 354
      script/download_print_out.py
  2. 57
      script/factory_sliceing/build_exe.bat
  3. 5
      script/factory_sliceing/requirements.txt
  4. 134
      script/factory_sliceing/type_setting_order.py
  5. 133
      script/type_setting_order.py

354
script/download_print_out.py

@ -0,0 +1,354 @@
import yaml
import oss2
import os
from tqdm import tqdm
import os
from pathlib import Path
import numpy as np
import os
import argparse
import open3d as o3d
def custom_mesh_transform(vertices, transform_matrix):
"""
手动实现网格变换对每个顶点应用齐次变换矩阵
参数:
vertices: 网格顶点数组 (N, 3)
transform_matrix: 4x4 齐次变换矩阵
返回:
变换后的顶点数组 (N, 3)
"""
# 1. 顶点转齐次坐标 (N, 3) → (N, 4)
homogeneous_vertices = np.hstack((vertices, np.ones((vertices.shape[0], 1))))
# 2. 应用变换矩阵:矩阵乘法 (4x4) * (4xN) → (4xN)
transformed_homogeneous = transform_matrix @ homogeneous_vertices.T
# 3. 转回非齐次坐标 (3xN) → (N, 3)
transformed_vertices = transformed_homogeneous[:3, :].T
return transformed_vertices
class DataTransfer:
'''
数据传输类
'''
def __init__(self, local_path: str, oss_path: str, oss_client: oss2.Bucket):
'''
local_path: 本地输出路径
oss_path: oss路径
oss_client: oss客户端
'''
self.local_path = local_path
self.oss_path = oss_path.lstrip('/')
self.oss_client = oss_client
order_id: str
pid: str
model_height: str
def download_data_rename_json(self, json_model_info):
"""
OSS 下载数据到本地保持原有目录结构
"""
# 列出所有对象
objects = []
prefix = self.oss_path.lstrip('/') # 移除开头的 '/' 以匹配 OSS 格式
for obj in oss2.ObjectIterator(self.oss_client, prefix=prefix):
if obj.key != prefix: # 跳过目录本身
objects.append(obj.key)
# 下载所有文件,添加进度条
for obj_key in tqdm(objects, desc="下载进度"):
if obj_key.endswith('/'):
continue
if "printId" in obj_key:
continue
# 计算相对路径
rel_path = obj_key[len(prefix):].lstrip('/')
file_dir, file_name = os.path.split(rel_path)
file_base, file_ext = os.path.splitext(file_name)
# 根据文件后缀名进行重命名
if file_ext.lower() in ['.mtl', '.jpg', '.jpeg', '.png']:
# 对于.mtl和图片文件,在原名前加order_id
new_file_name = f"{json_model_info.order_id}_{file_name}"
# new_file_name = file_name
elif file_ext.lower() == '.obj':
# 对于.obj文件,完全重命名
new_file_name = f"{json_model_info.obj_name}"
else:
# 其他文件类型保持原名
new_file_name = file_name
print("new_file_name=", new_file_name)
# 构建新的相对路径
if file_dir: # 如果有子目录
new_rel_path = os.path.join(file_dir, new_file_name)
else:
new_rel_path = new_file_name
# 构建本地完整路径
local_path = os.path.join(self.local_path, new_rel_path)
# 创建必要的目录
os.makedirs(os.path.dirname(local_path), exist_ok=True)
# 下载文件
self.oss_client.get_object_to_file(obj_key, local_path)
if file_ext == '.obj': # 10MB以上
try:
# 使用临时文件避免内存问题 [8](@ref)
temp_path = local_path + '.tmp'
with open(local_path, 'r', encoding='utf-8') as f_in, \
open(temp_path, 'w', encoding='utf-8') as f_out:
mtllib_modified = False
for line in f_in:
if not mtllib_modified and line.strip().startswith('mtllib '):
parts = line.split(' ', 1)
if len(parts) > 1:
old_mtl_name = parts[1].strip()
new_mtl_name = f"{json_model_info.order_id}_{old_mtl_name}"
f_out.write(f"mtllib {new_mtl_name}\n")
mtllib_modified = True
continue
f_out.write(line)
os.replace(temp_path, local_path) # 原子性替换
except IOError as e:
print(f"处理大文件 {local_path} 时出错: {e}")
if os.path.exists(temp_path):
os.remove(temp_path)
# 优化后的.obj文件处理逻辑
if file_ext == '.mtl':
try:
# 使用更高效的文件读取方式 [6,8](@ref)
with open(local_path, 'r', encoding='utf-8') as f:
content = f.read()
# 使用字符串方法直接查找和替换,避免不必要的循环 [9](@ref)
lines = content.split('\n')
mtllib_modified = False
for i, line in enumerate(lines):
stripped_line = line.strip()
if not mtllib_modified and stripped_line.startswith('map_Kd '):
# 更高效的分割方式 [9](@ref)
parts = line.split(' ', 1)
if len(parts) > 1:
old_name = parts[1].strip()
new_name = f"{json_model_info.order_id}_{old_name}"
lines[i] = f"map_Kd {new_name}"
mtllib_modified = True
print(f"已更新材质库引用: {old_name} -> {new_name}")
break # 找到第一个后立即退出
# 批量写入,减少I/O操作 [6](@ref)
with open(local_path, 'w', encoding='utf-8') as f:
f.write('\n'.join(lines))
except IOError as e:
print(f"处理文件 {local_path} 时出错: {e}")
except UnicodeDecodeError as e:
print(f"文件编码错误 {local_path}: {e}")
print(f"下载文件: {obj_key} -> {local_path}")
import requests
import json
import shutil
def get_api(url):
try:
response = requests.get(url)
response.raise_for_status() # 检查请求是否成功
response = json.loads(response.text)
if response.get("code") != 1000:
raise Exception(f"Error fetching URL {url}: {response.get('message')}")
else:
return response
except requests.exceptions.RequestException as e:
raise Exception(f"Error fetching URL {url}: {e}")
from dataclasses import dataclass
@dataclass
class JSONModelInfo:
obj_name: str
order_id: str
pid: str
model_height: str
def read_pids_from_json(pid_file):
"""从文件读取所有PID"""
json_path = pid_file
"""
加载JSON文件读取所有模型信息应用变换后返回模型列表
"""
# 检查JSON文件是否存在
if not os.path.exists(json_path):
print(f"错误: JSON文件不存在 - {json_path}")
return []
# 读取JSON文件
try:
with open(json_path, 'r') as f:
data = json.load(f)
except Exception as e:
print(f"读取JSON文件失败: {e}")
return []
list_model_info = []
# 处理每个模型
for model in data.get('models', []):
obj_name = model.get('file_name', '')
parts = obj_name.split('_')
order_id = parts[0]
pid = parts[1]
model_height = parts[3]
model_info = JSONModelInfo(
obj_name=obj_name,
order_id=order_id,
pid=pid,
model_height=model_height
)
list_model_info.append(model_info)
return list_model_info, data
def download_data_by_json(model_info, workdir, oss_client ):
try:
pid = model_info.pid
model_height = model_info.model_height
# target_dir = f"{workdir}/{pid}_image"
target_dir = f"{workdir}"
url = f"https://mp.api.suwa3d.com/api/order/getOssSuffixByOrderId?order_id={model_info.order_id}"
res = requests.get(url)
data = res.json()["data"]
# print("datas=",data)
data = data.replace("/init_obj", "")
print("target_dir=", target_dir)
download_textures = DataTransfer(target_dir, f"objs/download/print/{pid}/{data}/{model_height}/", oss_client)
download_textures.download_data_rename_json(model_info)
# 下载后检查目标文件夹是否为空
if os.path.exists(target_dir) and not os.listdir(target_dir):
shutil.rmtree(target_dir)
print(f"下载后检查发现目标文件夹为空,已删除: {target_dir}")
except Exception as e:
print(f"卡通图片下载失败: {pid}, 错误: {str(e)}")
pass
def get_oss_client(cfg_path):
# with open(os.path.expanduser(cfg_path), "r") as config:
# cfg = yaml.safe_load(config)
AccessKeyId_down = 'LTAI5tSReWm8hz7dSYxxth8f'
AccessKeySecret_down = '8ywTDF9upPAtvgXtLKALY2iMYHIxdS'
Endpoint_down = 'oss-cn-shanghai.aliyuncs.com'
Bucket_down = 'suwa3d-securedata'
oss_client = oss2.Bucket(
oss2.Auth(AccessKeyId_down, AccessKeySecret_down), Endpoint_down, Bucket_down
)
return oss_client
def download_datas_by_json(pid_file, workdir, oss_config):
oss_client = get_oss_client(oss_config)
# json_path = os.path.join(workdir, "3DPrintLayout.json")
json_path = pid_file
# 读取所有PID
list_model_info, data = read_pids_from_json(json_path)
print(f"从文件读取了 {len(list_model_info)} 个PID")
# 批量下载
for model_info in list_model_info:
print(f"开始下载PID: {model_info}")
download_data_by_json(model_info, args.workdir, oss_client)
return data
def download_transform_save_by_json(pid_file, workdir, oss_config):
layout_data = download_datas_by_json(pid_file, workdir, oss_config)
original_obj_pid_dir = workdir
cache_type_setting_dir = os.path.join(workdir, "arrange")
Path(cache_type_setting_dir).mkdir(exist_ok=True)
print(f"original_obj_pid_dir={original_obj_pid_dir}, cache_type_setting_dir={cache_type_setting_dir}")
transform_save(layout_data, original_obj_pid_dir, cache_type_setting_dir)
def transform_save(layout_data, original_obj_pid_dir, cache_type_setting_dir):
meshes = []
# 小打印机380*345,需要偏移-380,-345
need_offset = True
for model in layout_data["models"]:
transform = model.get('transform', {})
homo_matrix = transform["homo_matrix"] # 获取存储的列表
reconstructed_matrix = np.array(homo_matrix, dtype=np.float64)
obj_name = model.get('file_name', '')
obj_path = os.path.join(original_obj_pid_dir, obj_name)
# 加载网格
try:
mesh = o3d.io.read_triangle_mesh(obj_path, enable_post_processing=True)
if not mesh.has_vertices():
print(f"警告: 网格无有效顶点 - {obj_path}")
continue
except Exception as e:
print(f"加载模型失败: {obj_path} - {e}")
continue
original_vertices = np.asarray(mesh.vertices)
transformed_vertices = custom_mesh_transform(original_vertices, reconstructed_matrix)
# 如果 need_offset 为 True,应用额外的偏移
if need_offset:
# 应用偏移 (-380, -345, 0)
offset = np.array([-380, -345, 0])
transformed_vertices += offset
print(f"已对模型 {obj_name} 应用偏移: {offset}")
mesh.vertices = o3d.utility.Vector3dVector(transformed_vertices)
meshes.append(mesh)
# obj_path_arrange = os.path.join(original_obj_pid_dir, "arrange")
obj_path_arrange = cache_type_setting_dir
if not os.path.exists(obj_path_arrange):
os.mkdir(obj_path_arrange)
obj_path_arrange_obj = os.path.join(obj_path_arrange, obj_name)
print("obj_path_arrange_obj", obj_path_arrange_obj)
mesh.compute_vertex_normals()
o3d.io.write_triangle_mesh(obj_path_arrange_obj, mesh,write_triangle_uvs=True)
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("--pid_file", type=str, required=True, help="批次号, 也是json文件名")
parser.add_argument("--workdir", type=str, required=True, help="本代码文件所在的目录")
parser.add_argument("--oss_config", type=str, default="")
args = parser.parse_args()
download_transform_save_by_json(args.pid_file, args.workdir, args.oss_config)

57
script/factory_sliceing/build_exe.bat

@ -0,0 +1,57 @@
@echo off
chcp 65001 >nul
echo ========================================
echo 开始打包 type_setting_order.py 为 exe
echo ========================================
echo.
REM 检查 Python 是否安装
python --version >nul 2>&1
if errorlevel 1 (
echo [错误] 未找到 Python,请先安装 Python
pause
exit /b 1
)
echo [1/4] 检查并安装依赖包...
python -m pip install --upgrade pip
python -m pip install -r requirements.txt
if errorlevel 1 (
echo [错误] 依赖包安装失败
pause
exit /b 1
)
echo.
echo [2/4] 清理之前的构建文件...
if exist build rmdir /s /q build
if exist dist rmdir /s /q dist
if exist __pycache__ rmdir /s /q __pycache__
echo.
echo [3/4] 使用 PyInstaller 打包...
python -m PyInstaller build_exe.spec
if errorlevel 1 (
echo [错误] 打包失败
pause
exit /b 1
)
echo.
echo [4/4] 打包完成!
echo.
echo ========================================
echo 打包结果:
echo ========================================
echo exe 文件位置: dist\type_setting_order.exe
echo.
echo 注意事项:
echo 1. 确保 Windows 系统上已安装 Visual C++ Redistributable
echo 2. 确保 Redis 服务可访问(172.31.1.254:6379)
echo 3. 确保网络连接正常(访问 OSS 和 API)
echo 4. 如果程序需要调用 download_print_out.py,请确保该文件在同一目录
echo.
pause

5
script/factory_sliceing/requirements.txt

@ -0,0 +1,5 @@
redis>=4.0.0
oss2>=2.17.0
requests>=2.28.0
pyinstaller>=5.0.0

134
script/factory_sliceing/type_setting_order.py

@ -0,0 +1,134 @@
import os
import redis
import oss2,time,sys
import requests
currentDir = os.path.dirname(os.path.abspath(__file__))
ENV = 'dev'
url = 'https://mp.api.suwa3d.com'
if ENV == 'dev':
url = 'http://mp.dev.api.com'
elif ENV == 'prod':
url = 'https://mp.api.suwa3d.com'
# 连接oss - 单例模式
class OSSClientSingleton:
_instance = None
_client = None
def __new__(cls):
if cls._instance is None:
cls._instance = super(OSSClientSingleton, cls).__new__(cls)
return cls._instance
def get_client(self):
if self._client is None:
AccessKeyId = 'LTAI5tSReWm8hz7dSYxxth8f'
AccessKeySecret = '8ywTDF9upPAtvgXtLKALY2iMYHIxdS'
Endpoint = 'oss-cn-shanghai.aliyuncs.com'
Bucket = 'suwa3d-securedata'
self._client = oss2.Bucket(oss2.Auth(AccessKeyId, AccessKeySecret), Endpoint, Bucket)
return self._client
def ossClient():
"""获取OSS客户端单例"""
return OSSClientSingleton().get_client()
#连接redis,单例模式
class RedisClientSingleton:
_instance = None
_client = None
def __new__(cls):
if cls._instance is None:
cls._instance = super(RedisClientSingleton, cls).__new__(cls)
return cls._instance
def get_client(self):
if self._client is None:
self._client = redis.Redis(host='172.31.1.254', password='', port=6379, db=6)
return self._client
def redisClient():
"""获取Redis客户端单例"""
return RedisClientSingleton().get_client()
# 根据批次id,进行切片中的状态变更
def requestApiToUpdateSliceStatus(versionId):
api_url = f"{url}/api/printTypeSettingOrder/updateBatchSliceing?batch_id={versionId}"
print(f'发起状态变更请求url={api_url}, versionId={versionId}')
res = requests.post(api_url)
if res.status_code != 200:
print(f'状态变更请求失败, res={res.text}')
return False
print(f'状态变更请求成功, res={res.text}')
return True
#判断是否上传了 JSON 文件
def step1(versionId):
jsonFilePath = f'batchPrint/{versionId}/{versionId}.json'
#判断oss 上是否存在
if not ossClient().object_exists(jsonFilePath):
return False
#下载文件到项目所在目录
dirNow = currentDir + '/batchPrint/' + versionId + '/json'
if not os.path.exists(dirNow):
os.makedirs(dirNow)
localFilePath = dirNow + '/' + versionId + '.json'
ossClient().get_object_to_file(jsonFilePath, localFilePath)
return True
#根据下载下来的JSON文件传递,调用其它参数,进行处理
def step2(jsonFilePath, versionId):
print(f'jsonFilePath={jsonFilePath}, versionId={versionId} {currentDir}/download_print_out.py')
# 调用 download_print_out.py 脚本,下载数据
os.system(f'python {currentDir}/download_print_out.py --pid_file {jsonFilePath} --workdir {currentDir}/batchPrint/{versionId}/data')
# 读取 队列中一个数据出来
def main():
# 循环处理,直到队列为空
while True:
r = redisClient()
#检测队列是否有值
if r.llen('pb:sliceing') == 0:
print(f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())} 队列为空,等待10秒')
time.sleep(10)
continue
#获取队列中的值
data = r.lpop('pb:sliceing')
if data is None:
print(f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())} 取出的数据为空,等待10秒')
time.sleep(10)
continue
data = data.decode('utf-8')
#判断是否是数字
if not data.isdigit():
print(f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())} 取出的数据不是数字,等待10秒')
time.sleep(10)
continue
versionId = str(data)
print(f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())} 正在处理版次ID={versionId}')
res = step1(versionId)
if not res:
print(f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())} JSON文件下载数据失败,等待10秒')
time.sleep(10)
continue
jsonFilePath = currentDir + '/batchPrint/' + versionId + '/json/' + versionId + '.json'
step2(jsonFilePath, versionId)
print(f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())} 数据处理完成,等待10秒')
requestApiToUpdateSliceStatus(versionId)
time.sleep(10)
if __name__ == '__main__':
main()

133
script/type_setting_order.py

@ -0,0 +1,133 @@
import os
import redis
import oss2,time,sys
currentDir = os.path.dirname(os.path.abspath(__file__))
ENV = 'dev'
url = 'https://mp.api.suwa3d.com'
if ENV == 'dev':
url = 'http://mp.dev.api.com'
elif ENV == 'prod':
url = 'https://mp.api.suwa3d.com'
# 连接oss - 单例模式
class OSSClientSingleton:
_instance = None
_client = None
def __new__(cls):
if cls._instance is None:
cls._instance = super(OSSClientSingleton, cls).__new__(cls)
return cls._instance
def get_client(self):
if self._client is None:
AccessKeyId = 'LTAI5tSReWm8hz7dSYxxth8f'
AccessKeySecret = '8ywTDF9upPAtvgXtLKALY2iMYHIxdS'
Endpoint = 'oss-cn-shanghai.aliyuncs.com'
Bucket = 'suwa3d-securedata'
self._client = oss2.Bucket(oss2.Auth(AccessKeyId, AccessKeySecret), Endpoint, Bucket)
return self._client
def ossClient():
"""获取OSS客户端单例"""
return OSSClientSingleton().get_client()
#连接redis,单例模式
class RedisClientSingleton:
_instance = None
_client = None
def __new__(cls):
if cls._instance is None:
cls._instance = super(RedisClientSingleton, cls).__new__(cls)
return cls._instance
def get_client(self):
if self._client is None:
self._client = redis.Redis(host='172.31.1.254', password='', port=6379, db=6)
return self._client
def redisClient():
"""获取Redis客户端单例"""
return RedisClientSingleton().get_client()
# 根据批次id,进行切片中的状态变更
def requestApiToUpdateSliceStatus(versionId):
url = f"{url}/api/printTypeSettingOrder/updateBatchSliceing?batch_id={versionId}"
print(f'发起状态变更请求url={url}, versionId={versionId}')
res = requests.post(url)
if res.status_code != 200:
print(f'状态变更请求失败, res={res.text}')
return False
print(f'状态变更请求成功, res={res.text}')
return True
#判断是否上传了 JSON 文件
def step1(versionId):
jsonFilePath = f'batchPrint/{versionId}/{versionId}.json'
#判断oss 上是否存在
if not ossClient().object_exists(jsonFilePath):
return False
#下载文件到项目所在目录
dirNow = currentDir + '/batchPrint/' + versionId + '/json'
if not os.path.exists(dirNow):
os.makedirs(dirNow)
localFilePath = dirNow + '/' + versionId + '.json'
ossClient().get_object_to_file(jsonFilePath, localFilePath)
return True
#根据下载下来的JSON文件传递,调用其它参数,进行处理
def step2(jsonFilePath, versionId):
print(f'jsonFilePath={jsonFilePath}, versionId={versionId} {currentDir}/download_print_out.py')
# 调用 download_print_out.py 脚本,下载数据
os.system(f'python {currentDir}/download_print_out.py --pid_file {jsonFilePath} --workdir {currentDir}/batchPrint/{versionId}/data')
# 读取 队列中一个数据出来
def main():
# 循环处理,直到队列为空
while True:
r = redisClient()
#检测队列是否有值
if r.llen('pb:sliceing') == 0:
print(f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())} 队列为空,等待10秒')
time.sleep(10)
continue
#获取队列中的值
data = r.lpop('pb:sliceing')
if data is None:
print(f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())} 取出的数据为空,等待10秒')
time.sleep(10)
continue
data = data.decode('utf-8')
#判断是否是数字
if not data.isdigit():
print(f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())} 取出的数据不是数字,等待10秒')
time.sleep(10)
continue
versionId = str(data)
print(f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())} 正在处理版次ID={versionId}')
res = step1(versionId)
if not res:
print(f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())} JSON文件下载数据失败,等待10秒')
time.sleep(10)
continue
jsonFilePath = currentDir + '/batchPrint/' + versionId + '/json/' + versionId + '.json'
step2(jsonFilePath, versionId)
print(f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())} 数据处理完成,等待10秒')
requestApiToUpdateSliceStatus(versionId)
time.sleep(10)
if __name__ == '__main__':
main()
Loading…
Cancel
Save