Upload inference.py

inference.py

@@ -5,7 +5,7 @@
 Model Repository: https://huggingface.co/fromziro/MrPong
 
 Zero-dependency, standalone script for the public to:
-1. Play against MrPong live in the terminal (Flicker-free, 60 FPS continuous physics)
+1. Play against MrPong live in the terminal (Real-time hold-to-move keyboard control)
 2. Run match simulations against built-in AI baseline opponents
 
 Quickstart:
@@ -34,20 +34,20 @@ except ImportError:
 # Platform-specific real-time non-blocking keyboard input
 IS_WINDOWS = sys.platform.startswith("win")
 if IS_WINDOWS:
+    import ctypes
+    HAS_KEYBOARD = True
+else:
     try:
-        import msvcrt
+        import select
+        import tty
+        import termios
         HAS_KEYBOARD = True
     except ImportError:
         HAS_KEYBOARD = False
-else:
-    import select
-    import tty
-    import termios
-    HAS_KEYBOARD = True
 
 
 # =================================================================================================
-# STANDALONE PING PONG PHYSICS & ENVIRONMENT (ZERO EXTERNAL PROJECT DEPENDENCIES)
+# STANDALONE PING PONG PHYSICS & ENVIRONMENT
 # =================================================================================================
 
 @dataclass
@@ -71,9 +71,11 @@ class StandalonePongEnv:
     def __init__(self, phys: Optional[PhysicsConfig] = None, seed: Optional[int] = None):
         self.phys = phys or PhysicsConfig()
         self.rng = random.Random(seed)
+        self.ego_paddle_h = self.phys.paddle_height
+        self.opp_paddle_h = self.phys.paddle_height
         self.reset()
 
-    def reset(self, serve_direction: Optional[int] = None) -> np.ndarray:
+    def reset(self, serve_direction: Optional[int] = None, initial_speed: Optional[float] = None) -> np.ndarray:
         self.ego_y = self.phys.table_height / 2.0
         self.opp_y = self.phys.table_height / 2.0
         self.ego_vy = 0.0
@@ -86,8 +88,8 @@ class StandalonePongEnv:
         if serve_direction is None:
             serve_direction = 1 if self.rng.random() < 0.5 else -1
 
-        serve_angle = self.rng.uniform(-math.pi / 6.0, math.pi / 6.0)
-        speed = self.phys.ball_speed_initial
+        serve_angle = self.rng.uniform(-math.pi / 7.0, math.pi / 7.0)
+        speed = initial_speed or self.phys.ball_speed_initial
         self.ball_vx = serve_direction * speed * math.cos(serve_angle)
         self.ball_vy = speed * math.sin(serve_angle)
         self.rally_count = 0
@@ -117,9 +119,11 @@ class StandalonePongEnv:
         self.ego_vy = alpha * self.ego_vy + (1.0 - alpha) * ego_target_v
         self.opp_vy = alpha * self.opp_vy + (1.0 - alpha) * opp_target_v
 
-        half_h = self.phys.paddle_height / 2.0
-        self.ego_y = float(np.clip(self.ego_y + self.ego_vy, half_h, self.phys.table_height - half_h))
-        self.opp_y = float(np.clip(self.opp_y + self.opp_vy, half_h, self.phys.table_height - half_h))
+        ego_half_h = self.ego_paddle_h / 2.0
+        opp_half_h = self.opp_paddle_h / 2.0
+
+        self.ego_y = float(np.clip(self.ego_y + self.ego_vy, ego_half_h, self.phys.table_height - ego_half_h))
+        self.opp_y = float(np.clip(self.opp_y + self.opp_vy, opp_half_h, self.phys.table_height - opp_half_h))
 
         prev_ball_x = self.ball_x
         prev_ball_y = self.ball_y
@@ -141,10 +145,10 @@ class StandalonePongEnv:
             y_ball_at_impact = prev_ball_y + t * self.ball_vy
             y_ego_at_impact = prev_ego_y + t * (self.ego_y - prev_ego_y)
 
-            if abs(y_ball_at_impact - y_ego_at_impact) <= (half_h + r * 0.6):
+            if abs(y_ball_at_impact - y_ego_at_impact) <= (ego_half_h + r * 0.6):
                 hit_occurred = True
                 self.rally_count += 1
