Statistics observer

StatsDashboard.py

@@ -0,0 +1,135 @@
+import streamlit as st
+import pandas as pd
+
+# titles and initializations
+st.set_page_config(page_title="Game Stats Dashboard", layout="wide")
+
+st.title("Game Stats Dashboard")
+st.caption("Interactive dashboard for level outcomes, death causes, and player behavior.")
+
+deaths_df = pd.read_csv("level_clear_rate.csv")
+attempts_df = pd.read_csv("stats.csv")
+
+# Filters
+st.sidebar.header("Filters")
+
+levels = sorted(attempts_df["level"].unique(), key=lambda x: tuple(map(int, x.split("-"))))
+selected_levels = st.sidebar.multiselect("Select levels", levels, default=levels)
+
+filtered_df = attempts_df[
+    attempts_df["level"].isin(selected_levels)
+].copy()
+
+
+# metrics visualization
+total_attempts = len(filtered_df)
+success_count = (filtered_df["cause_of_death"] == "Success").sum()
+death_count = total_attempts - success_count
+success_rate = (success_count / total_attempts * 100) if total_attempts else 0
+avg_time = filtered_df["elapsed_time"].mean() if total_attempts else 0
+avg_velocity = filtered_df["avg_horizontal_velocity"].mean() if total_attempts else 0
+
+col1, col2, col3, col4, col5, col6 = st.columns(6)
+col1.metric("Attempts", total_attempts)
+col2.metric("Successes", success_count)
+col3.metric("Deaths", death_count)
+col4.metric("Success Rate", f"{success_rate:.1f}%")
+col5.metric("Avg Time", f"{avg_time:.2f}s")
+col6.metric("Avg Horizontal Velocity", f"{avg_velocity:.2f}")
+
+# Level table
+st.subheader("Level Table")
+
+level_table = filtered_df.groupby("level").agg(
+    attempts=("level", "count"),
+    successes=("cause_of_death", lambda x: (x == "Success").sum()),
+    avg_jumps=("jumps", "mean"),
+    avg_coins=("coins_collected", "mean"),
+    avg_enemies_killed=("enemies_killed", "mean"),
+    avg_time=("elapsed_time", "mean"),
+    avg_velocity=("avg_horizontal_velocity", "mean"),
+    avg_idle_ratio=("Idle_Ratio", "mean"),
+    avg_actions_per_second=("Actions_Per_Second", "mean"),
+).reset_index()
+
+level_table["success_rate"] = (
+    level_table["successes"] / level_table["attempts"] * 100
+).round(2)
+
+st.dataframe(level_table, use_container_width=True)
+
+# Attempts and Death Distribution Charts
+col1, col2 = st.columns(2)
+
+with col1:
+    st.subheader("Attempts by Level")
+    attempts_per_level = filtered_df["level"].value_counts().sort_index()
+    st.bar_chart(attempts_per_level)
+
+with col2:
+    st.subheader("Death Distribution")
+    death_counts = (
+        filtered_df[filtered_df["cause_of_death"] != "Success"]
+        ["cause_of_death"]
+        .value_counts())
+    st.bar_chart(death_counts)
+
+# Average Time and Velocity Charts
+col1, col2 = st.columns(2)
+
+with col1:
+    st.subheader("Average Elapsed Time by Level")
+    avg_time_by_level = filtered_df.groupby("level")["elapsed_time"].mean().sort_index()
+    st.bar_chart(avg_time_by_level)
+
+with col2:
+    st.subheader("Average Horizontal Velocity by Level")
+    avg_velocity_by_level = filtered_df.groupby("level")["avg_horizontal_velocity"].mean().sort_index()
+    st.bar_chart(avg_velocity_by_level)
+
+# Death Summary
+st.subheader("Death Summary from Aggregated CSV")
+st.dataframe(deaths_df, use_container_width=True)
+
+death_breakdown = deaths_df.set_index("level")[["Pit", "enemy", "stall"]]
+st.bar_chart(death_breakdown)
+
+# Action Usage Table
+st.subheader("Action Analysis")
+
+action_cols = [col for col in filtered_df.columns if col.startswith("action_")]
+
+if not filtered_df.empty:
+    action_totals = filtered_df[action_cols].sum().sort_values(ascending=False)
+
+    action_by_level = filtered_df.groupby("level")[action_cols].sum()
+    st.subheader("Action Usage by Level")
+    st.dataframe(action_by_level, use_container_width=True)
+
+    st.bar_chart(action_totals)
+
+# Detailed attempts Table
+
+st.subheader("Detailed Attempts")
+st.dataframe(filtered_df, use_container_width=True)
+
+# Quick Insights
+st.subheader("Quick Insights")
+
+if total_attempts > 0:
+    most_common_outcome = filtered_df["cause_of_death"].value_counts().idxmax()
+    hardest_level = (
+        filtered_df[filtered_df["cause_of_death"] != "Success"]["level"]
+        .value_counts()
+        .idxmax()
+        if not filtered_df[filtered_df["cause_of_death"] != "Success"].empty
+        else "N/A"
+    )
+
+    st.write(f"**Most common outcome:** {most_common_outcome}")
+    st.write(f"**Level with most failures:** {hardest_level}")
+
+    if "Success" in filtered_df["cause_of_death"].values:
+        success_df = filtered_df[filtered_df["cause_of_death"] == "Success"]
+        st.write(f"**Average time on successful runs:** {success_df['elapsed_time'].mean():.2f}s")
+        st.write(f"**Average idle ratio on successful runs:** {success_df['Idle_Ratio'].mean():.2f}")

