orchestrata_ultimate.refactor.codemod.analysis.complexity

Reference › orchestrata_ultimate › refactor › codemod › analysis › complexity

ComplexityAnalyzer computes complexity per function from the forest's ast overlay, using ComplexityMetrics. It takes the graph and the forest of LosslessModule objects from ctok; the graph is used only to resolve the target module, while the numbers come from the syntax trees. analyze(target) walks the functions of one module or of the whole forest (_walk_functions, _function_metrics) and returns a ComplexityReport of FunctionComplexity records holding the cyclomatic, nesting and cognitive values. It is read-only and imported by the analysis facade.

At a glance

Source src/orchestrata_ultimate/refactor/codemod/analysis/complexity.py
Lines of code 107
Classes 1
Functions 1
Imports (internal) orchestrata_ultimate.core.base.traces, orchestrata_ultimate.refactor.codemod._cig, orchestrata_ultimate.refactor.codemod._types, orchestrata_ultimate.refactor.codemod.analysis.metrics, orchestrata_ultimate.refactor.codemod.analysis.protocol, orchestrata_ultimate.refactor.ctok
Imported by orchestrata_ultimate.refactor.codemod.analysis
Tests that import it none

Class diagram

Class diagram orchestrata_ultimate.refactor.codemod.analysis.complexity

Dependency diagram

Dependency diagram orchestrata_ultimate.refactor.codemod.analysis.complexity

API

Classes

ComplexityAnalyzer

Decorators: traces(spec='110', story='US-110-006', reqs=['REQ-110-031', 'REQ-110-032']) · Line 31

Cyclomatic, nesting and cognitive complexity of every function.

Method Signature Summary
__init__ __init__(cig: CodeIntelligenceGraph, forest: Mapping[str, LosslessModule]) -> None
analyze analyze(target: str | None = None) -> ComplexityReport Complexity of every function in target (a module path) or the forest.
_function_metrics _function_metrics(path: str, tree: ast.Module) -> tuple[FunctionComplexity, ...]

Functions

Function Signature Summary
_walk_functions _walk_functions(node: ast.AST, prefix: str) -> list[tuple[str, ast.FunctionDef | ast.AsyncFunctionDef]] Every function/method under node with its dotted in-module name.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/refactor/codemod/analysis/complexity.py

"""ComplexityAnalyzer — per-function complexity from the forest's ``ast`` overlay.

GAP-6: no CIG query answers cyclomatic, nesting or cognitive complexity, so
the metrics come from ``ws.forest[path].tree`` (``ComplexityMetrics``). The
CIG only resolves the target module; each function's ``qualified_name`` is
``<path>:<dotted scope>`` taken from the AST walk.
"""

from __future__ import annotations

import ast
from collections.abc import Mapping
from typing import TYPE_CHECKING

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.refactor.codemod._types import (
    ComplexityReport,
    FunctionComplexity,
)
from orchestrata_ultimate.refactor.codemod.analysis.metrics import ComplexityMetrics
from orchestrata_ultimate.refactor.codemod.analysis.protocol import resolve_node_id

if TYPE_CHECKING:
    from orchestrata_ultimate.refactor.codemod._cig import CodeIntelligenceGraph
    from orchestrata_ultimate.refactor.ctok import LosslessModule

__all__ = ["ComplexityAnalyzer"]


@traces(spec="110", story="US-110-006", reqs=["REQ-110-031", "REQ-110-032"])
class ComplexityAnalyzer:
    """Cyclomatic, nesting and cognitive complexity of every function."""

    name = "complexity"

    @traces(spec="110", story="US-110-006", reqs=["REQ-110-031", "REQ-110-032"])
    def __init__(
        self, cig: CodeIntelligenceGraph, forest: Mapping[str, LosslessModule]
    ) -> None:
        self._cig = cig
        self._forest = forest

    @traces(spec="110", story="US-110-006", reqs=["REQ-110-031", "REQ-110-032"])
    def analyze(self, target: str | None = None) -> ComplexityReport:
        """Complexity of every function in ``target`` (a module path) or the forest.

        Raises:
            ValueError: ``target`` names no module in the graph.
        """
        if target is None:
            paths = sorted(self._forest)
        else:
            node_id = resolve_node_id(self._cig, target)
            if node_id is None:
                raise ValueError(f"module '{target}' not found in CIG")
            paths = [self._cig.nodes[node_id].path]
        functions: list[FunctionComplexity] = []
        for path in paths:
            module = self._forest.get(path)
            if module is not None:
                functions.extend(self._function_metrics(path, module.tree))
        cyclomatics = [f.cyclomatic for f in functions]
        return ComplexityReport(
            functions=tuple(functions),
            file_count=len(paths),
            max_cyclomatic=max(cyclomatics, default=0),
            mean_cyclomatic=(
                (sum(cyclomatics) / len(cyclomatics)) if cyclomatics else 0.0
            ),
        )

    @traces(spec="110", story="US-110-006", reqs=["REQ-110-031"])
    def _function_metrics(
        self, path: str, tree: ast.Module
    ) -> tuple[FunctionComplexity, ...]:
        found: list[FunctionComplexity] = []
        for scope, node in _walk_functions(tree, ""):
            found.append(
                FunctionComplexity(
                    # Dotted scope from the AST walk: unique per class even
                    # when two classes define a method of the same name.
                    qualified_name=f"{path}:{scope}",
                    path=path,
                    line=node.lineno,
                    cyclomatic=ComplexityMetrics.cyclomatic(node),
                    max_nesting=ComplexityMetrics.max_nesting(node),
                    cognitive=ComplexityMetrics.cognitive(node),
                )
            )
        return tuple(sorted(found, key=lambda f: f.line))


def _walk_functions(
    node: ast.AST, prefix: str
) -> list[tuple[str, ast.FunctionDef | ast.AsyncFunctionDef]]:
    """Every function/method under ``node`` with its dotted in-module name."""
    found: list[tuple[str, ast.FunctionDef | ast.AsyncFunctionDef]] = []
    for child in ast.iter_child_nodes(node):
        if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef)):
            name = f"{prefix}{child.name}"
            found.append((name, child))
            found.extend(_walk_functions(child, f"{name}."))
        elif isinstance(child, ast.ClassDef):
            found.extend(_walk_functions(child, f"{prefix}{child.name}."))
        else:
            found.extend(_walk_functions(child, prefix))
    return found