From 7b408127d158cb43c731bd881ca3d4fc114593bd Mon Sep 17 00:00:00 2001 From: Xuwznln <18435084+Xuwznln@users.noreply.github.com> Date: Fri, 1 Aug 2025 22:14:41 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B7=BB=E5=8A=A0=E7=8E=AF=E5=A2=83=E6=A3=80?= =?UTF-8?q?=E6=9F=A5=E5=8A=9F=E8=83=BD=EF=BC=8C=E6=94=AF=E6=8C=81=E8=87=AA?= =?UTF-8?q?=E5=8A=A8=E5=AE=89=E8=A3=85=E7=BC=BA=E5=A4=B1=E7=9A=84Python?= =?UTF-8?q?=E5=8C=85=EF=BC=9B=E6=9B=B4=E6=96=B0=E5=91=BD=E4=BB=A4=E8=A1=8C?= =?UTF-8?q?=E5=8F=82=E6=95=B0=E4=BB=A5=E8=B7=B3=E8=BF=87=E7=8E=AF=E5=A2=83?= =?UTF-8?q?=E6=A3=80=E6=9F=A5=EF=BC=9B=E4=BC=98=E5=8C=96=E4=BB=A3=E7=A0=81?= =?UTF-8?q?=E6=A0=BC=E5=BC=8F=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- unilabos/app/main.py | 30 ++++- unilabos/utils/__init__.py | 5 +- unilabos/utils/environment_check.py | 187 ++++++++++++++++++++++++++++ 3 files changed, 216 insertions(+), 6 deletions(-) create mode 100644 unilabos/utils/environment_check.py diff --git a/unilabos/app/main.py b/unilabos/app/main.py index e9d2f15..210f3e2 100644 --- a/unilabos/app/main.py +++ b/unilabos/app/main.py @@ -43,10 +43,11 @@ def convert_argv_dashes_to_underscores(args: argparse.ArgumentParser): for i, arg in enumerate(sys.argv): for option_string in option_strings: if arg.startswith(option_string): - new_arg = arg[:2] + arg[2:len(option_string)].replace("-", "_") + arg[len(option_string):] + new_arg = arg[:2] + arg[2 : len(option_string)].replace("-", "_") + arg[len(option_string) :] sys.argv[i] = new_arg break + def parse_args(): """解析命令行参数""" parser = argparse.ArgumentParser(description="Start Uni-Lab Edge server.") @@ -128,6 +129,11 @@ def parse_args(): default="", help="实验室唯一ID，也可通过环境变量 UNILABOS_MQCONFIG_LABID 设置或传入--config设置", ) + parser.add_argument( + "--skip_env_check", + action="store_true", + help="跳过启动时的环境依赖检查", + ) return parser @@ -138,6 +144,17 @@ def main(): convert_argv_dashes_to_underscores(args) args_dict = vars(args.parse_args()) + # 环境检查 - 检查并自动安装必需的包 (可选) + if not args_dict.get("skip_env_check", False): + from unilabos.utils.environment_check import check_environment + + print_status("正在进行环境依赖检查...", "info") + if not check_environment(auto_install=True): + print_status("环境检查失败，程序退出", "error") + os._exit(1) + else: + print_status("跳过环境依赖检查", "warning") + # 加载配置文件，优先加载config，然后从env读取 config_path = args_dict.get("config") if os.getcwd().endswith("unilabos_data"): @@ -151,17 +168,22 @@ def main(): if not os.path.exists(config_path): print_status( f"当前工作目录 {working_dir} 未找到local_config.py，请通过 --config 传入 local_config.py 文件路径", - "error") + "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"))): + 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") - shutil.copy(os.path.join(os.path.dirname(os.path.dirname(__file__)), "config", "example_config.py"), config_path) + shutil.copy( + os.path.join(os.path.dirname(os.path.dirname(__file__)), "config", "example_config.py"), config_path + ) print_status(f"已创建 local_config.py 路径： {config_path}", "info") print_status(f"请在文件夹中配置lab_id，放入下载的CA.crt、lab.crt、lab.key重新启动本程序", "info") os._exit(1) diff --git a/unilabos/utils/__init__.py b/unilabos/utils/__init__.py index 80b9573..8209b5a 100644 --- a/unilabos/utils/__init__.py +++ b/unilabos/utils/__init__.py @@ -1,7 +1,8 @@ from unilabos.utils.log import logger +from unilabos.utils.environment_check import check_environment, EnvironmentChecker # 确保日志配置在导入utils包时自动应用 # 这样任何导入utils包或其子模块的代码都会自动配置好日志 -# 导出logger，使其可以通过from unilabos.utils import logger直接导入 -__all__ = ['logger'] +# 导出logger和环境检查工具，使其可以直接导入 +__all__ = ["logger", "check_environment", "EnvironmentChecker"] diff --git a/unilabos/utils/environment_check.py b/unilabos/utils/environment_check.py new file mode 100644 index 0000000..cd83958 --- /dev/null +++ b/unilabos/utils/environment_check.py @@ -0,0 +1,187 @@ +""" +环境检查模块 +用于检查并自动安装 UniLabOS 运行所需的 Python 包 +""" + +import argparse +import importlib +import subprocess +import sys +from .banner_print import print_status + + +class EnvironmentChecker: + """环境检查器""" + + def __init__(self): + # 定义必需的包及其安装名称的映射 + self.required_packages = { + # 包导入名 : pip安装名 + "paho.mqtt": "paho-mqtt", + "opentrons_shared_data": "opentrons_shared_data", + } + + # 特殊安装包（需要特殊处理的包） + self.special_packages = {"pylabrobot": "git+https://github.com/Xuwznln/pylabrobot.git"} + + self.missing_packages = [] + self.failed_installs = [] + + def check_package_installed(self, package_name: str) -> bool: + """检查包是否已安装""" + try: + importlib.import_module(package_name) + return True + except ImportError: + return False + + def install_package(self, package_name: str, pip_name: str) -> bool: + """安装包""" + try: + print_status(f"正在安装 {package_name} ({pip_name})...", "info") + + # 构建安装命令 + cmd = [sys.executable, "-m", "pip", "install", pip_name] + + # 执行安装 + result = subprocess.run(cmd, capture_output=True, text=True, timeout=300) # 5分钟超时 + + if result.returncode == 0: + print_status(f"✅ {package_name} 安装成功", "success") + return True + else: + print_status(f"❌ {package_name} 安装失败: {result.stderr}", "error") + return False + + except subprocess.TimeoutExpired: + print_status(f"❌ {package_name} 安装超时", "error") + return False + except Exception as e: + print_status(f"❌ {package_name} 安装异常: {str(e)}", "error") + return False + + def check_all_packages(self) -> bool: + """检查所有必需的包""" + print_status("开始检查环境依赖...", "info") + + # 检查常规包 + for import_name, pip_name in self.required_packages.items(): + if not self.check_package_installed(import_name): + self.missing_packages.append((import_name, pip_name)) + + # 检查特殊包 + for package_name, install_url in self.special_packages.items(): + if not self.check_package_installed(package_name): + self.missing_packages.append((package_name, install_url)) + + if not self.missing_packages: + print_status("✅ 所有依赖包检查完成，环境正常", "success") + return True + + print_status(f"发现 {len(self.missing_packages)} 个缺失的包", "warning") + return False + + def install_missing_packages(self, auto_install: bool = True) -> bool: + """安装缺失的包""" + if not self.missing_packages: + return True + + if not auto_install: + print_status("缺失以下包:", "warning") + for import_name, pip_name in self.missing_packages: + print_status(f" - {import_name} (pip install {pip_name})", "warning") + return False + + print_status(f"开始自动安装 {len(self.missing_packages)} 个缺失的包...", "info") + + success_count = 0 + for import_name, pip_name in self.missing_packages: + if self.install_package(import_name, pip_name): + success_count += 1 + else: + self.failed_installs.append((import_name, pip_name)) + + if self.failed_installs: + print_status(f"有 {len(self.failed_installs)} 个包安装失败:", "error") + for import_name, pip_name in self.failed_installs: + print_status(f" - {import_name} (pip install {pip_name})", "error") + return False + + print_status(f"✅ 成功安装 {success_count} 个包", "success") + return True + + def verify_installation(self) -> bool: + """验证安装结果""" + if not self.missing_packages: + return True + + print_status("验证安装结果...", "info") + + failed_verification = [] + for import_name, pip_name in self.missing_packages: + if not self.check_package_installed(import_name): + failed_verification.append((import_name, pip_name)) + + if failed_verification: + print_status(f"有 {len(failed_verification)} 个包验证失败:", "error") + for import_name, pip_name in failed_verification: + print_status(f" - {import_name}", "error") + return False + + print_status("✅ 所有包验证通过", "success") + return True + + +def check_environment(auto_install: bool = True, show_details: bool = True) -> bool: + """ + 检查环境并自动安装缺失的包 + + Args: + auto_install: 是否自动安装缺失的包 + show_details: 是否显示详细信息 + + Returns: + bool: 环境检查是否通过 + """ + checker = EnvironmentChecker() + + # 检查包 + if checker.check_all_packages(): + return True + + # 安装缺失的包 + if not checker.install_missing_packages(auto_install): + if show_details: + print_status("请手动安装缺失的包后重新启动程序", "error") + return False + + # 验证安装 + if not checker.verify_installation(): + if show_details: + print_status("安装验证失败，请检查网络连接或手动安装", "error") + return False + + return True + + +if __name__ == "__main__": + # 命令行参数解析 + parser = argparse.ArgumentParser(description="UniLabOS 环境依赖检查工具") + parser.add_argument("--no-auto-install", action="store_true", help="仅检查环境，不自动安装缺失的包") + parser.add_argument("--silent", action="store_true", help="静默模式，不显示详细信息") + + args = parser.parse_args() + + # 执行环境检查 + auto_install = not args.no_auto_install + show_details = not args.silent + + success = check_environment(auto_install=auto_install, show_details=show_details) + + if not success: + if show_details: + print_status("环境检查失败", "error") + sys.exit(1) + else: + if show_details: + print_status("环境检查完成", "success")