pythonocc-step-editor/step_editor/scdm_status.py

from __future__ import annotations

from collections.abc import Mapping
from pathlib import Path

from .scdm_backend import ScdmBackendInfo, is_scdm_disabled, load_scdm_backend_cache
from .scdm_capabilities import CAPABILITY_DEFINITIONS, ScdmCapabilityDefinition


def cached_scdm_backend_payload(project_root: str | Path | None = None) -> dict[str, object] | None:
    if is_scdm_disabled():
        return {"disabled": True, "reason": "disabled", "message": "SCDM backend is disabled by environment."}
    backend = load_scdm_backend_cache(project_root_override=project_root)
    return backend.to_cache() if backend is not None else None


def summarize_scdm_runtime(
    *,
    backend: ScdmBackendInfo | Mapping[str, object] | None = None,
    cache_state: str = "",
    cache_message: str = "",
    feature_cache: Mapping[str, object] | None = None,
) -> dict[str, object]:
    backend_payload = _backend_payload(backend)
    disabled = _backend_disabled(backend)
    state = str(cache_state or "empty").strip().lower()
    message = _compact(str(cache_message or "").strip(), 120)
    count = _feature_cache_counts(feature_cache)

    if disabled:
        headline = "SCDM：已关闭，当前使用 OCCT/Analysis Situs 兜底"
        path = ""
    elif backend_payload:
        version = str(backend_payload.get("version") or "").strip()
        source = _source_label(str(backend_payload.get("source") or "").strip())
        version_text = f" {version}" if version else ""
        headline = f"SCDM：已配置{version_text}（{source}）"
        path = str(backend_payload.get("path") or "").strip()
    else:
        headline = "SCDM：未配置，当前可用 OCCT/Analysis Situs 兜底"
        path = ""

    if disabled:
        detail = "检测到 SCDM 禁用开关；本次不会启动 SpaceClaim.exe。"
    elif state == "running":
        detail = "正在后台识别可修改参数；界面可继续旋转查看模型。"
    elif state == "ready":
        object_text = f"{count['objects']} 个对象" if count["objects"] else "0 个对象"
        capability_text = f"{count['capabilities']} 项能力" if count["capabilities"] else "0 项能力"
        detail = f"识别缓存已就绪：{object_text}，{capability_text}。"
    elif state == "failed":
        reason = message or "未拿到 SCDM 识别结果"
        detail = f"识别未启用：{reason}；当前使用本软件已有能力。"
    elif state == "deferred":
        detail = message or "已延后 SCDM 全量识别，优先保证导入显示、旋转和点选流畅。"
    elif state == "stale":
        detail = message or "缓存已失效，用户选择对象后会按需重新识别。"
    else:
        detail = "导入 STEP 后先显示模型；用户选择对象后再按需启动 SCDM 识别。"

    tooltip_lines = [headline, detail]
    if path:
        tooltip_lines.append(f"路径：{path}")
    if backend_payload:
        run_script_ok = backend_payload.get("runScriptOk")
        license_ok = backend_payload.get("licenseOk")
        tooltip_lines.append(f"/RunScript：{_ok_text(run_script_ok)}")
        tooltip_lines.append(f"许可证：{_ok_text(license_ok)}")
    return {
        "headline": headline,
        "detail": detail,
        "tooltip": "\n".join(line for line in tooltip_lines if line),
        "backendReady": bool(backend_payload) and not disabled,
        "cacheState": state,
        "objectCount": count["objects"],
        "capabilityCount": count["capabilities"],
    }


