166 lines
5.8 KiB
Python
166 lines
5.8 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
OCCT 转换服务（常驻 HTTP，默认端口 8090）
|
||
|
||
只做一件事：读 STEP 文件 → 三角化 → 导出 glTF/GLB（网页 3D 轻量化用）。
|
||
零新依赖：HTTP 用标准库 http.server，几何内核用 OCP（OpenCASCADE Python 绑定）。
|
||
|
||
接口：
|
||
GET /health 存活探测 → {"ok":true,"port":8090}
|
||
POST /convert 请求体 = STEP 文件二进制（raw body），
|
||
query 参数 deflection=弦偏差(默认 0.5) → 返回 GLB 二进制
|
||
失败返回 JSON {"error": "..."}
|
||
|
||
说明：
|
||
- 单线程 HTTPServer：OCP 是 C++ 绑定、非线程安全，避免并发导致崩溃。
|
||
- 每次转换写临时文件，转换完清理。
|
||
- 本服务只做「STEP→glTF」转换，不做参数化建模（升级2 已决策不做）。
|
||
"""
|
||
import os
|
||
import sys
|
||
import time
|
||
import tempfile
|
||
from http.server import HTTPServer, BaseHTTPRequestHandler
|
||
from urllib.parse import urlparse, parse_qs
|
||
|
||
from OCP.STEPControl import STEPControl_Reader
|
||
from OCP.BRepMesh import BRepMesh_IncrementalMesh
|
||
from OCP.IFSelect import IFSelect_RetDone
|
||
# XCAF（glTF 导出用）
|
||
from OCP.XCAFApp import XCAFApp_Application
|
||
from OCP.TCollection import TCollection_ExtendedString
|
||
from OCP.TDocStd import TDocStd_Document
|
||
from OCP.XCAFDoc import XCAFDoc_DocumentTool, XCAFDoc_ShapeTool
|
||
from OCP.RWGltf import RWGltf_CafWriter
|
||
from OCP.TColStd import TColStd_IndexedDataMapOfStringString
|
||
from OCP.Message import Message_ProgressRange
|
||
|
||
PORT = int(os.environ.get("OCCT_PORT", "8090"))
|
||
|
||
|
||
def read_step(path):
|
||
"""读 STEP 文件 → TopoDS_Shape"""
|
||
reader = STEPControl_Reader()
|
||
status = reader.ReadFile(path)
|
||
if status != IFSelect_RetDone:
|
||
raise RuntimeError("STEP 读取失败 status=%s" % status)
|
||
reader.TransferRoots()
|
||
return reader.OneShape()
|
||
|
||
|
||
def tessellate(shape, deflection):
|
||
"""三角化（弦偏差 deflection 控制精度/体积平衡）"""
|
||
mesh = BRepMesh_IncrementalMesh(shape, deflection, False, 0.5, True)
|
||
mesh.Perform()
|
||
return shape
|
||
|
||
|
||
def export_glb(shape, out_path, deflection=0.5):
|
||
"""shape → 二进制 GLB（经 XCAF 文档承载）"""
|
||
tessellate(shape, deflection)
|
||
app = XCAFApp_Application.GetApplication_s()
|
||
doc = TDocStd_Document(TCollection_ExtendedString("XmlOcaf"))
|
||
app.InitDocument(doc)
|
||
st = XCAFDoc_DocumentTool.ShapeTool_s(doc.Main())
|
||
lbl = st.NewShape()
|
||
st.SetShape(lbl, shape)
|
||
w = RWGltf_CafWriter(out_path, True) # True = 二进制 GLB
|
||
info = TColStd_IndexedDataMapOfStringString()
|
||
prog = Message_ProgressRange()
|
||
return w.Perform(doc, info, prog)
|
||
|
||
|
||
class Handler(BaseHTTPRequestHandler):
|
||
server_version = "occt-service/1.0"
|
||
|
||
def log_message(self, fmt, *args):
|
||
sys.stderr.write("[%s] %s\n" % (time.strftime("%H:%M:%S"), fmt % args))
|
||
|
||
def _json(self, code, obj):
|
||
import json
|
||
body = json.dumps(obj).encode("utf-8")
|
||
self.send_response(code)
|
||
self.send_header("Content-Type", "application/json; charset=utf-8")
|
||
self.send_header("Content-Length", str(len(body)))
|
||
self.end_headers()
|
||
self.wfile.write(body)
|
||
|
||
def _send_glb(self, path):
|
||
size = os.path.getsize(path)
|
||
self.send_response(200)
|
||
self.send_header("Content-Type", "model/gltf-binary")
|
||
self.send_header("Content-Disposition", 'attachment; filename="model.glb"')
|
||
self.send_header("Content-Length", str(size))
|
||
self.end_headers()
|
||
with open(path, "rb") as f:
|
||
while True:
|
||
chunk = f.read(65536)
|
||
if not chunk:
|
||
break
|
||
self.wfile.write(chunk)
|
||
|
||
def do_GET(self):
|
||
path = urlparse(self.path).path
|
||
if path == "/health":
|
||
self._json(200, {"ok": True, "port": PORT})
|
||
else:
|
||
self._json(404, {"error": "not_found"})
|
||
|
||
def do_POST(self):
|
||
parsed = urlparse(self.path)
|
||
if parsed.path != "/convert":
|
||
self._json(404, {"error": "not_found"})
|
||
return
|
||
qs = parse_qs(parsed.query)
|
||
deflection = 0.5
|
||
if "deflection" in qs:
|
||
try:
|
||
deflection = float(qs["deflection"][0])
|
||
deflection = max(0.05, min(10.0, deflection))
|
||
except ValueError:
|
||
pass
|
||
# 读 raw body（整个请求体 = STEP 文件）
|
||
length = int(self.headers.get("Content-Length", "0"))
|
||
if length == 0:
|
||
self._json(400, {"error": "empty_body"})
|
||
return
|
||
data = self.rfile.read(length)
|
||
|
||
tmp_step = tempfile.mktemp(suffix=".step")
|
||
tmp_glb = tempfile.mktemp(suffix=".glb")
|
||
try:
|
||
with open(tmp_step, "wb") as f:
|
||
f.write(data)
|
||
t0 = time.time()
|
||
shape = read_step(tmp_step)
|
||
ok = export_glb(shape, tmp_glb, deflection)
|
||
if not ok:
|
||
self._json(500, {"error": "gltf_export_failed"})
|
||
return
|
||
elapsed = time.time() - t0
|
||
self.log_message("convert OK %.2fs deflection=%.2f size=%d", elapsed, deflection, os.path.getsize(tmp_glb))
|
||
self._send_glb(tmp_glb)
|
||
except Exception as e:
|
||
self._json(500, {"error": str(e)[:200]})
|
||
finally:
|
||
for p in (tmp_step, tmp_glb):
|
||
try:
|
||
if os.path.exists(p):
|
||
os.remove(p)
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
def main():
|
||
server = HTTPServer(("0.0.0.0", PORT), Handler)
|
||
sys.stderr.write("OCCT 转换服务已启动: http://0.0.0.0:%d (/convert, /health)\n" % PORT)
|
||
sys.stderr.flush()
|
||
try:
|
||
server.serve_forever()
|
||
except KeyboardInterrupt:
|
||
sys.stderr.write("\n服务已停止\n")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|