mirror of
https://github.com/dptech-corp/Uni-Lab-OS.git
synced 2025-12-15 13:44:39 +00:00
311 lines
12 KiB
Python
311 lines
12 KiB
Python
"""
|
||
环境检查模块
|
||
用于检查并自动安装 UniLabOS 运行所需的 Python 包
|
||
"""
|
||
|
||
import argparse
|
||
import importlib
|
||
import locale
|
||
import subprocess
|
||
import sys
|
||
|
||
from unilabos.utils.banner_print import print_status
|
||
|
||
|
||
class EnvironmentChecker:
|
||
"""环境检查器"""
|
||
|
||
def __init__(self):
|
||
# 定义必需的包及其安装名称的映射
|
||
self.required_packages = {
|
||
# 包导入名 : pip安装名
|
||
# "pymodbus.framer.FramerType": "pymodbus==3.9.2",
|
||
"websockets": "websockets",
|
||
"msgcenterpy": "msgcenterpy",
|
||
"opentrons_shared_data": "opentrons_shared_data",
|
||
"typing_extensions": "typing_extensions",
|
||
}
|
||
|
||
# 特殊安装包（需要特殊处理的包）
|
||
self.special_packages = {"pylabrobot": "git+https://github.com/Xuwznln/pylabrobot.git"}
|
||
|
||
# 包版本要求（包名: 最低版本）
|
||
self.version_requirements = {
|
||
"msgcenterpy": "0.1.5", # msgcenterpy 最低版本要求
|
||
}
|
||
|
||
self.missing_packages = []
|
||
self.failed_installs = []
|
||
self.packages_need_upgrade = []
|
||
|
||
# 检测系统语言
|
||
self.is_chinese = self._is_chinese_locale()
|
||
|
||
def _is_chinese_locale(self) -> bool:
|
||
"""检测系统是否为中文环境"""
|
||
try:
|
||
lang = locale.getdefaultlocale()[0]
|
||
if lang and ("zh" in lang.lower() or "chinese" in lang.lower()):
|
||
return True
|
||
except Exception:
|
||
pass
|
||
return False
|
||
|
||
def check_package_installed(self, package_name: str) -> bool:
|
||
"""检查包是否已安装"""
|
||
try:
|
||
importlib.import_module(package_name)
|
||
return True
|
||
except ImportError:
|
||
return False
|
||
|
||
def get_package_version(self, package_name: str) -> str | None:
|
||
"""获取已安装包的版本"""
|
||
try:
|
||
module = importlib.import_module(package_name)
|
||
return getattr(module, "__version__", None)
|
||
except (ImportError, AttributeError):
|
||
return None
|
||
|
||
def compare_version(self, current: str, required: str) -> bool:
|
||
"""
|
||
比较版本号
|
||
Returns:
|
||
True: current >= required
|
||
False: current < required
|
||
"""
|
||
try:
|
||
current_parts = [int(x) for x in current.split(".")]
|
||
required_parts = [int(x) for x in required.split(".")]
|
||
|
||
# 补齐长度
|
||
max_len = max(len(current_parts), len(required_parts))
|
||
current_parts.extend([0] * (max_len - len(current_parts)))
|
||
required_parts.extend([0] * (max_len - len(required_parts)))
|
||
|
||
return current_parts >= required_parts
|
||
except Exception:
|
||
return True # 如果无法比较，假设版本满足要求
|
||
|
||
def install_package(self, package_name: str, pip_name: str, upgrade: bool = False) -> bool:
|
||
"""安装包"""
|
||
try:
|
||
action = "升级" if upgrade else "安装"
|
||
print_status(f"正在{action} {package_name} ({pip_name})...", "info")
|
||
|
||
# 构建安装命令
|
||
cmd = [sys.executable, "-m", "pip", "install"]
|
||
|
||
# 如果是升级操作，添加 --upgrade 参数
|
||
if upgrade:
|
||
cmd.append("--upgrade")
|
||
|
||
cmd.append(pip_name)
|
||
|
||
# 如果是中文环境，使用清华镜像源
|
||
if self.is_chinese:
|
||
cmd.extend(["-i", "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/simple"])
|
||
|
||
# 执行安装
|
||
result = subprocess.run(cmd, capture_output=True, text=True, timeout=300) # 5分钟超时
|
||
|
||
if result.returncode == 0:
|
||
print_status(f"✓ {package_name} {action}成功", "success")
|
||
return True
|
||
else:
|
||
print_status(f"× {package_name} {action}失败: {result.stderr}", "error")
|
||
return False
|
||
|
||
except subprocess.TimeoutExpired:
|
||
print_status(f"× {package_name} {action}超时", "error")
|
||
return False
|
||
except Exception as e:
|
||
print_status(f"× {package_name} {action}异常: {str(e)}", "error")
|
||
return False
|
||
|
||
def upgrade_package(self, package_name: str, pip_name: str) -> bool:
|
||
"""升级包"""
|
||
return self.install_package(package_name, pip_name, upgrade=True)
|
||
|
||
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))
|
||
else:
|
||
# 检查版本要求
|
||
if import_name in self.version_requirements:
|
||
current_version = self.get_package_version(import_name)
|
||
required_version = self.version_requirements[import_name]
|
||
|
||
if current_version:
|
||
if not self.compare_version(current_version, required_version):
|
||
print_status(
|
||
f"{import_name} 版本过低 (当前: {current_version}, 需要: >={required_version})",
|
||
"warning",
|
||
)
|
||
self.packages_need_upgrade.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))
|
||
|
||
all_ok = not self.missing_packages and not self.packages_need_upgrade
|
||
|
||
if all_ok:
|
||
print_status("✓ 所有依赖包检查完成，环境正常", "success")
|
||
return True
|
||
|
||
if self.missing_packages:
|
||
print_status(f"发现 {len(self.missing_packages)} 个缺失的包", "warning")
|
||
if self.packages_need_upgrade:
|
||
print_status(f"发现 {len(self.packages_need_upgrade)} 个需要升级的包", "warning")
|
||
|
||
return False
|
||
|
||
def install_missing_packages(self, auto_install: bool = True) -> bool:
|
||
"""安装缺失的包"""
|
||
if not self.missing_packages and not self.packages_need_upgrade:
|
||
return True
|
||
|
||
if not auto_install:
|
||
if self.missing_packages:
|
||
print_status("缺失以下包:", "warning")
|
||
for import_name, pip_name in self.missing_packages:
|
||
print_status(f" - {import_name} (pip install {pip_name})", "warning")
|
||
if self.packages_need_upgrade:
|
||
print_status("需要升级以下包:", "warning")
|
||
for import_name, pip_name in self.packages_need_upgrade:
|
||
print_status(f" - {import_name} (pip install --upgrade {pip_name})", "warning")
|
||
return False
|
||
|
||
# 安装缺失的包
|
||
if self.missing_packages:
|
||
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))
|
||
|
||
print_status(f"✓ 成功安装 {success_count}/{len(self.missing_packages)} 个包", "success")
|
||
|
||
# 升级需要更新的包
|
||
if self.packages_need_upgrade:
|
||
print_status(f"开始自动升级 {len(self.packages_need_upgrade)} 个包...", "info")
|
||
|
||
upgrade_success_count = 0
|
||
for import_name, pip_name in self.packages_need_upgrade:
|
||
if self.upgrade_package(import_name, pip_name):
|
||
upgrade_success_count += 1
|
||
else:
|
||
self.failed_installs.append((import_name, pip_name))
|
||
|
||
print_status(f"✓ 成功升级 {upgrade_success_count}/{len(self.packages_need_upgrade)} 个包", "success")
|
||
|
||
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_name})", "error")
|
||
return False
|
||
|
||
return True
|
||
|
||
def verify_installation(self) -> bool:
|
||
"""验证安装结果"""
|
||
if not self.missing_packages and not self.packages_need_upgrade:
|
||
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))
|
||
|
||
# 验证升级的包
|
||
for import_name, pip_name in self.packages_need_upgrade:
|
||
if not self.check_package_installed(import_name):
|
||
failed_verification.append((import_name, pip_name))
|
||
elif import_name in self.version_requirements:
|
||
current_version = self.get_package_version(import_name)
|
||
required_version = self.version_requirements[import_name]
|
||
if current_version and not self.compare_version(current_version, required_version):
|
||
failed_verification.append((import_name, pip_name))
|
||
print_status(
|
||
f" {import_name} 版本仍然过低 (当前: {current_version}, 需要: >={required_version})",
|
||
"error",
|
||
)
|
||
|
||
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")
|