Initial implementation of minimum-violation LTL planner

Reproduces Tumova et al. (ACC 2013): given a set of LTL specs with
priority rewards, finds the lasso path that satisfies the highest-weight
subset when specs conflict.

- Büchi automata for G(!p), GF(p), F(p), G(p) implemented from scratch
- Product automaton built via BFS; SCCs found with Tarjan's algorithm
- Max-reward SCC selected; prefix + cycle reconstructed via BFS
- Gradio app with 3 preset scenarios, priority sliders, animated GIF output
- Ready for HuggingFace Spaces deployment

.gitignore

@@ -0,0 +1,11 @@
+context.md
+__pycache__/
+*.pyc
+*.pyo
+.DS_Store
+*.egg-info/
+dist/
+build/
+.env
+*.gif
+*.tmp

README.md

@@ -0,0 +1,53 @@
+---
+title: Minimum-Violation LTL Planning
+emoji: 🤖
+colorFrom: blue
+colorTo: red
+sdk: gradio
+sdk_version: "4.44.0"
+app_file: app.py
+pinned: false
+license: mit
+short_description: Interactive demo of Jana Tumova's minimum-violation LTL planning algorithm (ACC 2013)
+---
+
+# Minimum-Violation LTL Planning
+
+Interactive reproduction of:
+
+> Tumova, Reyes-Castro, Karaman, Frazzoli, Rus — **"Minimum-Violation LTL Planning with Conflicting Specifications"** — ACC 2013. [arXiv:1303.3679](https://arxiv.org/abs/1303.3679)
+
+## What it demonstrates
+
+When a robot's logical specifications conflict (e.g. "always avoid the danger zone" vs "reach the goal through the danger zone"), a standard planner simply fails. This algorithm instead finds the plan that **satisfies the highest-priority specs** and minimally violates the rest.
+
+**Core idea:**
+- Each spec φᵢ has a reward rᵢ (higher = harder to violate)
+- Build the product automaton: grid × Büchi(φ₁) × Büchi(φ₂) × ...
+- Find the max-reward strongly connected component (SCC) reachable from the start
+- Reconstruct a lasso path (prefix + repeating cycle) through that SCC
+
+**Try it:** drag the reward sliders to swap priorities — the planned path changes to satisfy whichever spec now has the highest weight.
+
+## Specs supported
+
+| Formula | Meaning |
+|---|---|
+| `G(!p)` | Safety: never visit region p |
+| `GF(p)` | Recurrence: visit p infinitely often |
+| `F(p)` | Reachability: eventually reach p |
+| `G(p)` | Invariance: always stay in p |
+
+## Implementation
+
+Pure Python, no external tools:
+- **Büchi automata** implemented directly for each formula pattern
+- **Product automaton** built lazily via BFS
+- **SCC detection** via Tarjan's algorithm
+- **Lasso reconstruction** via BFS within the winning SCC
+- **Visualization** via matplotlib → PIL → Gradio
+
+## Related work (Jana Tumova's group)
+
+- [KTH RPL Planiacs](https://github.com/KTH-RPL-Planiacs)
+- [Jana Tumova's homepage](https://sites.google.com/view/janatumova/home)

app.py

@@ -0,0 +1,173 @@
+"""
+Minimum-Violation LTL Planning — Interactive Demo
+Reproduces: Tumova et al., "Minimum-Violation LTL Planning with Conflicting
+Specifications", ACC 2013 (arXiv:1303.3679)
+
+Change spec priorities → watch the plan change to satisfy the highest-priority rules.
+"""
+
+import gradio as gr
+
+from src.grid_world import make_scenario, GridWorld, CELL_PROPS
+from src.automata import parse_spec, BuchiAut
+from src.planner import plan
+from src.visualize import render_static, render_animation, spec_table_html
+
+
+# ── Preset spec bundles per scenario ─────────────────────────────────────────
+
+SCENARIO_SPECS = {
+    "road": [
+        ("G(!danger)",  "Never cross double line (danger zone)", 80),
+        ("GF(zone_a)",  "Periodically visit pickup zone A",      50),
+        ("GF(zone_b)",  "Periodically visit dropoff zone B",     30),
+        ("F(goal)",     "Eventually reach the goal",             10),
+    ],
+    "patrol": [
+        ("G(!danger)",  "Stay away from danger zones",    100),
+        ("GF(zone_a)",  "Patrol zone A repeatedly",        60),
+        ("GF(zone_b)",  "Patrol zone B repeatedly",        40),
+    ],
+    "rescue": [
+        ("G(!danger)",  "Avoid hazardous areas",           90),
+        ("F(zone_a)",   "Reach survivor site A",           70),
+        ("F(zone_b)",   "Reach survivor site B",           50),
+        ("GF(zone_c)",  "Return to base (zone C) always",  20),
+    ],
+}
+
+SCENARIO_DESCRIPTIONS = {
+    "road": "🚗 Road network — robot must navigate around a danger zone to reach pickup/dropoff areas and a destination.",
+    "patrol": "🏭 Warehouse patrol — robot periodically covers two inspection zones while avoiding a hazardous area.",
+    "rescue": "🚁 Rescue mission — robot must reach two survivor sites and periodically return to base, avoiding hazards.",
+}
+
+
+def run_planning(scenario, r0, r1, r2, r3, animate):
+    grid = make_scenario(scenario)
+    specs_raw = SCENARIO_SPECS[scenario]
+    n_specs = len(specs_raw)
+    rewards_input = [r0, r1, r2, r3][:n_specs]
+
+    automata = []
+    rewards  = []
+    for i, (formula, _, _) in enumerate(specs_raw):
+        try:
+            aut = parse_spec(formula, label=formula)
+            automata.append(aut)
+            rewards.append(float(rewards_input[i]))
+        except ValueError as e:
+            return None, None, f"<p style='color:red'>Error: {e}</p>"
+
+    result = plan(grid, automata, rewards)
+    table_html = spec_table_html(result)
+
+    static_img = render_static(grid, result)
+
+    if animate and result.success:
+        gif_path = render_animation(grid, result, fps=3)
+        return static_img, gif_path, table_html
+    else:
+        return static_img, None, table_html
+
+
+def update_scenario_ui(scenario):
+    specs = SCENARIO_SPECS[scenario]
+    desc  = SCENARIO_DESCRIPTIONS[scenario]
+    n = len(specs)
+
+    updates = []
+    for i in range(4):
+        if i < n:
+            _, label, default_r = specs[i]
+            updates.append(gr.update(label=label, value=default_r, visible=True))
+        else:
+            updates.append(gr.update(visible=False))
+
+    return [gr.update(value=f"**{desc}**")] + updates
+
+
+# ── Build the Gradio UI ───────────────────────────────────────────────────────
+
+with gr.Blocks(title="Minimum-Violation LTL Planner", theme=gr.themes.Soft()) as demo:
+
+    gr.Markdown("""
+# Minimum-Violation LTL Planning
+**Reproducing:** Tumova et al., *"Minimum-Violation LTL Planning with Conflicting Specifications"*, ACC 2013
+
+When robot specs conflict, instead of failing, this planner finds the path that satisfies
+the **highest-priority** rules and minimally violates the rest.
+
+> **Try it:** drag the reward sliders to swap priorities — watch the planned path change.
+""")
+
+    with gr.Row():
+        with gr.Column(scale=1):
+            scenario_dd = gr.Dropdown(
+                choices=["road", "patrol", "rescue"],
+                value="road",
+                label="Scenario",
+            )
+            scenario_desc = gr.Markdown("**Loading...**")
+
+            gr.Markdown("### Spec Priorities (higher reward = harder to violate)")
+
+            sliders = []
+            default_specs = SCENARIO_SPECS["road"]
+            for i in range(4):
+                visible = i < len(default_specs)
+                _, lbl, val = default_specs[i] if visible else ("", f"Spec {i+1}", 10)
+                s = gr.Slider(
+                    minimum=0, maximum=200, step=5,
+                    value=val, label=lbl,
+                    visible=visible,
+                )
+                sliders.append(s)
+
+            animate_cb = gr.Checkbox(label="Generate animation (GIF)", value=True)
+            plan_btn   = gr.Button("▶  Synthesize Plan", variant="primary")
+
+            gr.Markdown("""
+---
+**How it works:**
+1. Each spec φ�� becomes a Büchi automaton
+2. Product automaton = grid × aut₁ × aut₂ × ...
+3. SCCs with accepting states for each spec are found
+4. Max-reward SCC is chosen → lasso path reconstructed
+
+🔵 Prefix path &nbsp;&nbsp; 🔴 Repeating cycle &nbsp;&nbsp; 🟣 Start &nbsp;&nbsp; 🟠 Robot
+""")
+
+        with gr.Column(scale=2):
+            grid_img  = gr.Image(label="Planned Path", type="pil", height=420)
+            anim_img  = gr.Image(label="Animation (GIF)", type="filepath", height=420)
+            result_md = gr.HTML(label="Spec Satisfaction")
+
+    # ── Event handlers ────────────────────────────────────────────────────────
+
+    scenario_dd.change(
+        fn=update_scenario_ui,
+        inputs=[scenario_dd],
+        outputs=[scenario_desc] + sliders,
+    )
+
+    plan_btn.click(
+        fn=run_planning,
+        inputs=[scenario_dd] + sliders + [animate_cb],
+        outputs=[grid_img, anim_img, result_md],
+    )
+
+    # Run on load with default scenario
+    demo.load(
+        fn=lambda: update_scenario_ui("road"),
+        outputs=[scenario_desc] + sliders,
+    )
+
+    demo.load(
+        fn=lambda: run_planning("road", 80, 50, 30, 10, False),
+        outputs=[grid_img, anim_img, result_md],
+    )
+
+
+if __name__ == "__main__":
+    demo.launch()

requirements.txt

@@ -0,0 +1,5 @@
+gradio>=4.44.1
+matplotlib>=3.7.0
+networkx>=3.0
+numpy>=1.24.0
+pillow>=9.0.0

src/automata.py

@@ -0,0 +1,132 @@
+"""
+Büchi automata for common LTL spec patterns.
+Implemented directly — no external tool needed.
+
+Supported patterns:
+  G(!p)   — safety: never visit p
+  GF(p)   — recurrence: visit p infinitely often
+  F(p)    — reachability: eventually visit p
+  G(p)    — invariance: always be in p
+
+Each automaton is a dict-based structure with:
+  states    : list of state names
+  initial   : initial state name
+  delta     : (state, label_frozenset) -> next_state | None  (None = sink/stuck)
+  accepting : set of accepting states
+"""
+
+from typing import Callable, FrozenSet, NamedTuple, Optional
+
+
+class BuchiAut:
+    def __init__(self, states, initial, delta_fn, accepting, name=""):
+        self.states = states          # list of hashable state ids
+        self.initial = initial
+        self._delta = delta_fn        # (state, frozenset) -> state | None
+        self.accepting = set(accepting)
+        self.name = name
+
+    def step(self, state, label: FrozenSet[str]) -> Optional[object]:
+        return self._delta(state, label)
+
+    def is_accepting(self, state) -> bool:
+        return state in self.accepting
+
+
+# ── factory functions ─────────────────────────────────────────────────────────
+
+def safety(prop: str, label: str = "") -> BuchiAut:
+    """G(!prop) — automaton rejects if prop is ever true."""
+    # States: q0 (safe), q_sink (violated, non-accepting)
+    def delta(state, lbl):
+        if state == "q0":
+            return "q_sink" if prop in lbl else "q0"
+        return "q_sink"  # once violated, stay violated
+
+    return BuchiAut(
+        states=["q0", "q_sink"],
+        initial="q0",
+        delta_fn=delta,
+        accepting=["q0"],  # must stay in q0 infinitely → never violated
+        name=label or f"G(!{prop})",
+    )
+
+
+def recurrence(prop: str, label: str = "") -> BuchiAut:
+    """GF(prop) — must visit prop infinitely often."""
+    # States: q0 (waiting), q1 (just saw prop — accepting)
+    def delta(state, lbl):
+        if state == "q0":
+            return "q1" if prop in lbl else "q0"
+        # q1: already accepted, loop back to wait for next occurrence
+        return "q0"
+
+    return BuchiAut(
+        states=["q0", "q1"],
+        initial="q0",
+        delta_fn=delta,
+        accepting=["q1"],
+        name=label or f"GF({prop})",
+    )
+
+
+def reachability(prop: str, label: str = "") -> BuchiAut:
+    """F(prop) — eventually reach prop (then stay accepting)."""
+    # States: q0 (searching), q1 (reached — accepting sink)
+    def delta(state, lbl):
+        if state == "q0":
+            return "q1" if prop in lbl else "q0"
+        return "q1"  # once reached, stay accepting
+
+    return BuchiAut(
+        states=["q0", "q1"],
+        initial="q0",
+        delta_fn=delta,
+        accepting=["q1"],
+        name=label or f"F({prop})",
+    )
+
+
+def invariance(prop: str, label: str = "") -> BuchiAut:
+    """G(prop) — always be in prop."""
+    def delta(state, lbl):
+        if state == "q0":
+            return "q0" if prop in lbl else "q_sink"
+        return "q_sink"
+
+    return BuchiAut(
+        states=["q0", "q_sink"],
+        initial="q0",
+        delta_fn=delta,
+        accepting=["q0"],
+        name=label or f"G({prop})",
+    )
+
+
+# ── simple formula parser ─────────────────────────────────────────────────────
+
+def parse_spec(formula: str, label: str = "") -> BuchiAut:
+    """
+    Parse simple LTL formula string into a BuchiAut.
+    Supported:
+      G(!p)   → safety
+      GF(p)   → recurrence
+      F(p)    → reachability
+      G(p)    → invariance
+    """
+    f = formula.strip().replace(" ", "")
+    lbl = label or formula
+
+    if f.startswith("GF(") and f.endswith(")"):
+        return recurrence(f[3:-1], lbl)
+    if f.startswith("G(!") and f.endswith(")"):
+        return safety(f[3:-1], lbl)
+    if f.startswith("G(") and f.endswith(")"):
+        return invariance(f[2:-1], lbl)
+    if f.startswith("F(") and f.endswith(")"):
+        return reachability(f[2:-1], lbl)
+
+    raise ValueError(
+        f"Unsupported formula: '{formula}'. "
+        "Supported: G(!p), GF(p), G(p), F(p)"
+    )

src/grid_world.py

@@ -0,0 +1,107 @@
+"""
+Grid world transition system.
+Each cell is labeled with a set of atomic propositions.
+Robots move N/S/E/W; obstacles block movement.
+"""
+
+from dataclasses import dataclass, field
+from typing import Dict, FrozenSet, List, Set, Tuple
+
+MOVES = {"N": (-1, 0), "S": (1, 0), "E": (0, 1), "W": (0, -1)}
+
+# Cell type → set of atomic propositions true at that cell
+CELL_PROPS = {
+    "free":     frozenset(),
+    "obstacle": frozenset({"obstacle"}),
+    "zone_a":   frozenset({"zone_a"}),
+    "zone_b":   frozenset({"zone_b"}),
+    "zone_c":   frozenset({"zone_c"}),
+    "danger":   frozenset({"danger"}),
+    "goal":     frozenset({"goal"}),
+    "start":    frozenset(),
+}
+
+
+@dataclass
+class GridWorld:
+    n: int
+    grid: List[List[str]] = field(default_factory=list)
+    start: Tuple[int, int] = (0, 0)
+
+    def __post_init__(self):
+        if not self.grid:
+            self.grid = [["free"] * self.n for _ in range(self.n)]
+
+    def label(self, pos: Tuple[int, int]) -> FrozenSet[str]:
+        r, c = pos
+        return CELL_PROPS.get(self.grid[r][c], frozenset())
+
+    def successors(self, pos: Tuple[int, int]) -> List[Tuple[str, Tuple[int, int]]]:
+        r, c = pos
+        result = []
+        for action, (dr, dc) in MOVES.items():
+            nr, nc = r + dr, c + dc
+            if 0 <= nr < self.n and 0 <= nc < self.n:
+                if self.grid[nr][nc] != "obstacle":
+                    result.append((action, (nr, nc)))
+        # also allow staying in place (needed for sync / waiting)
+        result.append(("stay", pos))
+        return result
+
+    def all_positions(self) -> List[Tuple[int, int]]:
+        return [
+            (r, c)
+            for r in range(self.n)
+            for c in range(self.n)
+            if self.grid[r][c] != "obstacle"
+        ]
+
+
+def make_scenario(name: str) -> GridWorld:
+    """Built-in demo scenarios."""
+    if name == "road":
+        # 8×8 road network
+        # danger=double-line zone, zone_a=pickup, zone_b=dropoff, goal=destination
+        n = 8
+        g = GridWorld(n=n)
+        g.start = (0, 0)
+        # vertical danger strip (double line)
+        for r in range(n):
+            g.grid[r][3] = "danger"
+        # obstacle block
+        for r in range(2, 5):
+            g.grid[r][5] = "obstacle"
+        g.grid[1][6] = "zone_a"
+        g.grid[6][1] = "zone_b"
+        g.grid[7][7] = "goal"
+        return g
+
+    if name == "patrol":
+        # 6×6 warehouse patrol
+        n = 6
+        g = GridWorld(n=n)
+        g.start = (0, 0)
+        g.grid[0][5] = "zone_a"
+        g.grid[5][0] = "zone_b"
+        g.grid[2][2] = "danger"
+        g.grid[2][3] = "danger"
+        g.grid[3][2] = "danger"
+        g.grid[3][3] = "danger"
+        return g
+
+    if name == "rescue":
+        # 7×7 rescue mission
+        n = 7
+        g = GridWorld(n=n)
+        g.start = (3, 0)
+        for c in range(1, 6):
+            g.grid[3][c] = "obstacle"
+        g.grid[3][3] = "free"
+        g.grid[0][6] = "zone_a"
+        g.grid[6][6] = "zone_b"
+        for r in [1, 2, 4, 5]:
+            g.grid[r][2] = "danger"
+        g.grid[0][0] = "zone_c"
+        return g
+
+    raise ValueError(f"Unknown scenario: {name}")

src/planner.py

@@ -0,0 +1,139 @@
+"""
+Maximum-reward lasso planner.
+
+Given a product automaton and per-spec rewards, finds:
+  1. The SCC with maximum total reward that is reachable from the initial state
+  2. A lasso path: prefix (initial → SCC) + cycle (within SCC visiting accepting states)
+
+Returns the path as a sequence of grid positions plus a result summary.
+"""
+
+from dataclasses import dataclass
+from typing import Dict, List, Optional, Set, Tuple
+
+from .grid_world import GridWorld
+from .automata import BuchiAut
+from .product import ProductGraph
+
+
+@dataclass
+class PlanResult:
+    path: List[Tuple[int, int]]       # grid positions (prefix + one full cycle)
+    cycle_start_idx: int               # index in path where cycle begins
+    satisfied: List[int]               # indices of satisfied specs
+    violated: List[int]                # indices of violated specs
+    total_reward: float
+    max_possible_reward: float
+    spec_names: List[str]
+    spec_rewards: List[float]
+    success: bool
+    message: str
+
+
+def plan(
+    grid: GridWorld,
+    automata: List[BuchiAut],
+    rewards: List[float],
+) -> PlanResult:
+    spec_names = [a.name for a in automata]
+    max_possible = sum(rewards)
+
+    pg = ProductGraph(grid, automata)
+    sccs = pg.compute_sccs()
+    reachable = pg.reachable_from_initial()
+
+    # Filter to nontrivial SCCs reachable from initial
+    init_idx = pg.state_index[pg.initial]
+    candidates = []
+    for scc in sccs:
+        scc_set = set(scc)
+        if not any(v in reachable for v in scc):
+            continue
+        if not pg.is_nontrivial_scc(scc):
+            continue
+        reward, satisfied_set = pg.scc_satisfied_specs(scc, rewards)
+        candidates.append((reward, satisfied_set, scc))
+
+    if not candidates:
+        return PlanResult(
+            path=[], cycle_start_idx=0,
+            satisfied=[], violated=list(range(len(automata))),
+            total_reward=0, max_possible_reward=max_possible,
+            spec_names=spec_names, spec_rewards=list(rewards),
+            success=False,
+            message="No reachable accepting cycle found. Check for obstacles blocking all paths.",
+        )
+
+    # Pick best SCC
+    candidates.sort(key=lambda x: x[0], reverse=True)
+    best_reward, satisfied_set, best_scc = candidates[0]
+    best_scc_set = set(best_scc)
+    violated_set = set(range(len(automata))) - satisfied_set
+
+    # Build accepting state sets per spec (restricted to the best SCC)
+    required_accepting = []
+    for i, aut in enumerate(automata):
+        if i in satisfied_set:
+            acc_in_scc = {
+                v for v in best_scc_set
+                if aut.is_accepting(pg.states[v][1 + i])
+            }
+            required_accepting.append(acc_in_scc)
+
+    # Find prefix: initial → any state in best SCC
+    prefix_path = pg.bfs_path(init_idx, best_scc_set)
+    if prefix_path is None:
+        return PlanResult(
+            path=[], cycle_start_idx=0,
+            satisfied=[], violated=list(range(len(automata))),
+            total_reward=0, max_possible_reward=max_possible,
+            spec_names=spec_names, spec_rewards=list(rewards),
+            success=False,
+            message="Could not find path to best SCC (graph error).",
+        )
+
+    # Find cycle within SCC through required accepting states
+    cycle_start_prod = prefix_path[-1]
+    cycle_scc = best_scc_set  # restrict to scc
+
+    # For cycle, we need to start from the endpoint of the prefix
+    cycle_entry = prefix_path[-1]
+    cycle = pg.find_cycle_through(best_scc_set, required_accepting)
+
+    if cycle is None:
+        return PlanResult(
+            path=[], cycle_start_idx=0,
+            satisfied=[], violated=list(range(len(automata))),
+            total_reward=0, max_possible_reward=max_possible,
+            spec_names=spec_names, spec_rewards=list(rewards),
+            success=False,
+            message="Could not construct cycle within SCC.",
+        )
+
+    # Connect prefix end to cycle start (they may differ)
+    if prefix_path[-1] != cycle[0]:
+        bridge = pg.bfs_path(prefix_path[-1], {cycle[0]})
+        if bridge is None:
+            bridge = [prefix_path[-1]]
+        full_prod_path = prefix_path[:-1] + bridge + cycle[1:]
+        cycle_start_idx = len(prefix_path[:-1] + bridge) - 1
+    else:
+        full_prod_path = prefix_path + cycle[1:]
+        cycle_start_idx = len(prefix_path) - 1
+
+    # Extract grid positions
+    grid_path = [pg.states[v][0] for v in full_prod_path]
+
+    return PlanResult(
+        path=grid_path,
+        cycle_start_idx=cycle_start_idx,
+        satisfied=sorted(satisfied_set),
+        violated=sorted(violated_set),
+        total_reward=best_reward,
+        max_possible_reward=max_possible,
+        spec_names=spec_names,
+        spec_rewards=list(rewards),
+        success=True,
+        message=f"Plan found! Satisfies {len(satisfied_set)}/{len(automata)} specs "
+                f"(reward {best_reward:.0f}/{max_possible:.0f}).",
+    )

src/product.py

@@ -0,0 +1,252 @@
+"""
+Product automaton: GridWorld × Büchi_1 × ... × Büchi_n
+
+A product state is (grid_pos, aut_state_1, ..., aut_state_n).
+We build the graph lazily via BFS, then run Tarjan's SCC algorithm.
+"""
+
+from collections import defaultdict, deque
+from typing import Dict, FrozenSet, List, Optional, Set, Tuple
+
+from .grid_world import GridWorld
+from .automata import BuchiAut
+
+
+# A product state is a tuple: (grid_pos, q1, q2, ..., qn)
+ProductState = tuple
+
+
+class ProductGraph:
+    def __init__(self, grid: GridWorld, automata: List[BuchiAut]):
+        self.grid = grid
+        self.automata = automata
+        self.n_aut = len(automata)
+
+        # Initial product state
+        init_aut = tuple(a.initial for a in automata)
+        self.initial: ProductState = (grid.start,) + init_aut
+
+        # Build graph
+        self.states: List[ProductState] = []
+        self.state_index: Dict[ProductState, int] = {}
+        self.adj: Dict[int, List[int]] = defaultdict(list)      # forward edges
+        self.radj: Dict[int, List[int]] = defaultdict(list)     # reverse edges
+        self._build()
+
+    # ── graph construction ────────────────────────────────────────────────────
+
+    def _build(self):
+        queue = deque([self.initial])
+        self._add_state(self.initial)
+
+        while queue:
+            ps = queue.popleft()
+            src_idx = self.state_index[ps]
+            grid_pos = ps[0]
+            aut_states = ps[1:]
+
+            label = self.grid.label(grid_pos)
+
+            for _, next_pos in self.grid.successors(grid_pos):
+                next_label = self.grid.label(next_pos)
+                # Advance each automaton on next_label (transition happens
+                # when entering the next cell, consistent with standard semantics)
+                next_aut = []
+                valid = True
+                for i, aut in enumerate(self.automata):
+                    nq = aut.step(aut_states[i], next_label)
+                    if nq is None:
+                        valid = False
+                        break
+                    next_aut.append(nq)
+
+                if not valid:
+                    continue
+
+                next_ps: ProductState = (next_pos,) + tuple(next_aut)
+                if next_ps not in self.state_index:
+                    self._add_state(next_ps)
+                    queue.append(next_ps)
+
+                dst_idx = self.state_index[next_ps]
+                self.adj[src_idx].append(dst_idx)
+                self.radj[dst_idx].append(src_idx)
+
+    def _add_state(self, ps: ProductState) -> int:
+        idx = len(self.states)
+        self.states.append(ps)
+        self.state_index[ps] = idx
+        return idx
+
+    # ── Tarjan's SCC ─────────────────────────────────────────────────────────
+
+    def compute_sccs(self) -> List[List[int]]:
+        """Returns list of SCCs (each a list of state indices), largest first."""
+        n = len(self.states)
+        index_counter = [0]
+        stack = []
+        lowlink = {}
+        index = {}
+        on_stack = {}
+        sccs = []
+
+        def strongconnect(v):
+            index[v] = index_counter[0]
+            lowlink[v] = index_counter[0]
+            index_counter[0] += 1
+            stack.append(v)
+            on_stack[v] = True
+
+            for w in self.adj[v]:
+                if w not in index:
+                    strongconnect(w)
+                    lowlink[v] = min(lowlink[v], lowlink[w])
+                elif on_stack.get(w):
+                    lowlink[v] = min(lowlink[v], index[w])
+
+            if lowlink[v] == index[v]:
+                scc = []
+                while True:
+                    w = stack.pop()
+                    on_stack[w] = False
+                    scc.append(w)
+                    if w == v:
+                        break
+                sccs.append(scc)
+
+        import sys
+        sys.setrecursionlimit(100000)
+
+        for v in range(n):
+            if v not in index:
+                strongconnect(v)
+
+        return sccs
+
+    # ── SCC reward analysis ───────────────────────────────────────────────────
+
+    def scc_satisfied_specs(self, scc: List[int], rewards: List[float]) -> Tuple[float, Set[int]]:
+        """
+        For an SCC, compute which specs have their accepting states inside it.
+        Returns (total_reward, set_of_satisfied_spec_indices).
+        """
+        satisfied = set()
+        for idx in scc:
+            ps = self.states[idx]
+            aut_states = ps[1:]
+            for i, aut in enumerate(self.automata):
+                if aut.is_accepting(aut_states[i]):
+                    satisfied.add(i)
+
+        total = sum(rewards[i] for i in satisfied)
+        return total, satisfied
+
+    def is_nontrivial_scc(self, scc: List[int]) -> bool:
+        """An SCC is nontrivial if it has >1 state, or 1 state with a self-loop."""
+        if len(scc) > 1:
+            return True
+        v = scc[0]
+        return v in self.adj[v]
+
+    # ── reachability ─────────────────────────────────────────────────────────
+
+    def reachable_from_initial(self) -> Set[int]:
+        visited = set()
+        queue = deque([self.state_index[self.initial]])
+        while queue:
+            v = queue.popleft()
+            if v in visited:
+                continue
+            visited.add(v)
+            for w in self.adj[v]:
+                if w not in visited:
+                    queue.append(w)
+        return visited
+
+    def bfs_path(self, src: int, targets: Set[int]) -> Optional[List[int]]:
+        """BFS from src to any state in targets. Returns list of state indices."""
+        if src in targets:
+            return [src]
+        parent = {src: None}
+        queue = deque([src])
+        while queue:
+            v = queue.popleft()
+            for w in self.adj[v]:
+                if w not in parent:
+                    parent[w] = v
+                    if w in targets:
+                        # reconstruct
+                        path = []
+                        cur = w
+                        while cur is not None:
+                            path.append(cur)
+                            cur = parent[cur]
+                        return list(reversed(path))
+                    queue.append(w)
+        return None
+
+    def find_cycle_through(self, scc_set: Set[int], required_accepting: List[Set[int]]) -> Optional[List[int]]:
+        """
+        Find a cycle within the SCC that passes through at least one accepting
+        state for each required spec.
+        Returns a list of state indices forming the cycle (first == last).
+        """
+        # Restrict graph to SCC nodes
+        # Strategy: chain BFS paths through each required accepting set
+        # Start from any state in scc, visit a state in required_accepting[0],
+        # then required_accepting[1], ..., then return to start.
+
+        if not scc_set:
+            return None
+
+        start = next(iter(scc_set))
+
+        # Build checkpoints: for each spec, one state in scc that is accepting
+        checkpoints = []
+        for acc_set in required_accepting:
+            candidates = acc_set & scc_set
+            if candidates:
+                checkpoints.append(next(iter(candidates)))
+
+        if not checkpoints:
+            # trivial cycle: just loop at start (if self-loop exists)
+            if start in self.adj.get(start, []):
+                return [start, start]
+            # find any 2-cycle
+            path = self._bfs_in_scc(start, {start}, scc_set)
+            return path
+
+        # chain: start -> cp0 -> cp1 -> ... -> cpN -> start
+        waypoints = [start] + checkpoints + [start]
+        full_path = []
+        for i in range(len(waypoints) - 1):
+            seg = self._bfs_in_scc(waypoints[i], {waypoints[i + 1]}, scc_set)
+            if seg is None:
+                return None
+            if full_path:
+                full_path.extend(seg[1:])  # skip duplicate junction
+            else:
+                full_path.extend(seg)
+
+        return full_path
+
+    def _bfs_in_scc(self, src: int, targets: Set[int], scc_set: Set[int]) -> Optional[List[int]]:
+        """BFS from src to any target, restricted to scc_set."""
+        if src in targets:
+            return [src]
+        parent = {src: None}
+        queue = deque([src])
+        while queue:
+            v = queue.popleft()
+            for w in self.adj[v]:
+                if w in scc_set and w not in parent:
+                    parent[w] = v
+                    if w in targets:
+                        path = []
+                        cur = w
+                        while cur is not None:
+                            path.append(cur)
+                            cur = parent[cur]
+                        return list(reversed(path))
+                    queue.append(w)
+        return None

src/visualize.py

@@ -0,0 +1,231 @@
+"""
+Visualization: static grid image + animated GIF of the planned path.
+"""
+
+import io
+from typing import List, Optional, Tuple
+
+import matplotlib
+matplotlib.use("Agg")
+import matplotlib.pyplot as plt
+import matplotlib.patches as mpatches
+from matplotlib.colors import ListedColormap
+import numpy as np
+from PIL import Image
+
+from .grid_world import GridWorld
+from .planner import PlanResult
+
+# Cell type → RGBA color
+CELL_COLORS = {
+    "free":     "#F5F5F5",
+    "obstacle": "#2C2C2C",
+    "zone_a":   "#A8D5FF",   # light blue
+    "zone_b":   "#A8FFB8",   # light green
+    "zone_c":   "#FFD6A8",   # light orange
+    "danger":   "#FFAAAA",   # light red
+    "goal":     "#FFE066",   # yellow
+    "start":    "#D0B4FF",   # purple
+}
+
+PATH_COLOR   = "#1A73E8"
+CYCLE_COLOR  = "#E83A1A"
+START_MARKER = "#7B2FBE"
+
+
+def _draw_grid(ax, grid: GridWorld, path: Optional[List[Tuple[int, int]]] = None,
+               cycle_start_idx: int = 0, step: int = -1, show_path: bool = True):
+    n = grid.n
+    ax.set_xlim(0, n)
+    ax.set_ylim(0, n)
+    ax.set_aspect("equal")
+    ax.set_xticks(range(n + 1))
+    ax.set_yticks(range(n + 1))
+    ax.tick_params(left=False, bottom=False, labelleft=False, labelbottom=False)
+    ax.grid(True, color="#CCCCCC", linewidth=0.5)
+
+    # Draw cells
+    for r in range(n):
+        for c in range(n):
+            cell_type = grid.grid[r][c]
+            color = CELL_COLORS.get(cell_type, "#F5F5F5")
+            rect = mpatches.FancyBboxPatch(
+                (c + 0.05, n - r - 1 + 0.05), 0.9, 0.9,
+                boxstyle="round,pad=0.02",
+                facecolor=color, edgecolor="#AAAAAA", linewidth=0.8,
+            )
+            ax.add_patch(rect)
+            # label
+            if cell_type not in ("free", "obstacle"):
+                short = {"zone_a": "A", "zone_b": "B", "zone_c": "C",
+                         "danger": "⚠", "goal": "★", "start": "S"}.get(cell_type, "")
+                ax.text(c + 0.5, n - r - 0.5, short,
+                        ha="center", va="center", fontsize=8,
+                        color="#444444", fontweight="bold")
+
+    # Start marker
+    sr, sc = grid.start
+    ax.plot(sc + 0.5, n - sr - 0.5, "o", color=START_MARKER,
+            markersize=10, zorder=5, markeredgecolor="white", markeredgewidth=1.5)
+
+    if not show_path or path is None or len(path) == 0:
+        return
+
+    static_mode = (step < 0)
+    display_path = path if static_mode else path[:step + 1]
+
+    def to_xy(pos):
+        r, c = pos
+        return c + 0.5, n - r - 0.5
+
+    # In static mode show full path split by color; in animation mode show
+    # only the portion reached so far.
+    if static_mode:
+        prefix = display_path[:cycle_start_idx + 1]
+        cycle  = display_path[cycle_start_idx:]
+
+        if len(prefix) > 1:
+            xs, ys = zip(*[to_xy(p) for p in prefix])
+            ax.plot(xs, ys, "-o", color=PATH_COLOR, linewidth=2,
+                    markersize=4, zorder=4, alpha=0.85)
+
+        if len(cycle) > 1:
+            xs, ys = zip(*[to_xy(p) for p in cycle])
+            ax.plot(xs, ys, "-o", color=CYCLE_COLOR, linewidth=2.5,
+                    markersize=4, zorder=4, alpha=0.9)
+
+        # Robot at end of path
+        if display_path:
+            rx, ry = to_xy(display_path[-1])
+            ax.plot(rx, ry, "D", color="#FF6B00", markersize=9, zorder=6,
+                    markeredgecolor="white", markeredgewidth=1.5)
+    else:
+        # Animation: colour prefix blue, cycle red, as steps accumulate
+        if step < cycle_start_idx:
+            # Still in prefix
+            seg = display_path
+            if len(seg) > 1:
+                xs, ys = zip(*[to_xy(p) for p in seg])
+                ax.plot(xs, ys, "-o", color=PATH_COLOR, linewidth=2,
+                        markersize=4, zorder=4, alpha=0.85)
+        else:
+            prefix_seg = path[:cycle_start_idx + 1]
+            cycle_seg  = display_path[cycle_start_idx:]
+            if len(prefix_seg) > 1:
+                xs, ys = zip(*[to_xy(p) for p in prefix_seg])
+                ax.plot(xs, ys, "-o", color=PATH_COLOR, linewidth=2,
+                        markersize=4, zorder=4, alpha=0.85)
+            if len(cycle_seg) > 1:
+                xs, ys = zip(*[to_xy(p) for p in cycle_seg])
+                ax.plot(xs, ys, "-o", color=CYCLE_COLOR, linewidth=2.5,
+                        markersize=4, zorder=4, alpha=0.9)
+
+        if display_path:
+            rx, ry = to_xy(display_path[-1])
+            ax.plot(rx, ry, "D", color="#FF6B00", markersize=9, zorder=6,
+                    markeredgecolor="white", markeredgewidth=1.5)
+
+
+def make_legend():
+    items = [
+        mpatches.Patch(color=CELL_COLORS["zone_a"], label="Zone A"),
+        mpatches.Patch(color=CELL_COLORS["zone_b"], label="Zone B"),
+        mpatches.Patch(color=CELL_COLORS["zone_c"], label="Zone C"),
+        mpatches.Patch(color=CELL_COLORS["danger"],  label="Danger"),
+        mpatches.Patch(color=CELL_COLORS["goal"],    label="Goal"),
+        mpatches.Patch(color=CELL_COLORS["obstacle"],label="Obstacle"),
+        mpatches.Patch(color=PATH_COLOR,  label="Prefix path"),
+        mpatches.Patch(color=CYCLE_COLOR, label="Cycle (repeating)"),
+    ]
+    return items
+
+
+def render_static(grid: GridWorld, result: PlanResult, dpi: int = 120) -> Image.Image:
+    """Render the full planned path as a static PIL image."""
+    fig, ax = plt.subplots(figsize=(5, 5))
+    _draw_grid(ax, grid, result.path, result.cycle_start_idx)
+    ax.legend(handles=make_legend(), loc="upper right", fontsize=6,
+              framealpha=0.85, ncol=2)
+    title = "PLAN FOUND" if result.success else "NO PLAN"
+    ax.set_title(title, fontsize=11, fontweight="bold",
+                 color="#1A73E8" if result.success else "#CC0000")
+    fig.tight_layout()
+    buf = io.BytesIO()
+    fig.savefig(buf, format="png", dpi=dpi, bbox_inches="tight")
+    plt.close(fig)
+    buf.seek(0)
+    return Image.open(buf).copy()
+
+
+def render_animation(grid: GridWorld, result: PlanResult,
+                     dpi: int = 100, fps: int = 4) -> Optional[str]:
+    """
+    Render an animated GIF of the robot following the path.
+    Returns file path to a temp GIF, or None on failure.
+    """
+    if not result.success or not result.path:
+        return None
+
+    frames = []
+    path = result.path
+    n_steps = len(path)
+
+    for step in range(n_steps):
+        fig, ax = plt.subplots(figsize=(5, 5))
+        _draw_grid(ax, grid, path, result.cycle_start_idx, step=step)
+        phase = "CYCLE" if step >= result.cycle_start_idx else "PREFIX"
+        ax.set_title(f"Step {step + 1}/{n_steps}  [{phase}]", fontsize=10)
+        fig.tight_layout()
+        buf = io.BytesIO()
+        fig.savefig(buf, format="png", dpi=dpi, bbox_inches="tight")
+        plt.close(fig)
+        buf.seek(0)
+        frames.append(Image.open(buf).copy())
+
+    import tempfile, os
+    tmp = tempfile.NamedTemporaryFile(suffix=".gif", delete=False)
+    tmp.close()
+    frames[0].save(
+        tmp.name,
+        save_all=True,
+        append_images=frames[1:],
+        loop=0,
+        duration=int(1000 / fps),
+        optimize=False,
+    )
+    return tmp.name
+
+
+def spec_table_html(result: PlanResult) -> str:
+    rows = []
+    for i, (name, reward) in enumerate(zip(result.spec_names, result.spec_rewards)):
+        ok = i in result.satisfied
+        icon  = "✅" if ok else "❌"
+        color = "#1a7a1a" if ok else "#aa0000"
+        rows.append(
+            f"<tr>"
+            f"<td style='padding:4px 10px;font-weight:bold;color:{color}'>{icon}</td>"
+            f"<td style='padding:4px 10px;font-family:monospace'>{name}</td>"
+            f"<td style='padding:4px 10px;text-align:right'>r = {reward:.0f}</td>"
+            f"<td style='padding:4px 10px;color:{color};font-weight:bold'>"
+            f"{'SATISFIED' if ok else 'VIOLATED'}</td>"
+            f"</tr>"
+        )
+    header = (
+        f"<div style='font-size:13px;margin-bottom:6px'>"
+        f"<b>Total reward:</b> {result.total_reward:.0f} / {result.max_possible_reward:.0f}"
+        f"</div>"
+    )
+    table = (
+        "<table style='border-collapse:collapse;width:100%;font-size:13px'>"
+        "<thead><tr>"
+        "<th style='padding:4px 10px'></th>"
+        "<th style='padding:4px 10px;text-align:left'>Spec</th>"
+        "<th style='padding:4px 10px;text-align:right'>Reward</th>"
+        "<th style='padding:4px 10px;text-align:left'>Result</th>"
+        "</tr></thead>"
+        f"<tbody>{''.join(rows)}</tbody></table>"
+    )
+    msg_color = "#1a7a1a" if result.success else "#aa0000"
+    msg = f"<p style='color:{msg_color};font-weight:bold;margin-top:8px'>{result.message}</p>"
+    return header + table + msg