Update app.py

app.py

@@ -49,7 +49,7 @@ def tau_eff_adaptive(
     """
     τ_eff is implemented here as a timing/decision delay modifier.
     - base: baseline τ_eff
-    - slow_by: explicit slow-down term (I wanted this behaviour: slow by 1.0)
+    - slow_by: explicit slow-down term
     - gain: reaction strength to uncertainty
     - cap: prevents absurd values
     """
@@ -178,9 +178,7 @@ def simulate_neo(
         "baseline_alerts": int(alerts_baseline),
         "rft_candidates": int(alerts_rft_raw),
         "rft_alerts_filtered": int(alerts_rft_filtered),
-        "false_positive_proxy_reduction_%": float(
-            100.0 * (1.0 - (alerts_rft_filtered / max(alerts_rft_raw, 1)))
-        ),
+        "false_positive_proxy_reduction_%": float(100.0 * (1.0 - (alerts_rft_filtered / max(alerts_rft_raw, 1)))),
         "ops_proxy": int(ops_proxy),
     }
 
@@ -301,7 +299,6 @@ def simulate_jitter(
 
 # -----------------------------
 # Starship-style Landing Harness (2D)
-# (Same logic you have now; only naming/wording updated)
 # -----------------------------
 def simulate_landing(
     seed: int,
@@ -476,7 +473,215 @@ def simulate_landing(
     return summary, [p_alt, p_x, p_w, p_a], csv_path
 
 # -----------------------------
-# Benchmarks
+# Predator Avoidance (2D)
+# -----------------------------
+def simulate_predator_avoidance(
+    seed: int,
+    steps: int,
+    dt: float,
+    world_size: float,
+    predator_speed: float,
+    agent_speed: float,
+    catch_radius: float,
+    sense_noise: float,
+    rft_k_attract: float,
+    rft_k_repulse: float,
+    baseline_k_attract: float,
+    baseline_k_repulse: float,
+    gate_threshold: float,
+    tau_gain: float,
+    goal_radius: float,
+    show_baseline: bool
+):
+    """
+    A simple pursuit/evasion harness:
+    - Predator chases the agent.
+    - Agent tries to reach a goal while avoiding capture.
+    Baseline: always-on potential field (attract goal + repulse predator).
+    Observer-gated RFT: same control, but can "wait" unless risk is high.
+    """
+    set_seed(seed)
+
+    # Initial positions
+    agent = np.array([-0.55 * world_size, -0.10 * world_size], dtype=float)
+    predator = np.array([0.35 * world_size, 0.25 * world_size], dtype=float)
+    goal = np.array([0.60 * world_size, -0.45 * world_size], dtype=float)
+
+    # Helper: normalize vector
+    def nrm(v):
+        s = float(np.linalg.norm(v))
+        if s < 1e-12:
+            return v * 0.0
+        return v / s
+
+    # Logging
+    rows = []
+    ops_proxy = 0
+
+    caught = False
+    reached_goal = False
+    t_end = int(steps)
+
+    # Simple "energy" proxy
+    energy_baseline = 0.0
+    energy_rft = 0.0
+
+    for t in range(int(steps)):
+        # Noisy sensed predator position (agent perception)
+        pred_meas = predator + np.random.normal(0.0, sense_noise, size=2)
+
+        # Distances
+        d_pa = float(np.linalg.norm(pred_meas - agent))   # perceived predator-agent
+        d_true = float(np.linalg.norm(predator - agent))  # true predator-agent
+        d_ag = float(np.linalg.norm(goal - agent))        # agent-goal
+
+        # Uncertainty proxy:
+        # - higher when predator is close (risk)
+        # - higher when sensing is noisy
+        # - mild term for distance-to-goal (to bias to act near the end)
+        risk = clamp((catch_radius / max(d_pa, 1e-6)) * 0.55, 0.0, 1.0)
+        noise_term = clamp((sense_noise / max(world_size, 1e-9)) * 8.0, 0.0, 1.0)
+        goal_term = clamp(1.0 - (d_ag / max(world_size, 1e-9)), 0.0, 1.0) * 0.15
+        uncertainty = clamp(risk + noise_term + goal_term, 0.0, 1.0)
+
+        tau = tau_eff_adaptive(uncertainty, base=1.0, slow_by=1.0, gain=tau_gain, cap=4.0)
+        conf = rft_confidence(uncertainty)
+
+        # Predator moves toward agent (pure pursuit)
+        predator_dir = nrm(agent - predator)
+        predator = predator + predator_dir * predator_speed * dt
+
+        # Baseline control (always on)
+        v_goal_b = nrm(goal - agent) * baseline_k_attract
+        v_away_b = nrm(agent - pred_meas) * (baseline_k_repulse / max(d_pa, 0.25))
+        v_base = v_goal_b + v_away_b
+        v_base = nrm(v_base) * agent_speed
+
+        # RFT (observer-gated) control
+        # Gate can be overridden if risk is high / predator is close / goal is near.
+        do_action = rft_gate(conf, tau, gate_threshold)
+        must_act = (d_pa < (3.0 * catch_radius)) or (d_ag < (2.5 * goal_radius))
+        do_action = bool(do_action or must_act)
+
+        v_rft = np.array([0.0, 0.0], dtype=float)
+        if do_action:
+            phase = clamp(1.0 - (d_ag / max(world_size, 1e-9)), 0.0, 1.0)
+            lookahead = 1.0 + 1.2 * phase
+
+            v_goal = nrm(goal - agent) * (rft_k_attract * lookahead)
+            v_away = nrm(agent - pred_meas) * ((rft_k_repulse * lookahead) / max(d_pa, 0.25))
+            v_rft = v_goal + v_away
+            v_rft = nrm(v_rft) * agent_speed
+
+        # Choose which agent motion to apply to the plant
+        if show_baseline:
+            agent_next = agent + v_base * dt
+            energy_baseline += float(np.linalg.norm(v_base)) * dt
+        else:
+            agent_next = agent + v_rft * dt
+            energy_rft += float(np.linalg.norm(v_rft)) * dt
+
+        # Clamp to bounds
+        agent_next[0] = clamp(agent_next[0], -world_size, world_size)
+        agent_next[1] = clamp(agent_next[1], -world_size, world_size)
+        agent = agent_next
+
+        # Termination checks
+        if d_true <= catch_radius:
+            caught = True
+            t_end = t + 1
+        if d_ag <= goal_radius:
+            reached_goal = True
+            t_end = t + 1
+
+        ops_proxy += 14
+
+        rows.append({
+            "t": t,
+            "agent_x": agent[0],
+            "agent_y": agent[1],
+            "pred_x": predator[0],
+            "pred_y": predator[1],
+            "goal_x": goal[0],
+            "goal_y": goal[1],
+            "d_pred_agent_true": d_true,
+            "d_pred_agent_meas": d_pa,
+            "d_agent_goal": d_ag,
+            "sense_noise": sense_noise,
+            "uncertainty": uncertainty,
+            "tau_eff": tau,
+            "confidence": conf,
+            "action_taken": int(do_action),
+            "mode": "baseline" if show_baseline else "observer_gated_rft",
+            "v_base_x": v_base[0],
+            "v_base_y": v_base[1],
+            "v_rft_x": v_rft[0],
+            "v_rft_y": v_rft[1],
+            "caught": int(caught),
+            "reached_goal": int(reached_goal),
+        })
+
+        if caught or reached_goal:
+            break
+
+    df = pd.DataFrame(rows)
+
+    # Plots
+    fig1 = plt.figure(figsize=(7.5, 7.0))
+    ax = fig1.add_subplot(111)
+    ax.plot(df["agent_x"], df["agent_y"], label="Agent")
+    ax.plot(df["pred_x"], df["pred_y"], label="Predator")
+    ax.scatter([df["goal_x"].iloc[0]], [df["goal_y"].iloc[0]], marker="x", s=80, label="Goal")
+    ax.set_title("Predator Avoidance: Trajectories")
+    ax.set_xlabel("x")
+    ax.set_ylabel("y")
+    ax.legend(loc="best")
+    p_traj = save_plot(fig1, f"predator_traj_seed{seed}_{'baseline' if show_baseline else 'rft'}.png")
+
+    fig2 = plt.figure(figsize=(10, 4))
+    ax = fig2.add_subplot(111)
+    ax.plot(df["t"], df["d_pred_agent_true"], label="True dist (pred→agent)")
+    ax.plot(df["t"], df["d_agent_goal"], label="Dist (agent→goal)")
+    ax.axhline(catch_radius, linestyle="--", label="Catch radius")
+    ax.axhline(goal_radius, linestyle="--", label="Goal radius")
+    ax.set_title("Predator Avoidance: Distances vs time")
+    ax.set_xlabel("t (step)")
+    ax.set_ylabel("distance")
+    ax.legend(loc="best")
+    p_dist = save_plot(fig2, f"predator_dist_seed{seed}_{'baseline' if show_baseline else 'rft'}.png")
+
+    fig3 = plt.figure(figsize=(10, 3))
+    ax = fig3.add_subplot(111)
+    ax.step(df["t"], df["action_taken"], where="post")
+    ax.set_title("Predator Avoidance: Action timeline (observer-gated)")
+    ax.set_xlabel("t (step)")
+    ax.set_ylabel("action (0/1)")
+    p_act = save_plot(fig3, f"predator_action_seed{seed}_{'baseline' if show_baseline else 'rft'}.png")
+
+    csv_path = df_to_csv_file(df, f"predator_log_seed{seed}_{'baseline' if show_baseline else 'rft'}.csv")
+
+    min_sep = float(df["d_pred_agent_true"].min()) if len(df) else 0.0
+    steps_ran = int(len(df))
+    success = bool(reached_goal and not caught)
+
+    summary = {
+        "seed": int(seed),
+        "mode": "baseline" if show_baseline else "observer_gated_rft",
+        "steps_ran": steps_ran,
+        "caught": bool(caught),
+        "reached_goal": bool(reached_goal),
+        "success": bool(success),
+        "min_separation_true": float(min_sep),
+        "final_dist_to_goal": float(df["d_agent_goal"].iloc[-1]) if len(df) else None,
+        "energy_proxy_baseline": float(energy_baseline),
+        "energy_proxy_rft": float(energy_rft),
+        "ops_proxy": int(ops_proxy),
+    }
+
+    return summary, [p_traj, p_dist, p_act], csv_path
+
+# -----------------------------
+# Benchmarks (NEO/Jitter/Landing only to keep UI stable)
 # -----------------------------
 def run_benchmarks(
     seed: int,
@@ -566,7 +771,7 @@ def run_benchmarks(
     return txt, score, score_path, all_imgs, [neo_csv, jit_csv, land_csv]
 
 # -----------------------------
-# UI text blocks (your voice, clean + friendly + no consciousness claims)
+# UI text blocks
 # -----------------------------
 HOME_MD = """
 # RFT — Observer Agent Console
@@ -583,10 +788,11 @@ What I’m demonstrating is a simple idea:
 **Decision timing matters.**
 RFT treats timing (τ_eff), uncertainty, and action “commit” as first-class controls.
 
-This Space contains three working agent harnesses:
+This Space contains working agent harnesses:
 - **NEO alerting** (filter noisy close-approach alerts)
 - **Satellite jitter reduction** (reduce actuator duty / chatter while keeping residual low)
-- **Starship-style landing harness** (simplified, but structured to test decision timing under wind/thrust disturbances)
+- **Starship-style landing harness** (simplified timing-control harness)
+- **Predator avoidance** (agent reaches goal while avoiding a pursuing predator)
 
 Every tab shows what it’s doing, why, and where it wins or loses.
 
@@ -667,6 +873,7 @@ I’m demonstrating a decision-timing framework that can be applied to:
 - alert filtering (NEO / tracking)
 - stabilisation (jitter reduction)
 - anomaly-aware control loops (landing harness)
+- pursuit/evasion behaviour (predator avoidance)
 
 This is a runnable harness:
 - you can reproduce results with seeds
@@ -675,9 +882,9 @@ This is a runnable harness:
 - you can change thresholds and see behaviour shift
 
 ## What I’m not claiming
-- I’m not claiming flight certification
+- I’m not claiming certification
 - I’m not claiming any company is using this
-- I’m not claiming this replaces aerospace validation pipelines
+- I’m not claiming this replaces validation pipelines
 
 ## What would make it production-grade
 - real sensor ingestion + timing constraints
@@ -749,6 +956,29 @@ def ui_run_landing(seed, steps, dt, wind_max, thrust_noise, kp_base, kp_rft, gat
     summary_txt = json.dumps(summary, indent=2)
     return summary_txt, imgs[0], imgs[1], imgs[2], imgs[3], csv_path
 
+def ui_run_predator(seed, steps, dt, world_size, predator_speed, agent_speed, catch_radius, sense_noise,
+                    rft_k_attract, rft_k_repulse, base_k_attract, base_k_repulse, gate_th, tau_gain, goal_radius, show_baseline):
+    summary, imgs, csv_path = simulate_predator_avoidance(
+        seed=int(seed),
+        steps=int(steps),
+        dt=float(dt),
+        world_size=float(world_size),
+        predator_speed=float(predator_speed),
+        agent_speed=float(agent_speed),
+        catch_radius=float(catch_radius),
+        sense_noise=float(sense_noise),
+        rft_k_attract=float(rft_k_attract),
+        rft_k_repulse=float(rft_k_repulse),
+        baseline_k_attract=float(base_k_attract),
+        baseline_k_repulse=float(base_k_repulse),
+        gate_threshold=float(gate_th),
+        tau_gain=float(tau_gain),
+        goal_radius=float(goal_radius),
+        show_baseline=bool(show_baseline),
+    )
+    summary_txt = json.dumps(summary, indent=2)
+    return summary_txt, imgs[0], imgs[1], imgs[2], csv_path
+
 def ui_run_bench(seed, neo_steps, neo_dt, neo_alert_km, neo_noise_km, jit_steps, jit_dt, jit_noise, land_steps, land_dt, land_wind, land_thrust_noise, tau_gain):
     txt, score_df, score_csv, imgs, logs = run_benchmarks(
         seed=int(seed),
@@ -766,7 +996,7 @@ def ui_run_bench(seed, neo_steps, neo_dt, neo_alert_km, neo_noise_km, jit_steps,
 # -----------------------------
 # Gradio UI
 # -----------------------------
-with gr.Blocks(title="RFT — Observer Agent Console (NEO / Jitter / Landing)") as demo:
+with gr.Blocks(title="RFT — Observer Agent Console (NEO / Jitter / Landing / Predator)") as demo:
     gr.Markdown(HOME_MD)
 
     with gr.Tabs():
@@ -822,17 +1052,12 @@ with gr.Blocks(title="RFT — Observer Agent Console (NEO / Jitter / Landing)")
             run_b.click(
                 ui_run_bench,
                 inputs=[seed_live, neo_steps, neo_dt, neo_alert, neo_noise, jit_steps, jit_dt, jit_noise, land_steps, land_dt, land_wind, land_thrust_noise, tau_gain_live],
-                outputs=[
-                    bench_txt, bench_table, bench_score_csv,
-                    img1, img2, img3, img4, img5, img6, img7, img8, img9, img10,
-                    neo_log, jit_log, land_log
-                ]
+                outputs=[bench_txt, bench_table, bench_score_csv, img1, img2, img3, img4, img5, img6, img7, img8, img9, img10, neo_log, jit_log, land_log]
             )
 
         with gr.Tab("NEO Observer Agent"):
             gr.Markdown(
                 "# Near-Earth Object (NEO) Observer Agent\n"
-                "This is a test harness for filtering close-approach alerts under noise.\n"
                 "Baseline: distance threshold only.\n"
                 "Observer-gated RFT: distance threshold + confidence + τ_eff decision gate.\n"
             )
@@ -927,13 +1152,54 @@ with gr.Blocks(title="RFT — Observer Agent Console (NEO / Jitter / Landing)")
                 outputs=[out_l_summary, out_l_img1, out_l_img2, out_l_img3, out_l_img4, out_l_csv]
             )
 
-        with gr.Tab("Benchmarks"):
+        with gr.Tab("Predator Avoidance"):
             gr.Markdown(
-                "# Benchmarks\n"
-                "Run full packs from the Live Console tab.\n"
-                "Everything is seeded, logged, and exportable.\n"
+                "# Predator Avoidance (Observer Agent)\n"
+                "Agent tries to reach a goal while avoiding a pursuing predator.\n"
+                "This is a simple pursuit/evasion harness designed to be inspectable and reproducible.\n"
+                "Toggle **Baseline** (always-on control) vs **Observer-gated RFT** (can wait unless risk is high).\n"
+            )
+            with gr.Row():
+                seed_p = gr.Number(value=7, precision=0, label="Seed")
+                steps_p = gr.Slider(50, 1500, value=450, step=1, label="Steps")
+                dt_p = gr.Slider(0.05, 1.0, value=0.20, step=0.01, label="dt")
+            with gr.Row():
+                world_size = gr.Slider(20.0, 300.0, value=120.0, step=5.0, label="World size (bounds)")
+                predator_speed = gr.Slider(0.1, 40.0, value=10.0, step=0.1, label="Predator speed")
+                agent_speed = gr.Slider(0.1, 40.0, value=12.0, step=0.1, label="Agent speed")
+            with gr.Row():
+                catch_radius = gr.Slider(0.5, 30.0, value=6.0, step=0.5, label="Catch radius")
+                goal_radius = gr.Slider(0.5, 30.0, value=6.0, step=0.5, label="Goal radius")
+                sense_noise = gr.Slider(0.0, 10.0, value=1.2, step=0.1, label="Sensor noise (predator)")
+            with gr.Row():
+                base_k_att = gr.Slider(0.1, 10.0, value=1.2, step=0.05, label="Baseline attract gain")
+                base_k_rep = gr.Slider(0.1, 30.0, value=10.0, step=0.1, label="Baseline repulse gain")
+            with gr.Row():
+                rft_k_att = gr.Slider(0.1, 10.0, value=1.2, step=0.05, label="RFT attract gain")
+                rft_k_rep = gr.Slider(0.1, 30.0, value=10.0, step=0.1, label="RFT repulse gain")
+            with gr.Row():
+                gate_p = gr.Slider(0.1, 0.95, value=0.55, step=0.01, label="Gate threshold")
+                tau_p = gr.Slider(0.0, 3.0, value=1.2, step=0.05, label="τ_eff gain")
+                mode_baseline = gr.Checkbox(value=False, label="Run Baseline mode (unchecked = Observer-gated RFT)")
+            run_p = gr.Button("Run Predator Avoidance")
+
+            out_p_summary = gr.Textbox(label="Summary JSON", lines=12)
+            with gr.Row():
+                out_p_img1 = gr.Image(label="Trajectories")
+                out_p_img2 = gr.Image(label="Distances vs time")
+                out_p_img3 = gr.Image(label="Action timeline")
+            out_p_csv = gr.File(label="Download Predator CSV log")
+
+            run_p.click(
+                ui_run_predator,
+                inputs=[seed_p, steps_p, dt_p, world_size, predator_speed, agent_speed, catch_radius, sense_noise,
+                        rft_k_att, rft_k_rep, base_k_att, base_k_rep, gate_p, tau_p, goal_radius, mode_baseline],
+                outputs=[out_p_summary, out_p_img1, out_p_img2, out_p_img3, out_p_csv]
             )
 
+        with gr.Tab("Benchmarks"):
+            gr.Markdown("# Benchmarks\nRun full packs from the Live Console tab.\nEverything is seeded, logged, and exportable.\n")
+
         with gr.Tab("Theory → Practice"):
             gr.Markdown(THEORY_PRACTICE_MD)