Dynamic Path Costing A*

The problem

Finding the most efficient path is difficult when some locations act as crowded hubs or bottlenecks. Standard navigation doesn't account for how busy or connected a specific point is.

What it does

It calculates paths by adjusting the 'cost' of a route based on how many connections a location has. It treats highly connected hubs as more expensive to move through.

Why it matters

It allows for pathfinding that accounts for network congestion and hub density rather than just physical distance.

Validation

It was run inside an isolated container with no network access. This is the exact command and the real output it produced — captured process output, not written by a model.

$ python3 dynamic_path_costing_astar.py
=== DPC-A* ===
Std A*    path=['a', 'h', 'far'] cost=11.00 explored=4
DPC-A*    path=['a', 'h', 'far'] cost=13.67 explored=5
Centrality: {'far': 0.167, 'a': 0.5, 'c': 0.5, 'd': 0.5, 'h': 1.0, 'e': 0.5, 'b': 0.5}
Same route. DPC cost 124.2% of standard (inflation via hub penalty).

A screenshot of that run.

A clean run proves this does what is shown above, in a CPU-only sandbox. It is a small research demo — not a production tool, and nothing here was published anywhere.

The code

All of it — 216 lines, one file, standard library only.

#!/usr/bin/env python3
"""
Dynamic Path Costing A* (DPC-A*)
Edge traversal cost = base_weight * (1 + alpha * target_node_centrality)

Standard A* on static graphs is recovered when alpha=0.
"""
from __future__ import annotations

import heapq
import math
from collections import defaultdict
from itertools import count
from typing import Any

Node = str

# ---------------------------------------------------------------------------
# Graph
# ---------------------------------------------------------------------------

class Graph:
    def __init__(self) -> None:
        self._adj: dict[Node, dict[Node, float]] = defaultdict(dict)
        self._nodes: set[Node] = set()

    def add_edge(self, u: Node, v: Node, weight: float) -> None:
        if weight < 0:
            raise ValueError(f"Negative edge weight ({u}--{v} = {weight}) not supported")
        self._adj[u][v] = weight
        self._adj[v][u] = weight
        self._nodes.update((u, v))

    def nodes(self) -> set[Node]:
        return set(self._nodes)

    def neighbours(self, u: Node) -> dict[Node, float]:
        return dict(self._adj.get(u, {}))

    def degree(self, u: Node) -> int:
        return len(self._adj.get(u, {}))

    def edge_weight(self, u: Node, v: Node) -> float | None:
        return self._adj.get(u, {}).get(v)


# ---------------------------------------------------------------------------
# Centrality
# ---------------------------------------------------------------------------

def degree_centrality(graph: Graph) -> dict[Node, float]:
    nodes = graph.nodes()
    n = len(nodes)
    if n <= 1:
        return {node: 0.0 for node in nodes}
    return {node: graph.degree(node) / (n - 1) for node in nodes}


def dynamic_cost(
    graph: Graph,
    u: Node,
    v: Node,
    centrality: dict[Node, float],
    alpha: float = 1.0,
) -> float:
    base = graph.edge_weight(u, v)
    if base is None:
        raise KeyError(f"edge {u}--{v} not in graph")
    return base * (1.0 + alpha * centrality[v])


# ---------------------------------------------------------------------------
# Heuristic helpers
# ---------------------------------------------------------------------------

def zero_heuristic(_u: Node, _goal: Node) -> float:
    return 0.0


def euclidean_heuristic(
    positions: dict[Node, tuple[float, float]],
) -> Any:
    def h(u: Node, goal: Node) -> float:
        x1, y1 = positions[u]
        x2, y2 = positions[goal]
        return math.hypot(x2 - x1, y2 - y1)
    return h


# ---------------------------------------------------------------------------
# DPC-A* algorithm
# ---------------------------------------------------------------------------

def dpc_astar(
    graph: Graph,
    start: Node,
    goal: Node,
    *,
    alpha: float = 1.0,
    heuristic: Any = None,
    positions: dict[Node, tuple[float, float]] | None = None,
) -> dict[str, Any]:
    centrality = degree_centrality(graph)

    if heuristic is None:
        heuristic = euclidean_heuristic(positions) if positions else zero_heuristic

    if start not in graph.nodes() or goal not in graph.nodes():
        raise ValueError(f"'{start}' or '{goal}' not in graph")

    counter = count()
    open_set: list[tuple[float, int, Node]] = [
        (heuristic(start, goal), next(counter), start)
    ]
    closed: set[Node] = set()
    came_from: dict[Node, tuple[Node | None, float]] = {start: (None, 0.0)}
    g_score: dict[Node, float] = {start: 0.0}

    while open_set:
        _, _, current = heapq.heappop(open_set)

        if current in closed:
            continue
        closed.add(current)

        if current == goal:
            path = _reconstruct(came_from, start, goal)
            return {
                "path": path,
                "cost": g_score[goal],
                "explored": len(closed),
                "centrality": centrality,
            }

        for neighbor, _ in graph.neighbours(current).items():
            if neighbor in closed:
                continue

            edge_cost = dynamic_cost(graph, current, neighbor, centrality, alpha)
            tent = g_score[current] + edge_cost

            if tent < g_score.get(neighbor, math.inf):
                g_score[neighbor] = tent
                f = tent + heuristic(neighbor, goal)
                heapq.heappush(open_set, (f, next(counter), neighbor))
                came_from[neighbor] = (current, edge_cost)

    return {
        "path": None,
        "cost": None,
        "explored": len(closed),
        "centrality": centrality,
    }


def _reconstruct(
    came_from: dict[Node, tuple[Node | None, float]],
    start: Node,
    goal: Node,
) -> list[Node]:
    path: list[Node] = []
    cur = goal
    while cur is not None:
        path.append(cur)
        cur = came_from[cur][0]
    path.reverse()
    return path


# ---------------------------------------------------------------------------
# Demo
# ---------------------------------------------------------------------------

def demo() -> None:
    g = Graph()

    # Outer pentagon
    for i in range(5):
        g.add_edge(chr(97 + i), chr(97 + (i + 1) % 5), weight=1.0)  # a-b, b-c, ...

    # Hub connections (makes hub high centrality)
    hub = "h"
    for ch in "abcde":
        g.add_edge(ch, hub, weight=1.0)

    # Far leaf
    g.add_edge(hub, "far", weight=10.0)

    xy = {
        "a": (1.0, 0.0),
        "b": (0.3, 0.95),
        "c": (-0.8, 0.6),
        "d": (-0.8, -0.6),
        "e": (0.3, -0.95),
        "h": (0.0, 0.0),
        "far": (10.0, 10.0),
    }

    print("=== DPC-A* ===")

    std = dpc_astar(g, "a", "far", alpha=0, positions=xy)
    dpc = dpc_astar(g, "a", "far", alpha=1, positions=xy)

    print(f"Std A*    path={std['path']} cost={std['cost']:.2f} explored={std['explored']}")
    print(f"DPC-A*    path={dpc['path']} cost={dpc['cost']:.2f} explored={dpc['explored']}")
    print(f"Centrality: { {k: round(v, 3) for k, v in dpc['centrality'].items()} }")

    if std["path"] != dpc["path"]:
        print("DPC chose a different (more sparse) route to avoid high-centrality nodes.")
    else:
        ratio = dpc["cost"] / std["cost"] * 100
        print(f"Same route. DPC cost {ratio:.1f}% of standard (inflation via hub penalty).")


if __name__ == "__main__":
    demo()