Fly-in

This project has been created as part of the 42 curriculum by vlnikola.

Fly-in

A Python simulation of drones moving through a graph of zones and connections while respecting turn-based movement, zone occupancy, and connection capacity rules.

The main goal is to route every drone from the start hub to the end hub in as few turns as possible while avoiding collisions and respecting all capacity constraints.

Description

Fly-in is built around a modular architecture:

Parser — Parser.parse() reads and validates a custom map format into a Network domain model.
Models — Zone, Connection, Drone, and Network represent the simulation data using Pydantic and dataclasses.
Algorithm — a Cooperative Space-Time A* pathfinder plans collision-free routes for every drone sequentially.
App — Application wires everything together; SimulationController runs the playback loop.
Renderer — two interchangeable implementations: a console printer and an interactive Pygame visualizer.

Project Structure

fly-in/
├── pyproject.toml             # Project metadata and dependencies (uv)
├── uv.lock                    # Dependency lockfile
├── Makefile                   # Build, run, test, lint, and clean targets
├── imgs/
│   └── drone.bmp              # Drone sprite for the Pygame visualizer
├── maps/                      # Map files
│   ├── README.md              #   Map format notes
│   ├── example.txt            #   Small 3-drone example map
│   ├── easy/                  #   Curriculum map sets:
│   ├── medium/                #   easy → medium → hard → challenger
│   ├── hard/
│   ├── challenger/
│   └── my_maps/               #   Hand-made edge-case maps (validation errors)
├── tests/                     # Pytest suite (make test)
│   ├── conftest.py            #   Test configuration
│   ├── builders.py            #   Map / network builders shared by tests
│   ├── test_animation.py      #   Renderer-side animation policy
│   ├── test_application.py    #   Composition root (Application.run)
│   ├── test_cli_logger.py     #   CLI output contract
│   ├── test_collision_manager.py  # Reservation table & capacity rules
│   ├── test_controller.py     #   Playback loop & clock
│   ├── test_hud_stats.py      #   HUD statistics
│   ├── test_network.py        #   Network validation & derivation
│   ├── test_parser.py         #   Golden error contract for Parser.parse
│   ├── test_plan_route.py     #   Route planning against reservations
│   ├── test_position.py       #   Position value object
│   ├── test_renderer_seam.py  #   Renderer protocols
│   └── test_zone.py           #   Zone capacity, cost, and transit
└── src/                       # Main source package
    ├── __init__.py
    ├── __main__.py            # Entry point: python -m src
    ├── app/
    │   ├── application.py     # Application — wires everything together
    │   └── controller.py      # SimulationController — playback loop
    ├── models/                # Domain models
    │   ├── zone.py            #   Zone + Hub + ZoneType enum
    │   ├── position.py        #   Position (zone, next_zone, progress)
    │   ├── connection.py      #   Connection (edge between zones)
    │   ├── drone.py           #   Drone (position_at / arrival_turn)
    │   ├── network.py         #   Network (validated graph + adjacency)
    │   └── temporal_state.py  #   TemporalState (A* search node)
    ├── parsers/               # Input parsing
    │   ├── arg_parser.py      #   ArgParser (argparse)
    │   └── parser.py          #   Parser — map file → Network
    ├── algorithm/             # Pathfinding
    │   └── coop_a_star/       #   Cooperative Space-Time A* implementation
    │       ├── algorithm.py   #     CooperativeAStar (plan_route)
    │       └── manager.py     #     CollisionManager (reservation table)
    └── renderer/              # Output / visualization
        ├── base.py            #   Renderer / InteractiveRenderer protocols
        ├── animation.py       #   Smoothstep + transit fade policy
        ├── cli_logger.py      #   CLILogger — prints turns to stdout
        ├── hud_stats.py       #   HUD statistics dataclass
        └── pygame/            #   Pygame interactive visualizer
            ├── colors.py      #     Color enum
            ├── config.py      #     Display configuration
            └── renderer.py    #     PygameRenderer

Instructions

Requirements