code/games/action_map.py

@@ -0,0 +1,5 @@
+ACTION_NAMES = {
+    0: "IDLE", 1: "LEFT", 2: "RIGHT", 3: "JUMP",
+    4: "RIGHT+JUMP", 5: "RUN+RIGHT", 6: "LEFT+JUMP", 7: "RUN+RIGHT+JUMP", 
+    8: "RUN+LEFT", 9: "RUN+LEFT+JUMP"
+}

code/games/modules/Objects/Player.py

@@ -4,6 +4,7 @@
 import pygame
 from typing import Dict, List, Any, Optional
 from enum import Enum, auto
+from ..Stats.StatsObserver import track
 
 # Imports from sibling packages
 from ..Parameters import Movement_parameters as MP
@@ -127,12 +128,11 @@ def _update_animation_logic(self, dt: float):
         # Hand the state (int) to the handler
         self.anim_handler.set_state(target_state.value)
 
-    def handle_input(self, a: int):    
-        agent_left = (a in (1,6))
-        agent_right = (a in (2,4,5,7))
-        agent_jump = (a in (3,4,6,7))
-        agent_run = (a in (5,7))
-
+    @track("horizontal_velocity")
+    def set_horizontal_velocity(self, vx):
+        self.vx = vx
+
+    def check_input(self, a: int):        
         kb_left = kb_right = kb_jump = kb_run = False
 
         if pygame.get_init():
@@ -141,7 +141,16 @@ def handle_input(self, a: int):
             kb_right = keys[pygame.K_RIGHT] or keys[pygame.K_d]
             kb_jump = keys[pygame.K_SPACE] or keys[pygame.K_w] or keys[pygame.K_UP]
             kb_run = keys[pygame.K_LSHIFT] or keys[pygame.K_RSHIFT]
-
+
+        self.handle_input(a, kb_left, kb_right, kb_jump, kb_run)
+
+    @track("input")
+    def handle_input(self, a:int, kb_left, kb_right, kb_jump, kb_run):
+        agent_left = (a in (1,6))
+        agent_right = (a in (2,4,5,7))
+        agent_jump = (a in (3,4,6,7))
+        agent_run = (a in (5,7))
+
         is_left = agent_left or kb_left
         is_right = agent_right or kb_right
         self.jump_pressed = agent_jump or kb_jump
@@ -155,6 +164,7 @@ def handle_input(self, a: int):
         if self.jump_pressed:
             self.jump_buffer = 6 
 
+
     def apply_physics(self, dt: float, ctx: PhysicsContext):
         if self.run_held:
             target_max = ctx.MAX_RUN_SPEED
@@ -171,14 +181,14 @@ def apply_physics(self, dt: float, ctx: PhysicsContext):
             skidding = (self.vx > 0 and self.input_dir < 0) or (self.vx < 0 and self.input_dir > 0)
 
             if self.on_ground and skidding:
-                self.vx += (self.input_dir * ctx.SKID_DECEL * dt)
+                self.set_horizontal_velocity(self.vx + (self.input_dir * ctx.SKID_DECEL * dt))
             else:
-                if self.input_dir > 0: self.vx = min(self.vx + (accel_rate * dt), target_max)
-                else: self.vx = max(self.vx - (accel_rate * dt), -target_max)
+                if self.input_dir > 0: self.set_horizontal_velocity(min(self.vx + (accel_rate * dt), target_max))
+                else: self.set_horizontal_velocity(max(self.vx - (accel_rate * dt), -target_max))
         else:
             friction = (ctx.GROUND_FRICTION if self.on_ground else ctx.AIR_FRICTION) * dt
-            if self.vx > 0: self.vx = max(0, self.vx - friction)
-            elif self.vx < 0: self.vx = min(0, self.vx + friction)
+            if self.vx > 0: self.set_horizontal_velocity(max(0, self.vx - friction))
+            elif self.vx < 0: self.set_horizontal_velocity(min(0, self.vx + friction))
 
         self.handle_jump(dt, ctx)
 
@@ -190,19 +200,27 @@ def handle_jump(self, dt: float, ctx: PhysicsContext):
         if self.jump_buffer > 0: self.jump_buffer -= 1
 
         if (self.coyote > 0) and (self.jump_hold == 0) and (self.jump_buffer > 0):
