文件
acoustic_algo_scheduler/src/env_detect.py
T

167 行
5.7 KiB
Python

from __future__ import annotations
import importlib.util
import os
import platform
import shutil
import sys
from typing import Any, Dict, List
def _has_module(module_name: str) -> bool:
"""安全检测 Python 模块是否可用"""
try:
return importlib.util.find_spec(module_name) is not None
except ModuleNotFoundError:
return False
def _get_lib_suffix() -> str:
"""获取当前平台的动态库后缀"""
system = platform.system().lower()
if system == "windows":
return ".dll"
elif system == "darwin":
return ".dylib"
return ".so"
def _check_ld_paths() -> List[str]:
"""检测动态库搜索路径"""
paths = []
for var in ["LD_LIBRARY_PATH", "DYLD_LIBRARY_PATH", "PATH"]:
val = os.environ.get(var)
if val:
paths.extend([p for p in val.split(os.pathsep) if p])
return paths[:10] # 限制输出长度
def _cpp_runtime_available() -> bool:
"""检测 C++ 运行时能力(不依赖编译器)"""
try:
import ctypes # noqa: F401
except Exception:
return False
return platform.system().lower() in {"linux", "darwin", "windows"}
def collect_env_info() -> Dict[str, Any]:
"""收集完整的环境信息"""
compilers = {
"gcc": shutil.which("gcc"),
"g++": shutil.which("g++"),
"clang": shutil.which("clang"),
"clang++": shutil.which("clang++"),
"cl": shutil.which("cl"), # MSVC
}
compile_available = any(compilers.values())
runtime_available = _cpp_runtime_available()
return {
"python": {
"version": sys.version.split()[0],
"executable": sys.executable,
},
"platform": {
"os": platform.system(),
"release": platform.release(),
"machine": platform.machine(),
"lib_suffix": _get_lib_suffix(),
},
"matlab": {
"engine_available": _has_module("matlab.engine") or _has_module("matlab"),
"oct2py_available": _has_module("oct2py"),
"octave_cli": shutil.which("octave") or shutil.which("octave-cli"),
},
"cpp": {
"compilers": {k: v for k, v in compilers.items() if v},
"compile_available": compile_available,
"runtime_available": runtime_available,
# Backward-compatible alias (historically used as "is cpp usable")
"any_available": runtime_available,
},
"ld_paths": _check_ld_paths(),
"scipy_available": _has_module("scipy"),
"numpy_available": _has_module("numpy"),
"web_available": _has_module("requests"),
}
def get_available_algo_types() -> List[str]:
"""根据环境返回可用的算法类型列表"""
env = collect_env_info()
types = ["python"] # Python 始终可用
if env["matlab"]["engine_available"] or env["matlab"]["oct2py_available"] or env["matlab"]["octave_cli"]:
types.append("matlab")
# C++ 运行时可用即视为可执行(编译器仅影响“从源码编译”能力)
if env["cpp"]["runtime_available"]:
types.append("cpp")
if env["web_available"]:
types.append("web")
return types
def check_endpoint_active(url: str, timeout: int = 5) -> Tuple[bool, str]:
"""检查远程端点是否存活"""
try:
import requests
# 对根路径发起 GET 请求进行探活
from urllib.parse import urljoin, urlparse
parsed = urlparse(url)
base_url = f"{parsed.scheme}://{parsed.netloc}"
resp = requests.get(base_url, timeout=timeout)
if resp.status_code < 500:
return True, f"Active (HTTP {resp.status_code})"
return False, f"Server Error (HTTP {resp.status_code})"
except Exception as e:
return False, f"Disconnected: {str(e)}"
def get_actionable_suggestions() -> List[Dict[str, str]]:
"""针对故障环境提供具体的修复指令建议"""
env = collect_env_info()
suggestions = []
# MATLAB/Octave 建议
if not (env["matlab"]["engine_available"] or env["matlab"]["oct2py_available"] or env["matlab"]["octave_cli"]):
suggestions.append({
"target": "MATLAB/Octave",
"reason": "未检测到 MATLAB 引擎或 Octave 环境",
"fix": "建议安装 Octave: `sudo apt install octave` 或安装相关 Python 库: `pip install oct2py`"
})
elif not env["matlab"]["engine_available"] and env["matlab"]["octave_cli"]:
suggestions.append({
"target": "MATLAB Engine",
"reason": "当前使用 Octave 降级执行,性能可能受限",
"fix": "若需原生性能,请安装 MATLAB 并运行 `cd \"$(matlabroot)/extern/engines/python\" && python setup.py install`"
})
# C++ 建议
if not env["cpp"]["runtime_available"]:
suggestions.append({
"target": "C++ Runtime",
"reason": "当前 Python 环境缺少 ctypes/动态链接支持",
"fix": "请使用包含 ctypes 的标准 Python 发行版,或联系交付方提供兼容运行环境。",
})
elif not env["cpp"]["compile_available"]:
suggestions.append({
"target": "C++ Compiler",
"reason": "未检测到 gcc/g++/clang 等编译器(仅影响“从源码编译”能力)",
"fix": "建议安装基础构建工具: `sudo apt install build-essential`"
})
# 核心库建议
if not env["scipy_available"]:
suggestions.append({
"target": "SciPy",
"reason": "缺少 scipy 库,影响 MATLAB 兜底执行",
"fix": "运行 `pip install scipy`"
})
return suggestions