Upload app.py

app.py

@@ -0,0 +1,306 @@
+# RPS_SIMPLE_DEMO/app.py
+import gradio as gr
+import uuid
+import httpx
+import os
+#from RockPaperScissor.repositories import CombinedStorage
+
+# FastAPI imports
+from fastapi import FastAPI, Request, Body
+from fastapi.responses import RedirectResponse
+from RockPaperScissor.routes import game_router
+from RockPaperScissor.services.service_instance import game_service
+from RockPaperScissor.services.LLM_service import LLMService
+
+# --- Game Constants and Types ---
+from enum import Enum
+
+class Move(Enum):
+    ROCK = "rock"
+    PAPER = "paper"
+    SCISSORS = "scissors"
+
+class GameResult(Enum):
+    PLAYER_WIN = "player_win"
+    AI_WIN = "ai_win"
+    DRAW = "draw"
+
+# --- Gradio Interface ---
+class RockPaperScissorsUI:
+    def __init__(self, game_service):
+        self.game_service = game_service
+        self.session_id = None  # Will be set from State
+        self.ai_models = list(self.game_service.ai_models.keys())
+        self.ai_descriptions = {
+            "random": "Random AI: Makes completely random moves.",
+            "adaptive_markov": "Adaptive Markov AI: Uses entropy-weighted Markov and frequency models to predict your next move."
+        }
+        self.last_move = None
+
+    async def reset_session(self, session_id: str):
+        # Use environment variable for base URL, fallback to localhost
+        base_url = os.getenv("HF_SPACE_URL", "http://localhost:7860")
+        async with httpx.AsyncClient() as client:
+            response = await client.post(
+                f"{base_url}/api/game/end",
+                json={"session_id": session_id}
+            )
+            print("[DEBUG] /api/game/end response:", response.text)
+        # Optionally, also clear the session in-memory (if needed)
+        await self.game_service.clear_session(session_id)
+        return {"status": "ok"}
+
+    def create_interface(self):
+        with gr.Blocks(theme=gr.themes.Soft(), title="Rock Paper Scissors 🎮") as demo:
+            gr.Markdown("# 🪨📄✂️ Rock Paper Scissors")
+            
+            with gr.Row():
+                with gr.Column(scale=1):
+                    gr.Markdown("### 🎮 Game Setup")
+                    ai_dropdown = gr.Dropdown(
+                        choices=self.ai_models,
+                        value=self.ai_models[0],
+                        label="Select AI Opponent"
+                    )
+                    ai_description = gr.Markdown(self.ai_descriptions[self.ai_models[0]])
+                    
+                    with gr.Row():
+                        rock_btn = gr.Button("🪨 Rock", variant="secondary", elem_classes=["move-btn"])
+                        paper_btn = gr.Button("📄 Paper", variant="secondary", elem_classes=["move-btn"])
+                        scissors_btn = gr.Button("✂️ Scissors", variant="secondary", elem_classes=["move-btn"])
+                    # Add End Game button
+                    end_btn = gr.Button("End Game", variant="stop", elem_id="end-game-btn")
+                    # Add Help button
+                    help_btn = gr.Button("💡 Get Help", variant="primary", elem_id="help-btn")
+                    # Add a dedicated box for the help answer, initially empty
+                    help_answer_box = gr.Markdown("", visible=True)
+
+                with gr.Column(scale=2):
+                    gr.Markdown("### 📊 Game Statistics")
+                    stats_display = gr.Markdown()
+                    result_display = gr.Markdown("Make your move!")
+                    end_result_display = gr.Markdown(visible=False)
+                    status_display = gr.Markdown(visible=True)
+
+            ai_dropdown.change(
+                fn=self.update_ai_description,
+                inputs=[ai_dropdown],
+                outputs=[ai_description]
+            )
+
+            # Use a Gradio State to store the session ID
+            move_state = gr.State("")
+            session_id_state = gr.State("")
+
+            async def play_rock(ai_type, session_id):
+                if not session_id:
+                    session_id = f"session_{uuid.uuid4()}"
+                self.session_id = session_id
+                stats, result = await self.play_round(ai_type, "rock")
+                return stats, result, session_id
+            async def play_paper(ai_type, session_id):
+                if not session_id:
+                    session_id = f"session_{uuid.uuid4()}"
+                self.session_id = session_id
+                stats, result = await self.play_round(ai_type, "paper")
+                return stats, result, session_id
+            async def play_scissors(ai_type, session_id):
+                if not session_id:
+                    session_id = f"session_{uuid.uuid4()}"
+                self.session_id = session_id
+                stats, result = await self.play_round(ai_type, "scissors")
+                return stats, result, session_id
+
+            # Add a hidden HTML block with JS to auto-save on tab close
+            gr.HTML("""
+    <script>
+    // Store session ID in localStorage whenever it changes
+    window.setRpsSessionId = function(session_id) {
+        if (session_id) {
+            localStorage.setItem('rps_session_id', session_id);
+        }
+    };
+    // On tab close, send session end to backend
+    window.onbeforeunload = function() {
+        let session_id = localStorage.getItem('rps_session_id');
+        if (session_id) {
+            navigator.sendBeacon("/game/end", JSON.stringify({session_id: session_id}));
+        }
+    };
+    </script>
+    """)
+
+            # Inject JS to click End Game button on tab close
+            gr.HTML("""
+<script>
+window.onbeforeunload = function() {
+    var btn = document.getElementById('end-game-btn');
+    if (btn) {
+        btn.click();
+    }
+};
+</script>
+""")
+
+            # After each move, update localStorage with the session ID
+            def update_session_id_js(session_id):
+                return f"window.setRpsSessionId('{session_id}');"
+
+            rock_btn.click(
+                fn=play_rock,
+                inputs=[ai_dropdown, session_id_state],
+                outputs=[stats_display, result_display, session_id_state],
+                js=update_session_id_js
+            )
+            paper_btn.click(
+                fn=play_paper,
+                inputs=[ai_dropdown, session_id_state],
+                outputs=[stats_display, result_display, session_id_state],
+                js=update_session_id_js
+            )
+            scissors_btn.click(
+                fn=play_scissors,
+                inputs=[ai_dropdown, session_id_state],
+                outputs=[stats_display, result_display, session_id_state],
+                js=update_session_id_js
+            )
+
+            # End Game button logic
+            async def end_game(session_id):
+                if not session_id:
+                    return "No session to end. Play a round first!"
+                result = await self.reset_session(session_id)
+                return f"End Game: {result['status']} - {result.get('message', '')}"
+
+            end_btn.click(
+                fn=end_game,
+                inputs=[session_id_state],
+                outputs=[end_result_display],
+            )
+            end_result_display.visible = True
+
+            # Help button logic
+            async def get_help(session_id):
+                # Show loading message while waiting
+                loading_message = "Generating answer, please wait..."
+                try:
+                    base_url = os.getenv("HF_SPACE_URL", "http://localhost:7860")
+                    async with httpx.AsyncClient() as client:
+                        response = await client.post(f"{base_url}/api/help", json={"session_id": session_id})
+                        if response.status_code == 200:
+                            data = response.json()
+                            suggestion = data.get("suggestion", "No suggestion available.")
+                        else:
+                            suggestion = "Sorry, I'm having trouble getting help right now."
+                except Exception as e:
+                    suggestion = f"Error getting help: {str(e)}"
+                return suggestion
+
+            # When help button is clicked, first show loading, then update with answer
+            def show_loading():
+                return "Generating answer, please wait..."
+
+            help_btn.click(
+                fn=show_loading,
+                inputs=[],
+                outputs=[help_answer_box],
+                queue=False
+            )
+            help_btn.click(
+                fn=get_help,
+                inputs=[session_id_state],
+                outputs=[help_answer_box],
+                queue=True
+            )
+
+        return demo
+
+    def update_ai_description(self, ai_type: str) -> str:
+        return self.ai_descriptions[ai_type]
+
+    async def play_round(self, ai_type: str, move: str):
+        if not self.session_id: 
+            return "Internal Error: Session ID missing in play_round.", "Error"
+        result = await self.game_service.play_round(self.session_id, move, ai_type)
+        stats = result["stats"]
+        stats_text = f"""
+        ### Game Statistics
+        - Total Rounds: {stats['total_rounds']}
+        - Player Wins: {stats['player_wins']} ({stats['player_win_rate']})
+        - AI Wins: {stats['ai_wins']} ({stats['ai_win_rate']})
+        - Draws: {stats['draws']}
+
+        ### Player Move Distribution
+        - Rock: {stats['player_moves']['rock']}
+        - Paper: {stats['player_moves']['paper']}
+        - Scissors: {stats['player_moves']['scissors']}
+
+        ### AI Move Distribution
+        - Rock: {stats['ai_moves']['rock']}
+        - Paper: {stats['ai_moves']['paper']}
+        - Scissors: {stats['ai_moves']['scissors']}
+        """
+        result_text = f"""
+        ### Round Result
+        You played: {result['player_move'].upper()}
+        AI played: {result['ai_move'].upper()}
+        Result: {result['result'].replace('_', ' ').title()}
+        """
+        return stats_text, result_text
+
+    async def clear_session(self, session_id: str):
+        await self.game_service.clear_session(session_id)
+        return {"status": "ok"}
+
+# Create FastAPI app
+app = FastAPI()
+
+# Initialize LLM Service
+llm_service = LLMService()
+
+# Add help endpoint
+@app.post("/api/help")
+async def get_help(request: Request, body: dict = Body(...)):
+    try:
+        session_id = body.get("session_id")
+        stats = None
+        if session_id:
+            # Try to get stats from the game service
+            session_data = game_service.cache.get_session(session_id)
+            if session_data:
+                # Calculate AI move distribution
+                ai_moves = [round['ai_move'] for round in session_data['rounds']]
+                total_moves = len(ai_moves)
+                if total_moves > 0:
+                    move_distribution = {
+                        "rock": int((ai_moves.count("rock") / total_moves) * 100),
+                        "paper": int((ai_moves.count("paper") / total_moves) * 100),
+                        "scissors": int((ai_moves.count("scissors") / total_moves) * 100)
+                    }
+                    stats = {"move_distribution": move_distribution}
+        suggestion = await llm_service.generate_response("help", stats=stats)
+        print("Help suggestion:", suggestion)  # Debug print
+        return {"suggestion": suggestion}
+    except Exception as e:
+        return {"error": str(e), "suggestion": "Sorry, I'm having trouble generating a suggestion right now."}
+
+# Create Gradio interface
+ui = RockPaperScissorsUI(game_service)
+demo = ui.create_interface()
+
+# Mount Gradio app
+app.include_router(game_router, prefix="/api/game")
+app = gr.mount_gradio_app(app, demo, path="/")
+
+# Initialize the app
+@app.on_event("startup")
+async def startup_event():
+    await game_service.initialize()
+
+@app.on_event("shutdown")
+async def shutdown_event():
+    await llm_service.close()
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run("app:app", host="0.0.0.0", port=7860, reload=False)