engine.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/complexity/engine.py
 
"""
Purpose
-------
Compute stable per-function cyclomatic complexity from Python syntax trees.
 
Responsibilities
----------------
- Apply Plan 214's pinned decision-counting convention.
- Score nested functions and lambdas independently from their parents.
- Preserve file, qualname, and source-span coordinates for every result.
 
Diagnostics
-----------
Domain: OCT-COMPLEXITY
Levels:
    L2 - lifecycle
    L3 - measurement details
    L4 - deep tracing
 
Contracts
---------
- Every function starts with complexity one.
- The engine imports only Python standard-library modules.
- Syntax errors propagate so orchestration can report them honestly.
 
Dependencies
------------
- ast (Python syntax tree)
- tokenize (declared source encodings)
"""
 
# OCT-LINT: disable-file=dbg_usage reason="Plan 214 P1 pins the engine as zero-dependency, so importing oc_diagnostics is forbidden"
# OCT-LINT: disable-file=dbg_import reason="Plan 214 P1 pins the engine as zero-dependency"
 
from __future__ import annotations
 
import ast
import tokenize
from dataclasses import asdict, dataclass
from pathlib import Path
 
 
@dataclass(frozen=True, slots=True)
class FunctionComplexity:
    """One stable per-function complexity measurement."""
 
    file: str
    qualname: str
    lineno: int
    end_lineno: int
    complexity: int
 
    def to_dict(self) -> dict[str, str | int]:
        """Return a JSON-ready representation with every pinned P1 field."""
        return asdict(self)
 
 
class _DecisionCounter(ast.NodeVisitor):
    """Count decisions inside one function while excluding nested scopes."""
 
    def __init__(self) -> None:
        self.complexity = 1
 
    def count_function(self, node: ast.FunctionDef | ast.AsyncFunctionDef) -> int:
        """Count decisions in a function body and return the final score."""
        for statement in node.body:
            self.visit(statement)
        return self.complexity
 
    def count_lambda(self, node: ast.Lambda) -> int:
        """Count decisions in a lambda expression and return its score."""
        self.visit(node.body)
        return self.complexity
 
    def visit_FunctionDef(self, node: ast.FunctionDef) -> None:
        """Do not fold a nested synchronous function into its parent."""
 
    def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None:
        """Do not fold a nested asynchronous function into its parent."""
 
    def visit_Lambda(self, node: ast.Lambda) -> None:
        """Do not fold a nested lambda into its parent."""
 
    def visit_If(self, node: ast.If) -> None:
        """Count each if node; an elif is represented as another if node."""
        self.complexity += 1
        self.generic_visit(node)
 
    def visit_IfExp(self, node: ast.IfExp) -> None:
        """Count one conditional-expression decision."""
        self.complexity += 1
        self.generic_visit(node)
 
    def visit_For(self, node: ast.For) -> None:
        """Count one synchronous loop decision."""
        self.complexity += 1
        self.generic_visit(node)
 
    def visit_AsyncFor(self, node: ast.AsyncFor) -> None:
        """Count one asynchronous loop decision."""
        self.complexity += 1
        self.generic_visit(node)
 
    def visit_While(self, node: ast.While) -> None:
        """Count one while-loop decision."""
        self.complexity += 1
        self.generic_visit(node)
 
    def visit_ExceptHandler(self, node: ast.ExceptHandler) -> None:
        """Count each except clause, including a bare except."""
        self.complexity += 1
        self.generic_visit(node)
 
    def visit_BoolOp(self, node: ast.BoolOp) -> None:
        """Count each written and/or operator in a boolean chain."""
        self.complexity += max(0, len(node.values) - 1)
        self.generic_visit(node)
 
    def visit_comprehension(self, node: ast.comprehension) -> None:
        """Count a comprehension for-clause and each attached if filter."""
        self.complexity += 1 + len(node.ifs)
        self.generic_visit(node)
 
    def visit_match_case(self, node: ast.match_case) -> None:
        """Count a match arm unless its pattern is the wildcard underscore."""
        if not _is_wildcard_pattern(node.pattern):
            self.complexity += 1
        self.generic_visit(node)
 
    def visit_Assert(self, node: ast.Assert) -> None:
        """Count one assert decision, then inspect its expressions."""
        self.complexity += 1
        self.generic_visit(node)
 
 
def _is_wildcard_pattern(pattern: ast.pattern) -> bool:
    """Return whether a match pattern is the bare wildcard underscore."""
    return (
        isinstance(pattern, ast.MatchAs)
        and pattern.pattern is None
        and pattern.name is None
    )
 
 
class _FunctionCollector(ast.NodeVisitor):
    """Collect independently-scored function records with lexical names."""
 
    def __init__(self, filename: str) -> None:
        self.filename = filename
        self.records: list[FunctionComplexity] = []
        self.scope: list[str] = []
 
    def visit_ClassDef(self, node: ast.ClassDef) -> None:
        """Include class names in method qualnames."""
        self.scope.append(node.name)
        self.generic_visit(node)
        self.scope.pop()
 
    def visit_FunctionDef(self, node: ast.FunctionDef) -> None:
        """Record and recurse through a synchronous function scope."""
        self._record_function(node)
        self.scope.append(node.name)
        self.generic_visit(node)
        self.scope.pop()
 
    def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None:
        """Record and recurse through an asynchronous function scope."""
        self._record_function(node)
        self.scope.append(node.name)
        self.generic_visit(node)
        self.scope.pop()
 
    def visit_Lambda(self, node: ast.Lambda) -> None:
        """Record a lambda with a source-coordinate-qualified stable name."""
        local_name = f"<lambda>@{node.lineno}:{node.col_offset}"
        qualname = ".".join([*self.scope, local_name])
        counter = _DecisionCounter()
        self.records.append(
            FunctionComplexity(
                file=self.filename,
                qualname=qualname,
                lineno=node.lineno,
                end_lineno=getattr(node, "end_lineno", node.lineno),
                complexity=counter.count_lambda(node),
            )
        )
        self.scope.append(local_name)
        self.generic_visit(node)
        self.scope.pop()
 
    def _record_function(
        self,
        node: ast.FunctionDef | ast.AsyncFunctionDef,
    ) -> None:
        """Append one scored named-function record."""
        counter = _DecisionCounter()
        self.records.append(
            FunctionComplexity(
                file=self.filename,
                qualname=".".join([*self.scope, node.name]),
                lineno=node.lineno,
                end_lineno=getattr(node, "end_lineno", node.lineno),
                complexity=counter.count_function(node),
            )
        )
 
 
def analyze_source(source: str, filename: str = "<unknown>") -> list[FunctionComplexity]:
    """Analyze Python source text and return records in source order.
 
    ``SyntaxError`` is intentionally allowed to reach the caller. Directory
    orchestration records that error explicitly instead of silently omitting
    the file.
    """
    tree = ast.parse(source, filename=filename)
    collector = _FunctionCollector(filename)
    collector.visit(tree)
    return sorted(
        collector.records,
        key=lambda item: (item.lineno, item.qualname),
    )
 
 
def analyze_file(path: str | Path) -> list[FunctionComplexity]:
    """Analyze one Python file using its declared source encoding."""
    resolved = Path(path).resolve()
    with tokenize.open(resolved) as source_file:
        source = source_file.read()
    return analyze_source(source, filename=str(resolved))