pythonocc-step-editor/step_editor/scdm_schema.py

from __future__ import annotations

import hashlib
import json
from collections.abc import Mapping
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path


SCDM_RAW_SCHEMA_VERSION = 1
SCDM_CACHE_SCHEMA_VERSION = 1


@dataclass(frozen=True)
class ScdmProbeJob:
    step_path: Path
    output_dir: Path
    raw_features_path: Path
    error_path: Path
    model_fingerprint: str
    unit: str = "model"
    scan_scope: str = "all"
    adapter: str = "spaceclaim-v1"
    backend_path: str = ""
    backend_version: str = ""
    created_at: str = ""

    def to_payload(self) -> dict[str, object]:
        return {
            "schemaVersion": SCDM_RAW_SCHEMA_VERSION,
            "adapter": self.adapter,
            "createdAt": self.created_at or utc_now(),
            "backend": {
                "name": "SCDM",
                "path": self.backend_path,
                "version": self.backend_version,
            },
            "model": {
                "path": str(self.step_path),
                "fingerprint": self.model_fingerprint,
                "unit": self.unit,
            },
            "scan": {
                "scope": self.scan_scope,
            },
            "outputs": {
                "rawFeatures": str(self.raw_features_path),
                "error": str(self.error_path),
            },
        }


def utc_now() -> str:
    return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")


def file_fingerprint(path: str | Path) -> str:
    source = Path(path)
    digest = hashlib.sha256()
    stat = source.stat()
    digest.update(str(source.resolve(strict=False)).encode("utf-8", errors="replace"))
    digest.update(str(stat.st_size).encode("ascii"))
    digest.update(str(stat.st_mtime_ns).encode("ascii"))
    with source.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def read_json(path: str | Path) -> dict[str, object]:
    payload = json.loads(Path(path).read_text(encoding="utf-8"))
    if not isinstance(payload, dict):
        raise ValueError(f"JSON payload must be an object: {path}")
    return payload


def write_json(path: str | Path, payload: Mapping[str, object]) -> Path:
    target = Path(path)
    target.parent.mkdir(parents=True, exist_ok=True)
    target.write_text(json.dumps(dict(payload), ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    return target


def default_scdm_work_dir(
    step_path: str | Path,
    *,
    project_root: str | Path | None = None,
    fingerprint: str | None = None,
) -> Path:
    root = Path(project_root).expanduser() if project_root else Path(__file__).resolve().parent.parent
    source = Path(step_path)
    short = (fingerprint or file_fingerprint(source))[:12]
    return root / "local" / "scdm" / f"{source.stem}_{short}"


def payload_model_fingerprint(payload: Mapping[str, object]) -> str:
    model = payload.get("model")
    if not isinstance(model, Mapping):
        return ""
    return str(model.get("fingerprint") or "")


def payload_backend_version(payload: Mapping[str, object]) -> str:
    backend = payload.get("backend")
    if not isinstance(backend, Mapping):
        return ""
    return str(backend.get("version") or "")


__all__ = [
    "SCDM_CACHE_SCHEMA_VERSION",
    "SCDM_RAW_SCHEMA_VERSION",
    "ScdmProbeJob",
    "default_scdm_work_dir",
    "file_fingerprint",
    "payload_backend_version",
    "payload_model_fingerprint",
    "read_json",
    "utc_now",
    "write_json",
]