07 - Complexity Assessment and Cognitive Load Metrics

# Cognitive complexity calculation
class CognitiveComplexityCalculator:
    """Calculates cognitive complexity following the SonarSource model"""

    def __init__(self):
        self.complexity = 0
        self.nesting_level = 0

    def calculate(self, ast_node):
        """Calculates the cognitive complexity of a function"""
        self.complexity = 0
        self.nesting_level = 0
        self._visit(ast_node)
        return self.complexity

    def _visit(self, node):
        """Recursive AST visit with incremental calculation"""
        import ast

        # Structural increments (B1): +1 for break in linear flow
        if isinstance(node, (ast.If, ast.For, ast.While)):
            self.complexity += 1  # base increment
            self.complexity += self.nesting_level  # nesting increment
            self.nesting_level += 1
            for child in ast.iter_child_nodes(node):
                self._visit(child)
            self.nesting_level -= 1
            return

        # Increments for compound logical operators
        if isinstance(node, ast.BoolOp):
            # Sequences of and/or count +1 per operator switch
            self.complexity += 1

        # Increments for flow breaks: else, elif, except
        if isinstance(node, ast.ExceptHandler):
            self.complexity += 1
            self.complexity += self.nesting_level

        # Recursion and goto (in languages that support them): +1
        # Nested lambdas: +1 + nesting
        if isinstance(node, ast.Lambda):
            self.complexity += 1 + self.nesting_level

        for child in ast.iter_child_nodes(node):
            self._visit(child)

# Example of AI code with high cognitive complexity
# vs refactored version

# HIGH: Cognitive Complexity = 21
def process_order_ai(order):
    if order:                                      # +1
        if order.status == "pending":              # +2 (nesting=1)
            for item in order.items:               # +3 (nesting=2)
                if item.quantity > 0:              # +4 (nesting=3)
                    if item.in_stock:              # +5 (nesting=4)
                        try:                       #
                            charge(item)           #
                        except PaymentError:       # +6 (nesting=4)
                            if item.retry_count < 3:  # +7 (nesting=5)
                                retry(item)
                            else:
                                cancel(item)
    return None

# LOW: Cognitive Complexity = 7
def process_order_refactored(order):
    if not order or order.status != "pending":     # +1
        return None

    for item in order.items:                       # +1
        process_single_item(item)                  # extracted function

def process_single_item(item):
    if item.quantity <= 0 or not item.in_stock:   # +1 (+1 operator)
        return
    try_charge_item(item)                          # extracted function

def try_charge_item(item):
    try:
        charge(item)
    except PaymentError:                           # +1
        if item.retry_count < 3:                   # +2 (nesting=1)
            retry(item)
        else:                                      # +1
            cancel(item)
      

import math

class HalsteadMetrics:
    """Calculates Halstead metrics for source code"""

    def __init__(self, operators, operands):
        # n1 = number of distinct operators
        # n2 = number of distinct operands
        # N1 = total number of operators
        # N2 = total number of operands
        self.n1 = len(set(operators))
        self.n2 = len(set(operands))
        self.N1 = len(operators)
        self.N2 = len(operands)

    def vocabulary(self):
        """n = n1 + n2: program vocabulary"""
        return self.n1 + self.n2

    def length(self):
        """N = N1 + N2: program length"""
        return self.N1 + self.N2

    def volume(self):
        """V = N * log2(n): program volume"""
        n = self.vocabulary()
        N = self.length()
        return N * math.log2(n) if n > 0 else 0

    def difficulty(self):
        """D = (n1/2) * (N2/n2): program difficulty"""
        if self.n2 == 0:
            return 0
        return (self.n1 / 2) * (self.N2 / self.n2)

    def effort(self):
        """E = D * V: implementation effort"""
        return self.difficulty() * self.volume()

    def time_to_program(self):
        """T = E / 18: estimated time in seconds (Stroud number)"""
        return self.effort() / 18

    def bugs_estimate(self):
        """B = V / 3000: bug estimate (empirical metric)"""
        return self.volume() / 3000

    def summary(self):
        """Complete metrics summary"""
        return {
            "vocabulary": self.vocabulary(),
            "length": self.length(),
            "volume": round(self.volume(), 2),
            "difficulty": round(self.difficulty(), 2),
            "effort": round(self.effort(), 2),
            "time_seconds": round(self.time_to_program(), 2),
            "estimated_bugs": round(self.bugs_estimate(), 3)
        }
      

# Architecture Fitness Functions for AI code
class ArchitectureFitness:
    """Evaluates architectural adherence of AI code"""

    def __init__(self, project_structure, architecture_rules):
        self.structure = project_structure
        self.rules = architecture_rules

    def evaluate_all(self):
        """Executes all fitness functions"""
        return {
            "coupling": self._check_coupling(),
            "cohesion": self._check_cohesion(),
            "layer_violations": self._check_layer_violations(),
            "circular_dependencies": self._check_circular_deps(),
            "component_size": self._check_component_size(),
            "overall_fitness": self._calculate_overall_score()
        }

    def _check_coupling(self):
        """Measures coupling between modules"""
        # Afferent coupling (Ca): who depends on me
        # Efferent coupling (Ce): who I depend on
        # Instability = Ce / (Ca + Ce)
        results = []
        for module in self.structure.modules:
            ca = len(module.dependents)
            ce = len(module.dependencies)
            instability = ce / (ca + ce) if (ca + ce) > 0 else 0
            results.append({
                "module": module.name,
                "afferent": ca,
                "efferent": ce,
                "instability": round(instability, 2),
                "status": "OK" if instability < 0.7 else "WARNING"
            })
        return results

    def _check_layer_violations(self):
        """Verifies that architectural layers are respected"""
        # E.g.: presentation must not import from data layer
        violations = []
        for rule in self.rules.get("layer_rules", []):
            source = rule["from_layer"]
            forbidden = rule["cannot_import"]
            imports = self.structure.get_imports(source)
            for imp in imports:
                if any(f in imp for f in forbidden):
                    violations.append({
                        "from": source,
                        "imports": imp,
                        "forbidden_layer": forbidden,
                        "severity": "HIGH"
                    })
        return violations