java-topology/defects/flightgear-0003/test/test_flightgear_0003.py

"""
flightgear-0003: Airway::Network::search2 A* linear findInOpen O(E*V)
    DEFECT: findInOpen scans open-node heap linearly per edge, O(E*V)
            Comment in source: "Inefficient (linear) helper"
    FIX:    unordered_map for O(1) open-node lookup

Simulates our airway A* pathfinding. Real-world airway networks have
thousands of navaids connected by airways.
"""
import time
import heapq
import random
import math


def heuristic(node, goal, positions):
    """Euclidean distance heuristic."""
    x1, y1 = positions[node]
    x2, y2 = positions[goal]
    return math.sqrt((x1 - x2)**2 + (y1 - y2)**2)


def build_airway_network(n_nodes):
    """Build a random airway network."""
    positions = {}
    for i in range(n_nodes):
        positions[i] = (random.uniform(0, 1000), random.uniform(0, 1000))

    adj = {i: [] for i in range(n_nodes)}
    for i in range(n_nodes):
        # Connect to 3-6 nearest neighbors
        dists = []
        for j in range(n_nodes):
            if i == j:
                continue
            d = heuristic(i, j, positions)
            dists.append((d, j))
        dists.sort()
        for d, j in dists[:random.randint(3, 6)]:
            adj[i].append((j, d))
            adj[j].append((i, d))
    return adj, positions


def astar_unpatched(adj, positions, start, goal):
    """Original: linear findInOpen per edge."""
    INF = float('inf')
    open_list = []  # (f_score, node, g_score, prev)
    open_set_list = []  # list of (node, ref_index) for linear findInOpen
    closed = set()

    g = {start: 0.0}
    h = heuristic(start, goal, positions)
    entry = [h, start, 0.0, None]
    open_list.append(entry)
    open_set_list.append(entry)
    heapq.heapify(open_list)

    ops = 0
    while open_list:
        f, current, g_curr, prev = heapq.heappop(open_list)
        if current in closed:
            continue
        closed.add(current)

        if current == goal:
            return g_curr, ops

        for neighbor, weight in adj[current]:
            ops += 1
            if neighbor in closed:
                continue

            tentative_g = g_curr + weight

            # Linear findInOpen: scan entire open list for neighbor
            found = None
            for entry in open_set_list:
                ops += 1
                if entry[1] == neighbor and entry[1] not in closed:
                    found = entry
                    break

            if found is not None:
                if tentative_g < found[2]:
                    # Update: rebuild heap
                    found[2] = tentative_g
                    found[0] = tentative_g + heuristic(neighbor, goal, positions)
                    heapq.heapify(open_list)
            else:
                if neighbor not in g or tentative_g < g[neighbor]:
                    g[neighbor] = tentative_g
                    h_val = heuristic(neighbor, goal, positions)
                    entry = [tentative_g + h_val, neighbor, tentative_g, current]
                    heapq.heappush(open_list, entry)
                    open_set_list.append(entry)

    return INF, ops


def astar_patched(adj, positions, start, goal):
    """Patched: hash map for O(1) open-node lookup."""
    INF = float('inf')
    open_list = []  # heap
    open_map = {}   # node -> entry (O(1) lookup)
    closed = set()

    g = {start: 0.0}
    h = heuristic(start, goal, positions)
    entry = [h, start, 0.0, None]
    open_list.append(entry)
    open_map[start] = entry
    heapq.heapify(open_list)

    ops = 0
    while open_list:
        f, current, g_curr, prev = heapq.heappop(open_list)
        if current in closed:
            continue
        closed.add(current)
        open_map.pop(current, None)

        if current == goal:
            return g_curr, ops

        for neighbor, weight in adj[current]:
            ops += 1
            if neighbor in closed:
                continue

            tentative_g = g_curr + weight

            # O(1) lookup instead of linear scan
            found = open_map.get(neighbor)
            ops += 1

            if found is not None:
                if tentative_g < found[2]:
                    found[2] = tentative_g
                    found[0] = tentative_g + heuristic(neighbor, goal, positions)
                    heapq.heapify(open_list)
            else:
                if neighbor not in g or tentative_g < g[neighbor]:
                    g[neighbor] = tentative_g
                    h_val = heuristic(neighbor, goal, positions)
                    entry = [tentative_g + h_val, neighbor, tentative_g, current]
                    heapq.heappush(open_list, entry)
                    open_map[neighbor] = entry

    return INF, ops


def test_correctness():
    """Verify both produce same path cost."""
    random.seed(42)
    n = 100
    adj, positions = build_airway_network(n)
    d_unpatched, _ = astar_unpatched(adj, positions, 0, n-1)
    d_patched, _ = astar_patched(adj, positions, 0, n-1)
    assert abs(d_unpatched - d_patched) < 1e-6, \
        f"Distances differ: {d_unpatched} vs {d_patched}"
    print(f"PASS correctness: distance={d_unpatched:.2f}")


def test_performance():
    """Benchmark at V=800 (realistic airway network size)."""
    random.seed(42)
    n = 800
    adj, positions = build_airway_network(n)

    iters = 5
    pairs = [(random.randint(0, n-1), random.randint(0, n-1)) for _ in range(iters)]

    t0 = time.perf_counter()
    ops_u = 0
    for s, e in pairs:
        _, o = astar_unpatched(adj, positions, s, e)
        ops_u += o
    t_unpatched = time.perf_counter() - t0

    t0 = time.perf_counter()
    ops_p = 0
    for s, e in pairs:
        _, o = astar_patched(adj, positions, s, e)
        ops_p += o
    t_patched = time.perf_counter() - t0

    ratio = t_unpatched / t_patched if t_patched > 0 else float('inf')
    op_ratio = ops_u / ops_p if ops_p > 0 else float('inf')
    print(f"V={n}, {iters} pathfinds")
    print(f"Unpatched: {t_unpatched*1e3:.1f} ms, {ops_u} ops")
    print(f"Patched:   {t_patched*1e3:.1f} ms, {ops_p} ops")
    print(f"Time ratio:  {ratio:.1f}x")
    print(f"Ops ratio:   {op_ratio:.1f}x")

    assert ratio > 2.0, f"Expected >2x speedup, got {ratio:.1f}x"
    print(f"PASS performance: {ratio:.1f}x speedup")


if __name__ == "__main__":
    test_correctness()
    test_performance()