def summarize_scdm_capability_progress(
    *,
    feature_cache: Mapping[str, object] | None = None,
    execution_ready: bool | set[str] | list[str] | tuple[str, ...] = False,
) -> dict[str, object]:
    ready_keys = _execution_ready_keys(execution_ready)
    # 这里把“识别到”“计划中”“几何 hint”和“可执行”分开统计。
    # 客户界面只展示能稳定解释的进度，不能把 SCDM/raw hint 直接包装成可改参数。
    detection_counts = _cache_capability_counts(feature_cache)
    blocked_counts = _cache_blocked_capability_counts(feature_cache)
    planned_counts = _planned_capability_counts(feature_cache)
    hint_counts = _geometry_candidate_hint_counts(feature_cache)
    discovered_summary = _discovered_not_productized_summary(feature_cache)
    probe_evidence = _probe_evidence_summary(feature_cache)
    rows: list[dict[str, object]] = []

    for key, definition in sorted(CAPABILITY_DEFINITIONS.items(), key=lambda item: (_stage_sort_key(item[1].roadmap_stage), item[0])):
        detected = int(detection_counts.get(key, 0))
        blocked = int(blocked_counts.get(key, 0))
        planned_detected = int(planned_counts.get(key, 0))
        hint_detected = int(hint_counts.get(key, 0))
        runner_ready = _capability_runner_ready(key, execution_ready, ready_keys)
        status, reason = _capability_progress_status(
            key,
            definition,
            detected=detected,
            blocked=blocked,
            planned_detected=planned_detected,
            hint_detected=hint_detected,
            runner_ready=runner_ready,
        )
        executable = detected if definition.productized and runner_ready else 0
        if blocked:
            executable = max(0, executable - blocked)
        rows.append(
            {
                "key": key,
                "displayName": definition.display_name,
                "roadmapStage": definition.roadmap_stage,
                "productized": definition.productized,
                "runnerReady": runner_ready,
                "detectedCount": detected,
                "plannedDetectedCount": planned_detected,
                "hintDetectedCount": hint_detected,
                "blockedCount": blocked,
                "executableCount": executable,
                "status": status,
                "reason": reason,
            }
        )

    executable_count = sum(int(row["executableCount"]) for row in rows)
    productized_count = sum(1 for row in rows if bool(row["productized"]))
    runner_ready_count = sum(1 for row in rows if bool(row["productized"]) and bool(row["runnerReady"]))
    planned_detected_total = sum(int(row["plannedDetectedCount"]) for row in rows)
    hint_detected_total = sum(int(row["hintDetectedCount"]) for row in rows)
    blocked_total = sum(int(row["blockedCount"]) for row in rows)
    return {
        "rows": rows,
        "summary": {
            "defined": len(rows),
            "productized": productized_count,
            "runnerReady": runner_ready_count,
            "detectedCapabilities": sum(int(row["detectedCount"]) for row in rows),
            "executableCapabilities": executable_count,
            "plannedDetected": planned_detected_total,
            "geometryHints": hint_detected_total,
            "backendBlocked": blocked_total,
            "discoveredNotProductized": discovered_summary["count"],
            "faceAdjacency": probe_evidence["faceAdjacency"],
            "circularEdges": probe_evidence["circularEdges"],
            "inventoryObjectTypes": probe_evidence["inventoryObjectTypes"],
            "inventoryOperationCandidates": probe_evidence["inventoryOperationCandidates"],
            "derivedFeatureCandidates": probe_evidence["derivedFeatureCandidates"],
        },
        "productizedLines": _capability_progress_lines(
            row for row in rows if bool(row["productized"])
        ),
        "plannedLines": _capability_progress_lines(
            row
            for row in rows
            if not bool(row["productized"])
            and (
                int(row["plannedDetectedCount"]) > 0
                or int(row["detectedCount"]) > 0
                or int(row["hintDetectedCount"]) > 0
            )
        ),
        "roadmapLines": _capability_progress_lines(
            row
            for row in rows
            if not bool(row["productized"])
            and int(row["plannedDetectedCount"]) <= 0
            and int(row["detectedCount"]) <= 0
            and int(row["hintDetectedCount"]) <= 0
        ),
        "discoveredNotProductized": discovered_summary,
        "probeEvidence": probe_evidence,
    }


