Upload 3 files

GUI-Light.py

@@ -0,0 +1,1040 @@
+# ==================== 导入标准库 ====================
+import json
+import os
+import re
+import sys
+from typing import List, Tuple, Optional
+
+# ==================== 设置路径（必须在导入 TextEnv_v2 之前）====================
+current_dir = os.path.dirname(os.path.abspath(__file__))
+lightenv_path = os.path.join(current_dir, "LightEnv")
+if os.path.exists(lightenv_path):
+    sys.path.insert(0, lightenv_path)
+
+# ==================== 修复 huggingface_hub 兼容性（必须在导入 gradio 之前）====================
+# Gradio 4.x 需要 HfFolder，但旧版本的 huggingface_hub 没有
+def _fix_huggingface_hub():
+    """修复 huggingface_hub 兼容性问题"""
+    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 gradio as gr
+
+# ==================== 导入本地模块（在路径设置之后）====================
+# fmt: on
+from TextEnv_v2 import LightBulbEnv
+
+# ------------------- 全局变量 -------------------
+current_env: Optional[LightBulbEnv] = None
+test_data: List[dict] = []
+current_env_idx: int = 0
+history_records: List[str] = []
+current_user_id: str = ""
+show_logic: bool = False  # 默认隐藏逻辑规则
+save_dir = os.path.join(current_dir, "user_progress")
+MAX_STEPS = 200  # lights 任务的步骤上限
+
+# ------------------- 工具函数 -------------------
+
+
+def load_test_data():
+    """加载测试数据"""
+    global test_data
+    # 支持相对路径和绝对路径
+    test_file = os.path.join(
+        current_dir, "test_data/turnonlights/test_turnonlights_lite_251030.json")
+    # 如果文件不存在，尝试相对路径
+    if not os.path.exists(test_file):
+        test_file = "test_data/turnonlights/test_turnonlights_lite_251030.json"
+
+    try:
+        with open(test_file, 'r', encoding='utf-8') as f:
+            test_data = json.load(f)
+        return f"✅ 成功加载 {len(test_data)} 个测试环境"
+    except FileNotFoundError:
+        return f"❌ 文件未找到: {test_file}\n提示: 请确保测试数据文件存在"
+    except Exception as e:
+        return f"❌ 加载失败: {str(e)}"
+
+
+def format_bulb_state(obs: List[bool]) -> str:
+    """格式化灯泡状态显示"""
+    state_str = " ".join(["💡" if b else "○" for b in obs])
+    indices = " ".join([f"{i:2d}" for i in range(len(obs))])
+    return f"{state_str}\n{indices}"
+
+
+def format_logic(logic_expr: dict) -> str:
+    """格式化逻辑表达式显示"""
+    lines = []
+    for bulb in sorted(logic_expr.keys()):
+        lines.append(f"{bulb}: {logic_expr[bulb]}")
+    return "\n".join(lines)
+
+
+def load_environment(env_idx_display: int) -> Tuple[str, str, str, str, str]:
+    """加载环境（如果有保存的进度就加载，否则初始化新环境）
+    Args:
+        env_idx_display: 用户输入的环境索引（1-30，显示用）
+    Returns: (环境信息, 状态显示, 逻辑规则, 历史记录, 进度摘要)
+    """
+    global current_env, current_env_idx, history_records, show_logic, test_data, current_user_id
+
+    if not current_user_id:
+        progress_summary_text = "点击'查看未完成问题'按钮查看进度"
+        return "❌ 请先输入用户 ID", "", "", "", progress_summary_text
+    
+    if not test_data:
+        progress_summary_text = get_user_progress_summary(current_user_id) if current_user_id else "点击'查看未完成问题'按钮查看进度"
+        return "❌ 请先加载测试数据", "", "", "", progress_summary_text
+
+    # 将用户输入的 1-30 转换为内部索引 0-29
+    env_idx = env_idx_display - 1
+
+    if env_idx < 0 or env_idx >= len(test_data):
+        progress_summary_text = get_user_progress_summary(current_user_id) if current_user_id else "点击'查看未完成问题'按钮查看进度"
+        return f"❌ 环境索引超出范围 (1-{len(test_data)})", "", "", "", progress_summary_text
+
+    # 检查是否有保存的进度（检查两个版本的文件）
+    has_saved_progress = False
+    saved_progress_data = None
+    found_with_logic = None
+    
+    # 检查带逻辑规则和不带逻辑规则的版本
+    for with_logic_flag in [True, False]:
+        save_path = get_save_path(current_user_id, with_logic=with_logic_flag)
+        if os.path.exists(save_path):
+            try:
+                with open(save_path, 'r', encoding='utf-8') as f:
+                    all_progress_data = json.load(f)
+                
+                # 处理新旧格式兼容
+                if "environments" in all_progress_data:
+                    # 新格式：所有环境的进度保存在一个文件中
+                    environments = all_progress_data.get("environments", {})
+                    if str(env_idx) in environments:
+                        saved_progress_data = environments[str(env_idx)]
+                        has_saved_progress = True
+                        found_with_logic = with_logic_flag
+                        break  # 找到就退出循环
+                else:
+                    # 旧格式：单个环境的进度
+                    old_env_idx = all_progress_data.get("env_idx", -1)
+                    if old_env_idx == env_idx:
+                        saved_progress_data = all_progress_data
+                        has_saved_progress = True
+                        found_with_logic = with_logic_flag
+                        break  # 找到就退出循环
+            except Exception:
+                continue
+    
+    # 如果找到了保存的进度，更新 show_logic 以匹配找到的文件
+    if has_saved_progress and found_with_logic is not None:
+        show_logic = found_with_logic
+    
+    # 如果有保存的进度，检查是否已完成
+    if has_saved_progress and saved_progress_data:
+        bulb_states = saved_progress_data.get("bulb_states", [])
+        num_steps = saved_progress_data.get("num_steps", 0)
+        is_completed = False
+        if bulb_states and all(bulb_states):
+            is_completed = True  # 成功完成
+        elif num_steps >= MAX_STEPS:
+            is_completed = True  # 达到上限，算作完成
+        
+        if is_completed:
+            # 已完成，不覆盖，直接加载
+            return load_specific_environment(env_idx, saved_progress_data)
+        else:
+            # 未完成，加载保存的进度
+            return load_specific_environment(env_idx, saved_progress_data)
+    
+    # 没有保存的进度，初始化新环境
+    current_env_idx = env_idx
+    d = test_data[env_idx]
+    current_env = LightBulbEnv(
+        custom_logic=d["custom_logic"], num_bulbs=d["level"])
+    history_records = []
+    
+    # 保存初始化后的状态
+    save_progress()
+
+    obs = current_env._get_obs()
+    state_display = format_bulb_state(obs)
+    # 初始化环境时总是显示逻辑规则（因为这是新环境）
+    logic_display = format_logic(d["custom_logic"])
+    show_logic = True  # 初始化时设置为显示
+    history_display = "环境已初始化（新环境）\n"
+
+    info = f"✅ 环境 {env_idx_display}/{len(test_data)} 已初始化（新环境）\n"
+    info += f"灯泡数量: {d['level']}\n"
+    info += f"初始状态: {state_display.split(chr(10))[0]}"
+
+    # 自动更新进度摘要
+    progress_summary_text = get_user_progress_summary(current_user_id) if current_user_id else "点击'查看未完成问题'按钮查看进度"
+    
+    return info, state_display, logic_display, history_display, progress_summary_text
+
+
+def load_specific_environment(env_idx: int, progress_data: dict) -> Tuple[str, str, str, str, str]:
+    """加载特定环境的保存进度
+    Args:
+        env_idx: 环境索引
+        progress_data: 保存的进度数据
+    Returns: (环境信息, 状态显示, 逻辑规则, 历史记录, 进度摘要)
+    """
+    global current_env, current_env_idx, history_records, show_logic, test_data, current_user_id
+    
+    current_env_idx = env_idx
+    env_idx_display = progress_data.get("env_idx_display", env_idx + 1)
+    bulb_states = progress_data.get("bulb_states", [])
+    history_records = progress_data.get("history", [])
+    level = progress_data.get("level", 0)
+    num_steps = progress_data.get("num_steps", len(history_records))
+    
+    # 获取逻辑规则
+    custom_logic = progress_data.get("custom_logic", {})
+    
+    # 根据数据是否包含逻辑规则设置显示状态
+    has_logic_in_data = bool(custom_logic and len(custom_logic) > 0)
+    if has_logic_in_data:
+        show_logic = True
+    else:
+        show_logic = False
+    
+    # 如果没有保存逻辑规则，从test_data中获取
+    if not custom_logic and env_idx < len(test_data):
+        custom_logic = test_data[env_idx].get("custom_logic", {})
+    
+    # 恢复环境
+    if env_idx < len(test_data) and level > 0:
+        current_env = LightBulbEnv(custom_logic=custom_logic, num_bulbs=level)
+        # 恢复环境的步数
+        current_env.steps = num_steps
+        # 恢复灯泡状态
+        for i, state in enumerate(bulb_states):
+            if i < current_env.num_bulbs:
+                bulb_name = f"B{i}"
+                if bulb_name in current_env.bulbs:
+                    current_env.bulbs[bulb_name] = state
+    
+    obs = current_env._get_obs()
+    state_display = format_bulb_state(obs)
+    
+    # 根据数据是否包含逻辑规则决定显示
+    if show_logic and has_logic_in_data:
+        logic_display = format_logic(progress_data.get("custom_logic", {}))
+    else:
+        logic_display = "（此进度文件不包含逻辑规则）"
+    
+    history_display = "\n".join(history_records) if history_records else "无历史记录"
+    
+    # 检查是否已完成
+    is_completed = False
+    if bulb_states and all(bulb_states):
+        is_completed = True
+    elif num_steps >= MAX_STEPS:
+        is_completed = True
+    
+    if is_completed:
+        if all(bulb_states):
+            status_text = "已加载（已完成）"
+        else:
+            status_text = "已加载（已达到步骤上限）"
+    else:
+        status_text = "已加载（继续之前的进度）"
+    
+    info = f"✅ 环境 {env_idx_display}/{len(test_data)} {status_text}\n"
+    info += f"灯泡数量: {level}\n"
+    info += f"步骤数: {len(history_records)}"
+    
+    # 自动更新进度摘要
+    progress_summary_text = get_user_progress_summary(current_user_id) if current_user_id else "点击'查看未完成问题'按钮查看进度"
+    
+    return info, state_display, logic_display, history_display, progress_summary_text
+
+
+def step_environment(action_str: str) -> Tuple[str, str, str, bool]:
+    """执行一步动作"""
+    global current_env, history_records
+
+    # 获取当前状态，用于错误时保持显示
+    current_state_display = ""
+    if current_env is not None:
+        obs = current_env._get_obs()
+        current_state_display = format_bulb_state(obs)
+
+    if current_env is None:
+        return "❌ 请先初始化环境", current_state_display if current_state_display else "请先初始化环境", "", False
+    
+    if not current_user_id:
+        return "❌ 请先输入用户 ID", current_state_display, "", False
+
+    # 解析动作
+    action = None
+    action_error = None
+    try:
+        action = int(action_str.strip())
+        if action < 0 or action >= current_env.num_bulbs:
+            action_error = f"动作超出范围 (0-{current_env.num_bulbs-1})"
+    except ValueError:
+        action_error = f"无效的动作格式: {action_str}"
+
+    # 检查是否已经达到步骤上限（在执行动作之前）
+    if current_env.steps >= MAX_STEPS:
+        # 已达到上限，任务算作完成（但失败了）
+        history_display = "\n".join(history_records) if history_records else ""
+        save_status = save_progress()
+        feedback_info = f"⚠️ 已达到步骤上限 ({MAX_STEPS} 步)\n"
+        feedback_info += "任务已结束（未能在规定步数内完成）\n"
+        feedback_info += "无法继续执行动作\n"
+        feedback_info += save_status
+        
+        return feedback_info, current_state_display, history_display, True
+
+    # 如果动作无效，也要算作一步并记录到历史
+    if action_error:
+        # 记录无效动作到历史
+        history_records.append(
+            f"步骤 {len(history_records) + 1}: 动作={action_str} (无效), 反馈={action_error}")
+        history_display = "\n".join(history_records)
+        
+        # 增加环境的步数（即使动作无效也算一步）
+        current_env.steps += 1
+        
+        # 检查是否达到上限（执行无效动作后）
+        if current_env.steps >= MAX_STEPS:
+            # 达到上限，任务算作完成（但失败了）
+            history_records.append(
+                f"步骤 {len(history_records) + 1}: 已达到步骤上限 ({MAX_STEPS} 步)，任务结束")
+            history_display = "\n".join(history_records)
+            
+            # 自动保存进度
+            save_status = save_progress()
+            
+            # 生成反馈信息
+            feedback_info = f"动作: {action_str}\n反馈: ❌ {action_error}\n"
+            feedback_info += f"⚠️ 已达到步骤上限 ({MAX_STEPS} 步)\n"
+            feedback_info += "任务已结束（未能在规定步数内完成）\n"
+            feedback_info += save_status
+            
+            return feedback_info, current_state_display, history_display, True
+        
+        # 自动保存进度
+        save_status = save_progress()
+        
+        # 生成反馈信息
+        feedback_info = f"动作: {action_str}\n反馈: ❌ {action_error}\n"
+        feedback_info += save_status
+        
+        return feedback_info, current_state_display, history_display, False
+
+    # 执行有效动作
+    obs, feedback, done, _ = current_env.step(action)
+    state_display = format_bulb_state(obs)
+
+    # 更新历史
+    history_records.append(
+        f"步骤 {len(history_records) + 1}: 动作={action}, 反馈={feedback}")
+    history_display = "\n".join(history_records)
+
+    # 检查是否达到上限（在执行动作后）
+    if current_env.steps >= MAX_STEPS:
+        done = True  # 达到上限，任务算作完成（但失败了）
+        if not all(obs):  # 如果还没完成所有灯泡
+            feedback = f"{feedback}\n⚠️ 已达到步骤上限 ({MAX_STEPS} 步)，任务结束（未能在规定步数内完成）"
+
+    # 自动保存进度
+    save_status = save_progress()
+
+    # 生成反馈信息
+    feedback_info = f"动作: {action}\n反馈: {feedback}\n"
+    if done:
+        if all(obs):  # 所有灯泡都点亮了
+            feedback_info += "🎉 任务完成！所有灯泡已点亮！\n"
+        else:  # 达到上限但未完成
+            feedback_info += f"⚠️ 任务已结束（已达到步骤上限 {MAX_STEPS} 步）\n"
+    feedback_info += save_status
+
+    return feedback_info, state_display, history_display, done
+
+
+def reset_environment() -> Tuple[str, str, str, str]:
+    """重置当前环境
+    Returns: (环境信息, 状态显示, 历史记录, 进度摘要)
+    """
+    global current_env, history_records, current_user_id
+
+    if current_env is None:
+        return "❌ 请先初始化环境", "", "", "点击'查看未完成问题'按钮查看进度"
+
+    current_env.reset()
+    history_records = []
+    
+    # 保存重置后的状态
+    save_progress()
+
+    obs = current_env._get_obs()
+    state_display = format_bulb_state(obs)
+    history_display = "环境已重置\n"
+
+    # 自动更新进度摘要
+    progress_summary_text = get_user_progress_summary(current_user_id) if current_user_id else "点击'查看未完成问题'按钮查看进度"
+
+    return "✅ 环境已重置", state_display, history_display, progress_summary_text
+
+
+def get_save_path(user_id: str, with_logic: bool = None) -> str:
+    """获取用户进度保存路径
+    Args:
+        user_id: 用户ID
+        with_logic: 是否包含逻辑规则。如果为None，则根据全局show_logic决定
+    """
+    os.makedirs(save_dir, exist_ok=True)
+    if with_logic is None:
+        with_logic = show_logic
+    
+    if with_logic:
+        return os.path.join(save_dir, f"user_{user_id}_with_logic.json")
+    else:
+        return os.path.join(save_dir, f"user_{user_id}_no_logic.json")
+
+
+def save_progress() -> str:
+    """保存当前进度（保存所有环境的进度到一个文件中）"""
+    global current_env, current_env_idx, history_records, current_user_id, test_data, show_logic
+    
+    if not current_user_id:
+        return "⚠️ 请先输入用户 ID"
+    
+    if current_env is None:
+        return "⚠️ 没有可保存的进度"
+    
+    try:
+        # 获取当前环境状态
+        obs = current_env._get_obs()
+        
+        # 根据是否显示逻辑规则保存到不同的文件
+        save_path = get_save_path(current_user_id, with_logic=show_logic)
+        
+        # 读取已有的进度数据（如果存在）
+        all_progress = {}
+        if os.path.exists(save_path):
+            try:
+                with open(save_path, 'r', encoding='utf-8') as f:
+                    existing_data = json.load(f)
+                    # 如果旧格式（单个环境），转换为新格式
+                    if "env_idx" in existing_data and "environments" not in existing_data:
+                        # 旧格式，转换为新格式
+                        old_env_idx = existing_data.get("env_idx", -1)
+                        all_progress["environments"] = {}
+                        all_progress["environments"][str(old_env_idx)] = existing_data
+                    else:
+                        # 新格式
+                        all_progress = existing_data
+            except Exception:
+                all_progress = {}
+        
+        # 确保 environments 字典存在
+        if "environments" not in all_progress:
+            all_progress["environments"] = {}
+        
+        # 构建当前环境的进度数据
+        env_progress = {
+            "user_id": current_user_id,
+            "env_idx": current_env_idx,
+            "env_idx_display": current_env_idx + 1,
+            "bulb_states": obs,
+            "history": history_records,
+            "num_steps": current_env.steps,
+            "level": current_env.num_bulbs,
+        }
+        
+        # 如果显示逻辑规则，才保存逻辑规则数据
+        if show_logic and current_env_idx < len(test_data):
+            env_progress["custom_logic"] = test_data[current_env_idx]["custom_logic"]
+        
+        # 保存当前环境的进度
+        all_progress["environments"][str(current_env_idx)] = env_progress
+        all_progress["user_id"] = current_user_id
+        all_progress["current_env_idx"] = current_env_idx
+        
+        # 保存到文件
+        with open(save_path, 'w', encoding='utf-8') as f:
+            json.dump(all_progress, f, ensure_ascii=False, indent=2)
+        
+        return f"✅ 进度已保存 (环境 {current_env_idx + 1}, 步骤 {len(history_records)})"
+    except Exception as e:
+        return f"❌ 保存失败: {str(e)}"
+
+
+def load_progress(user_id: str, with_logic: bool) -> Tuple[str, float, str, str, str, str]:
+    """加载用户进度
+    Args:
+        user_id: 用户ID
+        with_logic: 是否加载带逻辑规则的版本
+    Returns: (状态信息, 环境���引显示(数字), 灯泡状态, 逻辑规则, 历史记录, 进度摘要)
+    """
+    global current_env, current_env_idx, history_records, current_user_id, test_data, show_logic
+    
+    if not user_id or not user_id.strip():
+        progress_summary_text = "点击'查看未完成问题'按钮查看进度"
+        return "⚠️ 请输入用户 ID", 1.0, "请先初始化环境", "（暂无逻辑规则显示）", "", progress_summary_text
+    
+    user_id = user_id.strip()
+    save_path = get_save_path(user_id, with_logic=with_logic)
+    
+    if not os.path.exists(save_path):
+        # 尝试加载另一个版本的文件
+        alt_save_path = get_save_path(user_id, with_logic=not with_logic)
+        if os.path.exists(alt_save_path):
+            # 如果另一个版本存在，提示用户
+            version_type = "带逻辑规则" if not with_logic else "不带逻辑规则"
+            current_user_id = user_id
+            show_logic = False
+            progress_summary_text = get_user_progress_summary(user_id)
+            return f"ℹ️ 用户 {user_id} 的{version_type}版本进度存在，请选择对应的版本", 1.0, "请先初始化环境", "（暂无逻辑规则显示）", "", progress_summary_text
+        else:
+            current_user_id = user_id
+            show_logic = False  # 默认隐藏
+            progress_summary_text = get_user_progress_summary(user_id)
+            return f"ℹ️ 用户 {user_id} 没有保存的进度，请初始化新环境", 1.0, "请先初始化环境", "（暂无逻辑规则显示）", "", progress_summary_text
+    
+    try:
+        with open(save_path, 'r', encoding='utf-8') as f:
+            all_progress_data = json.load(f)
+        
+        # 处理新旧格式兼容
+        if "environments" in all_progress_data:
+            # 新格式：所有环境的进度保存在一个文件中
+            current_env_idx_from_file = all_progress_data.get("current_env_idx", 0)
+            environments = all_progress_data.get("environments", {})
+            # 加载当前环境的进度
+            if str(current_env_idx_from_file) in environments:
+                progress_data = environments[str(current_env_idx_from_file)]
+            else:
+                # 如果没有当前环境的进度，尝试加载第一个环境
+                if environments:
+                    first_env_key = sorted(environments.keys())[0]
+                    progress_data = environments[first_env_key]
+                else:
+                    current_user_id = user_id
+                    show_logic = False
+                    progress_summary_text = get_user_progress_summary(user_id)
+                    return f"⚠️ 进度文件格式错误", 1.0, "请先初始化环境", "（暂无逻辑规则显示）", "", progress_summary_text
+        else:
+            # 旧格式：单个环境的进度
+            progress_data = all_progress_data
+        
+        # 恢复数据
+        current_user_id = user_id
+        env_idx = progress_data.get("env_idx", 0)
+        env_idx_display = progress_data.get("env_idx_display", env_idx + 1)
+        bulb_states = progress_data.get("bulb_states", [])
+        history_records = progress_data.get("history", [])
+        level = progress_data.get("level", 0)
+        
+        # 恢复环境的步数
+        num_steps = progress_data.get("num_steps", len(history_records))
+        
+        # 获取逻辑规则
+        custom_logic = progress_data.get("custom_logic", {})
+        
+        # 根据加载的数据是否包含逻辑规则来自动设置显示状态
+        # 如果数据中包含 custom_logic 且不为空，自动显示逻辑规则
+        # 如果数据中不包含 custom_logic 或为空，自动隐藏逻辑规则
+        has_logic_in_data = bool(custom_logic and len(custom_logic) > 0)
+        if has_logic_in_data:
+            show_logic = True
+        else:
+            show_logic = False
+        
+        # 如果没有保存逻辑规则，从test_data中获取（用于环境恢复，但不影响显示状态）
+        if not custom_logic and env_idx < len(test_data):
+            custom_logic = test_data[env_idx].get("custom_logic", {})
+        
+        # 检查 test_data 是否已加载
+        if not test_data:
+            current_user_id = user_id
+            show_logic = False
+            progress_summary_text = get_user_progress_summary(user_id)
+            return f"⚠️ 请先加载测试数据，然后再加载用户进度", 1.0, "请先初始化环境", "（暂无逻辑规则显示）", "", progress_summary_text
+        
+        # 恢复环境
+        if env_idx < len(test_data) and level > 0:
+            current_env_idx = env_idx
+            current_env = LightBulbEnv(custom_logic=custom_logic, num_bulbs=level)
+            # 恢复环境的步数
+            current_env.steps = num_steps
+            # 恢复灯泡状态
+            for i, state in enumerate(bulb_states):
+                if i < current_env.num_bulbs:
+                    bulb_name = f"B{i}"
+                    if bulb_name in current_env.bulbs:
+                        current_env.bulbs[bulb_name] = state
+        else:
+            current_user_id = user_id
+            show_logic = False
+            progress_summary_text = get_user_progress_summary(user_id)
+            return f"⚠️ 环境数据无效，请重新初始化", 1.0, "请先初始化环境", "（暂无逻辑规则显示）", "", progress_summary_text
+        
+        # 格式化显示
+        if current_env is None:
+            current_user_id = user_id
+            show_logic = False
+            progress_summary_text = get_user_progress_summary(user_id)
+            return f"⚠️ 环境恢复失败，请重新初始化", 1.0, "请先初始化环境", "（暂无逻辑规则显示）", "", progress_summary_text
+        
+        obs = current_env._get_obs()
+        state_display = format_bulb_state(obs)
+        
+        # 根据数据中是否包含逻辑规则来决定是否显示
+        # 如果数据中包含逻辑规则，自动显示；否则隐藏
+        # 注意：只显示原始数据中的逻辑规则，不从test_data中获取用于显示
+        if show_logic and has_logic_in_data:
+            logic_display = format_logic(progress_data.get("custom_logic", {}))
+        else:
+            logic_display = "（此进度文件不包含逻辑规则）"
+        
+        history_display = "\n".join(history_records) if history_records else "无历史记录"
+        
+        # 根据数据中是否包含逻辑规则来确定版本类型
+        version_type = "带逻辑规则" if has_logic_in_data else "不带逻辑规则"
+        info = f"✅ 已加载用户 {user_id} 的{version_type}版本进度\n"
+        info += f"环境: {env_idx_display}/{len(test_data)}\n"
+        info += f"步骤数: {len(history_records)}"
+        if show_logic:
+            info += "\n逻辑规则已自动显示"
+        else:
+            info += "\n逻辑规则已隐藏"
+        
+        # 自动更新进度摘要
+        progress_summary_text = get_user_progress_summary(user_id)
+        
+        # 返回数字类型，而不是字符串
+        return info, float(env_idx_display), state_display, logic_display, history_display, progress_summary_text
+        
+    except Exception as e:
+        # 简化错误信息
+        error_msg = f"❌ 加载失败: {str(e)}"
+        progress_summary_text = get_user_progress_summary(user_id) if user_id else "点击'查看未完成问题'按钮查看进度"
+        return error_msg, 1.0, "请先初始化环境", "（暂无逻辑规则显示）", "", progress_summary_text
+
+
+def set_user_id(user_id: str, with_logic: bool) -> Tuple[str, float, str, str, str, str]:
+    """设置用户 ID 并尝试加载进度
+    Args:
+        user_id: 用户ID
+        with_logic: 是否加载带逻辑规则的版本
+    Returns: (状态信息, 环境索引显示(数字), 灯泡状态, 逻辑规则, 历史记录, 进度摘要)
+    """
+    global current_user_id
+    
+    try:
+        if not user_id or not str(user_id).strip():
+            progress_summary_text = "点击'查看未完成问题'按钮查看进度"
+            return "⚠️ 请输入有效的用户 ID", 1.0, "请先初始化环境", "（暂无逻辑规则显示）", "", progress_summary_text
+        
+        user_id = str(user_id).strip()
+        # 加载进度（根据with_logic选择对应的文件）
+        return load_progress(user_id, with_logic)
+    except Exception as e:
+        # 如果出错，至少设置用户ID
+        current_user_id = str(user_id).strip() if user_id else ""
+        progress_summary_text = get_user_progress_summary(current_user_id) if current_user_id else "点击'查看未完成问题'按钮查看进度"
+        return f"❌ 设置用户ID时出错: {str(e)}", 1.0, "请先初始化环境", "（暂无逻辑规则显示）", "", progress_summary_text
+
+
+def get_user_progress_summary(user_id: str) -> str:
+    """获取用户进度摘要，显示未完成的问题
+    Args:
+        user_id: 用户ID
+    Returns: 格式化的进度摘要字符串
+    """
+    global test_data
+    
+    if not user_id or not user_id.strip():
+        return "⚠️ 请输入用户 ID"
+    
+    user_id = user_id.strip()
+    
+    # 检查两个版本的文件
+    completed_envs = set()
+    all_progress_files = []
+    
+    # 检查带逻辑规则和不带逻辑规则的版本
+    for with_logic in [True, False]:
+        save_path = get_save_path(user_id, with_logic=with_logic)
+        if os.path.exists(save_path):
+            try:
+                with open(save_path, 'r', encoding='utf-8') as f:
+                    all_progress_data = json.load(f)
+                
+                # 处理新旧格式兼容
+                if "environments" in all_progress_data:
+                    # 新格式：所有环境的进度保存在一个文件中
+                    environments = all_progress_data.get("environments", {})
+                    # 遍历所有环境的进度
+                    for env_key, progress_data in environments.items():
+                        env_idx = progress_data.get("env_idx", -1)
+                        bulb_states = progress_data.get("bulb_states", [])
+                        num_steps = progress_data.get("num_steps", 0)
+                        
+                        # 检查是否完成：
+                        # 1. 所有灯泡都点亮（成功完成）
+                        # 2. 达到步骤上限（200步，算作完成但失败）
+                        is_completed = False
+                        if bulb_states and all(bulb_states):
+                            is_completed = True  # 成功完成
+                        elif num_steps >= MAX_STEPS:
+                            is_completed = True  # 达到上限，算作完成（但失败）
+                        
+                        if is_completed:
+                            completed_envs.add(env_idx)
+                        
+                        all_progress_files.append({
+                            "env_idx": env_idx,
+                            "env_idx_display": progress_data.get("env_idx_display", env_idx + 1),
+                            "completed": is_completed,
+                            "num_steps": num_steps,
+                            "with_logic": with_logic
+                        })
+                else:
+                    # 旧格式：单个环境的进度
+                    progress_data = all_progress_data
+                    env_idx = progress_data.get("env_idx", -1)
+                    bulb_states = progress_data.get("bulb_states", [])
+                    num_steps = progress_data.get("num_steps", 0)
+                    
+                    # 检查是否完成
+                    is_completed = False
+                    if bulb_states and all(bulb_states):
+                        is_completed = True
+                    elif num_steps >= MAX_STEPS:
+                        is_completed = True
+                    
+                    if is_completed:
+                        completed_envs.add(env_idx)
+                    
+                    all_progress_files.append({
+                        "env_idx": env_idx,
+                        "env_idx_display": progress_data.get("env_idx_display", env_idx + 1),
+                        "completed": is_completed,
+                        "num_steps": num_steps,
+                        "with_logic": with_logic
+                    })
+            except Exception:
+                pass
+    
+    if not all_progress_files:
+        return f"ℹ️ 用户 {user_id} 还没有任何进度记录"
+    
+    # 获取总环境数
+    total_envs = len(test_data) if test_data else 0
+    if total_envs == 0:
+        return "⚠️ 请先加载测试数据"
+    
+    # 找出未完成的环境
+    all_env_indices = set(range(total_envs))
+    incomplete_envs = sorted(all_env_indices - completed_envs)
+    
+    # 构建摘要信息
+    summary_lines = []
+    summary_lines.append(f"📊 用户 {user_id} 的进度摘要")
+    summary_lines.append(f"总环境数: {total_envs}")
+    summary_lines.append(f"已完成: {len(completed_envs)}/{total_envs}")
+    summary_lines.append(f"未完成: {len(incomplete_envs)}/{total_envs}")
+    
+    if incomplete_envs:
+        summary_lines.append("\n❌ 未完成的环境:")
+        # 每行显示5个环境索引
+        for i in range(0, len(incomplete_envs), 5):
+            env_display_list = [str(env_idx + 1) for env_idx in incomplete_envs[i:i+5]]
+            summary_lines.append("  " + ", ".join(env_display_list))
+    else:
+        summary_lines.append("\n🎉 恭喜！所有环境都已完成！")
+    
+    return "\n".join(summary_lines)
+
+
+def extract_action_from_llm_output(text: str) -> str:
+    """从 LLM 输出中提取动作"""
+    m = re.search(r"<action>(.*?)</action>", text, re.IGNORECASE | re.DOTALL)
+    if m:
+        return m.group(1).strip()
+    return ""
+
+
+def generate_prompt_for_llm(env: LightBulbEnv, history: List[str], feedback: str) -> str:
+    """生成 LLM 的输入 prompt"""
+    grid_text = env.return_obs()
+    history_text = "\n".join(history) if history else "无历史记录"
+
+    prompt = f"""
+You are an intelligent agent.
+
+### Goal:
+Your mission is to light on all the bulbs.
+However, the accessibility of the bulbs is based on the current condition of other bulbs.
+You need to learn the hidden rule behind the environment and complete the task.
+
+### Action Space:
+The action space is based on the index of bulbs. For example, you would like to light on / off the first bulb, you should \
+output <action>0</action> to toggle the state of the bulb. 
+
+### History Action and Feedback:
+{history_text}
+
+### Current State:
+{grid_text}
+
+Now think step by step and choose the next action to act in the environment.
+You are encouraged to act actively to derive the environment dynamics.
+Output ONLY one action in the format: <action>n</action>
+"""
+    return prompt.strip()
+
+# ------------------- Gradio 界面 -------------------
+
+
+def create_interface():
+    """创建 Gradio 界面"""
+
+    with gr.Blocks(title="灯泡环境交互界面", theme=gr.themes.Soft()) as demo:
+        gr.Markdown("""
+        # 💡 灯泡环境交互界面
+        
+        这是一个灯泡控制环境，你需要通过点击灯泡来点亮所有灯泡。
+        每个灯泡的可用性取决于其他灯泡的状态。
+        """)
+
+        with gr.Row():
+            with gr.Column(scale=1):
+                gr.Markdown("### 📝 逻辑规则")
+                logic_display = gr.Textbox(
+                    label="依赖逻辑",
+                    interactive=False,
+                    lines=10,
+                    value="逻辑规则将根据加载的数据自动显示或隐藏"
+                )
+                
+                gr.Markdown("### 👤 用户信息")
+                user_id_input = gr.Textbox(
+                    label="用户 ID",
+                    placeholder="请输入您的用户 ID",
+                    info="输入您的 ID 以保存和加载进度"
+                )
+                load_with_logic = gr.Checkbox(
+                    label="加载带逻辑规则的版本",
+                    value=False,
+                    info="勾选则加载带逻辑规则的进度文件，否则加载不带逻辑规则的进度文件"
+                )
+                load_user_btn = gr.Button("加载用户进度", variant="primary")
+                check_progress_btn = gr.Button("查看未完成问题", variant="secondary")
+                progress_summary = gr.Textbox(
+                    label="进度摘要",
+                    interactive=False,
+                    lines=10,
+                    value="点击'查看未完成问题'按钮查看进度"
+                )
+                
+                gr.Markdown("### 🎮 环境控制")
+                # 动态更新最大值（最多30个环境）
+                max_envs = min(30, len(test_data)) if test_data else 30
+                env_idx_input = gr.Number(
+                    label="环境索引",
+                    value=1,
+                    minimum=1,
+                    maximum=max_envs,
+                    precision=0,
+                    info=f"选择要加载的环境 (1-{max_envs})"
+                )
+                init_btn = gr.Button("加载环境", variant="primary")
+                reset_btn = gr.Button("重置环境")
+
+                env_info = gr.Textbox(label="环境信息", interactive=False, lines=5)
+
+            with gr.Column(scale=2):
+                gr.Markdown("### 💡 当前状态")
+                state_display = gr.Textbox(
+                    label="灯泡状态",
+                    interactive=False,
+                    lines=3,
+                    value="请先加载环境"
+                )
+                
+                gr.Markdown("### 🎯 动作输入")
+                action_input = gr.Textbox(
+                    label="输入动作 (灯泡索引)",
+                    placeholder="例如: 0",
+                    info="输入要切换的灯泡索引 (0 开始)"
+                )
+                step_btn = gr.Button("执行动作", variant="primary")
+
+                gr.Markdown("### 💬 环境反馈")
+                feedback_display = gr.Textbox(
+                    label="反馈信息",
+                    interactive=False,
+                    lines=5
+                )
+
+                gr.Markdown("### 📜 操作历史")
+                history_display = gr.Textbox(
+                    label="操作历史",
+                    interactive=False,
+                    lines=10
+                )
+
+        # 事件绑定
+        # 加载用户进度 - 添加错误处理包装
+        def load_user_wrapper(user_id, with_logic_flag):
+            """包装函数，处理用户ID加载，添加错误处理"""
+            try:
+                if not user_id or not str(user_id).strip():
+                    progress_summary_text = "点击'查看未完成问题'按钮查看进度"
+                    return "⚠️ 请输入有效的用户 ID", 1.0, "请先初始化环境", "（暂无逻辑规则显示）", "", progress_summary_text
+                result = set_user_id(str(user_id).strip(), bool(with_logic_flag))
+                # 确保返回6个值，且第二个值是数字
+                if not isinstance(result, tuple) or len(result) != 6:
+                    progress_summary_text = get_user_progress_summary(str(user_id).strip()) if user_id else "点击'查看未完成问题'按钮查看进度"
+                    return "❌ 加载函数返回格式错误", 1.0, "请先初始化环境", "（暂无逻辑规则显示）", "", progress_summary_text
+                # 确保第二个值是数字类型
+                if not isinstance(result[1], (int, float)):
+                    result = (result[0], float(result[1]) if result[1] else 1.0, result[2], result[3], result[4], result[5])
+                return result
+            except Exception as e:
+                # 简化错误信息
+                error_msg = f"❌ 加载用户进度时出错: {str(e)}"
+                progress_summary_text = get_user_progress_summary(str(user_id).strip()) if user_id else "点击'查看未完成问题'按钮查看进度"
+                return error_msg, 1.0, "请先初始化环境", "（暂无逻辑规则显示）", "", progress_summary_text
+        
+        load_user_btn.click(
+            fn=load_user_wrapper,
+            inputs=[user_id_input, load_with_logic],
+            outputs=[env_info, env_idx_input, state_display, logic_display, history_display, progress_summary]
+        )
+        
+        # 用户ID输入框回车键
+        user_id_input.submit(
+            fn=load_user_wrapper,
+            inputs=[user_id_input, load_with_logic],
+            outputs=[env_info, env_idx_input, state_display, logic_display, history_display, progress_summary]
+        )
+        
+        # 查看未完成问题
+        def check_progress_wrapper(user_id):
+            """包装函数，检查用户进度"""
+            try:
+                if not user_id or not str(user_id).strip():
+                    return "⚠️ 请输入用户 ID"
+                return get_user_progress_summary(str(user_id).strip())
+            except Exception as e:
+                return f"❌ 检查进度时出错: {str(e)}"
+        
+        check_progress_btn.click(
+            fn=check_progress_wrapper,
+            inputs=user_id_input,
+            outputs=progress_summary
+        )
+        
+        # 用户ID输入框回车键也可以触发进度检查（但优先加载进度）
+        # 这里不添加，避免冲突
+
+        init_btn.click(
+            fn=load_environment,
+            inputs=env_idx_input,
+            outputs=[env_info, state_display, logic_display, history_display, progress_summary]
+        )
+
+        reset_btn.click(
+            fn=reset_environment,
+            outputs=[env_info, state_display, history_display, progress_summary]
+        )
+
+        def step_wrapper(action_str):
+            feedback, state, history, done = step_environment(action_str)
+            # 如果完成，更新环境信息（显示索引从1开始）
+            if done:
+                env_info_text = f"🎉 任务完成！所有灯泡已点亮！\n环境索引: {current_env_idx + 1}/{len(test_data) if test_data else 0}"
+            else:
+                env_info_text = f"环境索引: {current_env_idx + 1}/{len(test_data) if test_data else 0}\n步骤数: {len(history_records)}"
+            return feedback, state, history, env_info_text
+
+        step_btn.click(
+            fn=step_wrapper,
+            inputs=action_input,
+            outputs=[feedback_display, state_display,
+                     history_display, env_info]
+        )
+
+        # 支持回车键执行
+        action_input.submit(
+            fn=step_wrapper,
+            inputs=action_input,
+            outputs=[feedback_display, state_display,
+                     history_display, env_info]
+        )
+
+        gr.Markdown("""
+        ### 📖 使用说明
+        
+        1. **输入用户 ID**: 在"用户 ID"框中输入您的 ID，点击"加载用户进度"（如果有保存的进度会自动加载）
+        2. **加载环境**: 输入环境索引（1-30），点击"加载环境"。如果有保存的进度会自动加载，否则初始化新环境（测试数据已自动加载）
+        3. **执行动作**: 在"输入动作"框中输入灯泡索引（0 开始），点击"执行动作"或按回车
+        4. **查看状态**: 观察灯泡状态变化和反馈信息
+        5. **重置环境**: 点击"重置环境"可以重新开始当前环境
+        6. **查看进度**: 点击"查看未完成问题"可以查看您的完成情况
+        
+        **进度保存**: 
+        - 每次执行动作后会自动保存进度
+        - 下次输入相同的用户 ID 可以继续之前的进度
+        - 进度保存在 `user_progress/` 目录下
+        
+        **提示**: 
+        - 💡 表示灯泡已点亮
+        - ○ 表示灯泡未点亮
+        - 每个灯泡的可用性取决于其他灯泡的状态（见逻辑规则）
+        """)
+
+    return demo
+
+
+# ------------------- 主函数 -------------------
+if __name__ == "__main__":
+    # 自动加载测试数据
+    load_test_data()
+
+    demo = create_interface()
+    # 适配 Hugging Face Spaces 和本地运行
+    # 如果在 Hugging Face Spaces 上运行，不设置 server_name 和 server_port
+    # 如果在本地运行，使用 127.0.0.1 或 localhost
+    if os.getenv("SPACE_ID") is None:
+        # 本地运行
+        demo.launch(
+            server_name="127.0.0.1",
+            server_port=7860,
+            share=False
+        )
+    else:
+        # Hugging Face Spaces 运行
+        demo.launch()