make
Python >= 3.10
uv — fast Python package manager (replaces pip/venv)
pygame >= 2.0.0
pydantic >= 2.5.0
Dev dependencies — pytest, flake8, mypy, ruff (installed via the dev dependency group)

Installation

make install

This uses uv sync to create a virtual environment and install all project dependencies from pyproject.toml.

Run

Running make run without a FILE argument auto-generates a temporary test map that covers all zone types, colors, and capacities. The temp file is cleaned up automatically after the simulation ends.

make run

You can also provide your own map or extra arguments:

# Run a specific map
make run FILE=maps/hard/02_capacity_hell.txt

# Enable the Pygame visualizer at 2x speed
make run ARGS="--renderer pygame --speed=2.0"

# Combine both
make run FILE=maps/hard/02_capacity_hell.txt ARGS="--renderer pygame"

Or run the simulator directly:

python -m src maps/easy/01_linear_path.txt
python -m src maps/medium/03_priority_puzzle.txt --renderer pygame
python -m src maps/hard/02_capacity_hell.txt --renderer pygame --speed 2.0

Tip

When running with --renderer pygame, you can pause/resume with Space, scrub time with Left/Right arrow keys, reset with R, or quit with Esc.

Note

Run from the repository root: the Pygame visualizer loads imgs/drone.bmp relative to the working directory.

Map Format

The parser expects a text file with:

nb_drones: <positive_integer> on the first meaningful line,
exactly one start_hub: entry,
exactly one end_hub: entry,
any number of hub: entries,
connection: entries between previously defined zones.

Zone metadata (inside [brackets]):

Key	Values	Default
`zone`	`normal`, `blocked`, `restricted`, `priority`	`normal`
`color`	any `Colors` enum name — see `src/renderer/pygame/colors.py` (e.g., `green`, `cyan`, `gold`)	`white`
`max_drones`	positive integer	`1`

Connection metadata:

Key	Values	Default
`max_link_capacity`	positive integer	`1`

Example map:

nb_drones: 4
start_hub: start 0 0 [zone=normal color=green max_drones=4]
end_hub: goal 6 0 [zone=normal color=gold]
hub: a 2 0 [zone=priority color=cyan]
hub: b 4 0 [zone=restricted color=purple]
connection: start-a [max_link_capacity=2]
connection: a-b [max_link_capacity=1]
connection: b-goal [max_link_capacity=1]

Validation rules:

Exactly one start_hub and one end_hub must be present.
Zone names cannot contain dashes (-), since dashes are used to delimit connection endpoints.
Duplicate zone names, coordinates, or connections are rejected.
Negative capacities or drone counts are rejected.
Connections to undefined or self-referencing zones are rejected.
A connection with max_link_capacity of 2 or more allows drones to traverse it in opposite directions at the same time (head-on traversal).
Invalid zone types (anything other than normal, blocked, restricted, priority) are rejected.
The max_drones metadata on start_hub and end_hub is ignored — these zones have unlimited capacity.
nb_drones must be between 1 and 1000.
The start_hub and end_hub cannot be blocked zone types.
The end_hub must be reachable from the start_hub through traversable zones (disconnected graphs are rejected).

Algorithm and Implementation Strategy

Application Pipeline

This is what happens when you run the simulator end-to-end:

flowchart TD
    A["__main__.py"] --> B["Application.run()"]
    B --> C["ArgParser.parse(argv)"]
    C --> D["Parser.parse(file)"]
    D --> E["Network\n(validated graph)"]
    E --> F["Drone.create_fleet()"]
    F --> G["CooperativeAStar\n+ CollisionManager"]
    G --> H["Sequential route planning\none drone at a time"]
    H --> I["SimulationController"]
    I --> J{{"--renderer pygame?"}}
    J -->|No| K["CLILogger\nPrint turns to stdout"]
    J -->|Yes| L["CLILogger + PygameRenderer\nTerminal output + Interactive window"]

Step by step:

Application.run() parses CLI arguments and reads the map file.
Parser.parse() validates the file line-by-line and builds a Network object (Pydantic model with validators).
Drone.create_fleet() builds the fleet from nb_drones, and Application creates one CooperativeAStar with a fresh CollisionManager.
Each drone is routed sequentially via plan_route(). Each planned path is immediately committed via CollisionManager.register_path(), so the next drone sees updated reservations.
Finally, a SimulationController drives the Renderer (either CLILogger alone, or CLILogger + PygameRenderer) to visualize the results.

Pathfinding: Cooperative Space-Time A*

The core problem is Multi-Agent Pathfinding (MAPF). To solve this efficiently without the exponential overhead of joint-state searching, the project uses Cooperative Space-Time A*.

Instead of searching in a standard 2D spatial graph, the algorithm searches in a 3D space-time graph where each node is (Zone, Turn).

flowchart TD
    subgraph "For each drone (sequential)"
        S["Start: zone=start, turn=0"] --> Q["Priority Queue (min-heap)"]
        Q --> POP["Pop lowest f_cost state"]
        POP --> GOAL{"Reached end_hub?"}
        GOAL -->|Yes| PATH["Reconstruct path via parent pointers"]
        PATH --> REG["CollisionManager.register_path()\nReserve zone+link slots"]
        GOAL -->|No| VIS{"Already visited\nthis (zone, turn)?"}
        VIS -->|Yes| Q
        VIS -->|No| GEN["generate_valid_neighbors()"]
        GEN --> WAIT["Wait action:\nsame zone, turn+1"]
        GEN --> MOVE["Move action:\nneighbor zone, turn + transit_time"]
        WAIT --> CAP1{"Zone has\ncapacity?"}
        MOVE --> CAP2{"Zone + link\nhave capacity?"}
        CAP1 -->|Yes| Q
        CAP2 -->|Yes| Q
        CAP1 -->|No| DROP1["Pruned"]
        CAP2 -->|No| DROP2["Pruned"]
    end

Key ideas:

Sequential Planning — drones are routed one at a time. Once a drone's path is found, it is locked in.
Reservation Table — the CollisionManager stores which zones and links are occupied at each turn. Future drones check this table before committing to a move.
Waiting — a drone can "wait" at its current zone (same zone, turn + 1) if the path ahead is blocked, provided the zone still has capacity.
Zone Types affect movement cost and transit time:

Zone Type	Transit Time	A* Cost	Behavior
`normal`	1 turn	1.0	Standard movement
`priority`	1 turn	0.8	Cheaper — A* prefers these paths
`restricted`	2 turns	2.0	Slow — drone is "mid-transit" for 2 turns
`blocked`	—	—	Impassable — completely pruned

How the Heuristic Works

A* uses the cost function f(n) = g(n) + h(n):

g(n) — the actual accumulated cost from start to node n.
h(n) — the heuristic estimate from node n to the goal.

For A* to find the optimal path, the heuristic must be admissible (never overestimates the true cost).

This project uses a hop-count heuristic:

h(n) = min_hops(n → goal) × 0.8

The minimum hop count from every zone to the goal is precomputed once with a breadth-first search over traversable connections.

Why is it admissible? Entering any zone costs at least 0.8 (the priority zone cost). The true remaining cost can therefore never be lower than 0.8 per remaining hop, so h(n) never overestimates. A coordinate-based heuristic cannot guarantee this, because connections may span arbitrarily large coordinate distances in a single move.

Generating Space-Time Neighbors

The _generate_valid_neighbors() method in CooperativeAStar is the core of the space-time adaptation. For every state it evaluates two types of moves:

Wait Action — the drone stays at its current zone while time advances by 1 turn. The CollisionManager checks that the zone still has capacity at turn + 1.
Move to Adjacent Zone — for each physically connected neighbor:
- Check is_traversable (blocked zones are pruned).
- Calculate transit_time and movement_cost from the zone type.
- Check CollisionManager for link capacity during every turn of transit.
- Check CollisionManager for zone capacity at the arrival turn.
- If everything passes, create a new TemporalState with updated costs and a parent pointer.