def _backend_payload(backend: ScdmBackendInfo | Mapping[str, object] | None) -> dict[str, object]:
    if isinstance(backend, ScdmBackendInfo):
        return backend.to_cache()
    if not isinstance(backend, Mapping):
        return {}
    nested = backend.get("backend")
    if isinstance(nested, ScdmBackendInfo):
        return nested.to_cache()
    if isinstance(nested, Mapping):
        return _backend_payload(nested)
    path = str(backend.get("path") or "").strip()
    if not path:
        return {}
    return {
        "path": path,
        "source": str(backend.get("source") or ""),
        "version": str(backend.get("version") or ""),
        "verifiedAt": str(backend.get("verifiedAt") or ""),
        "runScriptOk": backend.get("runScriptOk"),
        "licenseOk": backend.get("licenseOk"),
        "message": str(backend.get("message") or ""),
    }


def _backend_disabled(backend: ScdmBackendInfo | Mapping[str, object] | None) -> bool:
    return isinstance(backend, Mapping) and bool(backend.get("disabled"))


def _feature_cache_counts(feature_cache: Mapping[str, object] | None) -> dict[str, int]:
    if not isinstance(feature_cache, Mapping):
        return {"objects": 0, "capabilities": 0}
    objects = feature_cache.get("objects")
    if not isinstance(objects, list):
        return {"objects": 0, "capabilities": 0}
    capability_count = 0
    object_count = 0
    for item in objects:
        if not isinstance(item, Mapping):
            continue
        object_count += 1
        capabilities = item.get("capabilities")
        if isinstance(capabilities, list):
            capability_count += sum(1 for capability in capabilities if isinstance(capability, Mapping))
    return {"objects": object_count, "capabilities": capability_count}


def _cache_capability_counts(feature_cache: Mapping[str, object] | None) -> dict[str, int]:
    result: dict[str, int] = {}
    if not isinstance(feature_cache, Mapping):
        return result
    objects = feature_cache.get("objects")
    if not isinstance(objects, list):
        return result
    for item in objects:
        if not isinstance(item, Mapping):
            continue
        capabilities = item.get("capabilities")
        if not isinstance(capabilities, list):
            continue
        for capability in capabilities:
            if not isinstance(capability, Mapping):
                continue
            key = str(capability.get("key") or "").strip()
            if key:
                result[key] = result.get(key, 0) + 1
    return result


def _cache_blocked_capability_counts(feature_cache: Mapping[str, object] | None) -> dict[str, int]:
    result: dict[str, int] = {}
    if not isinstance(feature_cache, Mapping):
        return result
    objects = feature_cache.get("objects")
    if not isinstance(objects, list):
        return result
    for item in objects:
        if not isinstance(item, Mapping):
            continue
        object_block = str(item.get("blockReason") or "").strip()
        capabilities = item.get("capabilities")
        if not isinstance(capabilities, list):
            continue
        for capability in capabilities:
            if not isinstance(capability, Mapping):
                continue
            key = str(capability.get("key") or "").strip()
            if not key:
                continue
            capability_block = str(capability.get("blockReason") or "").strip()
            if object_block or capability_block or capability.get("editable") is False:
                result[key] = result.get(key, 0) + 1
    return result


def _planned_capability_counts(feature_cache: Mapping[str, object] | None) -> dict[str, int]:
    diagnostics = _cache_diagnostics(feature_cache)
    planned = diagnostics.get("planned_not_productized")
    result: dict[str, int] = {}
    if not isinstance(planned, list):
        return result
    for item in planned:
        if not isinstance(item, Mapping):
            continue
        key = str(item.get("capabilityKey") or "").strip()
        if key:
            result[key] = result.get(key, 0) + 1
    return result


