# ==================== Import Standard Libraries ==================== import os import sys import uuid import uuid import uuid # ==================== Set Paths ==================== current_dir = os.path.dirname(os.path.abspath(__file__)) # ==================== Fix huggingface_hub Compatibility ==================== def _fix_huggingface_hub(): """Fix huggingface_hub compatibility issues""" try: import huggingface_hub if not hasattr(huggingface_hub, 'HfFolder'): class HfFolder: @staticmethod def save_token(token): pass @staticmethod def get_token(): return None @staticmethod def get_token_path(): return None huggingface_hub.HfFolder = HfFolder if hasattr(huggingface_hub, '__all__'): if 'HfFolder' not in huggingface_hub.__all__: huggingface_hub.__all__.append('HfFolder') except Exception: pass _fix_huggingface_hub() # ==================== Import Third-Party Libraries ==================== import gradio as gr # ==================== Import Task Modules ==================== # Each task has an independent module file containing all logic and interface components for that task # Task module naming convention: GUI_{TaskName}_Task.py import GUI_Light_Task as light_task import GUI_Repo_Task as repo_task import GUI_Trade_Task as trade_task import GUI_Energy_Task as energy_task # ==================== Import Unified Progress Management Module ==================== import progress_manager # ------------------- Global Variables (for saving directories and other configurations) ------------------- # Use user_progress under project directory uniformly (consistent for local and remote) save_dir = os.path.join(current_dir, "user_progress") # Ensure directory exists os.makedirs(save_dir, exist_ok=True) # Detect running environment IS_HUGGINGFACE = progress_manager.is_running_on_huggingface() RUN_MODE = progress_manager.get_run_mode() # ==================== Unified Interface ==================== def create_interface(): """Create unified multi-task interface""" with gr.Blocks(title="Multi-Task Environment Interface") as demo: # Display different titles based on running environment if IS_HUGGINGFACE: gr.Markdown(""" # 🎮 Multi-Task Environment Interface (Hugging Face Version) Supports multiple task environments. Use task switching buttons to select different tasks. """) else: gr.Markdown(""" # 🎮 Multi-Task Environment Interface (Local Version) Supports multiple task environments. Use task switching buttons to select different tasks. """) # ==================== Create State Objects (Independent for Each User Session) ==================== # These state objects ensure data isolation during multi-user concurrency light_state = gr.State(light_task.create_light_state()) repo_state = gr.State(repo_task.create_repo_state()) trade_state = gr.State(trade_task.create_trade_state()) energy_state = gr.State(energy_task.create_energy_state()) # Current task type state current_task_type = gr.State("light") # Auto-generate default user ID for each session (no user input required) default_user_id = gr.State(f"user_{uuid.uuid4().hex[:8]}") # ==================== Task Switching Buttons ==================== gr.Markdown("---") gr.Markdown("### 🎯 Select Task") with gr.Row(): task_light_btn = gr.Button("💡 Light Task", variant="primary") task_repo_btn = gr.Button("📦 Repo Task", variant="secondary") task_trade_btn = gr.Button("💹 Trade Task", variant="secondary") task_energy_btn = gr.Button("⚡ Energy Task", variant="secondary") # ==================== Task Interface Area ==================== with gr.Row(): with gr.Column(scale=1): # Light Task Environment Control light_env_control_markdown = gr.Markdown("### 🎮 Light Environment Control", visible=True) light_env_idx_input = gr.Number( label="Environment Index", value=1, minimum=1, maximum=30, precision=0, info="Select environment to load (1-30)", visible=True ) light_init_btn = gr.Button("Load Environment", variant="primary", visible=True) light_reset_btn = gr.Button("Reset Environment", visible=True) light_env_info = gr.Textbox(label="Environment Info", interactive=False, lines=5, visible=True) # Energy Task Environment Control energy_env_control_markdown = gr.Markdown("### 🎮 Energy Environment Control", visible=False) energy_env_idx_input = gr.Number( label="Environment Index", value=1, minimum=1, maximum=30, precision=0, info="Select environment to load (1-30)", visible=False ) energy_init_btn = gr.Button("Load Environment", variant="primary", visible=False) energy_reset_btn = gr.Button("Reset Environment", visible=False) energy_env_info = gr.Textbox(label="Environment Info", interactive=False, lines=5, visible=False) # Repo Task Environment Control repo_env_control_markdown = gr.Markdown("### 🎮 Repo Environment Control", visible=False) repo_env_idx_input = gr.Number( label="Environment Index", value=1, minimum=1, maximum=30, precision=0, info="Select environment to load (1-30)", visible=False ) repo_init_btn = gr.Button("Load Environment", variant="primary", visible=False) repo_reset_btn = gr.Button("Reset Environment", visible=False) repo_env_info = gr.Textbox(label="Environment Info", interactive=False, lines=5, visible=False) # Trade Task Environment Control trade_env_control_markdown = gr.Markdown("### 🎮 Trade Environment Control", visible=False) trade_env_idx_input = gr.Number( label="Environment Index", value=1, minimum=1, maximum=30, precision=0, info="Select environment to load (1-30)", visible=False ) trade_init_btn = gr.Button("Load Environment", variant="primary", visible=False) trade_reset_btn = gr.Button("Reset Environment", visible=False) trade_env_info = gr.Textbox(label="Environment Info", interactive=False, lines=5, visible=False) # Example Display example_display = gr.Markdown( label="📖 Usage Example", value=light_task.LIGHT_EXAMPLE_TEXT, visible=True ) # ==================== Create Task Interface Components ==================== # Each task module provides a create_{task}_interface() function # Returns all Gradio components required for that task # Light Task Interface Components (light_interface, _, _, _, _, light_state_display, light_steps_info_text, light_action_input, light_step_btn, light_feedback_display, light_history_display) = \ light_task.create_light_interface(current_dir, save_dir, None) # Repo Task Interface Components (environment control components created in main interface) (repo_interface, _, _, _, _, repo_state_display, repo_steps_info_text, repo_action_input, repo_step_btn, repo_feedback_display, repo_history_display) = \ repo_task.create_repo_interface(current_dir, save_dir, None) # Trade Task Interface Components (environment control components created in main interface) (trade_interface, _, _, _, _, trade_state_display, trade_steps_info_text, trade_stock_inputs, trade_step_btn, trade_feedback_display, trade_history_display) = \ trade_task.create_trade_interface(current_dir, save_dir, None) # Energy Task Interface Components (environment control components created in main interface) (energy_interface, _, _, _, _, energy_state_display, energy_steps_info_text, energy_thermal_input, energy_wind_input, energy_solar_input, energy_battery_input, energy_cost_display, energy_step_btn, energy_feedback_display, energy_history_display) = \ energy_task.create_energy_interface(current_dir, save_dir, None) # ==================== Task Switching Logic ==================== # Each task module provides: # 1. load_{task}_test_data() - Load test data # 2. {TASK}_EXAMPLE_TEXT - Example text constant def switch_to_light(state): """Switch to Light task""" # Load test data state, _ = light_task.load_light_test_data(state, current_dir) return ( state, "light", # current_task_type gr.update(visible=True), # light_interface gr.update(visible=False), # repo_interface gr.update(visible=False), # trade_interface gr.update(visible=False), # energy_interface gr.update(variant="primary"), # task_light_btn gr.update(variant="secondary"), # task_repo_btn gr.update(variant="secondary"), # task_trade_btn gr.update(variant="secondary"), # task_energy_btn light_task.LIGHT_EXAMPLE_TEXT, # example_display # Environment control component visibility gr.update(visible=True), # light_env_control_markdown gr.update(visible=True), # light_env_idx_input gr.update(visible=True), # light_init_btn gr.update(visible=True), # light_reset_btn gr.update(visible=True), # light_env_info gr.update(visible=False), # energy_env_control_markdown gr.update(visible=False), # energy_env_idx_input gr.update(visible=False), # energy_init_btn gr.update(visible=False), # energy_reset_btn gr.update(visible=False), # energy_env_info gr.update(visible=False), # repo_env_control_markdown gr.update(visible=False), # repo_env_idx_input gr.update(visible=False), # repo_init_btn gr.update(visible=False), # repo_reset_btn gr.update(visible=False), # repo_env_info gr.update(visible=False), # trade_env_control_markdown gr.update(visible=False), # trade_env_idx_input gr.update(visible=False), # trade_init_btn gr.update(visible=False), # trade_reset_btn gr.update(visible=False) # trade_env_info ) def switch_to_repo(state): """Switch to Repo task""" state, _ = repo_task.load_repo_test_data(state, current_dir) return ( state, "repo", gr.update(visible=False), # light_interface gr.update(visible=True), # repo_interface gr.update(visible=False), # trade_interface gr.update(visible=False), # energy_interface gr.update(variant="secondary"), # task_light_btn gr.update(variant="primary"), # task_repo_btn gr.update(variant="secondary"), # task_trade_btn gr.update(variant="secondary"), # task_energy_btn repo_task.REPO_EXAMPLE_TEXT, # example_display # Environment control component visibility gr.update(visible=False), # light_env_control_markdown gr.update(visible=False), # light_env_idx_input gr.update(visible=False), # light_init_btn gr.update(visible=False), # light_reset_btn gr.update(visible=False), # light_env_info gr.update(visible=False), # energy_env_control_markdown gr.update(visible=False), # energy_env_idx_input gr.update(visible=False), # energy_init_btn gr.update(visible=False), # energy_reset_btn gr.update(visible=False), # energy_env_info gr.update(visible=True), # repo_env_control_markdown gr.update(visible=True), # repo_env_idx_input gr.update(visible=True), # repo_init_btn gr.update(visible=True), # repo_reset_btn gr.update(visible=True), # repo_env_info gr.update(visible=False), # trade_env_control_markdown gr.update(visible=False), # trade_env_idx_input gr.update(visible=False), # trade_init_btn gr.update(visible=False), # trade_reset_btn gr.update(visible=False) # trade_env_info ) def switch_to_trade(state): """Switch to Trade task""" state, _ = trade_task.load_trade_test_data(state, current_dir) return ( state, "trade", gr.update(visible=False), # light_interface gr.update(visible=False), # repo_interface gr.update(visible=True), # trade_interface gr.update(visible=False), # energy_interface gr.update(variant="secondary"), # task_light_btn gr.update(variant="secondary"), # task_repo_btn gr.update(variant="primary"), # task_trade_btn gr.update(variant="secondary"), # task_energy_btn trade_task.TRADE_EXAMPLE_TEXT, # example_display # Environment control component visibility gr.update(visible=False), # light_env_control_markdown gr.update(visible=False), # light_env_idx_input gr.update(visible=False), # light_init_btn gr.update(visible=False), # light_reset_btn gr.update(visible=False), # light_env_info gr.update(visible=False), # energy_env_control_markdown gr.update(visible=False), # energy_env_idx_input gr.update(visible=False), # energy_init_btn gr.update(visible=False), # energy_reset_btn gr.update(visible=False), # energy_env_info gr.update(visible=False), # repo_env_control_markdown gr.update(visible=False), # repo_env_idx_input gr.update(visible=False), # repo_init_btn gr.update(visible=False), # repo_reset_btn gr.update(visible=False), # repo_env_info gr.update(visible=True), # trade_env_control_markdown gr.update(visible=True), # trade_env_idx_input gr.update(visible=True), # trade_init_btn gr.update(visible=True), # trade_reset_btn gr.update(visible=True) # trade_env_info ) def switch_to_energy(state): """Switch to Energy task""" state, _ = energy_task.load_energy_test_data(state, current_dir) return ( state, "energy", gr.update(visible=False), # light_interface gr.update(visible=False), # repo_interface gr.update(visible=False), # trade_interface gr.update(visible=True), # energy_interface gr.update(variant="secondary"), # task_light_btn gr.update(variant="secondary"), # task_repo_btn gr.update(variant="secondary"), # task_trade_btn gr.update(variant="primary"), # task_energy_btn energy_task.ENERGY_EXAMPLE_TEXT, # example_display # Environment control component visibility gr.update(visible=False), # light_env_control_markdown gr.update(visible=False), # light_env_idx_input gr.update(visible=False), # light_init_btn gr.update(visible=False), # light_reset_btn gr.update(visible=False), # light_env_info gr.update(visible=True), # energy_env_control_markdown gr.update(visible=True), # energy_env_idx_input gr.update(visible=True), # energy_init_btn gr.update(visible=True), # energy_reset_btn gr.update(visible=True), # energy_env_info gr.update(visible=False), # repo_env_control_markdown gr.update(visible=False), # repo_env_idx_input gr.update(visible=False), # repo_init_btn gr.update(visible=False), # repo_reset_btn gr.update(visible=False), # repo_env_info gr.update(visible=False), # trade_env_control_markdown gr.update(visible=False), # trade_env_idx_input gr.update(visible=False), # trade_init_btn gr.update(visible=False), # trade_reset_btn gr.update(visible=False) # trade_env_info ) task_light_btn.click( fn=switch_to_light, inputs=[light_state], outputs=[light_state, current_task_type, light_interface, repo_interface, trade_interface, energy_interface, task_light_btn, task_repo_btn, task_trade_btn, task_energy_btn, example_display, light_env_control_markdown, light_env_idx_input, light_init_btn, light_reset_btn, light_env_info, energy_env_control_markdown, energy_env_idx_input, energy_init_btn, energy_reset_btn, energy_env_info, repo_env_control_markdown, repo_env_idx_input, repo_init_btn, repo_reset_btn, repo_env_info, trade_env_control_markdown, trade_env_idx_input, trade_init_btn, trade_reset_btn, trade_env_info] ) task_repo_btn.click( fn=switch_to_repo, inputs=[repo_state], outputs=[repo_state, current_task_type, light_interface, repo_interface, trade_interface, energy_interface, task_light_btn, task_repo_btn, task_trade_btn, task_energy_btn, example_display, light_env_control_markdown, light_env_idx_input, light_init_btn, light_reset_btn, light_env_info, energy_env_control_markdown, energy_env_idx_input, energy_init_btn, energy_reset_btn, energy_env_info, repo_env_control_markdown, repo_env_idx_input, repo_init_btn, repo_reset_btn, repo_env_info, trade_env_control_markdown, trade_env_idx_input, trade_init_btn, trade_reset_btn, trade_env_info] ) task_trade_btn.click( fn=switch_to_trade, inputs=[trade_state], outputs=[trade_state, current_task_type, light_interface, repo_interface, trade_interface, energy_interface, task_light_btn, task_repo_btn, task_trade_btn, task_energy_btn, example_display, light_env_control_markdown, light_env_idx_input, light_init_btn, light_reset_btn, light_env_info, energy_env_control_markdown, energy_env_idx_input, energy_init_btn, energy_reset_btn, energy_env_info, repo_env_control_markdown, repo_env_idx_input, repo_init_btn, repo_reset_btn, repo_env_info, trade_env_control_markdown, trade_env_idx_input, trade_init_btn, trade_reset_btn, trade_env_info] ) task_energy_btn.click( fn=switch_to_energy, inputs=[energy_state], outputs=[energy_state, current_task_type, light_interface, repo_interface, trade_interface, energy_interface, task_light_btn, task_repo_btn, task_trade_btn, task_energy_btn, example_display, light_env_control_markdown, light_env_idx_input, light_init_btn, light_reset_btn, light_env_info, energy_env_control_markdown, energy_env_idx_input, energy_init_btn, energy_reset_btn, energy_env_info, repo_env_control_markdown, repo_env_idx_input, repo_init_btn, repo_reset_btn, repo_env_info, trade_env_control_markdown, trade_env_idx_input, trade_init_btn, trade_reset_btn, trade_env_info] ) # ==================== Light Task Event Bindings ==================== def light_load_wrapper(state, env_idx, user_id): """Wrapper function for loading Light task environment""" state, info, state_display, logic, history, progress, steps = light_task.light_load_environment(state, env_idx, user_id, save_dir) return state, info, state_display, history, steps light_init_btn.click( fn=light_load_wrapper, inputs=[light_state, light_env_idx_input, default_user_id], outputs=[light_state, light_env_info, light_state_display, light_history_display, light_steps_info_text] ) def light_reset_wrapper(state, user_id): """Wrapper function for resetting Light task environment""" state, info, state_display, history, progress, steps = light_task.light_reset_environment(state, user_id, save_dir) return state, info, state_display, history, steps light_reset_btn.click( fn=light_reset_wrapper, inputs=[light_state, default_user_id], outputs=[light_state, light_env_info, light_state_display, light_history_display, light_steps_info_text] ) def light_step_wrapper(state, action_str, user_id): state, feedback, state_display, history, done, steps_info = light_task.light_step_environment(state, action_str, user_id, save_dir) test_data = light_task.get_light_test_data(state) current_env_idx = light_task.get_light_current_env_idx(state) history_records = light_task.get_light_history_records(state) if done: env_info_text = f"🎉 Task completed! All lights are on!\nEnvironment Index: {current_env_idx + 1}/{len(test_data) if test_data else 0}" else: env_info_text = f"Environment Index: {current_env_idx + 1}/{len(test_data) if test_data else 0}\nSteps: {len(history_records)}" return state, feedback, state_display, history, env_info_text, steps_info, "" light_step_btn.click( fn=light_step_wrapper, inputs=[light_state, light_action_input, default_user_id], outputs=[light_state, light_feedback_display, light_state_display, light_history_display, light_env_info, light_steps_info_text, light_action_input] ) light_action_input.submit( fn=light_step_wrapper, inputs=[light_state, light_action_input, default_user_id], outputs=[light_state, light_feedback_display, light_state_display, light_history_display, light_env_info, light_steps_info_text, light_action_input] ) # ==================== Repo Task Event Bindings ==================== def repo_load_wrapper(state, env_idx, user_id): """Wrapper function for loading Repo task environment""" state, info, state_display, logic, history, progress, steps = repo_task.repo_load_environment(state, env_idx, user_id, save_dir) return state, info, state_display, history, steps repo_init_btn.click( fn=repo_load_wrapper, inputs=[repo_state, repo_env_idx_input, default_user_id], outputs=[repo_state, repo_env_info, repo_state_display, repo_history_display, repo_steps_info_text] ) def repo_reset_wrapper(state, user_id): """Wrapper function for resetting Repo task environment""" state, info, state_display, history, progress, steps = repo_task.repo_reset_environment(state, user_id, save_dir) return state, info, state_display, history, steps repo_reset_btn.click( fn=repo_reset_wrapper, inputs=[repo_state, default_user_id], outputs=[repo_state, repo_env_info, repo_state_display, repo_history_display, repo_steps_info_text] ) def repo_step_wrapper(state, action_str, user_id): state, feedback, state_display, history, done, steps_info = repo_task.repo_step_environment(state, action_str, user_id, save_dir) test_data = repo_task.get_repo_test_data(state) current_env_idx = repo_task.get_repo_current_env_idx(state) history_records = repo_task.get_repo_history_records(state) if done: env_info_text = f"🎉 Task completed! Project runs successfully!\nEnvironment Index: {current_env_idx + 1}/{len(test_data) if test_data else 0}" else: env_info_text = f"Environment Index: {current_env_idx + 1}/{len(test_data) if test_data else 0}\nSteps: {len(history_records)}" return state, feedback, state_display, history, env_info_text, steps_info, "" repo_step_btn.click( fn=repo_step_wrapper, inputs=[repo_state, repo_action_input, default_user_id], outputs=[repo_state, repo_feedback_display, repo_state_display, repo_history_display, repo_env_info, repo_steps_info_text, repo_action_input] ) repo_action_input.submit( fn=repo_step_wrapper, inputs=[repo_state, repo_action_input, default_user_id], outputs=[repo_state, repo_feedback_display, repo_state_display, repo_history_display, repo_env_info, repo_steps_info_text, repo_action_input] ) # ==================== Trade Task Event Bindings ==================== def trade_load_wrapper(state, env_idx, user_id): """Wrapper function for loading Trade task environment""" state, info, state_display, logic, history, progress, steps = trade_task.trade_load_environment(state, env_idx, user_id, save_dir) # Update input boxes based on the number of stocks in the environment env = trade_task.get_trade_env(state) if env: stock_updates = trade_task.get_trade_stock_input_updates(env) else: stock_updates = [gr.update(visible=False) for _ in range(10)] return (state, info, state_display, history, steps) + tuple(stock_updates) # Collect all 10 stock input boxes (for output updates) all_trade_stock_inputs = [trade_stock_inputs.get(f"S{i}", None) for i in range(10)] # Filter out None values all_trade_stock_inputs = [inp for inp in all_trade_stock_inputs if inp is not None] trade_init_btn.click( fn=trade_load_wrapper, inputs=[trade_state, trade_env_idx_input, default_user_id], outputs=[trade_state, trade_env_info, trade_state_display, trade_history_display, trade_steps_info_text] + all_trade_stock_inputs ) def trade_reset_wrapper(state, user_id): """Wrapper function for resetting Trade task environment""" state, info, state_display, history, progress, steps = trade_task.trade_reset_environment(state, user_id, save_dir) # Update input boxes based on the number of stocks in the environment env = trade_task.get_trade_env(state) if env: stock_updates = trade_task.get_trade_stock_input_updates(env) else: stock_updates = [gr.update(visible=False) for _ in range(10)] return (state, info, state_display, history, steps) + tuple(stock_updates) trade_reset_btn.click( fn=trade_reset_wrapper, inputs=[trade_state, default_user_id], outputs=[trade_state, trade_env_info, trade_state_display, trade_history_display, trade_steps_info_text] + all_trade_stock_inputs ) def trade_step_wrapper(state, s0, s1, s2, s3, s4, s5, s6, s7, s8, s9, user_id): """Wrapper function for executing Trade task actions, collecting data from input boxes (positive for buy, negative for sell)""" # Dynamically build stock input dictionary using actual stock names from environment env = trade_task.get_trade_env(state) stock_inputs = {} if env: # Use actual stock names from environment stock_values = [s0, s1, s2, s3, s4, s5, s6, s7, s8, s9] for i, stock_name in enumerate(env.stocks): if i < len(stock_values): stock_inputs[stock_name] = stock_values[i] or 0 else: # If no environment, use default S0-S3 (backward compatibility) stock_inputs = { "S0": s0 or 0, "S1": s1 or 0, "S2": s2 or 0, "S3": s3 or 0 } state, feedback, state_display, history, done, steps_info = trade_task.trade_step_environment_from_inputs( state, stock_inputs, user_id, save_dir) test_data = trade_task.get_trade_test_data(state) current_env_idx = trade_task.get_trade_current_env_idx(state) history_records = trade_task.get_trade_history_records(state) if done: env_info_text = f"🎉 Task completed! All trading days ended!\nEnvironment Index: {current_env_idx + 1}/{len(test_data) if test_data else 0}" else: env_info_text = f"Environment Index: {current_env_idx + 1}/{len(test_data) if test_data else 0}\nSteps: {len(history_records)}" # Clear all input boxes return (state, feedback, state_display, history, env_info_text, steps_info, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0) # Collect all 10 input boxes (in order S0-S9) trade_inputs_list = [trade_stock_inputs.get(f"S{i}", None) for i in range(10)] # Filter out None values, ensure all input boxes exist trade_inputs_list = [inp for inp in trade_inputs_list if inp is not None] trade_step_btn.click( fn=trade_step_wrapper, inputs=[trade_state] + trade_inputs_list + [default_user_id], outputs=[trade_state, trade_feedback_display, trade_state_display, trade_history_display, trade_env_info, trade_steps_info_text] + trade_inputs_list ) # ==================== Energy Task Event Bindings ==================== def energy_load_wrapper(state, env_idx, user_id): """Wrapper function for loading Energy task environment""" state, info, state_display, logic, history, progress, steps = energy_task.energy_load_environment(state, env_idx, user_id, save_dir) cost_text = energy_task.calculate_estimated_cost(state, 0.0, 0.0, 0.0, 0.0) # Check if environment is completed env = energy_task.get_energy_env(state) is_done = env is not None and env.done return state, info, state_display, history, steps, cost_text, gr.update(interactive=not is_done) energy_init_btn.click( fn=energy_load_wrapper, inputs=[energy_state, energy_env_idx_input, default_user_id], outputs=[energy_state, energy_env_info, energy_state_display, energy_history_display, energy_steps_info_text, energy_cost_display, energy_step_btn] ) def energy_reset_wrapper(state, user_id): """Wrapper function for resetting Energy task environment""" state, info, state_display, history, progress, steps = energy_task.energy_reset_environment(state, user_id, save_dir) cost_text = energy_task.calculate_estimated_cost(state, 0.0, 0.0, 0.0, 0.0) return state, info, state_display, history, steps, cost_text, gr.update(interactive=True) energy_reset_btn.click( fn=energy_reset_wrapper, inputs=[energy_state, default_user_id], outputs=[energy_state, energy_env_info, energy_state_display, energy_history_display, energy_steps_info_text, energy_cost_display, energy_step_btn] ) def update_energy_cost(state, thermal, wind, solar, battery): """Update estimated cost in real-time""" return energy_task.calculate_estimated_cost(state, thermal, wind, solar, battery) def update_energy_state_display(state, thermal, wind, solar): """Update state display in real-time, including carbon emission ratio based on input values""" env = energy_task.get_energy_env(state) if env is None: return "Please initialize environment first" obs = env._get_obs() return energy_task.format_energy_state(state, obs, thermal_input=thermal, wind_input=wind, solar_input=solar) # Add change events to all input boxes to update estimated cost and state display in real-time for input_component in [energy_thermal_input, energy_wind_input, energy_solar_input, energy_battery_input]: input_component.change( fn=update_energy_cost, inputs=[energy_state, energy_thermal_input, energy_wind_input, energy_solar_input, energy_battery_input], outputs=[energy_cost_display] ) # Also update state display (only using thermal, wind, solar, excluding battery) input_component.change( fn=update_energy_state_display, inputs=[energy_state, energy_thermal_input, energy_wind_input, energy_solar_input], outputs=[energy_state_display] ) def energy_step_wrapper(state, thermal, wind, solar, battery, user_id): """Wrapper function for executing Energy task actions, collecting data from input boxes""" state, feedback, state_display, history, done, steps_info = energy_task.energy_step_environment_from_inputs( state, thermal, wind, solar, battery, user_id, save_dir) test_data = energy_task.get_energy_test_data(state) current_env_idx = energy_task.get_energy_current_env_idx(state) history_records = energy_task.get_energy_history_records(state) current_steps = len(history_records) if done: if current_steps < energy_task.ENERGY_MAX_STEPS: env_info_text = f"❌ Task failed (completed)\nEnvironment Index: {current_env_idx + 1}/{len(test_data) if test_data else 0}\nSteps: {current_steps} / {energy_task.ENERGY_MAX_STEPS}" else: env_info_text = f"🎉 Task completed!\nEnvironment Index: {current_env_idx + 1}/{len(test_data) if test_data else 0}\nSteps: {current_steps} / {energy_task.ENERGY_MAX_STEPS}" else: env_info_text = f"Environment Index: {current_env_idx + 1}/{len(test_data) if test_data else 0}\nSteps: {len(history_records)}" # Clear input boxes and update estimated cost cost_text = energy_task.calculate_estimated_cost(state, 0.0, 0.0, 0.0, 0.0) return (state, feedback, state_display, history, env_info_text, steps_info, 0.0, 0.0, 0.0, 0.0, cost_text, gr.update(interactive=not done)) energy_step_btn.click( fn=energy_step_wrapper, inputs=[energy_state, energy_thermal_input, energy_wind_input, energy_solar_input, energy_battery_input, default_user_id], outputs=[energy_state, energy_feedback_display, energy_state_display, energy_history_display, energy_env_info, energy_steps_info_text, energy_thermal_input, energy_wind_input, energy_solar_input, energy_battery_input, energy_cost_display, energy_step_btn] ) # ==================== Initialization ==================== # Automatically load test data for default task (Light) when page loads def init_light_data(state): state, _ = light_task.load_light_test_data(state, current_dir) return state demo.load( fn=init_light_data, inputs=[light_state], outputs=[light_state] ) return demo # ------------------- Main Function ------------------- if __name__ == "__main__": demo = create_interface() if os.getenv("SPACE_ID") is None: demo.launch( server_name="127.0.0.1", server_port=7860, share=False, theme=gr.themes.Soft() ) else: demo.launch(theme=gr.themes.Soft())