import os import json import uuid import shutil import subprocess from pathlib import Path import fcntl from flask import Flask, render_template, request, abort, Response from werkzeug.utils import secure_filename import ezdxf from ezdxf.addons.drawing import Frontend, RenderContext, svg, layout, config as ezconfig APP_DIR = Path(__file__).resolve().parent WORKER_DIR = APP_DIR / "worker" JOBS_DIR = APP_DIR / "_jobs" JOBS_DIR.mkdir(exist_ok=True) # In Docker: /usr/bin/freecadcmd, lokal (mac): via env setzen, wenn du willst FREECAD_TIMEOUT_SEC = int(os.environ.get("FREECAD_TIMEOUT_SEC", "1200")) FREECADCMD = (os.environ.get("FREECADCMD") or "").strip() or "/opt/conda/envs/fc/bin/freecadcmd" FREECADCMD = "/usr/bin/freecadcmd" ALLOWED_MATERIALS = [ ("stainless", "Edelstahl"), ("alu", "Aluminium"), ("copper", "Kupfer"), ] # Serialize FreeCAD runs (one at a time) LOCK_FILE = os.environ.get("FREECAD_LOCK_FILE", str(JOBS_DIR / ".freecad.lock")) app = Flask(__name__) def _safe_job_dir(job_id: str) -> Path: if not job_id or any(c not in "0123456789abcdef" for c in job_id.lower()): abort(404) job_dir = (JOBS_DIR / job_id).resolve() if not str(job_dir).startswith(str(JOBS_DIR.resolve())): abort(404) if not job_dir.exists(): abort(404) return job_dir class FreeCADLock: def __init__(self, lock_path: str): self.lock_path = lock_path self.fp = None def __enter__(self): self.fp = open(self.lock_path, "w") fcntl.flock(self.fp, fcntl.LOCK_EX) return self def __exit__(self, exc_type, exc, tb): try: fcntl.flock(self.fp, fcntl.LOCK_UN) finally: try: self.fp.close() except Exception: pass def run_job(job_dir: Path, filename: str, material: str, thickness_mm: str | None): """ Runs FreeCAD headless in a job dir. Keeps original filename. Produces: job_dir/result.json + flat.dxf (compat), plus named outputs by analyser. """ shutil.copy2(WORKER_DIR / "stepanalyser.py", job_dir / "stepanalyser.py") argv = ['"stepanalyser.py"', '"--input"', f'"{filename}"', '"--material"', f'"{material}"'] if thickness_mm: argv += ['"--thickness-mm"', f'"{thickness_mm}"'] runner_text = f"""import os, sys, json, traceback def write_result(payload): try: with open(\"result.json\", \"w\", encoding=\"utf-8\") as f: json.dump(payload, f, indent=2, ensure_ascii=False) except Exception: pass try: base = os.path.dirname(os.path.abspath(__file__)) os.chdir(base) if not os.path.exists({json.dumps(filename)}): raise SystemExit(\"Uploaded STEP file missing: \" + {json.dumps(filename)}) # Ensure FreeCAD can find user Mods (SheetMetal) # Linux default: ~/.local/share/FreeCAD/Mod mod_dir = os.path.expanduser("~/.local/share/FreeCAD/Mod") if os.path.isdir(mod_dir) and mod_dir not in sys.path: sys.path.append(mod_dir) # sm_dir = os.path.join(mod_dir, \"SheetMetal\") sm_dir = os.path.join(mod_dir, \"sheetmetal\") if os.path.isdir(sm_dir) and sm_dir not in sys.path: sys.path.append(sm_dir) sys.argv = [{", ".join(argv)}] code = open(\"stepanalyser.py\", \"r\", encoding=\"utf-8\").read() exec(compile(code, \"stepanalyser.py\", \"exec\"), {{\"__name__\": \"__main__\"}}) except BaseException as e: payload = {{ \"ok\": False, \"error_type\": type(e).__name__, \"error\": str(e), \"traceback\": traceback.format_exc() }} write_result(payload) print(\"RUNNER ERROR:\", payload[\"error_type\"], payload[\"error\"], flush=True) finally: os._exit(0) """ (job_dir / "run_stepanalyser.py").write_text(runner_text, encoding="utf-8") cmd = [FREECADCMD, "run_stepanalyser.py"] env = os.environ.copy() # Remove anything that can make embedded Python pick up the venv / wrong stdlib for k in list(env.keys()): if k.startswith("PYTHON"): env.pop(k, None) # Also de-venv-ify PATH if you want to be extra safe # (optional, but helpful if the venv puts shims first) # env["PATH"] = "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin" # Make locale sane (Py_EncodeLocale happens very early) env.setdefault("LANG", "C.UTF-8") env.setdefault("LC_ALL", "C.UTF-8") env.setdefault("LC_CTYPE", "C.UTF-8") # Your existing settings env["HOME"] = env.get("HOME") or "/config" env.setdefault("QT_QPA_PLATFORM", "offscreen") env["LANG"] = "C" env["LC_ALL"] = "C" env["LC_CTYPE"] = "C" env["PYTHONUTF8"] = "1" env["PYTHONCOERCECLOCALE"] = "1" timeout = FREECAD_TIMEOUT_SEC log_path = job_dir / "run.log" result_path = job_dir / "result.json" # One FreeCAD run at a time with FreeCADLock(LOCK_FILE): with open(log_path, "w", encoding="utf-8") as log_fp: log_fp.write("=== STEPANALYSER START ===\n") log_fp.write(f"Command: {cmd}\n") log_fp.flush() log_fp.write(f"Sanitized env: PYTHONHOME={env.get('PYTHONHOME')} PYTHONPATH={env.get('PYTHONPATH')}\n") log_fp.write(f"Locale: LANG={env.get('LANG')} LC_ALL={env.get('LC_ALL')} LC_CTYPE={env.get('LC_CTYPE')}\n") proc = subprocess.Popen( cmd, cwd=str(job_dir), stdin=subprocess.DEVNULL, stdout=log_fp, stderr=subprocess.STDOUT, text=True, env=env, ) try: proc.wait(timeout=timeout) except subprocess.TimeoutExpired: # Kill the process and persist a useful error result try: proc.kill() except Exception: pass try: log_fp.write(f"\nTIMEOUT: FreeCAD run exceeded {timeout} seconds and was killed.\n") log_fp.flush() except Exception: pass tail = "" try: tail = log_path.read_text(encoding="utf-8")[-8000:] except Exception: pass err = { "ok": False, "error_type": "TimeoutExpired", "error": f"FreeCAD job timed out after {timeout} seconds", "log_tail": tail, } try: result_path.write_text(json.dumps(err, indent=2, ensure_ascii=False), encoding="utf-8") except Exception: pass err["_job"] = { "id": job_dir.name, "dir": str(job_dir), "exit_code": None, "log_file": str(log_path), "compat_dxf": str(job_dir / "flat.dxf"), "compat_json": str(result_path), "svg_preview_url": f"/job/{job_dir.name}/dxf.svg", "dxf_download_url": f"/job/{job_dir.name}/flat.dxf", } return err # Normal completion path exit_code = proc.returncode if result_path.exists(): result = json.loads(result_path.read_text(encoding="utf-8")) else: tail = "" try: tail = log_path.read_text(encoding="utf-8")[-4000:] except Exception: pass result = { "ok": False, "error_type": "RunnerError", "error": "result.json not created", "log_tail": tail, } result["_job"] = { "id": job_dir.name, "dir": str(job_dir), "exit_code": exit_code, "log_file": str(log_path), "compat_dxf": str(job_dir / "flat.dxf"), "compat_json": str(result_path), "svg_preview_url": f"/job/{job_dir.name}/dxf.svg", "dxf_download_url": f"/job/{job_dir.name}/flat.dxf", } return result def dxf_to_svg_bytes(dxf_path: Path) -> bytes: doc = ezdxf.readfile(str(dxf_path)) msp = doc.modelspace() context = RenderContext(doc) backend = svg.SVGBackend() cfg = ezconfig.Configuration( background_policy=ezconfig.BackgroundPolicy.WHITE, color_policy=ezconfig.ColorPolicy.BLACK, ) frontend = Frontend(context, backend, config=cfg) frontend.draw_layout(msp) page = layout.Page(0, 0, layout.Units.mm, margins=layout.Margins.all(2)) try: settings = layout.Settings(scale=1, fit_page=True) svg_string = backend.get_string(page, settings=settings) except TypeError: svg_string = backend.get_string(page) return svg_string.encode("utf-8") @app.get("/") def index(): return render_template("index.html", materials=ALLOWED_MATERIALS) @app.post("/analyse") def analyse(): f = request.files.get("stepfile") material = (request.form.get("material", "stainless") or "stainless").strip().lower() thickness = (request.form.get("thickness_mm", "") or "").strip() if material not in {m for m, _ in ALLOWED_MATERIALS}: return "Invalid material", 400 if not f or not f.filename: return "Please upload a .step or .stp file", 400 filename = secure_filename(f.filename) if not filename.lower().endswith((".step", ".stp")): return "Please upload a .step or .stp file", 400 job_id = uuid.uuid4().hex[:12] job_dir = JOBS_DIR / job_id job_dir.mkdir(parents=True, exist_ok=True) step_dest = job_dir / filename f.save(str(step_dest)) thickness_mm = thickness if thickness else None result = run_job(job_dir, filename, material, thickness_mm) if not isinstance(result, dict): result = {"ok": False, "error_type": "InternalError", "error": "Unexpected result type"} return render_template("result.html", result=result) @app.get("/job//flat.dxf") def download_flat_dxf(job_id: str): job_dir = _safe_job_dir(job_id) dxf_path = (job_dir / "flat.dxf") if not dxf_path.exists(): abort(404) data = dxf_path.read_bytes() return Response( data, mimetype="application/dxf", headers={"Content-Disposition": f'inline; filename="{job_id}_flat.dxf"'}, ) @app.get("/job//dxf.svg") def preview_dxf_svg(job_id: str): job_dir = _safe_job_dir(job_id) dxf_path = (job_dir / "flat.dxf") if not dxf_path.exists(): abort(404) try: svg_bytes = dxf_to_svg_bytes(dxf_path) except Exception as e: msg = f"DXF->SVG failed: {type(e).__name__}: {e}" svg_fallback = f""" {msg} """ return Response(svg_fallback.encode("utf-8"), mimetype="image/svg+xml") return Response(svg_bytes, mimetype="image/svg+xml")