Stall-Aware Path-Value Density

The problem

Finding the shortest route is easy, but it often ignores the flow of movement, leading to paths that are jerky or inefficient to actually perform.

What it does

It calculates a route and ranks it based on both the shortest physical distance and the smoothness of the movement flow.

Why it matters

It ensures that the chosen path is both efficient and easy to execute without unnecessary interruptions.

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 stall_aware_pathfinding_v2.py
Path found with 14 nodes
-> (0, 0)
-> (10, 0)
-> (10, 2)
-> (12, 2)
-> (12, 4)
-> (13, 4)
-> (13, 8)
-> (16, 8)
-> (16, 16)
-> (17, 16)
-> (17, 18)
-> (18, 18)
-> (18, 19)
-> (19, 19)

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 — 124 lines, one file, standard library only.

import heapq


# Stall-Aware Path-Value Density Implementation with Dynamic Cost Weighting

class Node:
    def __init__(self, x, y, blocked=False):
        self.x = x
        self.y = y
        self.blocked = blocked
        self.g = 0
        self.h = 0
        self.f = 0
        self.parent = None
        self.visits = 0
        self.last_visited = 0
        self.loop_penalty = 0
        self.stall_threshold = 3 # Detect stalls after 3 repeated visits
        self.stall_sensitivity = 1.5 # New dynamic scaling factor

    def calculate_heuristic(self, end_node):
        return abs(self.x - end_node.x) + abs(self.y - end_node.y)

    def get_neighbors(self, grid):
        neighbors = []
        directions = [(1, 0), (-1, 0), (0, 1), (0, -1)]
        for dx, dy in directions:
            nx, ny = self.x + dx, self.y + dy
            if 0 <= nx < len(grid) and 0 <= ny < len(grid[0]):
                neighbor = grid[nx][ny]
                if not neighbor.blocked:
                    neighbors.append(neighbor)
        return neighbors

    def __lt__(self, other):
        return self.f < other.f


class StallAwarePathFinder:
    def __init__(self, grid_size, obstacles=[]):
        self.grid = [[Node(x, y, (x, y) in obstacles) for y in range(grid_size[1])] for x in range(grid_size[0])]
        self.agents = []

    def detect_stall(self, node, current_time):
        if node.visits >= node.stall_threshold:
            time_diff = current_time - node.last_visited
            if time_diff < 10:
                # Dynamic scaling of penalty based on sensitivity factor
                node.loop_penalty = node.loop_penalty * node.stall_sensitivity + 1
                node.visits += 1
                node.last_visited = current_time

    def smooth_path(self, path):
        simplified = []
        for i, node in enumerate(path):
            if i == 0 or i == len(path)-1:
                simplified.append(node)
            else:
                prev = path[i-1]
                next = path[i+1]
                if not (node.x == prev.x and node.x == next.x) and not (node.y == prev.y and node.y == next.y):
                    simplified.append(node)
        return simplified

    def find_path(self, start, end):
        open_list = []
        closed_list = set()
        current_time = 0
        start.g = 0
        start.h = start.calculate_heuristic(end)
        start.f = start.g + start.h
        heapq.heappush(open_list, start)

        while open_list:
            current = heapq.heappop(open_list)
            current_time += 1

            if current == end:
                return self.smooth_path(self.reconstruct_path(current))

            self.detect_stall(current, current_time)
            closed_list.add(current)

            for neighbor in current.get_neighbors(self.grid):
                if neighbor in closed_list:
                    continue

                neighbor.g = current.g + 1
                neighbor.h = neighbor.calculate_heuristic(end)

                # Dynamic cost weighting based on stall sensitivity
                neighbor.f = neighbor.g + neighbor.h + (neighbor.loop_penalty * 0.5)

                if any(n.x == neighbor.x and n.y == neighbor.y for n in open_list):
                    continue
                heapq.heappush(open_list, neighbor)
                neighbor.parent = current

        return None

    def reconstruct_path(self, end_node):
        path = []
        current = end_node
        while current:
            path.append(current)
            current = current.parent
        return path[::-1]


# Example usage with dynamic cost demonstration
if __name__ == '__main__':
    # Create a 20x20 grid with some obstacles
    obstacles = [(2, 3), (4, 5), (6, 5), (8, 5), (10, 5), (12, 5), (14, 5), (16, 5), (18, 5)]
    finder = StallAwarePathFinder(grid_size=(20, 20), obstacles=obstacles)
    start_node = finder.grid[0][0]
    end_node = finder.grid[19][19]
    path = finder.find_path(start_node, end_node)

    if path:
        print(f'Path found with {len(path)} nodes')
        for node in path:
            print(f'-> ({node.x}, {node.y}) [Loop Penalty: {node.loop_penalty}]')
    else:
        print('No path found')