Priority Queue (heapq)

The algorithm uses Python's heapq module (min-heap) for the open set.

heappop() always returns the state with the lowest f_cost — O(log N).
heappush() inserts a new state — O(log N).
The TemporalState dataclass orders by a sort_index field derived from f = g + h and uses @dataclass(order=True), so heapq can compare states directly without custom sort functions.

Complexity

Let S be the number of explored space-time states for a single drone.

Route planning per drone: O(S log S) (heap-based A*).
Neighbor generation: proportional to the degree of the current zone.
Reservation lookups: O(1) average (dictionary-based).
Total: scales linearly with the number of drones, since each drone is planned sequentially.

Caching and Recalculation

Routes are not globally cached between drones. Each drone gets a freshly planned path, and the CollisionManager accumulates the reservations from all previously planned drones. This avoids the complexity of invalidating a shared cache when later drones change the available space-time slots.

Memory Usage

Memory is mainly driven by:

the parsed graph structure (Network),
the CollisionManager reservation table (grows with scheduled zone-turn and link-turn entries, not with every possible turn),
the A* open set and visited states during planning,
one stored path per drone.

Advantages and Disadvantages

Advantages:

Simple to implement — the algorithm is a straightforward extension of single-agent A* with a shared reservation table. No complex conflict resolution logic is needed.
Fast in practice — each drone runs a standard A* search, and reservation lookups are O(1). On maps with enough capacity, all drones find paths quickly.
Deterministic — the same input always produces the same output, making debugging and testing straightforward.
Scalable for well-connected maps — when the graph has multiple disjoint paths, drones naturally spread across them because earlier drones reserve the fastest routes, pushing later drones to alternatives.

Disadvantages:

Greedy sequential ordering — the first drone planned always gets the globally optimal path. Every subsequent drone works with a more constrained reservation table. This means later drones may get significantly worse paths, even when a globally better solution exists where all drones share the cost more evenly. The total turn count depends heavily on the planning order.
No global optimality — because drones are planned one at a time, the algorithm cannot guarantee a globally optimal solution. For example, the first drone might occupy a chokepoint for 3 turns when rerouting it by 1 extra turn would free the chokepoint for 5 other drones, saving turns overall.
Conflict-Based Search (CBS) addresses this — CBS is an alternative MAPF algorithm that plans all agents simultaneously. Instead of sequential planning, CBS detects conflicts between agents and splits the search into branches where each branch resolves a specific conflict. This produces globally optimal solutions but at a higher computational cost (exponential in the worst case). For this project, Cooperative A* was chosen for its simplicity and good-enough performance on the provided maps.
Order-dependent results — shuffling the drone planning order can produce different total turn counts. A potential improvement would be to try multiple orderings and pick the best result, or to use priority-based ordering (e.g., plan drones with the longest shortest-path first).
Revisit penalty — revisiting a zone adds a +10 cost penalty to discourage back-and-forth looping. This can forgo a strictly cheaper path in rare cases, so results are good rather than provably optimal.

Simulation Output

The console renderer (CLILogger) prints one line per turn when drones move:

D1-a D2-b
D1-a-b D2-goal
D1-b
D1-goal

In-Transit Output: Notice D1-a-b — when a drone travels through a restricted zone (2 turns), it is shown as current_zone-next_zone to indicate it is mid-transit between two points.

When using --renderer pygame, the same turn log is still printed to the terminal alongside the interactive window.

Visual Representation

The optional Pygame visualizer (--renderer pygame) opens a live window that brings the simulation to life.

Key features:

Animated drone motion — positions are interpolated between turns at 60 FPS using smooth-step easing, so drones glide instead of jumping.
Node display — each zone is drawn as a colored circle showing its x,y coordinates, capacity number, and a type abbreviation (Start, End, R, P, B).
Connection rendering — edges are drawn with a link capacity label at the midpoint, rendered twice (dark under light) as a drop shadow so it stays readable over lines.
Drone markers — in-node drones get a red label; in-transit drones get a gray label. When multiple drones overlap, the label shows the count (e.g., 2D).
Heads-Up Display (HUD) — a bottom panel shows: total drones, active (moving) drones, average turns per drone, and total path cost.
Interactive controls:

