提高环境变量设定的优先级

recipes/ros-humble-unilabos-msgs/recipe.yaml

@@ -57,5 +57,5 @@ requirements:
     - robostack-staging::ros-humble-rosidl-default-runtime
     - robostack-staging::ros-humble-std-msgs
     - robostack-staging::ros-humble-geometry-msgs
-    - robostack-staging::ros2-distro-mutex=0.6.*
+#    - robostack-staging::ros2-distro-mutex=0.6.*
     - sel(osx and x86_64): __osx >={{ MACOSX_DEPLOYMENT_TARGET|default('10.14') }}

unilabos/app/main.py

@@ -140,10 +140,24 @@ def main():
 
     # 加载配置文件，优先加载config，然后从env读取
     config_path = args_dict.get("config")
-    working_dir = os.path.abspath(os.path.join(os.getcwd(), "unilabos_data"))
-    if not config_path and (not os.path.exists(working_dir) or not os.path.exists(os.path.join(working_dir, "local_config.py"))):
-        print_status(f"当前未指定config路径，非第一次使用请通过 --config 传入 local_config.py 文件路径", "info")
-        print_status(f"您是否为第一次使用？并将当前文件路径 {working_dir} 作为工作目录？ (Y/n)", "info")
+    if os.getcwd().endswith("unilabos_data"):
+        working_dir = os.path.abspath(os.getcwd())
+    else:
+        working_dir = os.path.abspath(os.path.join(os.getcwd(), "unilabos_data"))
+    if args_dict.get("working_dir"):
+        working_dir = args_dict.get("working_dir")
+        if config_path and not os.path.exists(config_path):
+            config_path = os.path.join(working_dir, "local_config.py")
+            if not os.path.exists(config_path):
+                print_status(
+                    f"当前工作目录 {working_dir} 未找到local_config.py，请通过 --config 传入 local_config.py 文件路径",
+                    "error")
+                os._exit(1)
+    elif os.path.exists(working_dir) and os.path.exists(os.path.join(working_dir, "local_config.py")):
+        config_path = os.path.join(working_dir, "local_config.py")
+    elif not config_path and (not os.path.exists(working_dir) or not os.path.exists(os.path.join(working_dir, "local_config.py"))):
+        print_status(f"未指定config路径，可通过 --config 传入 local_config.py 文件路径", "info")
+        print_status(f"您是否为第一次使用？并将当前路径 {working_dir} 作为工作目录？ (Y/n)", "info")
         if input() != "n":
             os.makedirs(working_dir, exist_ok=True)
             config_path = os.path.join(working_dir, "local_config.py")
@@ -153,16 +167,8 @@ def main():
             os._exit(1)
         else:
             os._exit(1)
-    else:
-        working_dir = args_dict.get("working_dir") or os.path.abspath(os.path.join(os.getcwd(), "unilabos_data"))
-        if working_dir:
-            if config_path and not os.path.exists(config_path):
-                config_path = os.path.join(working_dir, "local_config.py")
-                if not os.path.exists(config_path):
-                    print_status(f"当前工作目录 {working_dir} 未找到local_config.py，请通过 --config 传入 local_config.py 文件路径", "error")
-                    os._exit(1)
-            print_status(f"当前工作目录为 {working_dir}", "info")
     # 加载配置文件
+    print_status(f"当前工作目录为 {working_dir}", "info")
     load_config_from_file(config_path, args_dict["labid"])
 
     # 设置BasicConfig参数
@@ -204,10 +210,11 @@ def main():
             print_status("联网获取设备加载文件成功", "info")
         graph, data = read_node_link_json(request_startup_json)
     else:
-        if args_dict["graph"].endswith(".json"):
-            graph, data = read_node_link_json(args_dict["graph"])
+        file_path = args_dict["graph"]
+        if file_path.endswith(".json"):
+            graph, data = read_node_link_json(file_path)
         else:
-            graph, data = read_graphml(args_dict["graph"])
+            graph, data = read_graphml(file_path)
     import unilabos.resources.graphio as graph_res
 
     graph_res.physical_setup_graph = graph

unilabos/app/web/client.py

@@ -4,11 +4,12 @@
 提供与远程服务器通信的客户端功能，只有host需要用
 """
 import json
+import os
 from typing import List, Dict, Any, Optional
 
 import requests
 from unilabos.utils.log import info
-from unilabos.config.config import MQConfig, HTTPConfig
+from unilabos.config.config import MQConfig, HTTPConfig, BasicConfig
 from unilabos.utils import logger
 
 
@@ -189,7 +190,7 @@ def request_startup_json(self) -> Optional[Dict[str, Any]]:
             logger.error(f"请求启动配置失败: {response.status_code}, {response.text}")
         else:
             try:
-                with open("startup_config.json", "w", encoding="utf-8") as f:
+                with open(os.path.join(BasicConfig.working_dir, "startup_config.json"), "w", encoding="utf-8") as f:
                     f.write(response.text)
                 target_dict = json.loads(response.text)
                 if "data" in target_dict:

unilabos/config/config.py

@@ -163,12 +163,12 @@ def _update_config_from_env():
 def load_config(config_path=None, override_labid=None):
     # 如果提供了配置文件路径，从该文件导入配置
     if config_path:
-        _update_config_from_env()  # 允许config_path被env设定后读取
+        env_config_path = os.environ.get("UNILABOS.BASICCONFIG.CONFIG_PATH")
+        config_path = env_config_path if env_config_path else config_path
         BasicConfig.config_path = os.path.abspath(os.path.dirname(config_path))
         if not os.path.exists(config_path):
             logger.error(f"[ENV] 配置文件 {config_path} 不存在")
             exit(1)
-
         try:
             module_name = "lab_" + os.path.basename(config_path).replace(".py", "")
             spec = importlib.util.spec_from_file_location(module_name, config_path)
@@ -179,6 +179,7 @@ def load_config(config_path=None, override_labid=None):
             spec.loader.exec_module(module)  # type: ignore
             _update_config_from_module(module, override_labid)
             logger.info(f"[ENV] 配置文件 {config_path} 加载成功")
+            _update_config_from_env()
         except Exception as e:
             logger.error(f"[ENV] 加载配置文件 {config_path} 失败")
             traceback.print_exc()

unilabos/config/example_config.py

@@ -8,9 +8,9 @@ class MQConfig:
     broker_url = ""
     port = 1883
 
-    ca_file = "CA.crt"
-    cert_file = "lab.crt"
-    key_file = "lab.key"
+    ca_file = "./CA.crt"
+    cert_file = "./lab.crt"
+    key_file = "./lab.key"
 
 # HTTP配置
 class HTTPConfig: