Skip to content

analysis blast_radius

aakash-anko edited this page May 25, 2026 · 1 revision

analysis/blast_radius.py

Calculates how many files break when you change a given file. Uses BFS on a reversed dependency graph to find all direct and transitive dependents.


Key Concepts

Term Definition Example
vertex A node in a graph representing a single entity (a file, module, etc.). In a file graph, pipeline.py is one vertex.
edge A connection between two vertices in a graph, representing a relationship (e.g., an import). If pipeline.py imports scanner.py, there's a directed edge pipeline.py → scanner.py.
in-degree Number of edges pointing INTO a vertex (how many files import this file). utils.py with in-degree=15 means 15 other files import it.
blast radius All files that would be affected if a given file changes — found by following reverse import edges transitively. If A imports B and C imports A, changing B has blast radius = {A, C}.
transitive dependency An indirect dependency through a chain. If A imports B and B imports C, then A transitively depends on C. Changing C could break A even though A never directly imports C.
cosine distance Measures how different two vectors are. 0.0 = identical meaning, 1.0 = completely different, 2.0 = opposite. Query "scan files" has cosine distance 0.15 to scan_directory() (very similar) and 0.85 to grade_answer() (very different).
AST Abstract Syntax Tree — a tree representation of source code structure, where each node is a language construct (function, class, if-statement, etc.). def add(a, b): return a+b becomes a tree: FunctionDef → [args: a, b] → [body: Return → BinOp(a + b)].
igraph A high-performance C library for graph analysis with Python bindings. Much faster than pure-Python graph libraries. ig.Graph.TupleList([("a.py", "b.py"), ("b.py", "c.py")], directed=True) builds a graph with 3 vertices and 2 edges instantly.

Source: src/codewalk/analysis/blast_radius.py


build_reverse_graph

Reverses a dependency graph so edges point from "imported file" → "importing file" instead of "importer" → "imported".

Example

Input graph: {
    "pipeline.py": ["config.py", "scanner.py"],
    "scanner.py":  ["config.py"],
    "config.py":   [],
}

Line 14: internal_files = {"pipeline.py", "scanner.py", "config.py"}

Line 15: reverse = {"pipeline.py": [], "scanner.py": [], "config.py": []}

Lines 17–20: Loop through each file and its deps, appending the importer to the imported file's list:

  • file = "pipeline.py", deps = ["config.py", "scanner.py"]
    • dep = "config.py" → in internal_files ✓ → reverse["config.py"].append("pipeline.py")
    • dep = "scanner.py" → in internal_files ✓ → reverse["scanner.py"].append("pipeline.py")
  • file = "scanner.py", deps = ["config.py"]
    • dep = "config.py" → in internal_files ✓ → reverse["config.py"].append("scanner.py")
  • file = "config.py", deps = [] → nothing added

Return:

{
    "pipeline.py": [],
    "scanner.py":  ["pipeline.py"],
    "config.py":   ["pipeline.py", "scanner.py"],
}

get_blast_radius

Calculates blast radius for a single file — who breaks if this file changes. Supports two backends: GraphRuntime (igraph) and plain dict (legacy BFS).

Example (dict/legacy path)

Input target_file: "config.py"
Input graph: {
    "pipeline.py": ["config.py", "scanner.py"],
    "scanner.py":  ["config.py"],
    "config.py":   [],
}

Line 62: isinstance(graph, GraphRuntime) → False, falls into the else branch

Line 63: reverse = build_reverse_graph(graph)

{
    "pipeline.py": [],
    "scanner.py":  ["pipeline.py"],
    "config.py":   ["pipeline.py", "scanner.py"],
}

Line 64: internal_files = {"pipeline.py", "scanner.py", "config.py"}

Line 66: target_file = "config.py" is in internal_files ✓ (skip the early-return branch)

Line 75: visited = {"config.py"}
Line 76: queue = deque()
Line 77: impact_tree = {}

Lines 79–82: Seed the queue with direct dependents from reverse["config.py"] = ["pipeline.py", "scanner.py"]:

  • dependent = "pipeline.py" → not in visited → queue.append(("pipeline.py", 1)), visited = {"config.py", "pipeline.py"}
  • dependent = "scanner.py" → not in visited → queue.append(("scanner.py", 1)), visited = {"config.py", "pipeline.py", "scanner.py"}

BFS iteration 1:

  • Line 85: current_file = "pipeline.py", depth = 1
  • Line 86: impact_tree = {"pipeline.py": 1}
  • Lines 88–91: reverse["pipeline.py"] = [] → nothing to add

BFS iteration 2:

  • Line 85: current_file = "scanner.py", depth = 1
  • Line 86: impact_tree = {"pipeline.py": 1, "scanner.py": 1}
  • Lines 88–91: reverse["scanner.py"] = ["pipeline.py"]"pipeline.py" already in visited → skip

Line 93: direct = ["pipeline.py", "scanner.py"] (depth == 1)
Line 94: transitive = [] (depth > 1, none)
Line 95: total_affected = 2
Line 96: total_files = 3
Line 97: risk_level = _calculate_risk(2, 3)2/3 = 0.667 > 0.5"critical"

Return:

{
    "file": "config.py",
    "direct": ["pipeline.py", "scanner.py"],
    "transitive": [],
    "affected_files": 2,
    "risk_level": "critical",
    "impact_tree": {"pipeline.py": 1, "scanner.py": 1},
}

Example (GraphRuntime/igraph path)

When graph is a GraphRuntime instance:

Line 30: Gets vertex index for target_file using graph._find_vertex()
Line 42: Runs igraph's shortest_paths(source=idx, mode="in") — computes shortest distance from every vertex to target_file following incoming edges (who imports it)
Line 47–48: Loops through distances, skips self and infinity (unreachable), builds impact_tree with {filename: distance}
Line 50–51: Splits into direct (distance==1) and transitive (distance>1)

Same return shape as the dict path.


_calculate_risk

Assigns a risk label based on how many files are affected and what fraction of the codebase that represents.

Example

Input affected: 5
Input total: 20

Line 108: total = 20, not 0 → skip "none"
Line 109: ratio = 5 / 20 = 0.25
Line 110: 0.25 > 0.5? No. 5 >= 20? No → skip "critical"
Line 112: 0.25 > 0.25? No. 5 >= 10? No → skip "high"
Line 114: 0.25 > 0.10? Yes → Return: "moderate"

Thresholds

Risk Level Ratio OR Count
critical > 50% OR ≥ 20 files
high > 25% OR ≥ 10 files
moderate > 10% OR ≥ 4 files
low everything else

calculate_full_blast_map

Computes blast radius for EVERY file in the graph, then ranks them from most affected to least.

Example (dict/legacy path)

Input graph: {
    "pipeline.py": ["config.py", "scanner.py"],
    "scanner.py":  ["config.py"],
    "config.py":   [],
}

Line 179: reverse = build_reverse_graph(graph) → same as before
Line 180: internal_files = {"pipeline.py", "scanner.py", "config.py"}
Line 181: total_files = 3
Line 184: risk_counts = {"critical": 0, "high": 0, "moderate": 0, "low": 0, "none": 0}

Iteration: target_file = "pipeline.py":

  • BFS from "pipeline.py" through reverse graph
  • reverse["pipeline.py"] = [] → no dependents
  • impact_tree = {}, total_affected = 0
  • risk_level = _calculate_risk(0, 3)ratio = 0"low"
  • Appends {"file": "pipeline.py", "affected_files": 0, "risk_level": "low", "direct_count": 0, "transitive_count": 0}

Iteration: target_file = "scanner.py":

  • reverse["scanner.py"] = ["pipeline.py"]
  • BFS finds: impact_tree = {"pipeline.py": 1}, total_affected = 1
  • risk_level = _calculate_risk(1, 3)ratio = 0.33 > 0.25"high"
  • Appends {"file": "scanner.py", "affected_files": 1, "risk_level": "high", "direct_count": 1, "transitive_count": 0}

Iteration: target_file = "config.py":

  • BFS finds impact_tree = {"pipeline.py": 1, "scanner.py": 1}, total_affected = 2
  • risk_level = _calculate_risk(2, 3)"critical"
  • Appends {"file": "config.py", "affected_files": 2, "risk_level": "critical", "direct_count": 2, "transitive_count": 0}

Line 232: Sort by affected_files descending → [config.py(2), scanner.py(1), pipeline.py(0)]
Line 233: highest_risk = "config.py"

Return:

{
    "blast_map": [
        {"file": "config.py",   "affected_files": 2, "risk_level": "critical", "direct_count": 2, "transitive_count": 0},
        {"file": "scanner.py",  "affected_files": 1, "risk_level": "high",     "direct_count": 1, "transitive_count": 0},
        {"file": "pipeline.py", "affected_files": 0, "risk_level": "low",      "direct_count": 0, "transitive_count": 0},
    ],
    "stats": {
        "total_files": 3,
        "critical_files": 1,
        "high_files": 1,
        "moderate_files": 0,
        "low_files": 1,
    },
    "highest_risk": "config.py",
}

Clone this wiki locally