Spaces:

Jofthomas
/

Pokemon_showdown

Running

App Files Files Community

Update app.py

by Batnini - opened Sep 15

base: refs/heads/main

←

from: refs/pr/1

Discussion Files changed

+228

-61

Files changed (1) hide show

app.py +228 -61

app.py CHANGED Viewed

@@ -1,63 +1,230 @@
 import os
-from flask import Flask, render_template_string
-app = Flask(__name__)
-# HTML template using Jinja2 syntax to insert the URL safely
-# Added basic styling to make the iframe fill the page
-HTML_TEMPLATE = """
-<!DOCTYPE html>
-<html>
-<head>
-    <title>Iframe Viewer</title>
-    <style>
-        html, body {
-            margin: 0;
-            padding: 0;
-            height: 100%;
-            width: 100%;
-            overflow: hidden; /* Prevent scrollbars on the main page */
-        }
-        iframe {
-            display: block; /* Removes bottom space under iframe */
-            width: 100%;
-            height: 100%;
-            border: none; /* Remove default border */
-        }
-        .error {
-            padding: 20px;
-            color: red;
-            font-family: sans-serif;
-        }
-    </style>
-</head>
-<body>
-    {% if iframe_url %}
-        <iframe src="{{ iframe_url }}">
-            Your browser does not support iframes.
         </iframe>
-    {% else %}
-        <div class="error">
-            Error: The 'IFRAME_URL' secret is not set in the Space settings.
-            Please add it under Settings -> Repository secrets.
-        </div>
-    {% endif %}
-</body>
-</html>
-"""
-@app.route('/')
-def display_iframe():
-    # Read the URL from the environment variable 'IFRAME_URL'
-    # os.getenv returns None if the variable isn't set
-    url_from_env = os.getenv('IFRAME_URL')
-    print(url_from_env)
-    # Pass the URL to the template rendering function
-    # Flask's render_template_string automatically handles basic HTML escaping
-    # for security if used directly in text, but here it's okay for the src attribute.
-    return render_template_string(HTML_TEMPLATE, iframe_url=url_from_env)
-if __name__ == '__main__':
-    # Hugging Face Spaces expect the app to run on port 7860
-    # Binding to '0.0.0.0' makes it accessible from outside the container
-    app.run(host='0.0.0.0', port=7860)

+# app.py
+import gradio as gr
+import asyncio
 import os
