ocp/tools/inventory.py

"""Symbol inventory: what the app imports from OCP, and what this wheel provides.

Two jobs, one source of truth:

  --emit    parse the app's sources for every `from OCP.<mod> import <name>`
            plus the attributes reached on those names, and write inventory.json.
            That file is the coverage work-queue for the remaining increments
            and, re-run later, the drift detector for newly used symbols.

  --check   import the installed OCP and report what the inventory asks for but
            the wheel does not provide, grouped by module.

Static analysis only reaches names reached through an imported symbol
(`TopExp.MapShapes_s`), not methods called on instances (`shape.IsSame(...)`) —
those are covered by running the app's own suite in the parity venv, where a
gap is a loud AttributeError.  So a green --check is necessary, not sufficient.

    python tools/inventory.py --emit --app ../app/backend
    python tools/inventory.py --check
"""

from __future__ import annotations

import argparse
import ast
import collections
import json
import pathlib
import sys

HERE = pathlib.Path(__file__).resolve().parent
DEFAULT_INVENTORY = HERE.parent / "inventory.json"
DEFAULT_APP = HERE.parent.parent / "app" / "backend"

#: Symbols no app source imports, so the scan below cannot see them, but which
#: another consumer of this wheel needs.  Hand-maintained: add the symbol and
#: name who wants it, so a later reader can tell a real dependency from a
#: leftover.  Merged into the scan, so --emit keeps them.
EXTRA_SYMBOLS: dict[str, dict[str, list[str]]] = {
    # assay/scripts/gen_flow_fixtures.py builds spheres for its CFD fixtures.
    "BRepPrimAPI": {"BRepPrimAPI_MakeSphere": []},
}


def scan(app_root: pathlib.Path) -> dict:
    """Collect {module: {symbol: [attributes reached on it]}} from the app."""
    modules: dict[str, dict[str, set[str]]] = collections.defaultdict(
        lambda: collections.defaultdict(set)
    )
    files = 0

    for path in sorted(app_root.rglob("*.py")):
        if "__pycache__" in path.parts:
            continue
        try:
            tree = ast.parse(path.read_text(encoding="utf-8"))
        except (SyntaxError, UnicodeDecodeError):
            continue

        # local alias -> (module, original name)
        local: dict[str, tuple[str, str]] = {}
        for node in ast.walk(tree):
            if isinstance(node, ast.ImportFrom) and node.module:
                if node.module == "OCP" or node.module.startswith("OCP."):
                    mod = node.module[4:] or "OCP"
                    for alias in node.names:
                        local[alias.asname or alias.name] = (mod, alias.name)
                        modules[mod][alias.name]  # noqa: B018 — create the entry

        if local:
            files += 1

        for node in ast.walk(tree):
            if (
                isinstance(node, ast.Attribute)
                and isinstance(node.value, ast.Name)
                and node.value.id in local
            ):
                mod, name = local[node.value.id]
                modules[mod][name].add(node.attr)

    for mod, symbols in EXTRA_SYMBOLS.items():
        for name, attrs in symbols.items():
            modules[mod][name].update(attrs)

    return {
        "app_root": str(app_root),
        "files_importing_ocp": files,
        "modules": {
            mod: {name: sorted(attrs) for name, attrs in sorted(syms.items())}
            for mod, syms in sorted(modules.items())
        },
    }