def _geometry_candidate_hint_counts(feature_cache: Mapping[str, object] | None) -> dict[str, int]:
    diagnostics = _cache_diagnostics(feature_cache)
    hints = diagnostics.get("geometry_candidate_hints")
    result: dict[str, int] = {}
    if not isinstance(hints, list):
        return result
    for item in hints:
        if not isinstance(item, Mapping):
            continue
        key = str(item.get("capabilityKey") or "").strip()
        if not key:
            continue
        count = _int_value(item.get("evidenceCount"))
        result[key] = result.get(key, 0) + max(count, 1)
    return result


def _discovered_not_productized_summary(feature_cache: Mapping[str, object] | None) -> dict[str, object]:
    diagnostics = _cache_diagnostics(feature_cache)
    discovered = diagnostics.get("discovered_not_productized")
    by_type: dict[str, int] = {}
    if not isinstance(discovered, list):
        return {"count": 0, "byObjectType": {}, "lines": []}
    for item in discovered:
        if not isinstance(item, Mapping):
            continue
        object_type = str(item.get("objectType") or "object").strip() or "object"
        by_type[object_type] = by_type.get(object_type, 0) + 1
    lines = [f"{name}：{count} 个" for name, count in sorted(by_type.items(), key=lambda item: (-item[1], item[0]))[:6]]
    return {"count": sum(by_type.values()), "byObjectType": by_type, "lines": lines}


def _probe_evidence_summary(feature_cache: Mapping[str, object] | None) -> dict[str, object]:
    diagnostics = _cache_diagnostics(feature_cache)
    face_adjacency = diagnostics.get("face_adjacency")
    edge_summary = diagnostics.get("edge_geometry_summary")
    feature_inventory = diagnostics.get("feature_inventory")
    adjacency_count = len(face_adjacency) if isinstance(face_adjacency, list) else 0
    edge_summary = edge_summary if isinstance(edge_summary, Mapping) else {}
    feature_inventory = feature_inventory if isinstance(feature_inventory, Mapping) else {}
    edge_kind_counts = edge_summary.get("edgeKindCounts")
    edge_kind_counts = edge_kind_counts if isinstance(edge_kind_counts, Mapping) else {}
    object_type_counts = _mapping_count_dict(feature_inventory.get("objectTypeCounts"))
    surface_type_counts = _mapping_count_dict(feature_inventory.get("surfaceTypeCounts"))
    operation_counts = _mapping_count_dict(feature_inventory.get("operationCounts"))
    geometry_hints = diagnostics.get("geometry_candidate_hints")
    geometry_hint_lines = _geometry_candidate_hint_lines(geometry_hints)
    derived_candidates = diagnostics.get("derived_feature_candidates")
    derived_candidate_lines = _derived_feature_candidate_lines(derived_candidates)
    derived_candidate_count = len(derived_candidates) if isinstance(derived_candidates, list) else 0
    circular_edges = _int_value(edge_summary.get("circularEdgeCount"))
    if circular_edges <= 0:
        circular_edges = _int_value(edge_kind_counts.get("circular"))
    linear_edges = _int_value(edge_kind_counts.get("linear"))
    total_edges = _int_value(edge_summary.get("totalEdgeCount"))
    radius_buckets = edge_summary.get("circularRadiusBuckets")
    radius_bucket_count = len(radius_buckets) if isinstance(radius_buckets, list) else 0
    lines = []
    if adjacency_count:
        lines.append(f"Face 邻接 {adjacency_count} 组")
    if total_edges:
        lines.append(f"Edge {total_edges} 条")
    if circular_edges:
        lines.append(f"圆边 {circular_edges} 条")
    if linear_edges:
        lines.append(f"直边 {linear_edges} 条")
    if radius_bucket_count:
        lines.append(f"圆边半径分组 {radius_bucket_count} 类")
    object_lines = _count_summary_lines(object_type_counts, label="对象")
    surface_lines = _count_summary_lines(surface_type_counts, label="曲面")
    operation_lines = _count_summary_lines(operation_counts, label="命令候选")
    lines.extend(object_lines[:2])
    lines.extend(surface_lines[:2])
    lines.extend(operation_lines[:2])
    lines.extend(derived_candidate_lines[:3])
    lines.extend(geometry_hint_lines[:4])
    return {
        "faceAdjacency": adjacency_count,
        "totalEdges": total_edges,
        "circularEdges": circular_edges,
        "linearEdges": linear_edges,
        "radiusBucketCount": radius_bucket_count,
        "inventoryObjectTypes": sum(object_type_counts.values()),
        "inventorySurfaceTypes": sum(surface_type_counts.values()),
        "inventoryOperationCandidates": sum(operation_counts.values()),
        "derivedFeatureCandidates": derived_candidate_count,
        "derivedFeatureCandidateLines": derived_candidate_lines,
        "objectTypeCounts": object_type_counts,
        "surfaceTypeCounts": surface_type_counts,
        "operationCounts": operation_counts,
        "geometryHintLines": geometry_hint_lines,
        "lines": lines,
    }


