Flow-Constrained Spatial Pathfinder

The problem

Moving multiple items through a shared space is difficult because they can get stuck or create traffic jams. Standard pathfinding often fails because it doesn't account for how many items a single spot can hold at once.

What it does

It calculates routes for multiple agents on a grid while ensuring no location exceeds a set capacity. It prioritizes finding a path that actually works over simply finding the shortest route.

Why it matters

It ensures that movement remains possible and fluid by respecting the physical limits of the space.

Validation

It was run in the sandbox and it failed. run output shows an error/traceback — the artifact does NOT run clean.

$ python3 flow_constrained_pathfinder_v2.py
Traceback (most recent call last):
  File "/work/flow_constrained_pathfinder.py", line 119, in <module>
    main()
  File "/work/flow_constrained_pathfinder.py", line 20, in main
    {{"start": (0, 0), "goal": (4, 4)}},
    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
TypeError: unhashable type: 'dict'

No screenshot — there is nothing working to show. This is recorded as an unfinished sketch so the attempt stays visible instead of being quietly dropped.

The code

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

#!/usr/bin/env python3
from collections import defaultdict
import heapq
import json

def main():
    # Define grid and capacities
    grid_width = 5
    grid_height = 5
    capacities = [
        [3, 2, 1, 2, 3],
        [2, 1, 3, 1, 2],
        [1, 3, 2, 3, 1],
        [2, 1, 3, 1, 2],
        [3, 2, 1, 2, 3]
    ]
    
    # Define agents
    agents = [
        {"start": (0, 0), "goal": (4, 4)},
        {"start": (4, 0), "goal": (0, 4)}
    ]
    
    # Initialize grid_usage: track agent counts per cell per time step
    grid_usage = defaultdict(lambda: defaultdict(lambda: defaultdict(int)))
    paths = []
    
    # Process each agent with synchronized grid_usage
    for agent in agents:
start_pos = agent['start']
    goal_pos = agent['goal']
        
        # Run A* with shared grid_usage
        path = a_star(start_pos, goal_pos, grid_width, grid_height, capacities, grid_usage)
        
        if path is None:
            print(f'No path found for agent starting at {start}')
            continue
        
        # Update grid_usage with this agent's path
        for step, (x, y) in enumerate(path):
            grid_usage[x][y][step] += 1
        paths.append({"agent": agent, "path": path})
    
    # Output JSON
    output = {
        "grid": capacities,
        "agents": agents,
        "paths": [p['path'] for p in paths]
    }
    print(json.dumps(output, indent=2))

def a_star(start_pos, goal_pos, width, height, capacities, grid_usage):
    open_list = []
    start_h = abs(start_pos[0]-goal_pos[0]) + abs(start_pos[1]-goal_pos[1])
    heapq.heappush(open_list, (start_h, 0, start_pos[0], start_pos[1], 0, [start_pos]))
    closed_list = set()
    
    while open_list:
        current = heapq.heappop(open_list)
        current_x, current_y, current_time = current[2], current[3], current[4]
        current_path = current[5]
        
        if (current_x, current_y, current_time) in closed_list:
            continue
        
        if (current_x, current_y) == goal:
            return current_path
        
        closed_list.add((current_x, current_y, current_time))
        
        # Explore neighbors including wait (0,0)
        for dx, dy in [(-1,0), (1,0), (0,-1), (0,1), (0,0)]:
            new_x = current_x + dx
            new_y = current_y + dy
            new_time = current_time + 1
            
            if new_x < 0 or new_x >= width or new_y < 0 or new_y >= height:
                continue
            
            # Check capacity at (new_x, new_y) for new_time
            cell_capacity = capacities[new_x][new_y]
            current_usage = grid_usage[new_x][new_y][new_time]
            
            # Allow overriding capacity by 1 for synchronization
            if current_usage > cell_capacity - 1:
                continue
            
            # Check closed list
            if (new_x, new_y, new_time) in closed_list:
                continue
            
            # Calculate heuristic (Manhattan distance)
            h = abs(new_x - goal[0]) + abs(new_y - goal[1])
            g = current[1] + 1
            f = g + h
            
            # Add to open list
            new_path = current_path + [(new_x, new_y)]
            heapq.heappush(open_list, (f, g, new_x, new_y, new_time, new_path))
    
    return None

if __name__ == "__main__":
    main()