-                offset = float(np.clip((y_ball_at_impact - y_ego_at_impact) / half_h, -1.0, 1.0))
+                offset = float(np.clip((y_ball_at_impact - y_ego_at_impact) / ego_half_h, -1.0, 1.0))
                 bounce_angle = offset * (math.pi / 3.0)
                 current_speed = math.hypot(self.ball_vx, self.ball_vy)
                 new_speed = min(current_speed * self.phys.ball_acceleration, self.phys.ball_speed_max)
@@ -163,10 +167,10 @@ class StandalonePongEnv:
             y_ball_at_impact = prev_ball_y + t * self.ball_vy
             y_opp_at_impact = prev_opp_y + t * (self.opp_y - prev_opp_y)
 
-            if abs(y_ball_at_impact - y_opp_at_impact) <= (half_h + r * 0.6):
+            if abs(y_ball_at_impact - y_opp_at_impact) <= (opp_half_h + r * 0.6):
                 hit_occurred = True
                 self.rally_count += 1
-                offset = float(np.clip((y_ball_at_impact - y_opp_at_impact) / half_h, -1.0, 1.0))
+                offset = float(np.clip((y_ball_at_impact - y_opp_at_impact) / opp_half_h, -1.0, 1.0))
                 bounce_angle = offset * (math.pi / 3.0)
                 current_speed = math.hypot(self.ball_vx, self.ball_vy)
                 new_speed = min(current_speed * self.phys.ball_acceleration, self.phys.ball_speed_max)
@@ -254,7 +258,7 @@ class StandalonePongEnv:
         w, h = self.phys.table_width, self.phys.table_height
         v_max = self.phys.ball_speed_max
         pv_max = self.phys.paddle_speed
-        half_h = self.phys.paddle_height / 2.0
+        half_h = self.ego_paddle_h / 2.0
         ego_x = self.phys.paddle_width
 
         pred_intercept_y = self.calculate_intercept_y(ego_x, self.ball_x, self.ball_y, self.ball_vx, self.ball_vy)
@@ -289,7 +293,7 @@ class StandalonePongEnv:
         w, h = self.phys.table_width, self.phys.table_height
         v_max = self.phys.ball_speed_max
         pv_max = self.phys.paddle_speed
-        half_h = self.phys.paddle_height / 2.0
+        half_h = self.opp_paddle_h / 2.0
         opp_x = self.phys.table_width - self.phys.paddle_width
 
         pred_intercept_y = self.calculate_intercept_y(opp_x, self.ball_x, self.ball_y, self.ball_vx, self.ball_vy)
@@ -408,44 +412,77 @@ class RandomOpponent:
 
 
 # =================================================================================================
-# NON-BLOCKING KEYBOARD INPUT
+# REAL-TIME HOLD-TO-MOVE KEYBOARD CONTROLLER
 # =================================================================================================
 
 class KeyboardController:
+    """
+    Direct hardware key state listener.
+    Holding W/Up moves UP. Holding S/Down moves DOWN. Releasing stops (STAY).
+    """
     def __init__(self):
         self.is_windows = IS_WINDOWS
-        if not self.is_windows and HAS_KEYBOARD:
-            self.old_settings = termios.tcgetattr(sys.stdin)
-            tty.setcbreak(sys.stdin.fileno())
-
-    def get_action(self) -> Optional[int]:
-        if not HAS_KEYBOARD:
-            return None
-
         if self.is_windows:
-            if msvcrt.kbhit():
-                ch = msvcrt.getch()
-                if ch in [b'w', b'W', b'H']:  # 'w' or Up Arrow
-                    return 1
-                elif ch in [b's', b'S', b'P']:  # 's' or Down Arrow
-                    return 2
-                elif ch in [b' ', b'\r', b'\n']:
-                    return 0
-                elif ch in [b'q', b'Q']:
-                    return -1
+            self.user32 = ctypes.windll.user32
+            # Virtual Key Codes
+            self.VK_W = 0x57
+            self.VK_S = 0x53
+            self.VK_UP = 0x26
+            self.VK_DOWN = 0x28
+            self.VK_Q = 0x51
+            self.VK_ESCAPE = 0x1B
+        else:
+            self.decay_frames = 0
+            self.current_act = 0
+            if HAS_KEYBOARD:
+                self.old_settings = termios.tcgetattr(sys.stdin)
+                tty.setcbreak(sys.stdin.fileno())
+
+    def get_action(self) -> int:
+        """
+        Returns:
+          1: UP (while W / Up Arrow is held)
+          2: DOWN (while S / Down Arrow is held)
+          0: STAY (when released)
+         -1: QUIT (when Q / ESC is pressed)
+        """
+        if self.is_windows:
+            # Check Quit
+            if (self.user32.GetAsyncKeyState(self.VK_Q) & 0x8000) or (self.user32.GetAsyncKeyState(self.VK_ESCAPE) & 0x8000):
+                return -1
+
+            # Direct hardware physical key state check (0 latency)
+            w_held = bool((self.user32.GetAsyncKeyState(self.VK_W) & 0x8000) or (self.user32.GetAsyncKeyState(self.VK_UP) & 0x8000))
+            s_held = bool((self.user32.GetAsyncKeyState(self.VK_S) & 0x8000) or (self.user32.GetAsyncKeyState(self.VK_DOWN) & 0x8000))
+
+            if w_held and not s_held:
+                return 1
+            elif s_held and not w_held:
+                return 2
+            else:
+                return 0
         else:
+            # Unix non-blocking input with key-release decay
+            if not HAS_KEYBOARD:
+                return 0
             rlist, _, _ = select.select([sys.stdin], [], [], 0)
             if rlist:
                 ch = sys.stdin.read(1)
                 if ch in ['w', 'W']:
-                    return 1
+                    self.current_act = 1
+                    self.decay_frames = 5
                 elif ch in ['s', 'S']:
-                    return 2
-                elif ch in [' ', '\n']:
-                    return 0
+                    self.current_act = 2
+                    self.decay_frames = 5
                 elif ch in ['q', 'Q']:
                     return -1
-        return None
+
+            if self.decay_frames > 0:
+                self.decay_frames -= 1
+                return self.current_act
+            else:
+                self.current_act = 0
+                return 0
 
     def close(self):
         if not self.is_windows and HAS_KEYBOARD:
@@ -460,11 +497,9 @@ class KeyboardController:
 # =================================================================================================
 
 def resolve_model_path(path_or_id: Optional[str] = None) -> str:
-    # 1. If explicit local path exists, use its absolute path
     if path_or_id and os.path.exists(path_or_id):
         return os.path.abspath(path_or_id)
 