+import random
+import traceback
+import logging
+import threading
+from poke_env.player import Player, RandomPlayer
+from poke_env import AccountConfiguration, ServerConfiguration
+from agents import OpenAIAgent
+# --- Configuration ---
+POKE_SERVER_URL = "wss://jofthomas.com/showdown/websocket"
+POKE_AUTH_URL = "https://jofthomas.com/showdown/action.php"
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(threadName)s - %(levelname)s - %(message)s')
+# --- Constants ---
+RANDOM_PLAYER_BASE_NAME = "RandAgent"
+OPENAI_AGENT_BASE_NAME = "OpenAIAgent"
+DEFAULT_BATTLE_FORMAT = "gen9randombattle"
+custom_config = ServerConfiguration(POKE_SERVER_URL, POKE_AUTH_URL)
+# --- Agent Creation (Async - Required by poke-env) ---
+async def create_agent_async(agent_type: str, battle_format: str = DEFAULT_BATTLE_FORMAT) -> Player | str:
+    """
+    Creates and initializes a SINGLE agent instance with a unique username.
+    This function MUST be async because Player initialization involves async network setup.
+    Returns the Player object on success, or an error string on failure.
+    """
+    logging.info(f"Attempting to create agent of type: {agent_type}")
+    player: Player | None = None
+    error_message: str | None = None
+    username: str = "unknown_agent"
+    agent_suffix = random.randint(10000, 999999)
+    try:
+        if agent_type == "Random Player":
+            username = f"{RANDOM_PLAYER_BASE_NAME}{agent_suffix}"
+            account_config = AccountConfiguration(username, None)
+            logging.info(f"Initializing RandomPlayer with username: {username}")
+            player = RandomPlayer(
+                account_configuration=account_config,
+                server_configuration=custom_config,
+                battle_format=battle_format,
+                start_listening=True,
+            )
+        elif agent_type == "OpenAI Agent":
+            if not os.getenv("OPENAI_API_KEY"):
+                 error_message = "Error: Cannot create OpenAI Agent. OPENAI_API_KEY environment variable is missing."
+                 logging.error(error_message)
+                 return error_message
+            username = f"{OPENAI_AGENT_BASE_NAME}{agent_suffix}"
+            account_config = AccountConfiguration(username, None)
+            logging.info(f"Initializing OpenAIAgent with username: {username}")
+            player = OpenAIAgent(
+                account_configuration=account_config,
+                server_configuration=custom_config,
+                battle_format=battle_format,
+                start_listening=True,
+            )
+        else:
+            error_message = f"Error: Invalid agent type '{agent_type}' requested."
+            logging.error(error_message)
+            return error_message
+        logging.info(f"Agent object ({username}) created successfully.")
+        return player
+    except Exception as e:
+        error_message = f"Error creating agent {username}: {e}"
+        logging.error(error_message)
+        logging.error(traceback.format_exc())
+        return error_message
+# --- Battle Invitation (Async - Required by poke-env) ---
+async def send_battle_invite_async(player: Player, opponent_username: str, battle_format: str) -> str:
+    """
+    Sends a challenge using the provided player object.
+    This function MUST be async as sending challenges involves network I/O.
+    Returns a status string (success or error message).
+    """
+    if not isinstance(player, Player):
+         err_msg = f"Internal Error: Invalid object passed instead of Player: {type(player)}"
+         logging.error(err_msg)
+         # In background thread, we might just log this and exit thread
+         raise TypeError(err_msg) # Raise exception to be caught by the thread runner
+    player_username = getattr(player, 'username', 'unknown_agent')
+    try:
+        logging.info(f"Attempting to send challenge from {player_username} to {opponent_username} in format {battle_format}")
+        await player.send_challenges(opponent_username, n_challenges=1)
+        success_msg = f"Battle invitation ({battle_format}) sent to '{opponent_username}' from bot '{player_username}'."
+        logging.info(success_msg)
+        return success_msg # Indicate success
+    except Exception as e:
+        error_msg = f"Error sending challenge from {player_username} to {opponent_username}: {e}"
+        logging.error(error_msg)
+        logging.error(traceback.format_exc())
+        # Re-raise or return error indication for the thread runner
+        raise e # Raise exception to be caught by the thread runner
+# --- Background Task Runner (Runs in a separate thread) ---
+def run_invite_in_background(agent_choice: str, target_username: str, battle_format: str):
+    """
+    This function runs in a separate thread for each invite request.
+    It sets up and runs the asyncio operations needed for one invite.
+    """
+    thread_name = threading.current_thread().name
+    logging.info(f"Background thread '{thread_name}' started for {agent_choice} vs {target_username}.")
+    async def _run_async_challenge_steps():
+        """The async steps to be run via asyncio.run() in this thread."""
+        agent_or_error = await create_agent_async(agent_choice, battle_format)
+        if isinstance(agent_or_error, str):
+            # Agent creation failed, log the error message from create_agent_async
+            logging.error(f"[{thread_name}] Agent creation failed: {agent_or_error}")
+            # No further action needed in this thread
+            return
+        player_instance: Player = agent_or_error
+        player_username = getattr(player_instance, 'username', 'agent')
+        logging.info(f"[{thread_name}] Agent {player_username} created, proceeding to challenge {target_username}.")
+        try:
+            result = await send_battle_invite_async(player_instance, target_username, battle_format)
+            # Log the success message from send_battle_invite_async
+            logging.info(f"[{thread_name}] Challenge result: {result}")
+        except Exception as invite_error:
+            # Log errors from send_battle_invite_async
+            # Error message/traceback already logged inside send_battle_invite_async
+            logging.error(f"[{thread_name}] Failed to send challenge from {player_username} to {target_username}. Error: {invite_error}")
+        finally:
+            pass
+    try:
+        asyncio.run(_run_async_challenge_steps())
+        logging.info(f"Background thread '{thread_name}' finished successfully for {target_username}.")
+    except RuntimeError as e:
+         logging.error(f"[{thread_name}] asyncio RuntimeError: {e}")
+         logging.error(traceback.format_exc())
+    except Exception as e:
+        logging.error(f"[{thread_name}] Unexpected error in background task: {e}")
+        logging.error(traceback.format_exc())
+# --- Gradio Interface Logic (Starts the background thread) ---
+def start_invite_thread(agent_choice: str, username: str) -> str:
+    """
+    Handles the Gradio button click (Synchronous, but FAST).
+    Performs basic validation and starts a background thread to handle
+    the actual agent creation and invitation process.
+    Returns an immediate status message to Gradio.
+    """
+    username_clean = username.strip()
+    if not username_clean:
+        return "⚠️ Please enter your Showdown username."
+    if not agent_choice:
+        return "⚠️ Please select an agent type."
+    logging.info(f"Received request: Agent={agent_choice}, Opponent={username_clean}. Starting background thread.")
+    # Create and start the background thread
+    thread = threading.Thread(
+        target=run_invite_in_background,
+        args=(agent_choice, username_clean, DEFAULT_BATTLE_FORMAT),
+        daemon=True # Set as daemon so threads don't block app exit
+    )
+    thread.start()
+    # Return immediately to Gradio UI
+    return f"✅ Invite process for '{username_clean}' started in background. Check Pokémon Showdown and logs for status."
+# --- Gradio UI Definition ---
+# [ main_app function remains the same, but the button click now calls start_invite_thread ]
+def main_app():
+    """Creates and returns the Gradio application interface."""
+    agent_options = ["Random Player"]
+    agent_options.append("OpenAI Agent")
+    # Use a more descriptive title if possible
+    with gr.Blocks(title="Pokemon Showdown Multi-Challenger") as demo:
+        gr.Markdown("# Pokémon Battle Agent Challenger")
+        gr.Markdown(
+            "1. Choose a name in the Iframe, if you have an account, you can also connect.\n"
+            "2. Select an agent type.\n"
+            "3. Enter **your** Showdown username (the one you are logged in with below).\n"
+            "4. Click 'Send Battle Invitation'. You can click multiple times for different users.\n\n"
+            "A temporary bot will be created *in the background* to send the challenge in `gen9randombattle` format."
+        )
+        with gr.Row():
+            agent_dropdown = gr.Dropdown(
+                label="Select Agent", choices=agent_options, value=agent_options[0], scale=1
+            )
+            name_input = gr.Textbox(
+                label="Your Pokémon Showdown Username", placeholder="Enter username used in Showdown below", scale=2
+            )
+            #variant="primary"
+            battle_button = gr.Button("Send Battle Invitation",  scale=1)
+        gr.HTML("""
+        <iframe
+            src="https://jofthomas.com/play.pokemonshowdown.com/testclient.html"
+            width="100%" height="800" style="border: none;" referrerpolicy="no-referrer">
         </iframe>
+        """)
+        # *** IMPORTANT: Update the click handler ***
+        battle_button.click(
+            fn=start_invite_thread, # Calls the function that starts the thread
+            inputs=[agent_dropdown, name_input],
+        )
+    return demo
+# --- Application Entry Point ---
+# [ if __name__ == "__main__": block remains the same ]
+if __name__ == "__main__":
+    app = main_app()
+    app.launch()