cjw/geMoldInsight
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add freecad

Browse Source
This commit is contained in:
SZCJW
2026-03-08 02:53:27 +08:00
parent 6ee740853f
commit c79190e319
4 changed files with 385 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/migrate_multi_upload.py
+58 -3
View File
@@ -1,5 +1,5 @@
"""
数据库迁移脚本 - 添加多上传支持字段
数据库迁移脚本 - 添加多上传支持字段和验证字段
运行方式: python scripts/migrate_multi_upload.py
@@ -8,6 +8,7 @@
2. 添加 volume, surface_area, product_weight 快速查询字段到 stp_files 表
3. 移除 file_hash 字段的唯一约束（如果存在）
4. 为新字段创建索引
5. 添加验证结果字段到 analysis_metrics 表
"""
import asyncio
import sys
@@ -47,7 +48,7 @@ async def check_index_exists(conn, index_name: str) -> bool:
async def run_migration():
"""执行迁移"""
logger.info("开始数据库迁移 - 多上传支持...")
logger.info("开始数据库迁移 - 多上传支持和验证字段...")
await db_manager.connect()
@@ -58,6 +59,7 @@ async def run_migration():
async with db_manager.engine.begin() as conn:
migration_steps = []
# stp_files 表字段
if not await check_column_exists(conn, "stp_files", "upload_batch"):
migration_steps.append("添加 upload_batch 字段")
await conn.execute(text("""
@@ -90,6 +92,40 @@ async def run_migration():
"""))
logger.info("✓ 添加 product_weight 字段")
# analysis_metrics 表验证字段
if not await check_column_exists(conn, "analysis_metrics", "verification_status"):
migration_steps.append("添加 verification_status 字段")
await conn.execute(text("""
ALTER TABLE analysis_metrics
ADD COLUMN verification_status VARCHAR(20)
"""))
logger.info("✓ 添加 verification_status 字段")
if not await check_column_exists(conn, "analysis_metrics", "verification_volume_diff"):
migration_steps.append("添加 verification_volume_diff 字段")
await conn.execute(text("""
ALTER TABLE analysis_metrics
ADD COLUMN verification_volume_diff FLOAT
"""))
logger.info("✓ 添加 verification_volume_diff 字段")
if not await check_column_exists(conn, "analysis_metrics", "verification_area_diff"):
migration_steps.append("添加 verification_area_diff 字段")
await conn.execute(text("""
ALTER TABLE analysis_metrics
ADD COLUMN verification_area_diff FLOAT
"""))
logger.info("✓ 添加 verification_area_diff 字段")
if not await check_column_exists(conn, "analysis_metrics", "verification_details"):
migration_steps.append("添加 verification_details 字段")
await conn.execute(text("""
ALTER TABLE analysis_metrics
ADD COLUMN verification_details JSON
"""))
logger.info("✓ 添加 verification_details 字段")
# 索引
if not await check_index_exists(conn, "ix_stp_files_upload_batch"):
migration_steps.append("创建 upload_batch 索引")
await conn.execute(text("""
@@ -114,6 +150,7 @@ async def run_migration():
"""))
logger.info("✓ 创建 original_filename 索引")
# 删除唯一约束
try:
result = await conn.execute(text("""
SELECT conname
@@ -159,6 +196,7 @@ async def rollback_migration():
async with db_manager.engine.begin() as conn:
try:
# stp_files 表
if await check_index_exists(conn, "ix_stp_files_upload_batch"):
await conn.execute(text("DROP INDEX IF EXISTS ix_stp_files_upload_batch"))
logger.info("✓ 删除 upload_batch 索引")
@@ -183,6 +221,23 @@ async def rollback_migration():
await conn.execute(text("ALTER TABLE stp_files DROP COLUMN product_weight"))
logger.info("✓ 删除 product_weight 字段")
# analysis_metrics 表
if await check_column_exists(conn, "analysis_metrics", "verification_status"):
await conn.execute(text("ALTER TABLE analysis_metrics DROP COLUMN verification_status"))
logger.info("✓ 删除 verification_status 字段")
if await check_column_exists(conn, "analysis_metrics", "verification_volume_diff"):
await conn.execute(text("ALTER TABLE analysis_metrics DROP COLUMN verification_volume_diff"))
logger.info("✓ 删除 verification_volume_diff 字段")
if await check_column_exists(conn, "analysis_metrics", "verification_area_diff"):
await conn.execute(text("ALTER TABLE analysis_metrics DROP COLUMN verification_area_diff"))
logger.info("✓ 删除 verification_area_diff 字段")
if await check_column_exists(conn, "analysis_metrics", "verification_details"):
await conn.execute(text("ALTER TABLE analysis_metrics DROP COLUMN verification_details"))
logger.info("✓ 删除 verification_details 字段")
logger.info("回滚完成!")
except Exception as e:
logger.error(f"回滚失败: {e}")
@@ -195,7 +250,7 @@ async def rollback_migration():
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser(description="数据库迁移脚本 - 多上传支持")
parser = argparse.ArgumentParser(description="数据库迁移脚本 - 多上传支持和验证字段")
parser.add_argument("--rollback", action="store_true", help="回滚迁移")
args = parser.parse_args()
src/api/routes.py
+63
View File
@@ -358,6 +358,46 @@ async def _save_analysis_metrics(session, stp_file_id, analysis_result):
logger.info(f"分析指标保存成功: {metrics.id}")
async def _save_verification_metrics(session, stp_file_id, verification_result):
"""保存验证指标到数据库"""
from models.database import AnalysisMetrics
from sqlalchemy import select
# 查找现有的分析指标记录
result = await session.execute(
select(AnalysisMetrics).where(AnalysisMetrics.stp_file_id == stp_file_id)
)
metrics = result.scalar_one_or_none()
if metrics:
# 更新现有记录
metrics.verification_status = verification_result.get("status", "unknown")
comparison = verification_result.get("comparison", {})
volume_comparison = comparison.get("volume", {})
area_comparison = comparison.get("surface_area", {})
metrics.verification_volume_diff = volume_comparison.get("difference_percent", 0)
metrics.verification_area_diff = area_comparison.get("difference_percent", 0)
metrics.verification_details = verification_result
else:
# 创建新记录
comparison = verification_result.get("comparison", {})
volume_comparison = comparison.get("volume", {})
area_comparison = comparison.get("surface_area", {})
metrics = AnalysisMetrics(
stp_file_id=stp_file_id,
verification_status=verification_result.get("status", "unknown"),
verification_volume_diff=volume_comparison.get("difference_percent", 0),
verification_area_diff=area_comparison.get("difference_percent", 0),
verification_details=verification_result
)
session.add(metrics)
await session.commit()
logger.info(f"验证指标保存成功: stp_file_id={stp_file_id}")
async def process_file_core(
storage_service: StorageIntegrationService,
task_id: str,
@@ -633,6 +673,29 @@ async def process_file_core(
product_weight=product_weight_g
)
# 9.7 FreeCAD 几何验证
await storage_service.update_task_status(
db_session, task_id, "processing", 90, "FreeCAD几何验证"
)
verification_result = None
try:
from services.verification_service import verification_service
verification_result = await verification_service.verify_stp_file(file_path)
# 保存验证结果到数据库
if verification_result and analysis_result:
await _save_verification_metrics(
db_session,
stp_file_id,
verification_result
)
logger.info(f"FreeCAD验证完成: {verification_result.get('status', 'unknown') if verification_result else 'failed'}")
except Exception as ve:
logger.warning(f"FreeCAD验证失败（不影响主流程）: {ve}")
verification_result = {"status": "error", "error": str(ve)}
# 10. 完成处理
await storage_service.update_stp_file_status(db_session, stp_file_id, "completed")
await storage_service.update_task_status(
src/models/database.py
+6
View File
@@ -696,6 +696,12 @@ class AnalysisMetrics(Base):
# 分析摘要
analysis_summary = Column(Text, nullable=True)
# FreeCAD 验证结果
verification_status = Column(String(20), nullable=True) # passed, failed, pending, error
verification_volume_diff = Column(Float, nullable=True) # 体积差异百分比
verification_area_diff = Column(Float, nullable=True) # 表面积差异百分比
verification_details = Column(JSON, nullable=True) # 完整验证结果
# 时间戳
created_at = Column(DateTime, default=func.now())
src/services/verification_service.py
+258
View File
@@ -0,0 +1,258 @@
"""
几何验证服务
使用 FreeCAD 和 PythonOCC 交叉验证几何数据准确性
"""
import asyncio
import json
import subprocess
import sys
from pathlib import Path
from typing import Dict, Any, Optional
from datetime import datetime
from utils.logger import get_logger
logger = get_logger(__name__)
class GeometryVerificationService:
"""几何验证服务"""
def __init__(self):
self.verification_script = Path(__file__).parent.parent / "scripts" / "verify_stp.py"
async def verify_stp_file(self, stp_path: str) -> Dict[str, Any]:
"""
验证 STP 文件几何数据
Args:
stp_path: STP 文件路径
Returns:
验证结果字典
"""
try:
logger.info(f"开始验证 STP 文件: {stp_path}")
# 运行验证脚本
result = await self._run_verification_script(stp_path)
if result:
logger.info(f"验证完成: {result.get('status', 'unknown')}")
else:
logger.warning("验证脚本未返回结果")
return result
except Exception as e:
logger.error(f"验证失败: {e}")
return {
"status": "error",
"error": str(e),
"timestamp": datetime.now().isoformat()
}
async def _run_verification_script(self, stp_path: str) -> Optional[Dict[str, Any]]:
"""运行验证脚本"""
import tempfile
import os
# 创建临时输出文件
with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f:
output_path = f.name
try:
# 构建命令 - 使用 FreeCAD 命令行模式
cmd = None
# 尝试1: freecad.cmd (snap 安装的命令行模式)
for cmd_name in ['freecad.cmd', '/snap/bin/freecad.cmd', 'freecad-cmd']:
try:
result = subprocess.run(['which', cmd_name], capture_output=True, text=True)
if result.returncode == 0 and result.stdout.strip():
cmd = [cmd_name, str(self.verification_script), stp_path]
logger.debug(f"找到 FreeCAD 命令: {cmd_name}")
break
except:
pass
# 尝试2: freecad with offscreen
if not cmd:
try:
result = subprocess.run(['which', 'freecad'], capture_output=True, text=True)
if result.returncode == 0:
cmd = [
'env', 'QT_QPA_PLATFORM=offscreen',
'freecad', '-c',
str(self.verification_script), stp_path
]
except:
pass
# 尝试3: 使用 xvfb-run
if not cmd:
try:
result = subprocess.run(['which', 'xvfb-run'], capture_output=True, text=True)
if result.returncode == 0:
cmd = ['xvfb-run', 'freecad', '-c', str(self.verification_script), stp_path]
except:
pass
if not cmd:
logger.warning("FreeCAD 命令行工具不可用，跳过验证")
return {
"status": "skipped",
"reason": "FreeCAD not available",
"timestamp": datetime.now().isoformat()
}
logger.info(f"执行验证命令: {' '.join(cmd)}")
# 异步运行子进程
process = await asyncio.create_subprocess_exec(
*cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
cwd=str(Path(__file__).parent.parent)
)
stdout, stderr = await process.communicate()
if process.returncode == 0:
# 尝试读取生成的报告文件
report_path = Path(stp_path).stem + "_verification_report.json"
if Path(report_path).exists():
with open(report_path, 'r', encoding='utf-8') as f:
result = json.load(f)
# 删除临时报告文件
Path(report_path).unlink()
return result
else:
# 解析 stdout 获取结果
return self._parse_verification_output(stdout.decode('utf-8'))
else:
logger.error(f"验证脚本执行失败 (returncode={process.returncode}): {stderr.decode('utf-8')}")
return {
"status": "error",
"error": stderr.decode('utf-8'),
"timestamp": datetime.now().isoformat()
}
except FileNotFoundError:
logger.warning("FreeCAD 命令行工具不可用，跳过验证")
return {
"status": "skipped",
"reason": "FreeCAD not available",
"timestamp": datetime.now().isoformat()
}
except Exception as e:
logger.error(f"运行验证脚本失败: {e}")
return {
"status": "error",
"error": str(e),
"timestamp": datetime.now().isoformat()
}
finally:
# 清理临时文件
if os.path.exists(output_path):
os.unlink(output_path)
def _parse_verification_output(self, output: str) -> Dict[str, Any]:
"""解析验证脚本输出"""
result = {
"status": "unknown",
"timestamp": datetime.now().isoformat()
}
lines = output.split('\n')
for line in lines:
if '验证结果:' in line:
if '✅ 通过' in line:
result['status'] = 'passed'
elif '❌ 失败' in line:
result['status'] = 'failed'
elif '体积对比:' in line:
# 解析体积差异
pass
elif '表面积对比:' in line:
# 解析表面积差异
pass
return result
def verify_with_pythonocc(self, shape) -> Dict[str, Any]:
"""
使用 PythonOCC 验证几何数据（同步方法，用于内部验证）
Args:
shape: OCC 形状对象
Returns:
验证结果
"""
try:
from OCC.Core.GProp import GProp_GProps
from OCC.Core.BRepGProp import brepgprop_VolumeProperties, brepgprop_SurfaceProperties
from OCC.Core.Bnd import Bnd_Box
from OCC.Core.BRepBndLib import brepbndlib_Add
from OCC.Core.TopExp import TopExp_Explorer
from OCC.Core.TopAbs import TopAbs_FACE, TopAbs_EDGE, TopAbs_VERTEX, TopAbs_SOLID
# 计算体积
vol_props = GProp_GProps()
brepgprop_VolumeProperties(shape, vol_props)
volume_mm3 = vol_props.Mass()
com = vol_props.CentreOfMass()
# 计算表面积
surf_props = GProp_GProps()
brepgprop_SurfaceProperties(shape, surf_props)
surface_area_mm2 = surf_props.Mass()
# 计算边界框
bbox = Bnd_Box()
brepbndlib_Add(shape, bbox)
xmin, ymin, zmin, xmax, ymax, zmax = bbox.Get()
# 拓扑统计
def count_topology(shape, top_type):
explorer = TopExp_Explorer(shape, top_type)
count = 0
while explorer.More():
count += 1
explorer.Next()
return count
return {
"volume_mm3": float(volume_mm3),
"volume_cm3": float(volume_mm3 / 1000),
"surface_area_mm2": float(surface_area_mm2),
"surface_area_cm2": float(surface_area_mm2 / 100),
"bounding_box": {
"x_min": float(xmin),
"x_max": float(xmax),
"y_min": float(ymin),
"y_max": float(ymax),
"z_min": float(zmin),
"z_max": float(zmax),
"x_length": float(xmax - xmin),
"y_length": float(ymax - ymin),
"z_length": float(zmax - zmin),
"center": [float((xmin + xmax) / 2), float((ymin + ymax) / 2), float((zmin + zmax) / 2)]
},
"center_of_mass": [float(com.X()), float(com.Y()), float(com.Z())],
"topology": {
"faces": count_topology(shape, TopAbs_FACE),
"edges": count_topology(shape, TopAbs_EDGE),
"vertices": count_topology(shape, TopAbs_VERTEX),
"solids": count_topology(shape, TopAbs_SOLID)
}
}
except Exception as e:
logger.error(f"PythonOCC 验证失败: {e}")
return {"error": str(e)}
# 单例实例
verification_service = GeometryVerificationService()