app.py

@@ -0,0 +1,60 @@
+"""
+Hugging Face Spaces 入口文件
+这个文件是 Hugging Face Spaces 的标准入口点
+"""
+import os
+import sys
+
+# 设置工作目录
+script_dir = os.path.dirname(os.path.abspath(__file__))
+os.chdir(script_dir)
+
+# 添加 LightEnv 目录到 Python 路径
+lightenv_path = os.path.join(script_dir, "LightEnv")
+if os.path.exists(lightenv_path):
+    sys.path.insert(0, lightenv_path)
+
+# 修复 huggingface_hub 兼容性（必须在导入 gradio 之前）
+def _fix_huggingface_hub():
+    """修复 huggingface_hub 兼容性问题"""
+    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()
+
+# 导入 GUI-Light 模块
+# 由于文件名包含连字符，需要使用 importlib
+import importlib.util
+gui_light_path = os.path.join(script_dir, "GUI-Light.py")
+spec = importlib.util.spec_from_file_location("gui_light", gui_light_path)
+gui_light = importlib.util.module_from_spec(spec)
+sys.modules["gui_light"] = gui_light
+spec.loader.exec_module(gui_light)
+
+# 加载测试数据
+gui_light.load_test_data()
+
+# 创建 Gradio 应用
+demo = gui_light.create_interface()
+
+# Hugging Face Spaces 会自动调用 demo.launch()
+# 但为了确保兼容性，我们也可以显式调用
+if __name__ == "__main__":
+    demo.launch()

requirements.txt

@@ -0,0 +1,7 @@
+gradio>=4.0.0
+numpy
+huggingface_hub>=0.20.0
+requests
+
+
+