def _derived_feature_candidate_lines(value: object, *, limit: int = 4) -> list[str]:
    if not isinstance(value, list):
        return []
    counts: dict[str, int] = {}
    for item in value:
        if not isinstance(item, Mapping):
            continue
        object_type = str(item.get("objectType") or "object").strip() or "object"
        counts[object_type] = counts.get(object_type, 0) + 1
    rows = sorted(counts.items(), key=lambda item: (-int(item[1]), item[0]))[: max(1, int(limit))]
    return [f"派生候选 {name}:{count}" for name, count in rows]


def _geometry_candidate_hint_lines(value: object, *, limit: int = 4) -> list[str]:
    if not isinstance(value, list):
        return []
    best: dict[str, dict[str, object]] = {}
    for item in value:
        if not isinstance(item, Mapping):
            continue
        key = str(item.get("capabilityKey") or "").strip()
        if not key:
            continue
        count = max(_int_value(item.get("evidenceCount")), 1)
        existing = best.get(key)
        if existing is None or count > int(existing.get("evidenceCount") or 0):
            best[key] = {
                "displayName": str(item.get("displayName") or key),
                "evidenceCount": count,
                "confidence": str(item.get("confidence") or ""),
            }
    rows = sorted(best.items(), key=lambda item: (-int(item[1].get("evidenceCount") or 0), item[0]))[: max(1, int(limit))]
    return [
        f"几何候选 {payload['displayName']}:{payload['evidenceCount']}（{payload['confidence'] or 'unknown'}）"
        for _key, payload in rows
    ]


def _mapping_count_dict(value: object) -> dict[str, int]:
    if not isinstance(value, Mapping):
        return {}
    result: dict[str, int] = {}
    for key, count in value.items():
        text = str(key or "").strip() or "unknown"
        number = _int_value(count)
        if number > 0:
            result[text] = number
    return result


def _count_summary_lines(counts: Mapping[str, int], *, label: str, limit: int = 4) -> list[str]:
    if not counts:
        return []
    rows = sorted(counts.items(), key=lambda item: (-int(item[1]), item[0]))[: max(1, int(limit))]
    summary = "，".join(f"{name}:{count}" for name, count in rows)
    return [f"{label}分布 {summary}"]


def _int_value(value: object) -> int:
    try:
        return int(value)
    except (TypeError, ValueError):
        return 0


def _cache_diagnostics(feature_cache: Mapping[str, object] | None) -> Mapping[str, object]:
    if not isinstance(feature_cache, Mapping):
        return {}
    diagnostics = feature_cache.get("diagnostics")
    return diagnostics if isinstance(diagnostics, Mapping) else {}


def _execution_ready_keys(execution_ready: bool | set[str] | list[str] | tuple[str, ...]) -> set[str]:
    if isinstance(execution_ready, bool):
        return set()
    try:
        return {str(item) for item in execution_ready}
    except TypeError:
        return set()