def methods(app_root: pathlib.Path, only: str | None) -> int:
    """Report the *instance* methods the app calls on each OCP class.

    `scan` only sees names reached through an import, which is enough to know
    *which* classes to bind but not *what* to bind on them.  This fills that
    gap well enough to write a module in one pass instead of discovering the
    surface one AttributeError at a time: variables assigned straight from a
    constructor carry their class through the file, so `a = BRepAdaptor_Surface(f)`
    followed by `a.GetType()` is resolved.

    Chained calls are reported separately as `Klass.Outer() -> Inner`, because
    what they constrain is the *return* type — `adaptor.Cylinder().Radius()`
    says gp_Cylinder needs `Radius`, not that BRepAdaptor_Surface does.

    Heuristic by construction: it does not follow arguments, returns or
    attributes, so treat a quiet class as "look again", not "nothing needed".
    """
    direct: dict[str, set[str]] = collections.defaultdict(set)
    chained: dict[str, set[str]] = collections.defaultdict(set)

    for path in sorted(app_root.rglob("*.py")):
        if "__pycache__" in path.parts:
            continue
        try:
            tree = ast.parse(path.read_text(encoding="utf-8"))
        except (SyntaxError, UnicodeDecodeError):
            continue

        classes = {
            alias.asname or alias.name: alias.name
            for node in ast.walk(tree)
            if isinstance(node, ast.ImportFrom)
            and node.module
            and (node.module == "OCP" or node.module.startswith("OCP."))
            for alias in node.names
        }
        if not classes:
            continue

        # local variable -> OCP class, from `v = Klass(...)`
        env: dict[str, str] = {}
        for node in ast.walk(tree):
            if isinstance(node, ast.Assign) and isinstance(node.value, ast.Call):
                fn = node.value.func
                cls = classes.get(getattr(fn, "id", ""))
                if cls:
                    for tgt in node.targets:
                        if isinstance(tgt, ast.Name):
                            env[tgt.id] = cls

        def owner(node: ast.expr) -> str | None:
            """The OCP class an expression evaluates to, when knowable."""
            if isinstance(node, ast.Name):
                return env.get(node.id)
            if isinstance(node, ast.Call):
                return classes.get(getattr(node.func, "id", ""))
            return None

        for node in ast.walk(tree):
            if not isinstance(node, ast.Attribute):
                continue
            cls = owner(node.value)
            if cls:
                direct[cls].add(node.attr)
                continue
            # <known>.Outer().Inner  — constrains Outer's return type
            inner = node.value
            if isinstance(inner, ast.Call) and isinstance(inner.func, ast.Attribute):
                cls = owner(inner.func.value)
                if cls:
                    chained[f"{cls}.{inner.func.attr}()"].add(node.attr)

    def dump(title: str, data: dict[str, set[str]]) -> None:
        keys = [k for k in sorted(data) if only is None or only in k]
        if not keys:
            return
        print(f"\n{title}")
        for key in keys:
            print(f"  {key}: {', '.join(sorted(data[key]))}")

    dump("instance methods, by class:", direct)
    dump("chained calls (constrain the return type):", chained)
    return 0


def emit(app_root: pathlib.Path, out: pathlib.Path) -> int:
    data = scan(app_root)
    out.write_text(json.dumps(data, indent=2) + "\n")
    n_mod = len(data["modules"])
    n_sym = sum(len(v) for v in data["modules"].values())
    print(f"{n_sym} symbols across {n_mod} modules "
          f"({data['files_importing_ocp']} files) -> {out}")
    return 0


def check(inventory: pathlib.Path) -> int:
    import importlib

    data = json.loads(inventory.read_text())
    missing: dict[str, list[str]] = {}
    present = 0

    for mod, symbols in data["modules"].items():
        try:
            m = importlib.import_module(f"OCP.{mod}")
        except ImportError:
            missing[mod] = [f"<module missing> ({len(symbols)} symbols)"]
            continue

        for name, attrs in symbols.items():
            obj = getattr(m, name, None)
            if obj is None:
                missing.setdefault(mod, []).append(name)
                continue
            present += 1
            for attr in attrs:
                if not hasattr(obj, attr):
                    missing.setdefault(mod, []).append(f"{name}.{attr}")

    total = sum(len(v) for v in data["modules"].values())
    print(f"{present}/{total} imported symbols available")
    if missing:
        print(f"\nmissing, by module ({len(missing)} modules):")
        for mod in sorted(missing):
            print(f"  {mod}: {', '.join(sorted(missing[mod]))}")
        return 1
    print("complete")
    return 0


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--emit", action="store_true", help="rewrite the inventory")
    ap.add_argument("--check", action="store_true",
                    help="compare the installed OCP against the inventory")
    ap.add_argument("--methods", action="store_true",
                    help="report the instance methods the app calls per class")
    ap.add_argument("--only", help="with --methods: substring filter on the class")
    ap.add_argument("--app", type=pathlib.Path, default=DEFAULT_APP)
    ap.add_argument("--inventory", type=pathlib.Path, default=DEFAULT_INVENTORY)
    args = ap.parse_args()

    if args.methods:
        if not args.app.is_dir():
            print(f"app sources not found: {args.app}", file=sys.stderr)
            return 2
        return methods(args.app, args.only)
    if args.emit:
        if not args.app.is_dir():
            print(f"app sources not found: {args.app}", file=sys.stderr)
            return 2
        return emit(args.app, args.inventory)
    if args.check:
        if not args.inventory.exists():
            print(f"no inventory at {args.inventory}; run --emit first",
                  file=sys.stderr)
            return 2
        return check(args.inventory)

    ap.print_help()
    return 2


if __name__ == "__main__":
    raise SystemExit(main())