-    # 2. If local ./MrPong repository folder exists locally, prefer it over remote hub
     local_candidates = [
         os.path.abspath("./MrPong"),
         os.path.abspath("C:/Users/harley/MrPong"),
@@ -496,51 +531,71 @@ class PublicMrPongRunner:
         return self.model.act(obs_input, deterministic=deterministic)
 
     # ---------------------------------------------------------------------------------------------
-    # PLAY INTERACTIVELY IN TERMINAL (FLICKER-FREE 60 FPS CONTINUOUS SUB-FRAME RENDERING)
+    # PLAY INTERACTIVELY IN TERMINAL (HOLD TO MOVE, RELEASE TO STAY)
     # ---------------------------------------------------------------------------------------------
-    def play(self, points_to_win: int = 5, target_fps: int = 40):
-        env = StandalonePongEnv()
+    def play(self, points_to_win: int = 5, difficulty: str = "normal"):
+        phys = PhysicsConfig()
+
+        if difficulty == "easy":
+            phys.paddle_speed = 9.0
+            initial_ball_speed = 3.5
+            ego_paddle_h = 110.0
+            frame_delay = 0.028
+        elif difficulty == "hard":
+            phys.paddle_speed = 8.0
+            initial_ball_speed = 6.0
+            ego_paddle_h = 80.0
+            frame_delay = 0.022
+        else:  # normal
+            phys.paddle_speed = 8.5
+            initial_ball_speed = 4.2
+            ego_paddle_h = 95.0
+            frame_delay = 0.025
+
+        env = StandalonePongEnv(phys)
+        env.ego_paddle_h = ego_paddle_h
         kbd = KeyboardController()
+
         human_score = 0
         ai_score = 0
         max_rally = 0
-        dt = 1.0 / target_fps
 
         print("=" * 64)
-        print("         🏓 PLAY AGAINST MRPONG (FLICKER-FREE TERMINAL MODE)")
+        print(f"      🏓 PLAY AGAINST MRPONG (ARCADE MODE - {difficulty.upper()})")
         print("=" * 64)
         print(" Controls:")
-        print("   [W] / [Up Arrow]   : Move Paddle UP")
-        print("   [S] / [Down Arrow] : Move Paddle DOWN")
-        print("   [Space]            : STAY still")
-        print("   [Q]                : Quit match")
-        print(f"\n First to {points_to_win} points wins!")
-        input("\n Press [ENTER] to serve and start match...")
-
-        # Hide terminal cursor for crisp rendering
+        print("   Hold [W] / [Up Arrow]   : Move UP")
+        print("   Hold [S] / [Down Arrow] : Move DOWN")
+        print("   Release key             : STAY still")
+        print("   [Q] / [Esc]             : Quit match")
+        print(f"\n First player to {points_to_win} points wins!")
+        input("\n Press [ENTER] to start match...")
+
         sys.stdout.write("\033[?25l")
         sys.stdout.flush()
 
-        human_act = 0
         serve_dir = 1
         ai_act = 0
         substep_count = 0
 
         try:
             while human_score < points_to_win and ai_score < points_to_win:
-                obs = env.reset(serve_direction=serve_dir)
+                obs = env.reset(serve_direction=serve_dir, initial_speed=initial_ball_speed)
+
+                for cd in [3, 2, 1]:
+                    self._render_terminal_court(env, human_score, ai_score, 0, ai_act, points_to_win, banner=f"GET READY: SERVING IN {cd}...")
+                    time.sleep(0.6)
+
                 done = False
 
                 while not done:
                     t_start = time.perf_counter()
 
-                    # 1. Read non-blocking human input on every physics sub-step
-                    key_act = kbd.get_action()
-                    if key_act == -1:
+                    # 1. Real-time physical key check (Hold = Move, Release = Stay)
+                    human_act = kbd.get_action()
+                    if human_act == -1:
                         print("\n[!] Match aborted by player.")
                         return
-                    elif key_act is not None:
-                        human_act = key_act
 
                     # 2. Query AI policy every frame_skip sub-steps
                     if substep_count % env.phys.frame_skip == 0:
@@ -549,7 +604,7 @@ class PublicMrPongRunner:
 
                     substep_count += 1
 
-                    # 3. Advance exactly 1 continuous physics sub-step
+                    # 3. Advance continuous physics sub-step
                     done, info = env.physics_substep(ego_action=human_act, opp_action=ai_act)
 
                     if env.rally_count > max_rally:
@@ -558,28 +613,27 @@ class PublicMrPongRunner:
                     # 4. Render smooth flicker-free frame
                     self._render_terminal_court(env, human_score, ai_score, human_act, ai_act, points_to_win)
 
-                    # 5. Maintain rock-solid frame timing
+                    # 5. Precise frame timing
                     t_elapsed = time.perf_counter() - t_start
-                    if t_elapsed < dt:
-                        time.sleep(dt - t_elapsed)
+                    if t_elapsed < frame_delay:
+                        time.sleep(frame_delay - t_elapsed)
 
                 # Point completed
                 winner = info.get("winner")
                 if winner == "ego":
                     human_score += 1
                     serve_dir = 1
-                    banner = ">>> YOU SCORED! <<<"
+                    banner = ">>> POINT TO YOU! <<<"
                 elif winner == "opponent":
                     ai_score += 1
                     serve_dir = -1
-                    banner = ">>> MRPONG SCORED! <<<"
+                    banner = ">>> POINT TO MRPONG! <<<"
                 else:
                     banner = ">>> RALLY DRAW <<<"
 
-                self._render_terminal_court(env, human_score, ai_score, human_act, ai_act, points_to_win, banner=banner)
-                time.sleep(1.0)
+                self._render_terminal_court(env, human_score, ai_score, 0, ai_act, points_to_win, banner=banner)
+                time.sleep(1.2)
 
-            # Match Over
             sys.stdout.write("\033[?25h\033[H\033[J")
             print("\n" + "=" * 64)
             if human_score >= points_to_win:
@@ -594,25 +648,26 @@ class PublicMrPongRunner:
             kbd.close()
 
     def _render_terminal_court(self, env: StandalonePongEnv, s1: int, s2: int, a1: int, a2: int, target: int, banner: str = ""):
-        """Atomic flicker-free frame buffer rendering (overwrites cursor home without clear)."""
+        """Atomic flicker-free frame buffer rendering."""
         cols, rows = 60, 18
         bx = int(np.clip((env.ball_x / env.phys.table_width) * (cols - 1), 0, cols - 1))
         by = int(np.clip((env.ball_y / env.phys.table_height) * (rows - 1), 0, rows - 1))
         p1_y = int(np.clip((env.ego_y / env.phys.table_height) * (rows - 1), 0, rows - 1))
         p2_y = int(np.clip((env.opp_y / env.phys.table_height) * (rows - 1), 0, rows - 1))
-        ph = max(1, int((env.phys.paddle_height / env.phys.table_height) * (rows - 1) / 2))
+
+        p1_ph = max(1, int((env.ego_paddle_h / env.phys.table_height) * (rows - 1) / 2))
+        p2_ph = max(1, int((env.opp_paddle_h / env.phys.table_height) * (rows - 1) / 2))
 
         act_names = ["STAY", " UP ", "DOWN"]
         buf = []
 
-        # Cursor Home only (avoids screen erase strobe)
         buf.append("\033[H")
         buf.append("+" + "-" * (cols + 2) + "+\n")
         buf.append(f"|  YOU [P1]: {s1}/{target} ({act_names[a1]})" + " " * (cols - 41) + f"MRPONG [AI]: {s2}/{target} ({act_names[a2]})  |\n")
         buf.append("+" + "-" * (cols + 2) + "+\n")
 
         for r in range(rows):
-            line = ["|" if abs(r - p1_y) <= ph else " "]
+            line = ["#" if abs(r - p1_y) <= p1_ph else " "]
             for c in range(cols):
                 if r == by and c == bx:
                     line.append("O")
@@ -620,7 +675,7 @@ class PublicMrPongRunner:
                     line.append(":")
                 else:
                     line.append(" ")
-            line.append("|" if abs(r - p2_y) <= ph else " ")
+            line.append("|" if abs(r - p2_y) <= p2_ph else " ")
             buf.append("|" + "".join(line) + "|\n")
 
         buf.append("+" + "-" * (cols + 2) + "+\n")
@@ -631,7 +686,6 @@ class PublicMrPongRunner:
             buf.append(f"| Rally: {env.rally_count:2d} hits | Ball Speed: {speed:4.1f} px/f" + " " * (cols - 33) + "|\n")
         buf.append("+" + "-" * (cols + 2) + "+\n")
 
-        # Single atomic write
         sys.stdout.write("".join(buf))
         sys.stdout.flush()
 
@@ -703,8 +757,8 @@ def main():
     parser = argparse.ArgumentParser(description="MrPong Hugging Face Public Inference Runner (fromziro/MrPong)")
     parser.add_argument("--model", type=str, default="fromziro/MrPong", help="Hugging Face repo ID or local path (default: fromziro/MrPong)")
     parser.add_argument("--mode", type=str, choices=["play", "simulate"], default="play", help="Mode: 'play' to play against AI, 'simulate' for AI vs AI match simulations")
+    parser.add_argument("--difficulty", type=str, choices=["easy", "normal", "hard"], default="normal", help="Difficulty preset for human play mode (default: normal)")
     parser.add_argument("--points", type=int, default=5, help="Points to win in play mode (default: 5)")
-    parser.add_argument("--fps", type=int, default=40, help="Target frame rate for interactive gameplay (default: 40)")
     parser.add_argument("--opponent", type=str, default="realistic_hard", choices=["realistic_hard", "medium", "easy", "impossible_hard", "random"], help="Opponent type in simulate mode")
     parser.add_argument("--matches", type=int, default=20, help="Number of matches to simulate (default: 20)")
 
@@ -713,7 +767,7 @@ def main():
     runner = PublicMrPongRunner(model_id_or_path=args.model)
 
     if args.mode == "play":
-        runner.play(points_to_win=args.points, target_fps=args.fps)
+        runner.play(points_to_win=args.points, difficulty=args.difficulty)
     elif args.mode == "simulate":
         runner.simulate(opponent_type=args.opponent, num_matches=args.matches)