Key	Action
`Space`	Play / Pause
`Left` / `Right`	Scrub time
`R`	Reset to start
`Esc`	Quit

Limitations: The window is a fixed 1920×1080; the tile size adapts to fit the map's coordinate extremes (minimum 10 px). There is no zooming or panning, so extremely large maps can still overflow.

Useful Commands

make install       # Sync dependencies with uv
make run           # Run with auto-generated test map
make test          # Run the pytest suite
make debug         # Run with Python's pdb debugger
make lint          # flake8 + mypy type checking
make lint-strict   # Strict mypy mode
make lint-fix      # Auto-fix formatting and lint errors (ruff)
make clean         # Remove venv, caches, and temp files
make clean-cache   # Remove caches and temp files only
make help          # Show all available targets

Resources

Python documentation: https://docs.python.org/3/
heapq documentation: https://www.geeksforgeeks.org/python/heap-queue-or-heapq-in-python/
Pygame documentation: https://www.pygame.org/docs/
A* search overview: https://www.datacamp.com/tutorial/a-star-algorithm
Cooperative A*: David Silver. 2005. Cooperative pathfinding. In Proceedings of the First AAAI Conference on Artificial Intelligence and Interactive Digital Entertainment (AIIDE'05). AAAI Press, 117–122.
42 project subject and map files included in this repository

AI Usage

AI assistance was utilized during the development of this project for the following tasks:

Algorithm Comprehension: AI helped me better understand the algorithm, especially the space-time pathfinding part.
Debugging and Refactoring: Assisting in identifying edge cases within the space-time A* pathfinding implementation.
Documentation & Testing: Helping structure and proofread this README.md to ensure it meets all curriculum requirements, and generating PEP 257 compliant docstrings for classes and methods across the codebase.
Visuals & Rendering: AI helped implement a smoothstep (sigmoid-like) function for drone animation interpolation, suggested using pygame.gfxdraw for anti-aliased circles, and assisted in a rendering redesign to allow for fast color theme changes and reduced manual drawing work.
Design Patterns: AI explained and demonstrated SOLID and OOP principles in practice, helping me structure the codebase more effectively.

The core logical design, algorithmic choices, and constraints enforcement were driven by the developer, with AI acting as a supportive peer-programming tool.

Navigation Menu

Repository files navigation

Fly-in

Table of Contents

Description

Project Structure

Instructions

Requirements

Installation

Run

Map Format

Algorithm and Implementation Strategy

Application Pipeline

Pathfinding: Cooperative Space-Time A*

How the Heuristic Works

Generating Space-Time Neighbors

Priority Queue (heapq)

Complexity

Caching and Recalculation

Memory Usage

Advantages and Disadvantages

Simulation Output

Visual Representation

Useful Commands

Resources

AI Usage

About

Contributors

Languages

Name		Name	Last commit message	Last commit date
.github/workflows		.github/workflows
imgs		imgs
maps		maps
src		src
tests		tests
.gitignore		.gitignore
Makefile		Makefile
README.md		README.md
fly-in.pdf		fly-in.pdf
pyproject.toml		pyproject.toml
uv.lock		uv.lock

Latest commit

History

Folders and files

Repository files navigation

Fly-in

Table of Contents

Description

Project Structure

Instructions

Requirements

Installation

Run

Map Format

Algorithm and Implementation Strategy

Application Pipeline

Pathfinding: Cooperative Space-Time A*

How the Heuristic Works

Generating Space-Time Neighbors

Priority Queue (heapq)

Complexity

Caching and Recalculation

Memory Usage

Advantages and Disadvantages

Simulation Output

Visual Representation

Useful Commands

Resources

AI Usage

About

Topics

Resources

Stars

Watchers

Forks

Contributors

Languages