mirror of
https://github.com/deepmodeling/Uni-Lab-OS
synced 2026-04-24 12:20:00 +00:00
ccbf5378dd
update aksk desc print res query logs Fix skills exec error with action type Update Skills Update Skills addr Change uni-lab. to leap-lab. Support unit in pylabrobot Support async func. change to leap-lab backend. Support feedback interval. Reduce cocurrent lags. fix create_resource_with_slot update unilabos_formulation & batch-submit-exp scale multi exec thread up to 48 update handle creation api fit cocurrent gap add running status debounce allow non @topic_config support update skill add placeholder keys always free 提交实验技能 disable samples correct sample demo ret value 新增试剂reagent update registry 新增manual_confirm add workstation creation skill add virtual_sample_demo 样品追踪测试设备 add external devices param fix registry upload missing type fast registry load minor fix on skill & registry stripe ros2 schema desc add create-device-skill new registry system backwards to yaml remove not exist resource new registry sys exp. support with add device correct raise create resource error ret info fix revert ret info fix fix prcxi check add create_resource schema re signal host ready event add websocket connection timeout and improve reconnection logic add open_timeout parameter to websocket connection add TimeoutError and InvalidStatus exception handling implement exponential backoff for reconnection attempts simplify reconnection logic flow
368 lines
13 KiB
Python
368 lines
13 KiB
Python
"""
|
||
环境检查模块
|
||
用于检查并自动安装 UniLabOS 运行所需的 Python 包
|
||
"""
|
||
|
||
import argparse
|
||
import importlib
|
||
import locale
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
from pathlib import Path
|
||
from typing import List, Optional
|
||
|
||
from unilabos.utils.banner_print import print_status
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 底层安装工具
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _is_chinese_locale() -> bool:
|
||
try:
|
||
lang = locale.getdefaultlocale()[0]
|
||
return bool(lang and ("zh" in lang.lower() or "chinese" in lang.lower()))
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
_USE_UV: Optional[bool] = None
|
||
|
||
|
||
def _has_uv() -> bool:
|
||
global _USE_UV
|
||
if _USE_UV is None:
|
||
_USE_UV = shutil.which("uv") is not None
|
||
return _USE_UV
|
||
|
||
|
||
def _install_packages(
|
||
packages: List[str],
|
||
upgrade: bool = False,
|
||
label: str = "",
|
||
) -> bool:
|
||
"""
|
||
安装/升级一组包。优先 uv pip install，回退 sys pip。
|
||
逐个安装，任意一个失败不影响后续包。
|
||
|
||
Returns:
|
||
True if all succeeded, False otherwise.
|
||
"""
|
||
if not packages:
|
||
return True
|
||
|
||
is_chinese = _is_chinese_locale()
|
||
use_uv = _has_uv()
|
||
failed: List[str] = []
|
||
|
||
for pkg in packages:
|
||
action_word = "升级" if upgrade else "安装"
|
||
if label:
|
||
print_status(f"[{label}] 正在{action_word} {pkg}...", "info")
|
||
else:
|
||
print_status(f"正在{action_word} {pkg}...", "info")
|
||
|
||
if use_uv:
|
||
cmd = ["uv", "pip", "install"]
|
||
if upgrade:
|
||
cmd.append("--upgrade")
|
||
cmd.append(pkg)
|
||
if is_chinese:
|
||
cmd.extend(["--index-url", "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/simple"])
|
||
else:
|
||
cmd = [sys.executable, "-m", "pip", "install"]
|
||
if upgrade:
|
||
cmd.append("--upgrade")
|
||
cmd.append(pkg)
|
||
if is_chinese:
|
||
cmd.extend(["-i", "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/simple"])
|
||
|
||
try:
|
||
result = subprocess.run(cmd, capture_output=True, text=True, timeout=300)
|
||
if result.returncode == 0:
|
||
installer = "uv" if use_uv else "pip"
|
||
print_status(f"✓ {pkg} {action_word}成功 (via {installer})", "success")
|
||
else:
|
||
stderr_short = result.stderr.strip().split("\n")[-1] if result.stderr else "unknown error"
|
||
print_status(f"× {pkg} {action_word}失败: {stderr_short}", "error")
|
||
failed.append(pkg)
|
||
except subprocess.TimeoutExpired:
|
||
print_status(f"× {pkg} {action_word}超时 (300s)", "error")
|
||
failed.append(pkg)
|
||
except Exception as e:
|
||
print_status(f"× {pkg} {action_word}异常: {e}", "error")
|
||
failed.append(pkg)
|
||
|
||
if failed:
|
||
print_status(f"有 {len(failed)} 个包操作失败: {', '.join(failed)}", "error")
|
||
return False
|
||
return True
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# requirements.txt 安装（可多次调用）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def install_requirements_txt(req_path: str | Path, label: str = "") -> bool:
|
||
"""
|
||
读取一个 requirements.txt 文件，检查缺失的包并安装。
|
||
|
||
Args:
|
||
req_path: requirements.txt 文件路径
|
||
label: 日志前缀标签（如 "device_package_sim"）
|
||
|
||
Returns:
|
||
True if all ok, False if any install failed.
|
||
"""
|
||
req_path = Path(req_path)
|
||
if not req_path.exists():
|
||
return True
|
||
|
||
tag = label or req_path.parent.name
|
||
print_status(f"[{tag}] 检查依赖: {req_path}", "info")
|
||
|
||
reqs: List[str] = []
|
||
with open(req_path, "r", encoding="utf-8") as f:
|
||
for line in f:
|
||
line = line.strip()
|
||
if line and not line.startswith("#") and not line.startswith("-"):
|
||
reqs.append(line)
|
||
|
||
if not reqs:
|
||
return True
|
||
|
||
missing: List[str] = []
|
||
for req in reqs:
|
||
pkg_import = req.split(">=")[0].split("==")[0].split("<")[0].split("[")[0].split(">")[0].strip()
|
||
pkg_import = pkg_import.replace("-", "_")
|
||
try:
|
||
importlib.import_module(pkg_import)
|
||
except ImportError:
|
||
missing.append(req)
|
||
|
||
if not missing:
|
||
print_status(f"[{tag}] ✓ 依赖检查通过 ({len(reqs)} 个包)", "success")
|
||
return True
|
||
|
||
print_status(f"[{tag}] 缺失 {len(missing)} 个依赖: {', '.join(missing)}", "warning")
|
||
return _install_packages(missing, label=tag)
|
||
|
||
|
||
def check_device_package_requirements(devices_dirs: list[str]) -> bool:
|
||
"""
|
||
检查 --devices 指定的所有外部设备包目录中的 requirements.txt。
|
||
对每个目录查找 requirements.txt（先在目录内找，再在父目录找）。
|
||
"""
|
||
if not devices_dirs:
|
||
return True
|
||
|
||
all_ok = True
|
||
for d in devices_dirs:
|
||
d_path = Path(d).resolve()
|
||
req_file = d_path / "requirements.txt"
|
||
if not req_file.exists():
|
||
req_file = d_path.parent / "requirements.txt"
|
||
if not req_file.exists():
|
||
continue
|
||
if not install_requirements_txt(req_file, label=d_path.name):
|
||
all_ok = False
|
||
|
||
return all_ok
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# UniLabOS 核心环境检查
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class EnvironmentChecker:
|
||
"""环境检查器"""
|
||
|
||
def __init__(self):
|
||
self.required_packages = {
|
||
"websockets": "websockets",
|
||
"msgcenterpy": "msgcenterpy",
|
||
"orjson": "orjson",
|
||
"opentrons_shared_data": "opentrons_shared_data",
|
||
"typing_extensions": "typing_extensions",
|
||
"crcmod": "crcmod-plus",
|
||
}
|
||
|
||
self.special_packages = {"pylabrobot": "git+https://github.com/Xuwznln/pylabrobot.git"}
|
||
|
||
self.version_requirements = {
|
||
"msgcenterpy": "0.1.8",
|
||
}
|
||
|
||
self.missing_packages: List[tuple] = []
|
||
self.failed_installs: List[tuple] = []
|
||
self.packages_need_upgrade: List[tuple] = []
|
||
|
||
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:
|
||
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 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))
|
||
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):
|
||
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_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_name})", "warning")
|
||
return False
|
||
|
||
if self.missing_packages:
|
||
pkgs = [pip_name for _, pip_name in self.missing_packages]
|
||
if not _install_packages(pkgs, label="unilabos"):
|
||
self.failed_installs.extend(self.missing_packages)
|
||
|
||
if self.packages_need_upgrade:
|
||
pkgs = [pip_name for _, pip_name in self.packages_need_upgrade]
|
||
if not _install_packages(pkgs, upgrade=True, label="unilabos"):
|
||
self.failed_installs.extend(self.packages_need_upgrade)
|
||
|
||
return not self.failed_installs
|
||
|
||
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")
|