-            base = ctx.JUMP_VEL_MIN
-            bonus = min(2.2, abs(self.vx) * ctx.SPEED_JUMP_BONUS)
-            self.vy = base - bonus 
-            self.on_ground = False
-            self.coyote = 0
-            self.jump_hold = ctx.JUMP_HOLD_FRAMES
-            self.jump_buffer = 0
+            self.start_jump(ctx)
 
         if self.jump_hold > 0:
             if self.jump_pressed:
                 self.vy -= 0.30 * (dt * 60) 
             self.jump_hold -= 1
-
+
+    @track("jump")
+    def start_jump(self, ctx: PhysicsContext):
+        base = ctx.JUMP_VEL_MIN
+        bonus = min(2.2, abs(self.vx) * ctx.SPEED_JUMP_BONUS)
+        self.vy = base - bonus 
+        self.on_ground = False
+        self.coyote = 0
+        self.jump_hold = ctx.JUMP_HOLD_FRAMES
+        self.jump_buffer = 0
+
+# Player config
+# Tracking true or false 
+# What to track
+
     def render(self, surface: pygame.Surface, sx: float, sy: float, debug: bool = True):
         # 1. Get Sprite from Handler
         sprite = None

code/games/modules/Stats/LevelStatsObserver.py

@@ -0,0 +1,110 @@
+from ...action_map import ACTION_NAMES
+
+class LevelStatsObserver:
+    def __init__(self, levelName):
+        self.deathCause = "Success" 
+        self.jumps = 0
+        self.coins_collected = 0
+        self.enemies_killed = 0
+        self.actions = {i: 0 for i in ACTION_NAMES}
+        self.sum_vx = 0.0
+        self.count_vx = 0
+        self.levelName = levelName
+
+    def record(self, event_type, **data):
+        handler_name = f"record_{event_type}"
+        handler = getattr(self, handler_name, None)
+
+        if handler:
+            handler(**data)
+
+    def record_jump(self, **data):
+        self.jumps += 1
+
+    def record_death(self, cause, **data):
+        self.deathCause = cause
+
+    def record_coins_collected(self, **data):
+        self.coins_collected += 1
+
+    def record_enemies_killed(self, **data):
+        self.enemies_killed += 1
+
+    def record_horizontal_velocity(self, vx, **data):
+        self.sum_vx += abs(vx)
+        self.count_vx += 1
+
+
+    def record_input(self, **data):
+        # Extract the keyboard/agent booleans from data
+        kb_left  = data.get("kb_left", False)
+        kb_right = data.get("kb_right", False)
+        kb_jump  = data.get("kb_jump", False)
+        kb_run   = data.get("kb_run", False)
+
+        # Compute action index (0–9) based on the original ACTION_NAMES mapping
+        if kb_run:
+            if kb_left and kb_jump:
+                action_id = 9  # RUN+LEFT+JUMP
+            elif kb_left:
+                action_id = 8  # RUN+LEFT
+            elif kb_right and kb_jump:
+                action_id = 7  # RUN+RIGHT+JUMP
+            elif kb_right:
+                action_id = 5  # RUN+RIGHT
+            else:
+                action_id = 0  # IDLE / run alone not mapped
+        else:
+            if kb_left and kb_jump:
+                action_id = 6  # LEFT+JUMP
+            elif kb_left:
+                action_id = 1  # LEFT
+            elif kb_right and kb_jump:
+                action_id = 4  # RIGHT+JUMP
+            elif kb_right:
+                action_id = 2  # RIGHT
+            elif kb_jump:
+                action_id = 3  # JUMP
+            else:
+                action_id = 0  # IDLE
+        # Increment the counter
+        self.actions[action_id] += 1
+
+    def set_elapsed_time (self, elapsed_time):
+        self.elapsed_time = elapsed_time
+
+    def get_average_vx(self):
+        if (self.count_vx == 0):
+            return 0
+        else:
+            return self.sum_vx / self.count_vx
+
+    def to_dict(self):
+        row = {
+            "level": self.levelName,
+            "jumps": self.jumps,
+            "coins_collected": self.coins_collected,
+            "enemies_killed": self.enemies_killed,
+            "elapsed_time": round(self.elapsed_time, 2),
+            "avg_horizontal_velocity": round(self.get_average_vx(), 2),
+            "cause_of_death": self.deathCause
+        }
+
+        for action_id in ACTION_NAMES:
+            row[f"action_{ACTION_NAMES[action_id]}"] = self.actions.get(action_id, 0)
+
+        return row
+
+    def reset(self):
+        exclude = {"deaths"}  # keep elapsed time
+        for attr, value in self.__dict__.items():
+            if attr in exclude:
+                continue
+            if isinstance(value, int) or isinstance(value, float):
+                setattr(self, attr, 0)
+            elif isinstance(value, dict):
+                setattr(self, attr, {})
+            elif isinstance(value, type(self.actions)):
+                setattr(self, attr, {i: 0 for i in ACTION_NAMES})
+
+        self.actions = {i: 0 for i in ACTION_NAMES}