Files
pythonocc-step-editor/step_editor/scdm_edit_runner.py
T

2165 lines
82 KiB
Python
Raw Normal View History

from __future__ import annotations
import subprocess
from collections.abc import Mapping
from pathlib import Path
from typing import Callable
from .scdm_backend import ScdmBackendInfo, resolve_scdm_backend, save_scdm_backend_cache, scdm_run_script_command
from .scdm_capabilities import capability_definition
from .scdm_schema import default_scdm_work_dir, file_fingerprint, read_json, utc_now, write_json
SCDM_EDIT_SCHEMA_VERSION = 1
SCDM_EDIT_ADAPTER = "spaceclaim-v1"
def prepare_scdm_edit_job(
source_step: str | Path,
*,
capability_key: str,
target_value: object,
object_signature: Mapping[str, object] | None = None,
object_id: str = "",
backend_operation: str = "",
output_step: str | Path | None = None,
output_dir: str | Path | None = None,
project_root: str | Path | None = None,
backend: ScdmBackendInfo | None = None,
timeout_seconds: float = 180.0,
rollback_step: str | Path | None = None,
context: Mapping[str, object] | None = None,
) -> dict[str, object]:
source = Path(source_step).expanduser()
if not source.is_file():
return {"ok": False, "reason": "missing-step", "message": f"STEP file not found: {source}"}
definition = capability_definition(capability_key)
if definition is None:
return {
"ok": False,
"reason": "unsupported-capability",
"message": f"SCDM capability is not productized yet: {capability_key}",
}
if not definition.productized:
return {
"ok": False,
"reason": "capability-not-productized",
"message": definition.block_reason or f"SCDM capability is planned but not executable yet: {capability_key}",
"capability_key": definition.key,
"roadmap_stage": definition.roadmap_stage,
}
converted_target = _target_value_for_job(target_value, value_kind=definition.value_kind)
preflight = _preflight_scdm_edit_job(
definition.key,
converted_target,
object_signature if isinstance(object_signature, Mapping) else {},
)
if preflight.get("ok") is False:
return {
"ok": False,
"reason": str(preflight.get("reason") or "target-preflight-failed"),
"message": str(preflight.get("message") or "SCDM edit target is outside the supported range."),
"capability_key": definition.key,
"preflight": preflight,
}
fingerprint = file_fingerprint(source)
work_dir = _edit_work_dir(source, fingerprint=fingerprint, output_dir=output_dir, project_root=project_root).resolve(strict=False)
work_dir.mkdir(parents=True, exist_ok=True)
output_path = Path(output_step).expanduser().resolve(strict=False) if output_step else work_dir / "result.step"
result_path = work_dir / "result.json"
error_path = work_dir / "error.json"
job_path = work_dir / "scdm_edit_job.json"
script_path = work_dir / "scdm_edit.py"
operation = backend_operation or definition.backend_operation
_clear_stale_outputs((output_path, result_path, error_path))
job = {
"schemaVersion": SCDM_EDIT_SCHEMA_VERSION,
"adapter": SCDM_EDIT_ADAPTER,
"createdAt": utc_now(),
"backend": {
"name": "SCDM",
"path": str(backend.path) if backend else "",
"version": backend.version if backend else "",
},
"model": {
"sourceStep": str(source.resolve(strict=False)),
"rollbackStep": str(Path(rollback_step).expanduser().resolve(strict=False)) if rollback_step else str(source.resolve(strict=False)),
"fingerprint": fingerprint,
},
"object": {
"objectId": object_id,
"geometrySignature": _json_safe(dict(object_signature or {})),
},
"target": {
"capabilityKey": definition.key,
"displayName": definition.display_name,
"valueKind": definition.value_kind,
"value": converted_target,
"text": str(target_value),
"backendOperation": operation,
"postCheck": definition.post_check,
},
"execution": {
"mode": "SpaceClaim.exe /RunScript",
"timeoutSeconds": max(float(timeout_seconds), 0.1),
"isolatedProcess": True,
},
"outputs": {
"outputStep": str(output_path),
"result": str(result_path),
"error": str(error_path),
},
"context": _json_safe(dict(context or {})),
}
write_json(job_path, job)
script_path.write_text(generate_scdm_edit_script(job_path), encoding="utf-8")
return {
"ok": True,
"reason": "ok",
"work_dir": str(work_dir),
"job_path": str(job_path),
"script_path": str(script_path),
"output_step": str(output_path),
"result_path": str(result_path),
"error_path": str(error_path),
"model_fingerprint": fingerprint,
"capability_key": definition.key,
"backend_operation": operation,
}
def run_scdm_edit_job(
source_step: str | Path,
*,
capability_key: str,
target_value: object,
object_signature: Mapping[str, object] | None = None,
object_id: str = "",
backend_operation: str = "",
output_step: str | Path | None = None,
output_dir: str | Path | None = None,
project_root: str | Path | None = None,
backend: ScdmBackendInfo | None = None,
timeout_seconds: float = 180.0,
rollback_step: str | Path | None = None,
context: Mapping[str, object] | None = None,
runner: Callable[..., subprocess.CompletedProcess[str]] | None = None,
) -> dict[str, object]:
if backend is None:
resolved = resolve_scdm_backend(project_root_override=project_root, validate=False)
if not resolved.get("ok") or not isinstance(resolved.get("backend"), ScdmBackendInfo):
return {
"ok": False,
"reason": str(resolved.get("reason") or "missing-scdm"),
"message": str(resolved.get("message") or "SCDM backend is not available."),
}
backend = resolved["backend"] # type: ignore[assignment]
prepared = prepare_scdm_edit_job(
source_step,
capability_key=capability_key,
target_value=target_value,
object_signature=object_signature,
object_id=object_id,
backend_operation=backend_operation,
output_step=output_step,
output_dir=output_dir,
project_root=project_root,
backend=backend,
timeout_seconds=timeout_seconds,
rollback_step=rollback_step,
context=context,
)
if not prepared.get("ok"):
return prepared
result = run_prepared_scdm_edit_job(prepared, backend=backend, timeout_seconds=timeout_seconds, runner=runner)
if result.get("ok") is True:
try:
save_scdm_backend_cache(_edit_verified_backend(backend), project_root_override=project_root)
except Exception:
pass
return result
def run_prepared_scdm_edit_job(
prepared: Mapping[str, object],
*,
backend: ScdmBackendInfo,
timeout_seconds: float = 180.0,
runner: Callable[..., subprocess.CompletedProcess[str]] | None = None,
) -> dict[str, object]:
script_path = Path(str(prepared.get("script_path") or "")).expanduser()
result_path = Path(str(prepared.get("result_path") or "")).expanduser()
error_path = Path(str(prepared.get("error_path") or "")).expanduser()
output_step = Path(str(prepared.get("output_step") or "")).expanduser()
if not script_path.is_file():
return _with_prepared(
prepared,
ok=False,
reason="missing-script",
message=f"SCDM edit script not found: {script_path}",
backend=backend.to_cache(),
)
command = scdm_run_script_command(backend.path, script_path)
run = runner or subprocess.run
try:
completed = run(
command,
cwd=str(script_path.parent),
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
timeout=max(float(timeout_seconds), 0.1),
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
check=False,
)
except subprocess.TimeoutExpired:
_write_error(
error_path,
reason="timeout",
message="SCDM edit timed out.",
prepared=prepared,
)
return _with_prepared(prepared, ok=False, reason="timeout", message="SCDM edit timed out.")
except OSError as exc:
_write_error(
error_path,
reason="launch-failed",
message=str(exc),
prepared=prepared,
)
return _with_prepared(prepared, ok=False, reason="launch-failed", message=str(exc))
return _result_from_completed(completed, prepared=prepared, result_path=result_path, error_path=error_path, output_step=output_step, backend=backend)
def generate_scdm_edit_script(job_path: str | Path) -> str:
job_literal = repr(str(Path(job_path).expanduser()))
return _SCDM_EDIT_SCRIPT_TEMPLATE.replace("__STEP_EDITOR_SCDM_EDIT_JOB_PATH__", job_literal)
def _edit_work_dir(
source: Path,
*,
fingerprint: str,
output_dir: str | Path | None,
project_root: str | Path | None,
) -> Path:
if output_dir:
return Path(output_dir).expanduser()
return default_scdm_work_dir(source, project_root=project_root, fingerprint=fingerprint) / "edit"
def _clear_stale_outputs(paths: tuple[Path, ...]) -> None:
for path in paths:
try:
if path.exists():
path.unlink()
except OSError:
pass
def _target_value_for_job(value: object, *, value_kind: str) -> object:
if value_kind == "vector3":
parsed = _parse_vector3(value)
return parsed if parsed else _json_safe(value)
if value_kind in {"number", "length", "angle"}:
try:
return float(str(value).strip())
except (TypeError, ValueError):
return _json_safe(value)
if value_kind == "command":
return True if value in ("", None) else _json_safe(value)
return _json_safe(value)
def _preflight_scdm_edit_job(
capability_key: str,
target_value: object,
object_signature: Mapping[str, object],
) -> dict[str, object]:
if capability_key not in {"pattern.spacing", "pattern.segment_spacing"}:
return {"ok": True, "reason": "ok"}
if _is_body_pattern_signature(object_signature):
if not _body_pattern_has_component_locators(object_signature):
return {
"ok": False,
"reason": "body-pattern-spacing-missing-component-locators",
"message": (
"SCDM 已识别到实体/组件阵列间距,但缓存里没有每个阵列成员的组件实例定位。"
"已阻止执行,避免把共享实体定义或整列装配一起移走。"
),
}
try:
target = float(str(target_value).strip())
except (TypeError, ValueError):
return {"ok": True, "reason": "ok"}
if target <= 0:
return {
"ok": False,
"reason": "target-value-illegal",
"message": "目标阵列间距必须大于 0。",
}
fit = object_signature.get("supportPatternFit")
if not isinstance(fit, Mapping):
return {"ok": True, "reason": "ok"}
max_key = "maxSegmentSpacing" if capability_key == "pattern.segment_spacing" else "maxSpacing"
max_local_key = "maxSegmentSpacingLocal" if capability_key == "pattern.segment_spacing" else "maxSpacingLocal"
max_spacing = _float_or_none(fit.get(max_key))
if max_spacing is None or max_spacing <= 0:
return {"ok": True, "reason": "ok"}
tolerance = max(abs(max_spacing) * 1.0e-6, 1.0e-12)
if target <= max_spacing + tolerance:
return {"ok": True, "reason": "ok"}
max_local = _float_or_none(fit.get(max_local_key))
unit_scale = _float_or_none(fit.get("localUnitScale"))
if max_local is None and unit_scale is not None and unit_scale > 0:
max_local = max_spacing / unit_scale
target_local = target
if unit_scale is not None and unit_scale > 0:
target_local = target / unit_scale
support_ids = fit.get("supportFaceIds")
label = "局部阵列间距" if capability_key == "pattern.segment_spacing" else "阵列间距"
max_label = "该段最大安全间距" if capability_key == "pattern.segment_spacing" else "保持阵列中心不变时最大安全间距"
return {
"ok": False,
"reason": "pattern-spacing-exceeds-support",
"message": (
f"目标{label} {target_local:g} 会超出支撑面范围;"
f"{max_label} {max_local if max_local is not None else max_spacing:g}。"
f"支撑面 Face: {support_ids or 'unknown'}。"
),
"targetSpacing": target,
"targetSpacingLocal": target_local,
max_key: max_spacing,
max_local_key: max_local,
"supportFaceIds": support_ids,
}
def _float_or_none(value: object) -> float | None:
try:
return float(str(value).strip())
except (TypeError, ValueError):
return None
def _int_or_none_value(value: object) -> int | None:
try:
return int(value)
except (TypeError, ValueError):
return None
def _is_body_pattern_signature(signature: Mapping[str, object]) -> bool:
pattern_kind = str(signature.get("patternKind") or "").strip().lower()
instance_kind = str(signature.get("instanceKind") or "").strip().lower()
if pattern_kind == "body" or instance_kind in {"body", "part", "component"}:
return True
body_indices = signature.get("bodyIndices")
if isinstance(body_indices, (list, tuple)) and body_indices:
return True
instances = signature.get("patternInstances")
if isinstance(instances, (list, tuple)):
for item in instances:
if not isinstance(item, Mapping):
continue
item_kind = str(item.get("instanceKind") or "").strip().lower()
if item_kind in {"body", "part", "component"}:
return True
locators = item.get("bodyLocators")
if isinstance(locators, (list, tuple)) and locators:
return True
return False
def _body_pattern_has_component_locators(signature: Mapping[str, object]) -> bool:
instances = signature.get("patternInstances")
if not isinstance(instances, (list, tuple)) or len(instances) < 3:
return False
seen_paths: set[tuple[int, ...]] = set()
for item in instances:
if not isinstance(item, Mapping):
return False
locators = item.get("componentLocators")
if not isinstance(locators, (list, tuple)):
locators = item.get("bodyLocators")
path = _first_component_path(locators)
if not path:
return False
if path in seen_paths:
return False
seen_paths.add(path)
return True
def _first_component_path(value: object) -> tuple[int, ...]:
if not isinstance(value, (list, tuple)):
return ()
for locator in value:
if not isinstance(locator, Mapping):
continue
raw_path = locator.get("componentPath")
if isinstance(raw_path, (list, tuple)):
try:
path = tuple(int(item) for item in raw_path)
except (TypeError, ValueError):
path = ()
if path:
return path
component_index = _int_or_none_value(locator.get("componentIndex"))
if component_index is not None:
return (component_index,)
return ()
def _parse_vector3(value: object) -> list[float]:
if isinstance(value, (list, tuple)):
items = list(value)
elif isinstance(value, str):
text = value.strip().strip("()[]")
if not text:
return []
items = text.replace(",", " ").split()
else:
return []
if len(items) != 3:
return []
try:
return [float(item) for item in items]
except (TypeError, ValueError):
return []
def _json_safe(value: object) -> object:
if isinstance(value, Path):
return str(value)
if isinstance(value, Mapping):
return {str(key): _json_safe(item) for key, item in value.items()}
if isinstance(value, (list, tuple)):
return [_json_safe(item) for item in value]
if isinstance(value, (str, int, float, bool)) or value is None:
return value
return str(value)
def _result_from_completed(
completed: subprocess.CompletedProcess[str],
*,
prepared: Mapping[str, object],
result_path: Path,
error_path: Path,
output_step: Path,
backend: ScdmBackendInfo,
) -> dict[str, object]:
returncode = int(getattr(completed, "returncode", -1))
stdout = str(getattr(completed, "stdout", "") or "")
stderr = str(getattr(completed, "stderr", "") or "")
if error_path.is_file():
error = _read_json_or_message(error_path)
reason = str(error.get("reason") or "scdm-edit-failed")
message = str(error.get("message") or stderr or stdout or "SCDM edit failed.")
return {
**dict(prepared),
"ok": False,
"reason": reason,
"message": message,
"returncode": returncode,
"stdout": stdout,
"stderr": stderr,
"error": error,
"backend": backend.to_cache(),
}
if returncode != 0:
message = (stderr or stdout or f"SCDM returned {returncode}.").strip()
_write_error(error_path, reason="edit-failed", message=message, prepared=prepared, returncode=returncode)
return {
**dict(prepared),
"ok": False,
"reason": "edit-failed",
"message": message,
"returncode": returncode,
"stdout": stdout,
"stderr": stderr,
"backend": backend.to_cache(),
}
if not result_path.is_file():
message = "SCDM edit finished but did not write result.json."
_write_error(error_path, reason="missing-result", message=message, prepared=prepared, returncode=returncode)
return _with_prepared(prepared, ok=False, reason="missing-result", message=message, returncode=returncode, backend=backend.to_cache())
result = _read_json_or_message(result_path)
if result.get("ok") is not True:
reason = str(result.get("reason") or "negative-result")
message = str(result.get("message") or "SCDM edit result was not successful.")
return _with_prepared(prepared, ok=False, reason=reason, message=message, returncode=returncode, result=result, backend=backend.to_cache())
if not output_step.is_file():
message = f"SCDM edit reported success but output STEP was not written: {output_step}"
_write_error(error_path, reason="missing-output-step", message=message, prepared=prepared, returncode=returncode)
return {
**dict(prepared),
"ok": False,
"reason": "missing-output-step",
"message": message,
"returncode": returncode,
"result": result,
"backend": backend.to_cache(),
}
try:
output_size = output_step.stat().st_size
except OSError:
output_size = -1
if output_size <= 0:
message = f"SCDM edit reported success but output STEP is empty: {output_step}"
_write_error(error_path, reason="empty-output-step", message=message, prepared=prepared, returncode=returncode)
return {
**dict(prepared),
"ok": False,
"reason": "empty-output-step",
"message": message,
"returncode": returncode,
"result": result,
"backend": backend.to_cache(),
}
return {
**dict(prepared),
"ok": True,
"reason": "ok",
"message": str(result.get("message") or "SCDM edit finished."),
"returncode": returncode,
"stdout": stdout,
"stderr": stderr,
"result": result,
"output_step": str(output_step),
"backend": _edit_verified_backend(backend).to_cache(),
}
def _edit_verified_backend(backend: ScdmBackendInfo) -> ScdmBackendInfo:
return ScdmBackendInfo(
path=backend.path,
source=backend.source,
version=backend.version,
verified_at=utc_now(),
run_script_ok=True,
license_ok=True,
message="SCDM edit completed.",
)
def _with_prepared(prepared: Mapping[str, object], **payload: object) -> dict[str, object]:
result = dict(prepared)
result.update(payload)
return result
def _write_error(path: Path, *, reason: str, message: str, prepared: Mapping[str, object], **extra: object) -> None:
payload = {
"ok": False,
"reason": reason,
"message": message,
"createdAt": utc_now(),
"jobPath": str(prepared.get("job_path") or ""),
"scriptPath": str(prepared.get("script_path") or ""),
"outputStep": str(prepared.get("output_step") or ""),
}
payload.update(extra)
write_json(path, payload)
def _read_json_or_message(path: Path) -> dict[str, object]:
try:
return read_json(path)
except Exception as exc:
return {"ok": False, "reason": "bad-json", "message": str(exc)}
_SCDM_EDIT_SCRIPT_TEMPLATE = r'''
from __future__ import print_function
import json
import traceback
JOB_PATH = __STEP_EDITOR_SCDM_EDIT_JOB_PATH__
def _write_json(path, payload):
handle = open(path, 'w')
try:
handle.write(json.dumps(payload, indent=2))
finally:
handle.close()
def _read_job():
handle = open(JOB_PATH, 'r')
try:
return json.loads(handle.read())
finally:
handle.close()
def _items(collection):
if collection is None:
return []
try:
return list(collection)
except Exception:
result = []
try:
count = int(collection.Count)
for index in range(count):
result.append(collection[index])
except Exception:
pass
return result
def _maybe_call(target, name, *args):
try:
func = getattr(target, name)
except Exception:
return None
try:
return func(*args)
except Exception:
return None
def _open_step(path):
errors = []
document_open = globals().get('DocumentOpen')
if document_open is not None:
for args in ((path,),):
try:
return document_open.Execute(*args)
except Exception as exc:
errors.append('DocumentOpen.Execute: ' + str(exc))
application = globals().get('Application')
if application is not None:
for name in ('OpenDocument', 'OpenFile'):
try:
return getattr(application, name)(path)
except Exception as exc:
errors.append('Application.' + name + ': ' + str(exc))
raise Exception('open_step_failed: ' + '; '.join(errors))
def _root_part():
get_root_part = globals().get('GetRootPart')
if get_root_part is not None:
try:
return get_root_part()
except Exception:
pass
application = globals().get('Application')
if application is not None:
for expr in ('ActiveWindow.Document.MainPart', 'ActiveDocument.MainPart', 'Document.MainPart'):
value = application
ok = True
for name in expr.split('.'):
try:
value = getattr(value, name)
except Exception:
ok = False
break
if ok and value is not None:
return value
raise Exception('root_part_not_found')
def _bodies(root):
for name in ('Bodies', 'GetAllBodies'):
value = _maybe_call(root, name)
if value is None:
try:
value = getattr(root, name)
except Exception:
value = None
items = _items(value)
if items:
return items
return []
def _immediate_components(part):
for name in ('Components',):
value = _maybe_call(part, name)
if value is None:
try:
value = getattr(part, name)
except Exception:
value = None
items = _items(value)
if items:
return items
return []
def _component_content(component):
for name in ('Content', 'ContentMaster', 'Template', 'Part'):
try:
value = getattr(component, name)
if value is not None:
return value
except Exception:
pass
return None
def _component_entries(root):
result = []
queue = [(root, [])]
while queue:
part, path = queue.pop(0)
if part is None or len(path) > 8:
continue
components = _immediate_components(part)
for child_index, component in enumerate(components):
component_path = list(path) + [child_index]
content = _component_content(component)
result.append({
'component': component,
'content': content,
'componentIndex': len(result),
'componentPath': component_path,
})
if content is not None:
queue.append((content, component_path))
return result
def _component_path(locator):
raw = locator.get('componentPath') if isinstance(locator, dict) else None
if isinstance(raw, list) and raw:
result = []
for item in raw:
number = _int_or_none(item)
if number is None:
return []
result.append(number)
return result
return []
def _locate_one_component(root, locator):
if not isinstance(locator, dict):
return None
entries = _component_entries(root)
wanted_path = _component_path(locator)
if wanted_path:
for entry in entries:
if entry.get('componentPath') == wanted_path:
return entry.get('component')
wanted_index = _int_or_none(locator.get('componentIndex'))
if wanted_index is not None and 0 <= wanted_index < len(entries):
return entries[wanted_index].get('component')
return None
def _faces(body):
for name in ('Faces', 'GetFaces'):
value = _maybe_call(body, name)
if value is None:
try:
value = getattr(body, name)
except Exception:
value = None
items = _items(value)
if items:
return items
return []
def _flat_faces(bodies):
result = []
for body in bodies:
result.extend(_faces(body))
return result
def _locate_one_body(bodies, body_index):
body_index = _int_or_none(body_index)
if body_index is not None and 0 <= body_index < len(bodies):
return bodies[body_index]
return None
def _int_or_none(value):
try:
return int(value)
except Exception:
return None
def _append_unique_face(result, face):
for existing in result:
try:
if existing == face:
return
except Exception:
pass
try:
if existing is face:
return
except Exception:
pass
result.append(face)
def _locate_one_face(bodies, body_index, face_ordinal, global_face_ordinal):
if body_index is not None and face_ordinal is not None and 0 <= body_index < len(bodies):
faces = _faces(bodies[body_index])
if 0 <= face_ordinal < len(faces):
return faces[face_ordinal]
flat_ordinal = global_face_ordinal if global_face_ordinal is not None else face_ordinal
if flat_ordinal is not None:
flat = _flat_faces(bodies)
if 0 <= flat_ordinal < len(flat):
return flat[flat_ordinal]
return None
def _locate_faces(signature):
root = _root_part()
bodies = _bodies(root)
body_index = _int_or_none(signature.get('bodyIndex'))
face_ordinal = _int_or_none(signature.get('faceOrdinal'))
global_face_ordinal = _int_or_none(signature.get('globalFaceOrdinal'))
result = []
locators = signature.get('scdmFaceLocators')
if isinstance(locators, list):
for locator in locators:
if not isinstance(locator, dict):
continue
face = _locate_one_face(
bodies,
_int_or_none(locator.get('bodyIndex')),
_int_or_none(locator.get('faceOrdinal')),
_int_or_none(locator.get('globalFaceOrdinal')),
)
if face is not None:
_append_unique_face(result, face)
face_ordinals = signature.get('faceOrdinals')
if isinstance(face_ordinals, list) and body_index is not None:
for item in face_ordinals:
face = _locate_one_face(bodies, body_index, _int_or_none(item), None)
if face is not None:
_append_unique_face(result, face)
global_face_ordinals = signature.get('globalFaceOrdinals')
if isinstance(global_face_ordinals, list):
for item in global_face_ordinals:
face = _locate_one_face(bodies, None, None, _int_or_none(item))
if face is not None:
_append_unique_face(result, face)
if result:
return result
face = _locate_one_face(bodies, body_index, face_ordinal, global_face_ordinal)
if face is not None:
return [face]
raise Exception('object_not_found: geometrySignature does not locate a unique SCDM face')
def _locate_named_faces(signature, locator_keys, face_ordinal_keys, global_face_ordinal_keys):
root = _root_part()
bodies = _bodies(root)
body_index = _int_or_none(signature.get('bodyIndex'))
result = []
for key in locator_keys:
locators = signature.get(key)
if not isinstance(locators, list):
continue
for locator in locators:
if not isinstance(locator, dict):
continue
face = _locate_one_face(
bodies,
_int_or_none(locator.get('bodyIndex')),
_int_or_none(locator.get('faceOrdinal')),
_int_or_none(locator.get('globalFaceOrdinal')),
)
if face is not None:
_append_unique_face(result, face)
for key in face_ordinal_keys:
ordinals = signature.get(key)
if not isinstance(ordinals, list) or body_index is None:
continue
for item in ordinals:
face = _locate_one_face(bodies, body_index, _int_or_none(item), None)
if face is not None:
_append_unique_face(result, face)
for key in global_face_ordinal_keys:
ordinals = signature.get(key)
if not isinstance(ordinals, list):
continue
for item in ordinals:
face = _locate_one_face(bodies, None, None, _int_or_none(item))
if face is not None:
_append_unique_face(result, face)
return result
def _locate_height_faces(signature):
faces = _locate_named_faces(
signature,
('heightFaceLocators', 'topFaceLocators'),
('heightFaceOrdinals', 'topFaceOrdinals'),
('globalHeightFaceOrdinals', 'globalTopFaceOrdinals'),
)
if faces:
return faces
raise Exception('object_signature_missing_height_face_locator')
def _locate_depth_faces(signature):
faces = _locate_named_faces(
signature,
('depthFaceLocators', 'bottomFaceLocators'),
('depthFaceOrdinals', 'bottomFaceOrdinals'),
('globalDepthFaceOrdinals', 'globalBottomFaceOrdinals'),
)
if faces:
return faces
raise Exception('object_signature_missing_depth_face_locator')
def _locate_pattern_instance_faces(instance, fallback_body_index):
signature = dict(instance)
if signature.get('bodyIndex') is None and fallback_body_index is not None:
signature['bodyIndex'] = fallback_body_index
instance_kind = str(signature.get('instanceKind') or '').lower()
root = _root_part()
bodies = _bodies(root)
if instance_kind in ('body', 'part', 'component'):
body_locators = signature.get('bodyLocators')
if isinstance(body_locators, list):
for locator in body_locators:
if not isinstance(locator, dict):
continue
body = _locate_one_body(bodies, locator.get('bodyIndex'))
if body is not None:
return [body]
body = _locate_one_body(bodies, signature.get('bodyIndex'))
if body is not None:
return [body]
try:
return _locate_faces(signature)
except Exception:
body = _locate_one_body(bodies, signature.get('bodyIndex'))
if body is not None and instance_kind in ('body', 'part', 'component'):
return [body]
raise
def _locate_pattern_instance_items(instance, fallback_body_index):
signature = dict(instance)
if signature.get('bodyIndex') is None and fallback_body_index is not None:
signature['bodyIndex'] = fallback_body_index
root = _root_part()
component_locators = signature.get('componentLocators')
if not isinstance(component_locators, list):
component_locators = signature.get('bodyLocators')
if isinstance(component_locators, list):
components = []
for locator in component_locators:
component = _locate_one_component(root, locator)
if component is not None:
components.append(component)
if len(components) == 1:
return {'kind': 'component', 'items': components}
if len(components) > 1:
raise Exception('object_not_unique: pattern instance resolves to multiple SCDM components')
return {'kind': 'face', 'items': _locate_pattern_instance_faces(signature, fallback_body_index)}
def _locate_diameter_faces(signature):
faces = _locate_named_faces(
signature,
('diameterFaceLocators', 'sideFaceLocators'),
('diameterFaceOrdinals', 'sideFaceOrdinals'),
('globalDiameterFaceOrdinals', 'globalSideFaceOrdinals'),
)
if faces:
return faces
raise Exception('object_signature_missing_diameter_face_locator')
def _selection(items):
selection = globals().get('Selection')
if selection is not None:
try:
return selection.Create(items)
except Exception:
pass
if len(items) == 1:
try:
return selection.Create(items[0])
except Exception:
pass
return items[0] if len(items) == 1 else items
def _import_attr(module_name, attr_name):
try:
module = __import__(module_name, fromlist=[attr_name])
return getattr(module, attr_name)
except Exception:
return None
def _api_versions():
return ('V23', 'V22', 'V21', 'V20', 'V19', 'V18', 'V17', 'V16')
def _command_type(name):
value = globals().get(name)
if value is not None:
return value
for version in _api_versions():
value = _import_attr('SpaceClaim.Api.' + version + '.Scripting.Commands', name)
if value is not None:
return value
return None
def _command_option_type(name):
value = globals().get(name)
if value is not None:
return value
for version in _api_versions():
value = _import_attr('SpaceClaim.Api.' + version + '.Scripting.Commands.CommandOptions', name)
if value is not None:
return value
return None
def _selection_type():
value = globals().get('Selection')
if value is not None:
return value
for version in _api_versions():
value = _import_attr('SpaceClaim.Api.' + version + '.Scripting.Selection', 'Selection')
if value is not None:
return value
return None
def _geometry_type(name):
value = globals().get(name)
if value is not None:
return value
for version in _api_versions():
value = _import_attr('SpaceClaim.Api.' + version + '.Geometry', name)
if value is not None:
return value
return None
def _new_options(name):
cls = _command_option_type(name)
if cls is None:
return None
try:
options = cls()
except Exception:
return None
for attr, value in (('Copy', False), ('DetachFirst', False), ('CreatePatterns', False)):
try:
if hasattr(options, attr):
setattr(options, attr, value)
except Exception:
pass
return options
def _empty_selection():
selection = _selection_type()
if selection is None:
return None
for name in ('Empty', 'Create'):
try:
return getattr(selection, name)()
except Exception:
pass
return None
def _fill_mode_three_d():
fill_mode = _command_type('FillMode')
if fill_mode is None:
return None
for name in ('ThreeD', 'ThreeDMode'):
try:
return getattr(fill_mode, name)
except Exception:
pass
return None
def _make_vector(x, y, z):
vector_type = _geometry_type('Vector')
if vector_type is None:
return None
for maker in ('Create', 'CreateVector'):
try:
return getattr(vector_type, maker)(x, y, z)
except Exception:
pass
try:
return vector_type(x, y, z)
except Exception:
return None
def _make_translation_matrix(delta):
matrix_type = _geometry_type('Matrix')
if matrix_type is None:
return None
vector = _make_vector(delta[0], delta[1], delta[2])
if vector is None:
return None
try:
return matrix_type.CreateTranslation(vector)
except Exception:
return None
def _translate_component_occurrence(component, delta):
matrix = _make_translation_matrix(delta)
if matrix is None:
raise Exception('capability_not_implemented: Matrix.CreateTranslation command not available')
try:
component.Transform(matrix)
return {'command': 'Component.Transform', 'delta': delta, 'apiSignature': 'Component.Transform(Matrix.CreateTranslation)'}
except Exception as exc:
raise Exception('capability_not_implemented: Component.Transform failed; ' + str(exc))
def _make_direction(x, y, z):
direction_type = _geometry_type('Direction')
if direction_type is None:
return None
for maker in ('Create', 'CreateDirection'):
try:
return getattr(direction_type, maker)(x, y, z)
except Exception:
pass
try:
return direction_type(x, y, z)
except Exception:
return None
def _vector_length(values):
return (values[0] * values[0] + values[1] * values[1] + values[2] * values[2]) ** 0.5
def _unit_direction_and_distance(values):
distance = _vector_length(values)
if distance <= 1e-12:
return None, 0.0
direction = _make_direction(values[0] / distance, values[1] / distance, values[2] / distance)
return direction, distance
def _signature_direction(signature):
for key in ('normal', 'axis'):
value = signature.get(key)
if isinstance(value, list) and len(value) == 3:
try:
direction = _make_direction(float(value[0]), float(value[1]), float(value[2]))
if direction is not None:
return direction
except Exception:
pass
return None
def _call_variants(label, func, variants):
errors = []
for args in variants:
if args is None:
continue
try:
result = func(*args)
return {'ok': True, 'result': result, 'signature': label + '(' + str(len(args)) + ' args)'}
except Exception as exc:
errors.append(label + '(' + str(len(args)) + ' args): ' + str(exc))
return {'ok': False, 'errors': errors}
def _offset_faces(selection, distance, signature):
offset_faces = _command_type('OffsetFaces')
if offset_faces is None:
raise Exception('capability_not_implemented: OffsetFaces command not available')
options = _new_options('OffsetFaceOptions')
direction = _signature_direction(signature)
variants = []
if direction is not None:
variants.append((selection, distance, direction, options, None))
variants.append((selection, distance, direction, options))
variants.append((selection, distance, options, None))
variants.append((selection, distance, options))
result = _call_variants('OffsetFaces.Execute', offset_faces.Execute, variants)
if result.get('ok'):
return {'command': 'OffsetFaces.Execute', 'distance': distance, 'apiSignature': result.get('signature')}
raise Exception('capability_not_implemented: OffsetFaces.Execute failed; ' + '; '.join(result.get('errors') or []))
def _translate_selection(selection, delta):
move = _command_type('Move')
if move is None:
raise Exception('capability_not_implemented: Move command not available')
options = _new_options('MoveOptions')
vector = _make_vector(delta[0], delta[1], delta[2])
variants = []
if vector is not None:
variants.append((selection, vector, options, None))
variants.append((selection, vector, options))
direction, distance = _unit_direction_and_distance(delta)
if direction is not None:
variants.append((selection, direction, distance, options, None))
variants.append((selection, direction, distance, options))
if vector is not None:
variants.append((selection, vector))
result = _call_variants('Move.Translate', move.Translate, variants)
if result.get('ok'):
return {'command': 'Move.Translate', 'delta': delta, 'apiSignature': result.get('signature')}
execute = getattr(move, 'Execute', None)
if execute is not None and vector is not None:
fallback = _call_variants('Move.Execute', execute, ((selection, vector, options), (selection, vector)))
if fallback.get('ok'):
return {'command': 'Move.Execute', 'delta': delta, 'apiSignature': fallback.get('signature')}
result['errors'].extend(fallback.get('errors') or [])
raise Exception('capability_not_implemented: Move.Translate failed; ' + '; '.join(result.get('errors') or []))
def _fill_selection(selection):
fill = _command_type('Fill')
if fill is None:
raise Exception('capability_not_implemented: Fill command not available')
options = _new_options('FillOptions')
for attr, value in (('AutoExtendFillArea', True), ('PatchBlend', True), ('ZipLaminarEdges', True)):
try:
if options is not None and hasattr(options, attr):
setattr(options, attr, value)
except Exception:
pass
secondary = _empty_selection()
mode = _fill_mode_three_d()
variants = []
if mode is not None:
variants.append((selection, secondary, options, mode, None))
variants.append((selection, secondary, options, mode))
variants.append((selection, None, options, mode, None))
variants.append((selection, None, options, mode))
variants.append((selection, secondary, options, None, None))
variants.append((selection, secondary, options, None))
variants.append((selection, options, None))
variants.append((selection, options))
variants.append((selection, None))
variants.append((selection,))
result = _call_variants('Fill.Execute', fill.Execute, variants)
if result.get('ok'):
return {'command': 'Fill.Execute', 'apiSignature': result.get('signature')}
delete = _command_type('Delete')
if delete is not None:
fallback = _call_variants('Delete.Execute', delete.Execute, ((selection,),))
if fallback.get('ok'):
return {'command': 'Delete.Execute', 'apiSignature': fallback.get('signature'), 'fillErrors': result.get('errors') or []}
result['errors'].extend(fallback.get('errors') or [])
raise Exception('capability_not_implemented: Fill.Execute failed; ' + '; '.join(result.get('errors') or []))
def _float_value(value):
try:
return float(value)
except Exception:
raise Exception('target_value_not_numeric: ' + str(value))
def _vector3(value):
if not isinstance(value, list) or len(value) != 3:
raise Exception('target_value_not_vector3: ' + str(value))
return [float(value[0]), float(value[1]), float(value[2])]
def _save_step(path):
errors = []
for name in ('DocumentSave', 'DocumentSaveAs'):
command = globals().get(name)
if command is None:
command = _command_type(name)
if command is None:
continue
try:
command.Execute(path)
return {'command': name + '.Execute'}
except Exception as exc:
errors.append(name + '.Execute: ' + str(exc))
application = globals().get('Application')
if application is not None:
for expr in ('ActiveWindow.Document', 'ActiveDocument', 'Document'):
document = application
ok = True
for name in expr.split('.'):
try:
document = getattr(document, name)
except Exception:
ok = False
break
if not ok or document is None:
continue
for method in ('SaveAs', 'SaveCopyAs', 'Export'):
try:
getattr(document, method)(path)
return {'command': expr + '.' + method}
except Exception as exc:
errors.append(expr + '.' + method + ': ' + str(exc))
raise Exception('save_step_failed: ' + '; '.join(errors))
def change_hole_diameter(job):
target = _float_value(job['target']['value'])
if target <= 0:
raise Exception('target_value_illegal: diameter must be positive')
radius = target / 2.0
signature = job['object'].get('geometrySignature') or {}
faces = _locate_faces(signature)
selection = _selection(faces)
errors = []
standard_holes = _command_type('StandardHoles')
if standard_holes is not None:
for args in ((selection, radius, None), (selection, radius), (faces, radius, None), (faces, radius)):
try:
standard_holes.ModifyHoleRadius(*args)
return {'command': 'StandardHoles.ModifyHoleRadius', 'targetRadius': radius, 'apiSignature': str(len(args)) + ' args'}
except Exception as exc:
errors.append('StandardHoles.ModifyHoleRadius: ' + str(exc))
current_radius = None
for key in ('radius',):
try:
current_radius = float(signature.get(key))
break
except Exception:
pass
if current_radius is None:
try:
current_radius = float(signature.get('diameter')) / 2.0
except Exception:
current_radius = None
if current_radius is not None:
# For internal cylindrical hole walls, SpaceClaim's positive face offset
# follows the face normal into the void and reduces the measured radius.
radial_delta = current_radius - radius
if abs(radial_delta) <= 1e-12:
return {'command': 'noop', 'targetRadius': radius, 'reason': 'target already reached'}
try:
applied = _offset_faces(selection, radial_delta, signature)
applied['targetRadius'] = radius
applied['radialDelta'] = radial_delta
applied['standardHoleErrors'] = errors
return applied
except Exception as exc:
errors.append('OffsetFaces.Execute: ' + str(exc))
raise Exception('capability_not_implemented: hole.diameter adapter failed; ' + '; '.join(errors))
def move_hole_axis(job):
target = _vector3(job['target']['value'])
signature = job['object'].get('geometrySignature') or {}
center = signature.get('center') or []
if not isinstance(center, list) or len(center) != 3:
raise Exception('object_signature_missing_center')
delta = [float(target[index]) - float(center[index]) for index in range(3)]
if _vector_length(delta) <= 1e-12:
return {'command': 'noop', 'delta': delta, 'reason': 'target already reached'}
faces = _locate_faces(signature)
selection = _selection(faces)
try:
return _translate_selection(selection, delta)
except Exception as exc:
raise Exception('capability_not_implemented: hole.position adapter failed; ' + str(exc))
def move_slot(job):
target = _vector3(job['target']['value'])
signature = job['object'].get('geometrySignature') or {}
center = signature.get('center') or signature.get('axisCenter') or []
if not isinstance(center, list) or len(center) != 3:
raise Exception('object_signature_missing_center')
delta = [float(target[index]) - float(center[index]) for index in range(3)]
if _vector_length(delta) <= 1e-12:
return {'command': 'noop', 'delta': delta, 'reason': 'target already reached'}
faces = _locate_faces(signature)
selection = _selection(faces)
try:
applied = _translate_selection(selection, delta)
applied['targetCenter'] = target
return applied
except Exception as exc:
raise Exception('capability_not_implemented: slot.position adapter failed; ' + str(exc))
def change_slot_width(job):
target = _float_value(job['target']['value'])
if target <= 0:
raise Exception('target_value_illegal: slot width must be positive')
signature = job['object'].get('geometrySignature') or {}
current_width = None
try:
current_width = float(signature.get('width'))
except Exception:
current_width = None
if current_width is None:
try:
current_width = float(signature.get('diameter'))
except Exception:
current_width = None
if current_width is None:
try:
current_width = float(signature.get('radius')) * 2.0
except Exception:
current_width = None
if current_width is None or current_width <= 0:
raise Exception('object_signature_missing_width')
# Open slots are usually internal cut walls; SpaceClaim positive OffsetFaces
# follows the wall normal into the void, which reduces the measured width.
half_delta = (current_width - target) / 2.0
if abs(half_delta) <= 1e-12:
return {'command': 'noop', 'targetWidth': target, 'reason': 'target already reached'}
faces = _locate_faces(signature)
selection = _selection(faces)
try:
applied = _offset_faces(selection, half_delta, signature)
applied['targetWidth'] = target
applied['widthDelta'] = target - current_width
applied['halfOffset'] = half_delta
return applied
except Exception as exc:
raise Exception('capability_not_implemented: slot.width adapter failed; ' + str(exc))
def change_slot_depth(job):
target = _float_value(job['target']['value'])
if target <= 0:
raise Exception('target_value_illegal: slot depth must be positive')
signature = job['object'].get('geometrySignature') or {}
current_depth = None
try:
current_depth = float(signature.get('depth'))
except Exception:
current_depth = None
if current_depth is None or current_depth <= 0:
raise Exception('object_signature_missing_depth')
depth_axis = signature.get('depthAxis') or signature.get('depthDirection') or []
if not isinstance(depth_axis, list) or len(depth_axis) != 3:
raise Exception('object_signature_missing_depth_axis')
axis_length = _vector_length([float(depth_axis[0]), float(depth_axis[1]), float(depth_axis[2])])
if axis_length <= 1e-12:
raise Exception('object_signature_missing_depth_axis')
delta_distance = target - current_depth
if abs(delta_distance) <= 1e-12:
return {'command': 'noop', 'targetDepth': target, 'reason': 'target already reached'}
faces = _locate_depth_faces(signature)
selection = _selection(faces)
delta = [float(depth_axis[index]) / axis_length * delta_distance for index in range(3)]
errors = []
try:
applied = _translate_selection(selection, delta)
applied['targetDepth'] = target
applied['depthDelta'] = delta_distance
applied['depthFaceCount'] = len(faces)
return applied
except Exception as exc:
errors.append('Move.Translate: ' + str(exc))
try:
depth_signature = dict(signature)
depth_signature['normal'] = [float(depth_axis[index]) / axis_length for index in range(3)]
applied = _offset_faces(selection, delta_distance, depth_signature)
applied['targetDepth'] = target
applied['depthDelta'] = delta_distance
applied['depthFaceCount'] = len(faces)
applied['moveErrors'] = errors
return applied
except Exception as exc:
errors.append('OffsetFaces.Execute: ' + str(exc))
raise Exception('capability_not_implemented: slot.depth adapter failed; ' + '; '.join(errors))
def move_boss(job):
target = _vector3(job['target']['value'])
signature = job['object'].get('geometrySignature') or {}
center = signature.get('center') or signature.get('axisCenter') or []
if not isinstance(center, list) or len(center) != 3:
raise Exception('object_signature_missing_center')
delta = [float(target[index]) - float(center[index]) for index in range(3)]
if _vector_length(delta) <= 1e-12:
return {'command': 'noop', 'delta': delta, 'reason': 'target already reached'}
faces = _locate_faces(signature)
selection = _selection(faces)
try:
applied = _translate_selection(selection, delta)
applied['targetCenter'] = target
return applied
except Exception as exc:
raise Exception('capability_not_implemented: boss.position adapter failed; ' + str(exc))
def _component_locator_key(locator):
if not isinstance(locator, dict):
return ''
path = _component_path(locator)
if path:
return 'path:' + '.'.join(str(item) for item in path)
component_index = _int_or_none(locator.get('componentIndex'))
if component_index is not None:
return 'index:' + str(component_index)
return ''
def _signature_has_component_pattern_locators(signature):
instances = signature.get('patternInstances')
if not isinstance(instances, list) or len(instances) < 3:
return False
seen = set()
for instance in instances:
if not isinstance(instance, dict):
return False
locators = instance.get('componentLocators')
if not isinstance(locators, list):
locators = instance.get('bodyLocators')
if not isinstance(locators, list):
return False
keys = []
for locator in locators:
key = _component_locator_key(locator)
if key:
keys.append(key)
if len(keys) != 1:
return False
if keys[0] in seen:
return False
seen.add(keys[0])
return True
def change_pattern_spacing(job):
target = _float_value(job['target']['value'])
if target <= 0:
raise Exception('target_value_illegal: pattern spacing must be positive')
signature = job['object'].get('geometrySignature') or {}
pattern_kind = str(signature.get('patternKind') or '').strip().lower()
instance_kind = str(signature.get('instanceKind') or '').strip().lower()
body_indices = signature.get('bodyIndices')
body_pattern = pattern_kind == 'body' or instance_kind in ('body', 'part', 'component') or (isinstance(body_indices, list) and body_indices)
if body_pattern and not _signature_has_component_pattern_locators(signature):
raise Exception(
'object_signature_missing_component_locator: body pattern spacing requires one component occurrence locator per instance'
)
current_spacing = None
for key in ('spacing', 'pitch'):
try:
current_spacing = float(signature.get(key))
break
except Exception:
pass
if current_spacing is None or current_spacing <= 0:
raise Exception('object_signature_missing_pattern_spacing')
support_fit = signature.get('supportPatternFit') or {}
if isinstance(support_fit, dict):
max_spacing = None
try:
max_spacing = float(support_fit.get('maxSpacing'))
except Exception:
max_spacing = None
if max_spacing is not None and max_spacing > 0:
tolerance = max(abs(max_spacing) * 1e-6, 1e-12)
if target > max_spacing + tolerance:
local_unit_scale = None
try:
local_unit_scale = float(support_fit.get('localUnitScale'))
except Exception:
local_unit_scale = None
target_local = target / local_unit_scale if local_unit_scale and local_unit_scale > 0 else target
max_local = support_fit.get('maxSpacingLocal')
try:
max_local = float(max_local)
except Exception:
max_local = max_spacing / local_unit_scale if local_unit_scale and local_unit_scale > 0 else max_spacing
raise Exception(
'target_value_illegal: pattern spacing exceeds support face range; '
+ 'target=' + str(target_local)
+ ', max=' + str(max_local)
+ ', supportFaceIds=' + str(support_fit.get('supportFaceIds'))
)
axis = signature.get('axis') or []
if not isinstance(axis, list) or len(axis) != 3:
raise Exception('object_signature_missing_pattern_axis')
axis_length = _vector_length([float(axis[0]), float(axis[1]), float(axis[2])])
if axis_length <= 1e-12:
raise Exception('object_signature_missing_pattern_axis')
axis_unit = [float(axis[index]) / axis_length for index in range(3)]
instances = signature.get('patternInstances')
if not isinstance(instances, list) or len(instances) < 3:
raise Exception('object_signature_missing_pattern_instances')
body_index = _int_or_none(signature.get('bodyIndex'))
located = []
for instance in instances:
if not isinstance(instance, dict):
continue
center = instance.get('center') or instance.get('instanceCenter') or []
if not isinstance(center, list) or len(center) != 3:
continue
located_items = _locate_pattern_instance_items(instance, body_index) if body_pattern else {'kind': 'face', 'items': _locate_pattern_instance_faces(instance, body_index)}
projection = sum(float(center[index]) * axis_unit[index] for index in range(3))
located.append({
'center': [float(center[0]), float(center[1]), float(center[2])],
'items': located_items.get('items') or [],
'kind': located_items.get('kind') or 'face',
'projection': projection,
})
if len(located) < 3:
raise Exception('object_signature_missing_pattern_instances')
located.sort(key=lambda item: item['projection'])
if abs(target - current_spacing) <= 1e-12:
return {'command': 'noop', 'targetSpacing': target, 'reason': 'target already reached'}
center_projection = sum(item['projection'] for item in located) / len(located)
mid_index = (len(located) - 1) * 0.5
moves = []
for index, item in enumerate(located):
desired_projection = center_projection + (index - mid_index) * target
delta_distance = desired_projection - item['projection']
delta = [axis_unit[axis_index] * delta_distance for axis_index in range(3)]
if _vector_length(delta) <= 1e-12:
continue
moves.append({'items': item['items'], 'kind': item['kind'], 'delta': delta, 'itemCount': len(item['items'])})
if not moves:
return {'command': 'noop', 'targetSpacing': target, 'reason': 'instance centers already satisfy target spacing'}
applied_moves = []
errors = []
for move_index, move in enumerate(moves):
try:
if move.get('kind') == 'component':
items = move.get('items') or []
if len(items) != 1:
raise Exception('component_pattern_instance_not_unique')
applied = _translate_component_occurrence(items[0], move['delta'])
else:
applied = _translate_selection(_selection(move['items']), move['delta'])
applied['instanceIndex'] = move_index + 1
applied['targetKind'] = move.get('kind')
applied['itemCount'] = move['itemCount']
applied_moves.append(applied)
except Exception as exc:
errors.append('Move.Translate instance ' + str(move_index + 1) + ': ' + str(exc))
break
if errors:
raise Exception('capability_not_implemented: pattern.spacing adapter failed; ' + '; '.join(errors))
command_name = 'Component.Transform instances' if all(move.get('targetKind') == 'component' for move in applied_moves) else 'Move.Translate instances'
return {
'command': command_name,
'targetSpacing': target,
'previousSpacing': current_spacing,
'spacingMode': 'centered',
'instanceCount': len(located),
'movedInstanceCount': len(applied_moves),
'moves': applied_moves,
}
def change_pattern_segment_spacing(job):
target = _float_value(job['target']['value'])
if target <= 0:
raise Exception('target_value_illegal: pattern segment spacing must be positive')
signature = job['object'].get('geometrySignature') or {}
pattern_kind = str(signature.get('patternKind') or '').strip().lower()
instance_kind = str(signature.get('instanceKind') or '').strip().lower()
body_indices = signature.get('bodyIndices')
body_pattern = pattern_kind == 'body' or instance_kind in ('body', 'part', 'component') or (isinstance(body_indices, list) and body_indices)
if body_pattern and not _signature_has_component_pattern_locators(signature):
raise Exception(
'object_signature_missing_component_locator: body pattern segment spacing requires one component occurrence locator per instance'
)
segment_index = _int_or_none(signature.get('segmentIndex'))
if segment_index is None or segment_index < 0:
raise Exception('object_signature_missing_pattern_segment_index')
support_fit = signature.get('supportPatternFit') or {}
if isinstance(support_fit, dict):
max_segment = None
try:
max_segment = float(support_fit.get('maxSegmentSpacing'))
except Exception:
max_segment = None
if max_segment is not None and max_segment > 0:
tolerance = max(abs(max_segment) * 1e-6, 1e-12)
if target > max_segment + tolerance:
local_unit_scale = None
try:
local_unit_scale = float(support_fit.get('localUnitScale'))
except Exception:
local_unit_scale = None
target_local = target / local_unit_scale if local_unit_scale and local_unit_scale > 0 else target
max_local = support_fit.get('maxSegmentSpacingLocal')
try:
max_local = float(max_local)
except Exception:
max_local = max_segment / local_unit_scale if local_unit_scale and local_unit_scale > 0 else max_segment
raise Exception(
'target_value_illegal: pattern segment spacing exceeds support face range; '
+ 'target=' + str(target_local)
+ ', max=' + str(max_local)
+ ', supportFaceIds=' + str(support_fit.get('supportFaceIds'))
)
axis = signature.get('axis') or []
if not isinstance(axis, list) or len(axis) != 3:
raise Exception('object_signature_missing_pattern_axis')
axis_length = _vector_length([float(axis[0]), float(axis[1]), float(axis[2])])
if axis_length <= 1e-12:
raise Exception('object_signature_missing_pattern_axis')
axis_unit = [float(axis[index]) / axis_length for index in range(3)]
instances = signature.get('patternInstances')
if not isinstance(instances, list) or len(instances) < 2:
raise Exception('object_signature_missing_pattern_instances')
body_index = _int_or_none(signature.get('bodyIndex'))
located = []
for instance in instances:
if not isinstance(instance, dict):
continue
center = instance.get('center') or instance.get('instanceCenter') or []
if not isinstance(center, list) or len(center) != 3:
continue
located_items = _locate_pattern_instance_items(instance, body_index) if body_pattern else {'kind': 'face', 'items': _locate_pattern_instance_faces(instance, body_index)}
projection = sum(float(center[index]) * axis_unit[index] for index in range(3))
located.append({
'center': [float(center[0]), float(center[1]), float(center[2])],
'items': located_items.get('items') or [],
'kind': located_items.get('kind') or 'face',
'projection': projection,
})
if len(located) < 2:
raise Exception('object_signature_missing_pattern_instances')
located.sort(key=lambda item: item['projection'])
if segment_index >= len(located) - 1:
raise Exception('object_signature_invalid_pattern_segment_index')
current_spacing = located[segment_index + 1]['projection'] - located[segment_index]['projection']
if current_spacing <= 0:
raise Exception('object_signature_invalid_pattern_segment_spacing')
delta_distance = target - current_spacing
if abs(delta_distance) <= 1e-12:
return {
'command': 'noop',
'targetSpacing': target,
'previousSpacing': current_spacing,
'segmentIndex': segment_index,
'reason': 'target already reached',
}
moving_side = str(signature.get('movingSide') or 'after').strip().lower()
moves = []
spacing_mode = 'segment_after'
if moving_side in ('after', 'right'):
delta = [axis_unit[index] * delta_distance for index in range(3)]
moves = [
{'items': item['items'], 'kind': item['kind'], 'delta': delta, 'itemCount': len(item['items'])}
for item in located[segment_index + 1:]
]
elif moving_side in ('before', 'left'):
delta = [axis_unit[index] * -delta_distance for index in range(3)]
moves = [
{'items': item['items'], 'kind': item['kind'], 'delta': delta, 'itemCount': len(item['items'])}
for item in located[:segment_index + 1]
]
spacing_mode = 'segment_before'
elif moving_side in ('split', 'both', 'center'):
left_delta = [axis_unit[index] * (-delta_distance * 0.5) for index in range(3)]
right_delta = [axis_unit[index] * (delta_distance * 0.5) for index in range(3)]
moves = [
{'items': item['items'], 'kind': item['kind'], 'delta': left_delta, 'itemCount': len(item['items'])}
for item in located[:segment_index + 1]
]
moves.extend(
{'items': item['items'], 'kind': item['kind'], 'delta': right_delta, 'itemCount': len(item['items'])}
for item in located[segment_index + 1:]
)
spacing_mode = 'segment_split'
else:
raise Exception('capability_not_implemented: unsupported pattern segment spacing motion semantics: ' + moving_side)
if not moves:
raise Exception('object_signature_invalid_pattern_segment_index')
applied_moves = []
errors = []
for move_index, move in enumerate(moves):
try:
if move.get('kind') == 'component':
items = move.get('items') or []
if len(items) != 1:
raise Exception('component_pattern_instance_not_unique')
applied = _translate_component_occurrence(items[0], move['delta'])
else:
applied = _translate_selection(_selection(move['items']), move['delta'])
applied['relativeMoveIndex'] = move_index + 1
applied['targetKind'] = move.get('kind')
applied['itemCount'] = move['itemCount']
applied_moves.append(applied)
except Exception as exc:
errors.append('Move.Translate segment instance ' + str(move_index + 1) + ': ' + str(exc))
break
if errors:
raise Exception('capability_not_implemented: pattern.segment_spacing adapter failed; ' + '; '.join(errors))
command_name = 'Component.Transform segment' if all(move.get('targetKind') == 'component' for move in applied_moves) else 'Move.Translate segment'
return {
'command': command_name,
'targetSpacing': target,
'segmentSpacing': target,
'previousSpacing': current_spacing,
'segmentIndex': segment_index,
'movingSide': moving_side,
'spacingMode': spacing_mode,
'instanceCount': len(located),
'movedInstanceCount': len(applied_moves),
'moves': applied_moves,
}
def change_boss_height(job):
target = _float_value(job['target']['value'])
if target <= 0:
raise Exception('target_value_illegal: boss height must be positive')
signature = job['object'].get('geometrySignature') or {}
current_height = None
try:
current_height = float(signature.get('height'))
except Exception:
current_height = None
if current_height is None or current_height <= 0:
raise Exception('object_signature_missing_height')
axis = signature.get('axis') or []
if not isinstance(axis, list) or len(axis) != 3:
raise Exception('object_signature_missing_axis')
axis_length = _vector_length([float(axis[0]), float(axis[1]), float(axis[2])])
if axis_length <= 1e-12:
raise Exception('object_signature_missing_axis')
delta_distance = target - current_height
if abs(delta_distance) <= 1e-12:
return {'command': 'noop', 'targetHeight': target, 'reason': 'target already reached'}
faces = _locate_height_faces(signature)
selection = _selection(faces)
delta = [float(axis[index]) / axis_length * delta_distance for index in range(3)]
errors = []
try:
applied = _translate_selection(selection, delta)
applied['targetHeight'] = target
applied['heightDelta'] = delta_distance
applied['heightFaceCount'] = len(faces)
return applied
except Exception as exc:
errors.append('Move.Translate: ' + str(exc))
try:
applied = _offset_faces(selection, delta_distance, signature)
applied['targetHeight'] = target
applied['heightDelta'] = delta_distance
applied['heightFaceCount'] = len(faces)
applied['moveErrors'] = errors
return applied
except Exception as exc:
errors.append('OffsetFaces.Execute: ' + str(exc))
raise Exception('capability_not_implemented: boss.height adapter failed; ' + '; '.join(errors))
def change_boss_diameter(job):
target = _float_value(job['target']['value'])
if target <= 0:
raise Exception('target_value_illegal: boss diameter must be positive')
signature = job['object'].get('geometrySignature') or {}
current_radius = None
try:
current_radius = float(signature.get('radius'))
except Exception:
current_radius = None
if current_radius is None:
try:
current_radius = float(signature.get('diameter')) / 2.0
except Exception:
current_radius = None
if current_radius is None or current_radius <= 0:
raise Exception('object_signature_missing_diameter')
target_radius = target / 2.0
radial_delta = target_radius - current_radius
if abs(radial_delta) <= 1e-12:
return {'command': 'noop', 'targetDiameter': target, 'reason': 'target already reached'}
faces = _locate_diameter_faces(signature)
selection = _selection(faces)
try:
applied = _offset_faces(selection, radial_delta, signature)
applied['targetDiameter'] = target
applied['targetRadius'] = target_radius
applied['radialDelta'] = radial_delta
applied['diameterFaceCount'] = len(faces)
return applied
except Exception as exc:
raise Exception('capability_not_implemented: boss.diameter adapter failed; ' + str(exc))
def _constant_round_radius(selection, faces, target):
constant_round = _command_type('ConstantRound')
if constant_round is None:
raise Exception('capability_not_implemented: ConstantRound command not available')
options = _new_options('ConstantRoundOptions')
errors = []
for label, method in (
('ConstantRound.ModifyRadius', 'ModifyRadius'),
('ConstantRound.SetRadius', 'SetRadius'),
('ConstantRound.ChangeRadius', 'ChangeRadius'),
('ConstantRound.Execute', 'Execute'),
):
func = getattr(constant_round, method, None)
if func is None:
continue
variants = []
if options is not None:
variants.append((selection, target, options, None))
variants.append((selection, target, options))
variants.append((faces, target, options, None))
variants.append((faces, target, options))
variants.append((selection, target, None))
variants.append((selection, target))
variants.append((faces, target, None))
variants.append((faces, target))
result = _call_variants(label, func, variants)
if result.get('ok'):
return {'command': label, 'targetRadius': target, 'apiSignature': result.get('signature')}
errors.extend(result.get('errors') or [])
raise Exception('capability_not_implemented: ConstantRound radius edit failed; ' + '; '.join(errors))
def change_round_radius(job):
target = _float_value(job['target']['value'])
if target <= 0:
raise Exception('target_value_illegal: round radius must be positive')
signature = job['object'].get('geometrySignature') or {}
if signature.get('isConstantRound') is not True:
raise Exception('object_signature_missing_constant_round_evidence')
current_radius = None
try:
current_radius = float(signature.get('radius'))
except Exception:
current_radius = None
if current_radius is None or current_radius <= 0:
raise Exception('object_signature_missing_round_radius')
if abs(target - current_radius) <= 1e-12:
return {'command': 'noop', 'targetRadius': target, 'reason': 'target already reached'}
faces = _locate_faces(signature)
selection = _selection(faces)
try:
applied = _constant_round_radius(selection, faces, target)
applied['roundFaceCount'] = len(faces)
applied['previousRadius'] = current_radius
return applied
except Exception as exc:
raise Exception('capability_not_implemented: round.radius adapter failed; ' + str(exc))
def _chamfer_distance(selection, faces, target):
chamfer = _command_type('Chamfer')
if chamfer is None:
raise Exception('capability_not_implemented: Chamfer command not available')
options = _new_options('ChamferOptions')
errors = []
for label, method in (
('Chamfer.ModifyDistance', 'ModifyDistance'),
('Chamfer.SetDistance', 'SetDistance'),
('Chamfer.ChangeDistance', 'ChangeDistance'),
('Chamfer.Execute', 'Execute'),
):
func = getattr(chamfer, method, None)
if func is None:
continue
variants = []
if options is not None:
variants.append((selection, target, options, None))
variants.append((selection, target, options))
variants.append((selection, target, target, options, None))
variants.append((selection, target, target, options))
variants.append((faces, target, options, None))
variants.append((faces, target, options))
variants.append((faces, target, target, options, None))
variants.append((faces, target, target, options))
variants.append((selection, target, None))
variants.append((selection, target))
variants.append((selection, target, target, None))
variants.append((selection, target, target))
variants.append((faces, target, None))
variants.append((faces, target))
variants.append((faces, target, target, None))
variants.append((faces, target, target))
result = _call_variants(label, func, variants)
if result.get('ok'):
return {'command': label, 'targetDistance': target, 'apiSignature': result.get('signature')}
errors.extend(result.get('errors') or [])
raise Exception('capability_not_implemented: Chamfer distance edit failed; ' + '; '.join(errors))
def change_chamfer_distance(job):
target = _float_value(job['target']['value'])
if target <= 0:
raise Exception('target_value_illegal: chamfer distance must be positive')
signature = job['object'].get('geometrySignature') or {}
if signature.get('isEqualDistanceChamfer') is not True:
raise Exception('object_signature_missing_equal_distance_chamfer_evidence')
current_distance = None
try:
current_distance = float(signature.get('distance'))
except Exception:
current_distance = None
if current_distance is None or current_distance <= 0:
raise Exception('object_signature_missing_chamfer_distance')
if abs(target - current_distance) <= 1e-12:
return {'command': 'noop', 'targetDistance': target, 'reason': 'target already reached'}
faces = _locate_faces(signature)
selection = _selection(faces)
try:
applied = _chamfer_distance(selection, faces, target)
applied['chamferFaceCount'] = len(faces)
applied['previousDistance'] = current_distance
return applied
except Exception as exc:
raise Exception('capability_not_implemented: chamfer.distance adapter failed; ' + str(exc))
def pull_face_offset(job):
target = _float_value(job['target']['value'])
signature = job['object'].get('geometrySignature') or {}
current = None
try:
current = float(signature.get('planeOffset'))
except Exception:
current = None
distance = target - current if current is not None else target
if abs(distance) <= 1e-12:
return {'command': 'noop', 'distance': distance, 'reason': 'target already reached'}
faces = _locate_faces(signature)
selection = _selection(faces)
try:
applied = _offset_faces(selection, distance, signature)
applied['targetPlaneOffset'] = target
return applied
except Exception as exc:
raise Exception('capability_not_implemented: face.offset adapter failed; ' + str(exc))
def fill_feature(job):
faces = _locate_faces(job['object'].get('geometrySignature') or {})
selection = _selection(faces)
try:
return _fill_selection(selection)
except Exception as exc:
raise Exception('capability_not_implemented: feature.fill adapter failed; ' + str(exc))
def delete_round_or_chamfer(job):
faces = _locate_faces(job['object'].get('geometrySignature') or {})
selection = _selection(faces)
try:
applied = _fill_selection(selection)
applied['targetFeatureRemoved'] = True
return applied
except Exception as exc:
raise Exception('capability_not_implemented: feature.delete_round_or_chamfer adapter failed; ' + str(exc))
def _apply_edit(job):
operation = job['target'].get('backendOperation') or ''
if operation == 'change_hole_diameter':
return change_hole_diameter(job)
if operation == 'move_hole_axis':
return move_hole_axis(job)
if operation == 'move_slot':
return move_slot(job)
if operation == 'change_slot_width':
return change_slot_width(job)
if operation == 'change_slot_depth':
return change_slot_depth(job)
if operation == 'move_boss':
return move_boss(job)
if operation == 'change_pattern_spacing':
return change_pattern_spacing(job)
if operation == 'change_pattern_segment_spacing':
return change_pattern_segment_spacing(job)
if operation == 'change_boss_height':
return change_boss_height(job)
if operation == 'change_boss_diameter':
return change_boss_diameter(job)
if operation == 'change_round_radius':
return change_round_radius(job)
if operation == 'change_chamfer_distance':
return change_chamfer_distance(job)
if operation == 'pull_face_offset':
return pull_face_offset(job)
if operation == 'fill_feature':
return fill_feature(job)
if operation == 'delete_round_or_chamfer':
return delete_round_or_chamfer(job)
raise Exception('unsupported_backend_operation: ' + operation)
def main():
job = _read_job()
outputs = job.get('outputs') or {}
error_path = outputs.get('error') or 'error.json'
result_path = outputs.get('result') or 'result.json'
output_step = outputs.get('outputStep') or 'result.step'
try:
_open_step(job['model']['sourceStep'])
applied = _apply_edit(job)
saved = _save_step(output_step)
_write_json(result_path, {
'ok': True,
'reason': 'ok',
'message': 'SCDM edit finished.',
'capabilityKey': job['target'].get('capabilityKey'),
'backendOperation': job['target'].get('backendOperation'),
'objectId': job['object'].get('objectId'),
'targetValue': job['target'].get('value'),
'outputStep': output_step,
'applied': applied,
'saved': saved,
})
except Exception as exc:
_write_json(error_path, {
'ok': False,
'reason': str(exc).split(':', 1)[0].replace('_', '-'),
'message': str(exc),
'traceback': traceback.format_exc(),
'capabilityKey': (job.get('target') or {}).get('capabilityKey') if 'job' in locals() else '',
'backendOperation': (job.get('target') or {}).get('backendOperation') if 'job' in locals() else '',
'objectId': (job.get('object') or {}).get('objectId') if 'job' in locals() else '',
'outputStep': output_step if 'output_step' in locals() else '',
})
raise
main()
'''
__all__ = [
"SCDM_EDIT_ADAPTER",
"SCDM_EDIT_SCHEMA_VERSION",
"generate_scdm_edit_script",
"prepare_scdm_edit_job",
"run_prepared_scdm_edit_job",
"run_scdm_edit_job",
]