def _capability_runner_ready(
    key: str,
    execution_ready: bool | set[str] | list[str] | tuple[str, ...],
    ready_keys: set[str],
) -> bool:
    return bool(execution_ready) if isinstance(execution_ready, bool) else key in ready_keys


def _capability_progress_status(
    key: str,
    definition: ScdmCapabilityDefinition,
    *,
    detected: int,
    blocked: int,
    planned_detected: int,
    hint_detected: int,
    runner_ready: bool,
) -> tuple[str, str]:
    if definition.productized and runner_ready and detected > blocked:
        return "已开放", "已识别到对象时会显示在特征参数表。"
    if definition.productized and runner_ready and blocked:
        return "已开放但被后端阻止", "当前模型里识别到该能力，但 SCDM 命令、对象状态或安全守门暂时阻止执行。"
    if definition.productized and runner_ready:
        return "已开放待识别", "执行链路已接入，当前 cache 还没有识别到可执行对象。"
    if definition.productized and detected:
        return "已识别待执行器", "能力已进入产品字典，但当前 UI 执行器还未开放。"
    if definition.productized:
        return "已产品化待对象", "能力已定义，等待 SCDM 在当前模型中识别到对象。"
    if planned_detected or detected:
        return "已识别待验证", definition.block_reason or "已识别到候选，但还没有完成真实 STEP 回测。"
    if hint_detected:
        return "几何证据待分类", "SCDM probe 已看到相关曲面/边/命令线索，但还没有确认成可执行特征对象。"
    return "路线中待接入", definition.block_reason or f"{key} 还没有接入可执行闭环。"


def _capability_progress_lines(rows: object) -> list[str]:
    result: list[str] = []
    for row in rows:  # type: ignore[assignment]
        if not isinstance(row, Mapping):
            continue
        display = str(row.get("displayName") or row.get("key") or "").strip()
        status = str(row.get("status") or "").strip()
        detected = int(row.get("detectedCount") or 0)
        planned = int(row.get("plannedDetectedCount") or 0)
        hinted = int(row.get("hintDetectedCount") or 0)
        blocked = int(row.get("blockedCount") or 0)
        suffix_parts = []
        if detected:
            suffix_parts.append(f"cache {detected}")
        if planned:
            suffix_parts.append(f"候选 {planned}")
        if hinted:
            suffix_parts.append(f"证据 {hinted}")
        if blocked:
            suffix_parts.append(f"阻止 {blocked}")
        suffix = f"（{'，'.join(suffix_parts)}）" if suffix_parts else ""
        result.append(f"- {display}：{status}{suffix}")
    return result


def _stage_sort_key(stage: str) -> tuple[int, int, str]:
    text = str(stage or "")
    numbers: list[int] = []
    for part in text.replace("S", "").split("."):
        try:
            numbers.append(int(part))
        except ValueError:
            pass
    while len(numbers) < 2:
        numbers.append(0)
    return numbers[0], numbers[1], text


def _source_label(source: str) -> str:
    if source.startswith("registry:"):
        return "注册表"
    if source.startswith("env:"):
        return "环境变量"
    if source.startswith("common:"):
        return "常见安装目录"
    if source.lower() == "path":
        return "PATH"
    if source == "manual":
        return "手动配置"
    if source == "cache":
        return "缓存"
    return source or "未知来源"


def _ok_text(value: object) -> str:
    if value is True:
        return "可用"
    if value is False:
        return "不可用"
    return "未验证"


def _compact(text: str, limit: int) -> str:
    text = " ".join(text.split())
    if len(text) <= limit:
        return text
    return text[: max(limit - 1, 0)].rstrip() + "…"


__all__ = ["cached_scdm_backend_payload", "summarize_scdm_capability_progress", "summarize_scdm_runtime"]