dongchangxi
3 weeks ago
6 changed files with 652 additions and 0 deletions
@ -0,0 +1,100 @@
@@ -0,0 +1,100 @@
|
||||
import os,time |
||||
import re |
||||
from .logs import log |
||||
|
||||
|
||||
def changeMtlFile(localMtlPath,newName): |
||||
beginTime = time.time() |
||||
try: |
||||
# 使用更高效的文件读取方式 [6,8](@ref) |
||||
with open(localMtlPath, 'r', encoding='utf-8') as f: |
||||
content = f.read() |
||||
|
||||
# 使用字符串方法直接查找和替换，避免不必要的循环 [9](@ref) |
||||
lines = content.split('\n') |
||||
for i, line in enumerate(lines): |
||||
stripped_line = line.strip() |
||||
if stripped_line.startswith('map_Kd '): |
||||
lines[i] = f"map_Kd {newName}" |
||||
break |
||||
# 批量写入，减少I/O操作 [6](@ref) |
||||
with open(localMtlPath, 'w', encoding='utf-8') as f: |
||||
f.write('\n'.join(lines)) |
||||
|
||||
except IOError as e: |
||||
log(f"处理文件 {localMtlPath} 时出错: {e}") |
||||
return False |
||||
except UnicodeDecodeError as e: |
||||
log(f"文件编码错误 {localMtlPath}: {e}") |
||||
return False |
||||
endTime = time.time() |
||||
log(f"修改文件文件内容 {localMtlPath} : 耗时{endTime - beginTime}秒") |
||||
return True |
||||
|
||||
|
||||
|
||||
|
||||
def changeObjFile(localObjPath,newName): |
||||
beginTime = time.time() |
||||
try: |
||||
# 优化策略：mtllib通常在文件前几行，先只读取前100行 |
||||
# 对于大文件（如155万行），这样可以大幅减少内存使用和处理时间 |
||||
with open(localObjPath, 'r', encoding='utf-8') as f: |
||||
first_100_lines = [] |
||||
mtllib_line_idx = -1 |
||||
|
||||
# 读取前100行查找mtllib |
||||
for i in range(100): |
||||
line = f.readline() |
||||
if not line: |
||||
break |
||||
first_100_lines.append(line) |
||||
if mtllib_line_idx == -1 and line.strip().startswith('mtllib '): |
||||
mtllib_line_idx = i |
||||
|
||||
# 如果在前100行找到了mtllib，直接修改 |
||||
if mtllib_line_idx != -1: |
||||
parts = first_100_lines[mtllib_line_idx].split(' ', 1) |
||||
if len(parts) > 1: |
||||
#old_mtl_name = parts[1].strip() |
||||
new_mtl_name = f"{newName}" |
||||
first_100_lines[mtllib_line_idx] = f"mtllib {new_mtl_name}\n" |
||||
# 读取剩余内容并合并 |
||||
remaining_content = f.read() |
||||
# 写回文件 |
||||
with open(localObjPath, 'w', encoding='utf-8') as f_out: |
||||
f_out.writelines(first_100_lines) |
||||
if remaining_content: |
||||
f_out.write(remaining_content) |
||||
|
||||
endTime = time.time() |
||||
log(f"修改文件文件内容 {localObjPath} : 耗时{endTime - beginTime}秒") |
||||
return True |
||||
|
||||
log(f"浪费时间读取完整的文件，进行替换mtllib路径, localObjPath={localObjPath}, newName={newName}") |
||||
# 如果前100行没找到，读取整个文件（很少见的情况） |
||||
f.seek(0) # 重置文件指针 |
||||
content = f.read() |
||||
# 使用正则表达式一次性替换（更高效） |
||||
pattern = r'^mtllib\s+(.+)$' |
||||
def replace_mtllib(match): |
||||
old_mtl_name = match.group(1).strip() |
||||
new_mtl_name = f"{newName}_{old_mtl_name}" |
||||
return f"mtllib {new_mtl_name}" |
||||
|
||||
new_content = re.sub(pattern, replace_mtllib, content, flags=re.MULTILINE) |
||||
if new_content != content: |
||||
with open(localObjPath, 'w', encoding='utf-8') as f_out: |
||||
f_out.write(new_content) |
||||
endTime = time.time() |
||||
log(f"完整读取修改文件文件内容 {localObjPath} : 耗时{endTime - beginTime}秒") |
||||
return True |
||||
|
||||
except IOError as e: |
||||
print(f"处理文件 {localObjPath} 时出错: {e}") |
||||
return False |
||||
except UnicodeDecodeError as e: |
||||
print(f"文件编码错误 {localObjPath}: {e}") |
||||
return False |
||||
|
||||
return False |
||||
@ -0,0 +1,304 @@
@@ -0,0 +1,304 @@
|
||||
import requests,time |
||||
import os |
||||
import shutil |
||||
import json |
||||
from .oss_redis import ossClient |
||||
from .logs import log |
||||
from .oss_func import download_file_with_check, checkFileExists |
||||
from .changeFiles import changeObjFile, changeMtlFile |
||||
from .small_machine_transform import transform_save |
||||
ENV = 'prod' |
||||
url = 'https://mp.api.suwa3d.com' |
||||
if ENV == 'dev': |
||||
url = 'http://mp.api.dev.com' |
||||
elif ENV == 'prod': |
||||
url = 'https://mp.api.suwa3d.com' |
||||
|
||||
|
||||
#根据打印ID 获取下载目录 |
||||
def getDownloadDirByPrintId(printIds): |
||||
#调用接口获取下载的路径 |
||||
api_url = f"{url}/api/printOrder/getInfoByPrintIds" |
||||
res = requests.post(api_url,json={"print_ids":printIds}) |
||||
res = res.json() |
||||
print(f"根据打印ID 获取下载目录, res={res}") |
||||
if res["code"] == 1000: |
||||
return res["data"] |
||||
else: |
||||
return False |
||||
|
||||
#根据批次id,进行切片中的状态变更 |
||||
def requestApiToUpdateSliceStatus(versionId,downloadCounts): |
||||
api_url = f"{url}/api/printTypeSettingOrder/updateBatchSliceing?batch_id={versionId}&download_counts="+str(downloadCounts) |
||||
log(f'发起状态变更请求url={api_url}, versionId={versionId}') |
||||
try: |
||||
# 添加超时参数，防止请求时间过长 |
||||
res = requests.post(api_url, timeout=60) # 60秒超时 |
||||
if res.status_code != 200: |
||||
log(f'状态变更请求失败, res={res.text}') |
||||
return False |
||||
log(f'状态变更请求成功, res={res.text}') |
||||
return True |
||||
except requests.exceptions.Timeout: |
||||
log(f'状态变更请求超时, url={api_url}') |
||||
return False |
||||
except requests.exceptions.RequestException as e: |
||||
log(f'状态变更请求异常, error={str(e)}') |
||||
return False |
||||
|
||||
#判断是否上传了 JSON 文件 |
||||
def checkJsonFileExists(versionId): |
||||
log(f"检测文件和图片是否存在, versionId={versionId}") |
||||
jsonFilePath = f'batchPrint/{versionId}/{versionId}.json' |
||||
#判断oss 上是否存在 |
||||
jpgFilePath = f'batchPrint/{versionId}/{versionId}.jpg' |
||||
|
||||
if not ossClient().object_exists(jsonFilePath): |
||||
log(f'JSON文件不存在: {jsonFilePath}') |
||||
return False,False |
||||
|
||||
|
||||
if not ossClient().object_exists(jpgFilePath): |
||||
log(f'JPG文件不存在: {jpgFilePath}') |
||||
return False,False |
||||
log(f"文件和图片检测成功,存在, versionId={versionId}") |
||||
return jsonFilePath,jpgFilePath |
||||
|
||||
|
||||
|
||||
#检测本地文件是否存在 |
||||
def checkLocalFileExists(localFilePath): |
||||
if not os.path.exists(localFilePath): |
||||
return False |
||||
return True |
||||
|
||||
#读取JSON文件内容 |
||||
def readJsonFile(localFilePath): |
||||
with open(localFilePath, 'r', encoding='utf-8') as f: |
||||
jsonData = json.load(f) |
||||
return jsonData |
||||
|
||||
#根据批次ID,获取批次信息 |
||||
def getBatchInfo(versionId): |
||||
url1 = f"{url}/api/printTypeSettingOrder/getBatchInfoAndPrintMachineInfoByBatchId?batch_id={versionId}" |
||||
res = requests.get(url1) |
||||
res = res.json() |
||||
log(f"获取批次信息和打印机信息, url={url1}, res={res}") |
||||
if res["code"] == 1000: |
||||
return res["data"] |
||||
else: |
||||
return False |
||||
|
||||
#下载文件，读取文件内容，并且文件迁移至正确的目录里，不再放在临时目录里 |
||||
def downloadJsonAndJpgFileAndMoveToCorrectDir(versionId,currentDir): |
||||
jsonFilePath,jpgFilePath = checkJsonFileExists(versionId) |
||||
if jsonFilePath == False or jpgFilePath == False: |
||||
return False,False |
||||
|
||||
#将文件下载到临时目录，待会要判断是否什么类型机型 |
||||
tempDir = os.path.join(currentDir, 'batchPrint', 'temp', versionId) |
||||
if not os.path.exists(tempDir): |
||||
os.makedirs(tempDir) |
||||
localFilePath = os.path.join(tempDir, f'{versionId}.json') |
||||
# 使用带完整性校验的下载方法下载JSON文件 |
||||
ok = download_file_with_check(jsonFilePath, localFilePath) |
||||
if not ok: |
||||
log(f"JSON 文件下载失败或不完整, versionId={versionId}") |
||||
return False,False |
||||
#下载JPG文件 |
||||
localJpgFilePath = os.path.join(tempDir, f'{versionId}.jpg') |
||||
ok = download_file_with_check(jpgFilePath, localJpgFilePath) |
||||
if not ok: |
||||
log(f"JPG 文件下载失败或不完整, versionId={versionId}") |
||||
return False,False |
||||
#根据批次ID,获取批次信息和打印机信息 |
||||
batchMachineInfo = getBatchInfo(versionId) |
||||
if not batchMachineInfo: |
||||
log(f"获取批次信息和打印机信息失败, versionId={versionId}") |
||||
return False,False |
||||
# "batch_info": batchInfo, |
||||
# "print_machine_info": printMachineInfo, |
||||
machineInfo = batchMachineInfo["print_machine_info"] |
||||
print(f"machineInfo={machineInfo['id']}") |
||||
dirNewName = "" |
||||
if str(machineInfo["machine_type"]) == '1': |
||||
dirNewName = os.path.join(currentDir, 'batchPrint', versionId + '_big_No'+str(machineInfo['id'])) |
||||
else: |
||||
dirNewName = os.path.join(currentDir, 'batchPrint', versionId + '_small_No'+str(machineInfo['id'])) |
||||
|
||||
#判断目录是否存在，存在就删除 |
||||
if os.path.exists(dirNewName): |
||||
shutil.rmtree(dirNewName) |
||||
#创建目录 |
||||
os.makedirs(dirNewName) |
||||
|
||||
#创建json子目录 |
||||
jsonSubDir = os.path.join(dirNewName, 'json') |
||||
if not os.path.exists(jsonSubDir): |
||||
os.makedirs(jsonSubDir) |
||||
|
||||
|
||||
#将数据移动过来 |
||||
shutil.move(localFilePath, os.path.join(jsonSubDir, '3DPrintLayout.json')) |
||||
shutil.move(localJpgFilePath, os.path.join(jsonSubDir, f'{versionId}.jpg')) |
||||
#检测文件是否移动成功 |
||||
if not os.path.exists(os.path.join(jsonSubDir, '3DPrintLayout.json')): |
||||
log(f"JSON文件不存在, versionId={versionId}") |
||||
return False,False |
||||
if not os.path.exists(os.path.join(jsonSubDir, f'{versionId}.jpg')): |
||||
log(f"JPG文件不存在, versionId={versionId}") |
||||
return False,False |
||||
log(f"文件移动成功, versionId={versionId}") |
||||
#返回目录路径 |
||||
return dirNewName,machineInfo |
||||
|
||||
|
||||
#整合json文件，读取对应的数据，返回数据结构 |
||||
def getJsonData(dirNewName): |
||||
jsonFilePath = os.path.join(dirNewName, 'json', '3DPrintLayout.json') |
||||
if not os.path.exists(jsonFilePath): |
||||
log(f"JSON文件不存在, dirNewName={dirNewName}") |
||||
return False |
||||
#读取JSON文件内容 |
||||
jsonData = readJsonFile(jsonFilePath) |
||||
#读取models |
||||
models = jsonData.get('models', []) |
||||
if not models: |
||||
log(f"models不存在, dirNewName={dirNewName}") |
||||
return False |
||||
|
||||
listData = [] |
||||
#遍历models |
||||
for model in models: |
||||
file_name = model.get('file_name', '') |
||||
#分割数据 |
||||
arrFileName = file_name.split('_') |
||||
orderId = arrFileName[0] |
||||
pid = arrFileName[1] |
||||
printId = arrFileName[2].replace("P", "") |
||||
size = arrFileName[3] |
||||
counts = arrFileName[4].replace("x", "").replace(".obj", "") |
||||
|
||||
#检测这些数据 |
||||
if not orderId or not pid or not printId or not size or not counts: |
||||
log(f"数据不完整, orderId={orderId}, pid={pid}, printId={printId}, size={size}, counts={counts}") |
||||
return False |
||||
|
||||
|
||||
|
||||
#创建数据结构 |
||||
modelInfo = { |
||||
"orderId": orderId, |
||||
"printId": printId, |
||||
"pid": pid, |
||||
"size": size, |
||||
"counts": counts, |
||||
} |
||||
listData.append(modelInfo) |
||||
|
||||
#检测数据长度 |
||||
if len(listData) == 0: |
||||
log(f"数据长度为0, dirNewName={dirNewName}") |
||||
return False |
||||
#返回数据结构 |
||||
return listData |
||||
|
||||
|
||||
# 读取 json 文件，获取 homo_matrix 数据，根据 file_name 获取 homo_matrix 数据 |
||||
def getHomoMatrixByFileName(dirNewName,fileName): |
||||
jsonFilePath = os.path.join(dirNewName, 'json', '3DPrintLayout.json') |
||||
if not os.path.exists(jsonFilePath): |
||||
log(f"JSON文件不存在, dirNewName={dirNewName}") |
||||
return False |
||||
#读取JSON文件内容 |
||||
jsonData = readJsonFile(jsonFilePath) |
||||
if not jsonData: |
||||
log(f"读取JSON文件内容失败, dirNewName={dirNewName}") |
||||
return False |
||||
|
||||
for model in jsonData["models"]: |
||||
log(f"jsonData={model['file_name']} == {fileName}") |
||||
if model["file_name"] == fileName: |
||||
log(f"model={model['transform']}") |
||||
return model["transform"]["homo_matrix"] |
||||
return False |
||||
|
||||
|
||||
#json文件进行下载对应的数据 和转换数据,传递目录路径 |
||||
def downloadDataByOssAndTransformSave(dirNewName,isSmallMachine=False): |
||||
listData = getJsonData(dirNewName) |
||||
if not listData: |
||||
log(f"获取数据失败, dirNewName={dirNewName}") |
||||
return False |
||||
#遍历数据 |
||||
arrPrintId = [] |
||||
for modelInfo in listData: |
||||
arrPrintId.append(modelInfo.get('printId')) |
||||
|
||||
#调用接口获取下载的路径 |
||||
arrDownloadPath = getDownloadDirByPrintId(arrPrintId) |
||||
if not arrDownloadPath: |
||||
log(f"获取下载路径失败, arrPrintId={arrPrintId}") |
||||
return False |
||||
|
||||
dirPath = os.path.join(dirNewName, 'data') |
||||
if not os.path.exists(dirPath): |
||||
os.makedirs(dirPath) |
||||
|
||||
#遍历数据 |
||||
for info in arrDownloadPath: |
||||
filePath = info["path"] |
||||
pid = info["pid"] |
||||
orderId = info["order_id"] |
||||
printId = info["print_order_id"] |
||||
size = info["real_size"] |
||||
counts = info["quantity"] |
||||
|
||||
#判断文件是否存在 |
||||
ossJpgFilePath = f"{filePath}/printId_{printId}Tex1.jpg" |
||||
if not checkFileExists(ossJpgFilePath): |
||||
ossJpgFilePath = f"{filePath}/{pid}Tex1.jpg" |
||||
|
||||
localJpgName = os.path.join(f"{orderId}_{pid}Tex1.jpg") |
||||
loaclMtlName = os.path.join(f"{orderId}_{pid}.mtl") |
||||
localObjName = os.path.join(f"{orderId}_{pid}_P{printId}_{size}_x{counts}.obj") |
||||
arrDownloadFiles = [ |
||||
{"ossPath": ossJpgFilePath, "localPath": os.path.join(dirPath, localJpgName)}, |
||||
{"ossPath": f"{filePath}/{pid}.obj", "localPath": os.path.join(dirPath, localObjName)}, |
||||
{"ossPath": f"{filePath}/{pid}.mtl", "localPath": os.path.join(dirPath, loaclMtlName)}, |
||||
] |
||||
#遍历下载文件 |
||||
beginTime = time.time() |
||||
objsLocal = "" |
||||
for objFiles in arrDownloadFiles: |
||||
downloadOk = download_file_with_check(objFiles["ossPath"], objFiles["localPath"]) |
||||
if not downloadOk: |
||||
log(f"下载文件失败, ossPath={objFiles["ossPath"]}, localPath={objFiles["localPath"]}") |
||||
return False |
||||
|
||||
#下载成功之后要修改文件之间的关联路径 |
||||
if objFiles["localPath"].endswith(".obj"): |
||||
objsLocal = objFiles["localPath"] |
||||
changeObjFile(objFiles["localPath"], f"{orderId}_{pid}.mtl") |
||||
|
||||
if objFiles["localPath"].endswith(".mtl"): |
||||
changeMtlFile(objFiles["localPath"], f"{orderId}_{pid}Tex1.jpg") |
||||
|
||||
endTime = time.time() |
||||
log(f"下载文件和修改文件之间的关联路径 : 耗时{endTime - beginTime}秒") |
||||
|
||||
#如果是小机台，则要转换数据 |
||||
if not isSmallMachine: |
||||
continue |
||||
|
||||
timeBegin = time.time() |
||||
homo_matrix = getHomoMatrixByFileName(dirNewName, localObjName) |
||||
if not homo_matrix: |
||||
log(f"获取homo_matrix失败, dirNewName={dirNewName}, objsLocal={objsLocal}") |
||||
return False |
||||
transform_save(objFiles["localPath"], homo_matrix) |
||||
timeEnd = time.time() |
||||
log(f"转换数据时间{objsLocal}: 耗时{timeEnd - timeBegin}秒") |
||||
|
||||
|
||||
return True |
||||
@ -0,0 +1,4 @@
@@ -0,0 +1,4 @@
|
||||
import time |
||||
|
||||
def log(message): |
||||
print(f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())} {message}') |
||||
@ -0,0 +1,45 @@
@@ -0,0 +1,45 @@
|
||||
import os |
||||
from .oss_redis import ossClient |
||||
from .logs import log |
||||
|
||||
# 从 OSS 下载文件到本地，并做完整性校验（状态码 + 文件大小） |
||||
def download_file_with_check(ossFilePath, localFilePath): |
||||
try: |
||||
log(f"开始从 OSS 下载文件: {ossFilePath} -> {localFilePath}") |
||||
if not checkFileExists(ossFilePath): |
||||
log(f"文件不存在: {ossFilePath}") |
||||
return False |
||||
# 同步阻塞下载，下载完成后才会返回 |
||||
result = ossClient().get_object_to_file(ossFilePath, localFilePath) |
||||
|
||||
# 1. 检查 HTTP 状态码 |
||||
status = getattr(result, "status", None) |
||||
if status != 200: |
||||
log(f"下载失败，HTTP 状态码异常: status={status}") |
||||
return False |
||||
|
||||
# 2. 远端 / 本地文件大小对比，作为二次校验 |
||||
remote_meta = ossClient().head_object(ossFilePath) |
||||
remote_size = getattr(remote_meta, "content_length", None) |
||||
if remote_size is None: |
||||
log("无法获取远端文件大小，放弃本次下载结果") |
||||
return False |
||||
|
||||
if not os.path.exists(localFilePath): |
||||
log("本地文件不存在，下载可能失败") |
||||
return False |
||||
|
||||
local_size = os.path.getsize(localFilePath) |
||||
if remote_size != local_size: |
||||
log(f"文件大小不一致，下载可能不完整: remote={remote_size}, local={local_size}") |
||||
return False |
||||
|
||||
log("文件下载成功且完整性校验通过") |
||||
return True |
||||
except Exception as e: |
||||
log(f"下载文件出现异常: {str(e)}") |
||||
return False |
||||
|
||||
|
||||
def checkFileExists(ossFilePath): |
||||
return ossClient().object_exists(ossFilePath) |
||||
@ -0,0 +1,79 @@
@@ -0,0 +1,79 @@
|
||||
import oss2,redis |
||||
|
||||
# 连接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: |
||||
# 添加超时参数，防止连接超时 |
||||
# socket_timeout: 每次操作的超时时间（秒） |
||||
# socket_connect_timeout: 连接超时时间（秒） |
||||
# socket_keepalive: 启用 TCP keepalive |
||||
# socket_keepalive_options: keepalive 选项 |
||||
self._client = redis.Redis( |
||||
host='mp.api.suwa3d.com', |
||||
password='kcV2000', |
||||
port=6379, |
||||
db=6, |
||||
socket_timeout=30, # 操作超时30秒 |
||||
socket_connect_timeout=10, # 连接超时10秒 |
||||
socket_keepalive=True, # 启用 TCP keepalive |
||||
socket_keepalive_options={}, # keepalive 选项 |
||||
health_check_interval=30 # 健康检查间隔30秒 |
||||
) |
||||
else: |
||||
# 检查连接是否有效，如果断开则重新连接 |
||||
try: |
||||
self._client.ping() |
||||
except (redis.ConnectionError, redis.TimeoutError, AttributeError): |
||||
# 连接断开，重新创建连接 |
||||
self._client = None |
||||
self._client = redis.Redis( |
||||
host='mp.api.suwa3d.com', |
||||
password='kcV2000', |
||||
port=6379, |
||||
db=6, |
||||
socket_timeout=30, |
||||
socket_connect_timeout=10, |
||||
socket_keepalive=True, |
||||
socket_keepalive_options={}, |
||||
health_check_interval=30 |
||||
) |
||||
return self._client |
||||
|
||||
|
||||
|
||||
def redisClient(): |
||||
"""获取Redis客户端单例""" |
||||
return RedisClientSingleton().get_client() |
||||
@ -0,0 +1,120 @@
@@ -0,0 +1,120 @@
|
||||
import os,time |
||||
import numpy as np |
||||
import subprocess |
||||
from .logs import log |
||||
|
||||
|
||||
def findBpyModule(): |
||||
# /Applications/Blender.app/Contents/Resources/4.4/python/bin |
||||
blender_bin_path = '/Applications/Blender.app/Contents/Resources/4.4/python/bin/blender' |
||||
return blender_bin_path |
||||
|
||||
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 |
||||
|
||||
# { |
||||
# "file_name": "888687_324912_P117183_13.8cm_x1.obj", |
||||
# "transform": { |
||||
# "homo_matrix": [ |
||||
# [-0.09437,0.995195,0.026116,77.975499], |
||||
# [-0.58016,-0.033658,-0.813807,322.60588], |
||||
# [-0.809017,-0.09195,0.580549,24.503686], |
||||
# [0.0,0.0,0.0,1.0] |
||||
# ] |
||||
# } |
||||
# }, |
||||
def transform_save(obj_path,homo_matrix): |
||||
# 延迟导入 bpy，只在 Blender 环境中可用 |
||||
try: |
||||
blender_bin_path = findBpyModule() |
||||
subprocess.run([blender_bin_path, "--background", "--python-expr", "import bpy"]) |
||||
except ImportError: |
||||
log("错误: bpy 模块不可用。此函数只能在 Blender 环境中运行。") |
||||
log("请使用 Blender 的 Python 解释器运行此脚本，或通过 Blender 命令行运行。") |
||||
raise ImportError("bpy 模块不可用。此函数只能在 Blender 环境中运行。") |
||||
|
||||
obj_name = obj_path.split("/")[-1] |
||||
# 清除场景 |
||||
bpy.ops.object.select_all(action='SELECT') |
||||
bpy.ops.object.delete(use_global=False) |
||||
|
||||
## 尺寸调整，环境设置 |
||||
bpy.ops.object.delete(use_global=False, confirm=False) |
||||
bpy.context.scene.unit_settings.length_unit = 'CENTIMETERS' |
||||
bpy.context.scene.unit_settings.scale_length = 0.001 |
||||
bpy.context.scene.unit_settings.mass_unit = 'GRAMS' |
||||
|
||||
# meshes = [] |
||||
# 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) |
||||
|
||||
#mtl_name_temp = obj_name |
||||
#separator = "_P" |
||||
#index = mtl_name_temp.find(separator) |
||||
# if index != -1: |
||||
# old_mtl_name = mtl_name_temp[:index] |
||||
# else: |
||||
# old_mtl_name = mtl_name_temp # 或者你希望的其他处理逻辑，比如直接使用原字符串或报错 |
||||
# old_mtl_name = f"{old_mtl_name}.mtl" |
||||
#old_mtl_path = os.path.join(original_obj_pid_dir, old_mtl_name) |
||||
|
||||
bpy.ops.wm.obj_import(filepath=obj_path) |
||||
imported_object = bpy.context.object |
||||
|
||||
if imported_object is None or imported_object.type != 'MESH': |
||||
print(f"警告: 未能成功导入网格对象 {obj_name}。跳过。") |
||||
return |
||||
|
||||
mesh_data = imported_object.data |
||||
mesh_data.calc_loop_triangles() |
||||
original_vertices = np.empty(len(mesh_data.vertices) * 3, dtype=np.float64) |
||||
mesh_data.vertices.foreach_get('co', original_vertices) |
||||
original_vertices = original_vertices.reshape(-1, 3) |
||||
|
||||
transformed_vertices = custom_mesh_transform(original_vertices, reconstructed_matrix) |
||||
|
||||
offset = np.array([-380, -345, 0]) |
||||
transformed_vertices += offset |
||||
print(f"已对模型 {obj_name} 应用偏移: {offset}") |
||||
|
||||
flattened_verts = transformed_vertices.reshape(-1) |
||||
mesh_data.vertices.foreach_set('co', flattened_verts) |
||||
|
||||
mesh_data.update() |
||||
|
||||
# meshes.append(imported_object) |
||||
|
||||
# 导出前确保选中当前对象 |
||||
bpy.ops.object.select_all(action='DESELECT') |
||||
imported_object.select_set(True) |
||||
bpy.context.view_layer.objects.active = imported_object |
||||
bpy.ops.wm.obj_export( |
||||
filepath=obj_path, |
||||
export_selected_objects=True, |
||||
export_materials=True, |
||||
path_mode='AUTO' |
||||
) |
||||
|
||||
#os.remove(old_mtl_path) |
||||
Loading…
Reference in new issue