From 95971b5ecc872802c1598acab93989db82b70583 Mon Sep 17 00:00:00 2001 From: tatp-yf Date: Fri, 21 Aug 2026 15:57:20 +0800 Subject: [PATCH] refactor: move device info collection into library layer (#1240) --- scripts/benchmark/core/device_info.py | 295 +----------------------- src/unilab/training/experiment.py | 40 +--- src/unilab/utils/device.py | 292 +++++++++++++++++++++++ tests/benchmark/test_device_info.py | 88 ------- tests/test_library_import_boundary.py | 31 +++ tests/utils/test_device.py | 94 ++++++++ tests/utils/test_experiment_tracking.py | 15 +- 7 files changed, 426 insertions(+), 429 deletions(-) delete mode 100644 tests/benchmark/test_device_info.py create mode 100644 tests/test_library_import_boundary.py diff --git a/scripts/benchmark/core/device_info.py b/scripts/benchmark/core/device_info.py index 7503f19cb..21312bc67 100644 --- a/scripts/benchmark/core/device_info.py +++ b/scripts/benchmark/core/device_info.py @@ -1,294 +1,7 @@ -from __future__ import annotations - -import platform -import re -import subprocess -from functools import lru_cache -from typing import Dict - - -def _is_macos() -> bool: - return platform.system() == "Darwin" - - -def _is_linux() -> bool: - return platform.system() == "Linux" - - -def _is_windows() -> bool: - return platform.system() == "Windows" - - -def _get_device_info_macos() -> Dict[str, str]: - """Collect hardware info on macOS via system_profiler.""" - info: Dict[str, str] = { - "chip": "unknown", - "cpu_total_cores": "unknown", - "cpu_performance_cores": "unknown", - "cpu_efficiency_cores": "unknown", - "gpu_cores": "unknown", - "memory": "unknown", - } - try: - hw_text = subprocess.check_output( - ["system_profiler", "SPHardwareDataType"], text=True, stderr=subprocess.DEVNULL - ) - disp_text = subprocess.check_output( - ["system_profiler", "SPDisplaysDataType"], text=True, stderr=subprocess.DEVNULL - ) - except Exception: - return info - - chip_match = re.search(r"Chip:\s*(.+)", hw_text) - if chip_match: - info["chip"] = chip_match.group(1).strip() +"""Thin re-export of the library device-info helpers (moved in issue #1240).""" - mem_match = re.search(r"Memory:\s*(.+)", hw_text) - if mem_match: - info["memory"] = mem_match.group(1).strip() - - # Apple Silicon core descriptions vary by generation: - # M3/M4: "10 performance and 4 efficiency" - # M5 Pro/Max: "6 super and 12 performance" - cpu_match = re.search( - r"Total Number of Cores:\s*(\d+)\s*\(\s*(\d+)\s*(\w+)\s+and\s+(\d+)\s*(\w+)\s*\)", - hw_text, - ) - if cpu_match: - total, count1, type1, count2, type2 = cpu_match.groups() - info["cpu_total_cores"] = total - info["cpu_core_type_1"] = type1 - info["cpu_core_count_1"] = count1 - info["cpu_core_type_2"] = type2 - info["cpu_core_count_2"] = count2 - # Backward-compat keys for legacy P+E format - if type1 == "performance" and type2 == "efficiency": - info["cpu_performance_cores"] = count1 - info["cpu_efficiency_cores"] = count2 - elif type1 == "super" and type2 == "performance": - info["cpu_super_cores"] = count1 - info["cpu_performance_cores"] = count2 - else: - cpu_total_match = re.search(r"Total Number of Cores:\s*(\d+)", hw_text) - if cpu_total_match: - info["cpu_total_cores"] = cpu_total_match.group(1) - - gpu_match = re.search(r"Type:\s*GPU[\s\S]*?Total Number of Cores:\s*(\d+)", disp_text) - if gpu_match: - info["gpu_cores"] = gpu_match.group(1) - - return info - - -def _get_device_info_linux() -> Dict[str, str]: - """Collect hardware info on Linux via /proc and common CLI tools.""" - info: Dict[str, str] = { - "chip": "unknown", - "cpu_total_cores": "unknown", - "cpu_performance_cores": "unknown", - "cpu_efficiency_cores": "unknown", - "gpu_name": "unknown", - "memory": "unknown", - } - # CPU model - try: - with open("/proc/cpuinfo", encoding="utf-8") as f: - cpuinfo = f.read() - model_match = re.search(r"^model name\s*:\s*(.+)$", cpuinfo, re.MULTILINE) - if model_match: - info["chip"] = model_match.group(1).strip() - # Count physical cores (unique core id per physical id) - pairs = re.findall(r"physical id\s*:\s*(\d+).*?core id\s*:\s*(\d+)", cpuinfo, re.DOTALL) - if pairs: - info["cpu_total_cores"] = str(len(set(pairs))) - else: - processor_count = len(re.findall(r"^processor\s*:", cpuinfo, re.MULTILINE)) - if processor_count: - info["cpu_total_cores"] = str(processor_count) - except Exception: - pass - # Total memory - try: - with open("/proc/meminfo", encoding="utf-8") as f: - meminfo = f.read() - mem_match = re.search(r"MemTotal:\s*(\d+)\s*kB", meminfo) - if mem_match: - mem_gb = int(mem_match.group(1)) / 1024 / 1024 - info["memory"] = f"{mem_gb:.1f} GB" - except Exception: - pass - # GPU via nvidia-smi - try: - gpu_out = subprocess.check_output( - ["nvidia-smi", "--query-gpu=name,memory.total", "--format=csv,noheader"], - text=True, - stderr=subprocess.DEVNULL, - ).strip() - if gpu_out: - # Take the first GPU line - first_line = gpu_out.splitlines()[0] - parts = [p.strip() for p in first_line.split(",")] - info["gpu_name"] = parts[0] - if len(parts) > 1: - info["gpu_memory"] = parts[1] - except Exception: - pass - # GPU via rocm-smi (AMD) - if info["gpu_name"] == "unknown": - try: - rocm_out = subprocess.check_output( - ["rocm-smi", "--showproductname"], - text=True, - stderr=subprocess.DEVNULL, - ) - gpu_match = re.search(r"Card series\s*:\s*(.+)", rocm_out, re.IGNORECASE) - if gpu_match: - info["gpu_name"] = gpu_match.group(1).strip() - except Exception: - pass - # GPU memory via amd-smi (AMD ROCm). On unified-memory APUs this reports - # the BIOS-allocated visible VRAM slice, e.g. 96 GB out of 128 GB. - try: - amd_smi_out = subprocess.check_output( - ["amd-smi", "metric"], - text=True, - stderr=subprocess.DEVNULL, - ) - vram_match = re.search(r"TOTAL_VISIBLE_VRAM:\s*(\d+)\s*MB", amd_smi_out) - if vram_match: - info["gpu_memory"] = f"{int(vram_match.group(1))} MB" - gtt_match = re.search(r"TOTAL_GTT:\s*(\d+)\s*MB", amd_smi_out) - if gtt_match: - info["gpu_gtt_memory"] = f"{int(gtt_match.group(1))} MB" - except Exception: - pass - # Fallback GPU via lspci (AMD/ATI, Intel iGPU/Arc, and others) - if info["gpu_name"] == "unknown": - try: - lspci_out = subprocess.check_output(["lspci"], text=True, stderr=subprocess.DEVNULL) - for line in lspci_out.splitlines(): - if "VGA" in line or "Display" in line or "3D" in line: - if "AMD" in line or "ATI" in line: - match = re.search(r"\[AMD/ATI\]\s*(.+)", line) - if match: - name = match.group(1).strip() - name = re.sub(r"\s*\(rev.*\)", "", name) - info["gpu_name"] = name - break - elif "Intel" in line: - # e.g. "Intel Corporation Meteor Lake-P [Intel Arc Graphics] (rev 08)" - match = re.search(r"\[([^\]]+)\]", line) - if match: - info["gpu_name"] = match.group(1).strip() - break - except Exception: - pass - # If GPU name is still generic/unknown, try to infer from CPU model (APUs) - if info["gpu_name"] in ("unknown", "AMD Radeon Graphics"): - chip = info.get("chip", "") - match = re.search(r"w(?:ith)?/\s*(Radeon\s+[\w\s\+]+)", chip, re.IGNORECASE) - if match: - info["gpu_name"] = match.group(1).strip() - return info - - -def _get_device_info_windows() -> Dict[str, str]: - """Collect hardware info on Windows via wmic.""" - info: Dict[str, str] = { - "chip": "unknown", - "cpu_total_cores": "unknown", - "cpu_performance_cores": "unknown", - "cpu_efficiency_cores": "unknown", - "gpu_name": "unknown", - "memory": "unknown", - } - try: - cpu_out = subprocess.check_output( - ["wmic", "cpu", "get", "Name,NumberOfCores", "/format:csv"], - text=True, - stderr=subprocess.DEVNULL, - ) - lines = [l for l in cpu_out.splitlines() if l.strip() and not l.strip().startswith("Node")] - if lines: - parts = lines[0].split(",") - if len(parts) >= 3: - info["cpu_total_cores"] = parts[1].strip() - info["chip"] = parts[2].strip() - except Exception: - pass - # Memory - try: - mem_out = subprocess.check_output( - ["wmic", "ComputerSystem", "get", "TotalPhysicalMemory", "/format:csv"], - text=True, - stderr=subprocess.DEVNULL, - ) - lines = [l for l in mem_out.splitlines() if l.strip() and not l.strip().startswith("Node")] - if lines: - parts = lines[0].split(",") - if len(parts) >= 2: - mem_gb = int(parts[1].strip()) / 1024**3 - info["memory"] = f"{mem_gb:.1f} GB" - except Exception: - pass - # GPU via nvidia-smi (also available on Windows) - try: - gpu_out = subprocess.check_output( - ["nvidia-smi", "--query-gpu=name,memory.total", "--format=csv,noheader"], - text=True, - stderr=subprocess.DEVNULL, - ).strip() - if gpu_out: - first_line = gpu_out.splitlines()[0] - parts = [p.strip() for p in first_line.split(",")] - info["gpu_name"] = parts[0] - if len(parts) > 1: - info["gpu_memory"] = parts[1] - except Exception: - pass - return info - - -@lru_cache(maxsize=1) -def get_device_info_dict() -> Dict[str, str]: - base: Dict[str, str] = {"platform": platform.platform()} - if _is_macos(): - base.update(_get_device_info_macos()) - elif _is_linux(): - base.update(_get_device_info_linux()) - elif _is_windows(): - base.update(_get_device_info_windows()) - return base +from __future__ import annotations +from unilab.utils.device import get_device_info_dict, get_device_info_line -def get_device_info_line() -> str: - d = get_device_info_dict() - if _is_macos(): - # Build core-type summary dynamically so M5 (super+performance) is shown correctly - if d.get("cpu_core_type_1") and d.get("cpu_core_type_2"): - t1 = d["cpu_core_type_1"][0].upper() - t2 = d["cpu_core_type_2"][0].upper() - core_summary = f"{d['cpu_core_count_1']}{t1}+{d['cpu_core_count_2']}{t2}" - elif ( - d.get("cpu_performance_cores") != "unknown" - and d.get("cpu_efficiency_cores") != "unknown" - ): - core_summary = f"{d['cpu_performance_cores']}P+{d['cpu_efficiency_cores']}E" - else: - core_summary = "unknown" - return ( - f"Device: {d.get('chip', 'unknown')} | " - f"CPU: {d.get('cpu_total_cores', 'unknown')} cores " - f"({core_summary}) | " - f"GPU: {d.get('gpu_cores', 'unknown')} cores | " - f"Memory: {d.get('memory', 'unknown')}" - ) - else: - gpu_part = d.get("gpu_name", "unknown") - if "gpu_memory" in d: - gpu_part += f" ({d['gpu_memory']})" - return ( - f"CPU: {d.get('chip', 'unknown')} ({d.get('cpu_total_cores', 'unknown')} cores) | " - f"GPU: {gpu_part} | " - f"Memory: {d.get('memory', 'unknown')}" - ) +__all__ = ["get_device_info_dict", "get_device_info_line"] diff --git a/src/unilab/training/experiment.py b/src/unilab/training/experiment.py index 580940c87..ffc9cda44 100644 --- a/src/unilab/training/experiment.py +++ b/src/unilab/training/experiment.py @@ -5,10 +5,8 @@ import dataclasses import getpass import importlib -import importlib.util import json import os -import platform import socket import subprocess import time @@ -19,6 +17,7 @@ from omegaconf import OmegaConf from unilab.training.sim2sim import extract_contract_snapshot +from unilab.utils.device import get_device_info_dict def _cfg_get(cfg: Any, key: str, default: Any = None) -> Any: @@ -60,43 +59,6 @@ def _json_safe(value: Any) -> Any: return str(value) -def _fallback_device_info_dict() -> dict[str, str]: - return { - "platform": platform.platform(), - "chip": platform.processor() or "unknown", - "cpu_total_cores": str(os.cpu_count() or "unknown"), - "gpu_name": "unknown", - "memory": "unknown", - } - - -def _benchmark_device_info_path() -> Path | None: - for parent in Path(__file__).resolve().parents: - candidate = parent / "scripts" / "benchmark" / "core" / "device_info.py" - if candidate.is_file(): - return candidate - return None - - -def get_device_info_dict() -> dict[str, str]: - try: - module_path = _benchmark_device_info_path() - if module_path is None: - return _fallback_device_info_dict() - spec = importlib.util.spec_from_file_location( - "unilab_benchmark_device_info", - module_path, - ) - if spec is None or spec.loader is None: - raise ImportError(f"Unable to load device info module from {module_path}") - module = importlib.util.module_from_spec(spec) - spec.loader.exec_module(module) - getter = getattr(module, "get_device_info_dict") - return dict(getter()) - except Exception: - return _fallback_device_info_dict() - - def get_git_info(root_dir: str | Path) -> dict[str, Any]: root = Path(root_dir) diff --git a/src/unilab/utils/device.py b/src/unilab/utils/device.py index f5a4cdcee..004d64b37 100644 --- a/src/unilab/utils/device.py +++ b/src/unilab/utils/device.py @@ -1,5 +1,9 @@ from __future__ import annotations +import platform +import re +import subprocess +from functools import lru_cache from typing import Callable, cast import torch @@ -120,3 +124,291 @@ def resolve_torch_device_alias(device: str | None, *, default: str = "cpu") -> s ) raise ValueError(f"Unsupported device alias {original!r}; expected cpu, gpu, cuda, mps, or xpu") + + +def _is_macos() -> bool: + return platform.system() == "Darwin" + + +def _is_linux() -> bool: + return platform.system() == "Linux" + + +def _is_windows() -> bool: + return platform.system() == "Windows" + + +def _get_device_info_macos() -> dict[str, str]: + """Collect hardware info on macOS via system_profiler.""" + info: dict[str, str] = { + "chip": "unknown", + "cpu_total_cores": "unknown", + "cpu_performance_cores": "unknown", + "cpu_efficiency_cores": "unknown", + "gpu_cores": "unknown", + "memory": "unknown", + } + try: + hw_text = subprocess.check_output( + ["system_profiler", "SPHardwareDataType"], text=True, stderr=subprocess.DEVNULL + ) + disp_text = subprocess.check_output( + ["system_profiler", "SPDisplaysDataType"], text=True, stderr=subprocess.DEVNULL + ) + except Exception: + return info + + chip_match = re.search(r"Chip:\s*(.+)", hw_text) + if chip_match: + info["chip"] = chip_match.group(1).strip() + + mem_match = re.search(r"Memory:\s*(.+)", hw_text) + if mem_match: + info["memory"] = mem_match.group(1).strip() + + # Apple Silicon core descriptions vary by generation: + # M3/M4: "10 performance and 4 efficiency" + # M5 Pro/Max: "6 super and 12 performance" + cpu_match = re.search( + r"Total Number of Cores:\s*(\d+)\s*\(\s*(\d+)\s*(\w+)\s+and\s+(\d+)\s*(\w+)\s*\)", + hw_text, + ) + if cpu_match: + total, count1, type1, count2, type2 = cpu_match.groups() + info["cpu_total_cores"] = total + info["cpu_core_type_1"] = type1 + info["cpu_core_count_1"] = count1 + info["cpu_core_type_2"] = type2 + info["cpu_core_count_2"] = count2 + # Backward-compat keys for legacy P+E format + if type1 == "performance" and type2 == "efficiency": + info["cpu_performance_cores"] = count1 + info["cpu_efficiency_cores"] = count2 + elif type1 == "super" and type2 == "performance": + info["cpu_super_cores"] = count1 + info["cpu_performance_cores"] = count2 + else: + cpu_total_match = re.search(r"Total Number of Cores:\s*(\d+)", hw_text) + if cpu_total_match: + info["cpu_total_cores"] = cpu_total_match.group(1) + + gpu_match = re.search(r"Type:\s*GPU[\s\S]*?Total Number of Cores:\s*(\d+)", disp_text) + if gpu_match: + info["gpu_cores"] = gpu_match.group(1) + + return info + + +def _get_device_info_linux() -> dict[str, str]: + """Collect hardware info on Linux via /proc and common CLI tools.""" + info: dict[str, str] = { + "chip": "unknown", + "cpu_total_cores": "unknown", + "cpu_performance_cores": "unknown", + "cpu_efficiency_cores": "unknown", + "gpu_name": "unknown", + "memory": "unknown", + } + # CPU model + try: + with open("/proc/cpuinfo", encoding="utf-8") as f: + cpuinfo = f.read() + model_match = re.search(r"^model name\s*:\s*(.+)$", cpuinfo, re.MULTILINE) + if model_match: + info["chip"] = model_match.group(1).strip() + # Count physical cores (unique core id per physical id) + pairs = re.findall(r"physical id\s*:\s*(\d+).*?core id\s*:\s*(\d+)", cpuinfo, re.DOTALL) + if pairs: + info["cpu_total_cores"] = str(len(set(pairs))) + else: + processor_count = len(re.findall(r"^processor\s*:", cpuinfo, re.MULTILINE)) + if processor_count: + info["cpu_total_cores"] = str(processor_count) + except Exception: + pass + # Total memory + try: + with open("/proc/meminfo", encoding="utf-8") as f: + meminfo = f.read() + mem_match = re.search(r"MemTotal:\s*(\d+)\s*kB", meminfo) + if mem_match: + mem_gb = int(mem_match.group(1)) / 1024 / 1024 + info["memory"] = f"{mem_gb:.1f} GB" + except Exception: + pass + # GPU via nvidia-smi + try: + gpu_out = subprocess.check_output( + ["nvidia-smi", "--query-gpu=name,memory.total", "--format=csv,noheader"], + text=True, + stderr=subprocess.DEVNULL, + ).strip() + if gpu_out: + # Take the first GPU line + first_line = gpu_out.splitlines()[0] + parts = [p.strip() for p in first_line.split(",")] + info["gpu_name"] = parts[0] + if len(parts) > 1: + info["gpu_memory"] = parts[1] + except Exception: + pass + # GPU via rocm-smi (AMD) + if info["gpu_name"] == "unknown": + try: + rocm_out = subprocess.check_output( + ["rocm-smi", "--showproductname"], + text=True, + stderr=subprocess.DEVNULL, + ) + gpu_match = re.search(r"Card series\s*:\s*(.+)", rocm_out, re.IGNORECASE) + if gpu_match: + info["gpu_name"] = gpu_match.group(1).strip() + except Exception: + pass + # GPU memory via amd-smi (AMD ROCm). On unified-memory APUs this reports + # the BIOS-allocated visible VRAM slice, e.g. 96 GB out of 128 GB. + try: + amd_smi_out = subprocess.check_output( + ["amd-smi", "metric"], + text=True, + stderr=subprocess.DEVNULL, + ) + vram_match = re.search(r"TOTAL_VISIBLE_VRAM:\s*(\d+)\s*MB", amd_smi_out) + if vram_match: + info["gpu_memory"] = f"{int(vram_match.group(1))} MB" + gtt_match = re.search(r"TOTAL_GTT:\s*(\d+)\s*MB", amd_smi_out) + if gtt_match: + info["gpu_gtt_memory"] = f"{int(gtt_match.group(1))} MB" + except Exception: + pass + # Fallback GPU via lspci (AMD/ATI, Intel iGPU/Arc, and others) + if info["gpu_name"] == "unknown": + try: + lspci_out = subprocess.check_output(["lspci"], text=True, stderr=subprocess.DEVNULL) + for line in lspci_out.splitlines(): + if "VGA" in line or "Display" in line or "3D" in line: + if "AMD" in line or "ATI" in line: + match = re.search(r"\[AMD/ATI\]\s*(.+)", line) + if match: + name = match.group(1).strip() + name = re.sub(r"\s*\(rev.*\)", "", name) + info["gpu_name"] = name + break + elif "Intel" in line: + # e.g. "Intel Corporation Meteor Lake-P [Intel Arc Graphics] (rev 08)" + match = re.search(r"\[([^\]]+)\]", line) + if match: + info["gpu_name"] = match.group(1).strip() + break + except Exception: + pass + # If GPU name is still generic/unknown, try to infer from CPU model (APUs) + if info["gpu_name"] in ("unknown", "AMD Radeon Graphics"): + chip = info.get("chip", "") + match = re.search(r"w(?:ith)?/\s*(Radeon\s+[\w\s\+]+)", chip, re.IGNORECASE) + if match: + info["gpu_name"] = match.group(1).strip() + return info + + +def _get_device_info_windows() -> dict[str, str]: + """Collect hardware info on Windows via wmic.""" + info: dict[str, str] = { + "chip": "unknown", + "cpu_total_cores": "unknown", + "cpu_performance_cores": "unknown", + "cpu_efficiency_cores": "unknown", + "gpu_name": "unknown", + "memory": "unknown", + } + try: + cpu_out = subprocess.check_output( + ["wmic", "cpu", "get", "Name,NumberOfCores", "/format:csv"], + text=True, + stderr=subprocess.DEVNULL, + ) + lines = [l for l in cpu_out.splitlines() if l.strip() and not l.strip().startswith("Node")] + if lines: + parts = lines[0].split(",") + if len(parts) >= 3: + info["cpu_total_cores"] = parts[1].strip() + info["chip"] = parts[2].strip() + except Exception: + pass + # Memory + try: + mem_out = subprocess.check_output( + ["wmic", "ComputerSystem", "get", "TotalPhysicalMemory", "/format:csv"], + text=True, + stderr=subprocess.DEVNULL, + ) + lines = [l for l in mem_out.splitlines() if l.strip() and not l.strip().startswith("Node")] + if lines: + parts = lines[0].split(",") + if len(parts) >= 2: + mem_gb = int(parts[1].strip()) / 1024**3 + info["memory"] = f"{mem_gb:.1f} GB" + except Exception: + pass + # GPU via nvidia-smi (also available on Windows) + try: + gpu_out = subprocess.check_output( + ["nvidia-smi", "--query-gpu=name,memory.total", "--format=csv,noheader"], + text=True, + stderr=subprocess.DEVNULL, + ).strip() + if gpu_out: + first_line = gpu_out.splitlines()[0] + parts = [p.strip() for p in first_line.split(",")] + info["gpu_name"] = parts[0] + if len(parts) > 1: + info["gpu_memory"] = parts[1] + except Exception: + pass + return info + + +@lru_cache(maxsize=1) +def get_device_info_dict() -> dict[str, str]: + """Collect static hardware metadata (chip, cores, GPU, memory) for run records.""" + base: dict[str, str] = {"platform": platform.platform()} + if _is_macos(): + base.update(_get_device_info_macos()) + elif _is_linux(): + base.update(_get_device_info_linux()) + elif _is_windows(): + base.update(_get_device_info_windows()) + return base + + +def get_device_info_line() -> str: + d = get_device_info_dict() + if _is_macos(): + # Build core-type summary dynamically so M5 (super+performance) is shown correctly + if d.get("cpu_core_type_1") and d.get("cpu_core_type_2"): + t1 = d["cpu_core_type_1"][0].upper() + t2 = d["cpu_core_type_2"][0].upper() + core_summary = f"{d['cpu_core_count_1']}{t1}+{d['cpu_core_count_2']}{t2}" + elif ( + d.get("cpu_performance_cores") != "unknown" + and d.get("cpu_efficiency_cores") != "unknown" + ): + core_summary = f"{d['cpu_performance_cores']}P+{d['cpu_efficiency_cores']}E" + else: + core_summary = "unknown" + return ( + f"Device: {d.get('chip', 'unknown')} | " + f"CPU: {d.get('cpu_total_cores', 'unknown')} cores " + f"({core_summary}) | " + f"GPU: {d.get('gpu_cores', 'unknown')} cores | " + f"Memory: {d.get('memory', 'unknown')}" + ) + else: + gpu_part = d.get("gpu_name", "unknown") + if "gpu_memory" in d: + gpu_part += f" ({d['gpu_memory']})" + return ( + f"CPU: {d.get('chip', 'unknown')} ({d.get('cpu_total_cores', 'unknown')} cores) | " + f"GPU: {gpu_part} | " + f"Memory: {d.get('memory', 'unknown')}" + ) diff --git a/tests/benchmark/test_device_info.py b/tests/benchmark/test_device_info.py deleted file mode 100644 index 96c082d38..000000000 --- a/tests/benchmark/test_device_info.py +++ /dev/null @@ -1,88 +0,0 @@ -from __future__ import annotations - -import scripts.benchmark.core.device_info as device_info - - -def test_linux_device_info_reads_amd_visible_vram(monkeypatch): - cpuinfo = "model name\t: AMD RYZEN AI MAX+ 395 w/ Radeon 8060S\nprocessor\t: 0\n" - meminfo = "MemTotal: 32486180 kB\n" - amd_smi_metric = """ -GPU: 0 - MEM_USAGE: - TOTAL_VRAM: 98304 MB - USED_VRAM: 3603 MB - FREE_VRAM: 94701 MB - TOTAL_VISIBLE_VRAM: 98304 MB - USED_VISIBLE_VRAM: 3603 MB - FREE_VISIBLE_VRAM: 94701 MB - TOTAL_GTT: 15862 MB - USED_GTT: 158 MB - FREE_GTT: 15704 MB -""" - - def fake_open(path, *args, **kwargs): - del args, kwargs - if path == "/proc/cpuinfo": - from io import StringIO - - return StringIO(cpuinfo) - if path == "/proc/meminfo": - from io import StringIO - - return StringIO(meminfo) - raise FileNotFoundError(path) - - def fake_check_output(cmd, *args, **kwargs): - del args, kwargs - if cmd[0] == "nvidia-smi": - raise FileNotFoundError(cmd[0]) - if cmd[:2] == ["rocm-smi", "--showproductname"]: - return "Card series: AMD Radeon Graphics\n" - if cmd[:2] == ["amd-smi", "metric"]: - return amd_smi_metric - raise FileNotFoundError(cmd[0]) - - monkeypatch.setattr(device_info, "open", fake_open, raising=False) - monkeypatch.setattr(device_info.subprocess, "check_output", fake_check_output) - - info = device_info._get_device_info_linux() - - assert info["gpu_name"] == "Radeon 8060S" - assert info["gpu_memory"] == "98304 MB" - assert info["gpu_gtt_memory"] == "15862 MB" - assert info["memory"] == "31.0 GB" - - -def test_linux_device_info_reads_intel_igpu_from_lspci(monkeypatch): - cpuinfo = "model name\t: Intel(R) Core(TM) Ultra 9 185H\nprocessor\t: 0\n" - meminfo = "MemTotal: 31692928 kB\n" - lspci_out = ( - "00:02.0 VGA compatible controller: " - "Intel Corporation Meteor Lake-P [Intel Arc Graphics] (rev 08)\n" - ) - - def fake_open(path, *args, **kwargs): - del args, kwargs - from io import StringIO - - if path == "/proc/cpuinfo": - return StringIO(cpuinfo) - if path == "/proc/meminfo": - return StringIO(meminfo) - raise FileNotFoundError(path) - - def fake_check_output(cmd, *args, **kwargs): - del args, kwargs - if cmd[0] == "lspci": - return lspci_out - # No nvidia-smi, rocm-smi, or amd-smi on Intel iGPU systems - raise FileNotFoundError(cmd[0]) - - monkeypatch.setattr(device_info, "open", fake_open, raising=False) - monkeypatch.setattr(device_info.subprocess, "check_output", fake_check_output) - - info = device_info._get_device_info_linux() - - assert info["gpu_name"] == "Intel Arc Graphics" - assert info["chip"] == "Intel(R) Core(TM) Ultra 9 185H" - assert info["memory"] == "30.2 GB" diff --git a/tests/test_library_import_boundary.py b/tests/test_library_import_boundary.py new file mode 100644 index 000000000..1e974d4cc --- /dev/null +++ b/tests/test_library_import_boundary.py @@ -0,0 +1,31 @@ +"""Library-layer import boundary tests (issue #1240).""" + +from __future__ import annotations + +import ast +from pathlib import Path + +_REPO_ROOT = Path(__file__).resolve().parents[1] +_LIBRARY_PACKAGE = _REPO_ROOT / "src" / "unilab" + + +def _imports(path: Path) -> set[str]: + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + modules: set[str] = set() + for node in ast.walk(tree): + if isinstance(node, ast.ImportFrom) and node.module: + modules.add(node.module) + elif isinstance(node, ast.Import): + modules.update(alias.name for alias in node.names) + return modules + + +def test_library_does_not_import_scripts() -> None: + violations = [ + (path.relative_to(_REPO_ROOT).as_posix(), module) + for path in sorted(_LIBRARY_PACKAGE.rglob("*.py")) + for module in sorted(_imports(path)) + if module == "scripts" or module.startswith("scripts.") + ] + + assert violations == [], "src/unilab must not import scripts/ modules" diff --git a/tests/utils/test_device.py b/tests/utils/test_device.py index db3967a74..8283b7df5 100644 --- a/tests/utils/test_device.py +++ b/tests/utils/test_device.py @@ -92,3 +92,97 @@ def test_resolve_torch_device_alias_rejects_unavailable_accelerator( with pytest.raises(ValueError, match="none is available"): device_mod.resolve_torch_device_alias("gpu") + + +def test_linux_device_info_reads_amd_visible_vram(monkeypatch: pytest.MonkeyPatch) -> None: + cpuinfo = "model name\t: AMD RYZEN AI MAX+ 395 w/ Radeon 8060S\nprocessor\t: 0\n" + meminfo = "MemTotal: 32486180 kB\n" + amd_smi_metric = """ +GPU: 0 + MEM_USAGE: + TOTAL_VRAM: 98304 MB + USED_VRAM: 3603 MB + FREE_VRAM: 94701 MB + TOTAL_VISIBLE_VRAM: 98304 MB + USED_VISIBLE_VRAM: 3603 MB + FREE_VISIBLE_VRAM: 94701 MB + TOTAL_GTT: 15862 MB + USED_GTT: 158 MB + FREE_GTT: 15704 MB +""" + + def fake_open(path, *args, **kwargs): + del args, kwargs + if path == "/proc/cpuinfo": + from io import StringIO + + return StringIO(cpuinfo) + if path == "/proc/meminfo": + from io import StringIO + + return StringIO(meminfo) + raise FileNotFoundError(path) + + def fake_check_output(cmd, *args, **kwargs): + del args, kwargs + if cmd[0] == "nvidia-smi": + raise FileNotFoundError(cmd[0]) + if cmd[:2] == ["rocm-smi", "--showproductname"]: + return "Card series: AMD Radeon Graphics\n" + if cmd[:2] == ["amd-smi", "metric"]: + return amd_smi_metric + raise FileNotFoundError(cmd[0]) + + monkeypatch.setattr(device_mod, "open", fake_open, raising=False) + monkeypatch.setattr(device_mod.subprocess, "check_output", fake_check_output) + + info = device_mod._get_device_info_linux() + + assert info["gpu_name"] == "Radeon 8060S" + assert info["gpu_memory"] == "98304 MB" + assert info["gpu_gtt_memory"] == "15862 MB" + assert info["memory"] == "31.0 GB" + + +def test_linux_device_info_reads_intel_igpu_from_lspci(monkeypatch: pytest.MonkeyPatch) -> None: + cpuinfo = "model name\t: Intel(R) Core(TM) Ultra 9 185H\nprocessor\t: 0\n" + meminfo = "MemTotal: 31692928 kB\n" + lspci_out = ( + "00:02.0 VGA compatible controller: " + "Intel Corporation Meteor Lake-P [Intel Arc Graphics] (rev 08)\n" + ) + + def fake_open(path, *args, **kwargs): + del args, kwargs + from io import StringIO + + if path == "/proc/cpuinfo": + return StringIO(cpuinfo) + if path == "/proc/meminfo": + return StringIO(meminfo) + raise FileNotFoundError(path) + + def fake_check_output(cmd, *args, **kwargs): + del args, kwargs + if cmd[0] == "lspci": + return lspci_out + # No nvidia-smi, rocm-smi, or amd-smi on Intel iGPU systems + raise FileNotFoundError(cmd[0]) + + monkeypatch.setattr(device_mod, "open", fake_open, raising=False) + monkeypatch.setattr(device_mod.subprocess, "check_output", fake_check_output) + + info = device_mod._get_device_info_linux() + + assert info["gpu_name"] == "Intel Arc Graphics" + assert info["chip"] == "Intel(R) Core(TM) Ultra 9 185H" + assert info["memory"] == "30.2 GB" + + +def test_get_device_info_dict_reports_core_fields() -> None: + info = device_mod.get_device_info_dict() + assert info["platform"] + assert "chip" in info + assert "cpu_total_cores" in info + assert "memory" in info + assert "gpu_name" in info or "gpu_cores" in info diff --git a/tests/utils/test_experiment_tracking.py b/tests/utils/test_experiment_tracking.py index 87c871b29..7c683ae20 100644 --- a/tests/utils/test_experiment_tracking.py +++ b/tests/utils/test_experiment_tracking.py @@ -332,17 +332,10 @@ def test_build_wandb_settings_defaults_for_shared_workspace(): assert "mujoco" in settings["tags"] -def test_experiment_device_info_uses_benchmark_helper(): - helper_path = experiment_module._benchmark_device_info_path() - assert helper_path is not None - assert helper_path.name == "device_info.py" - - info = experiment_module.get_device_info_dict() - assert info["platform"] - assert "chip" in info - assert "cpu_total_cores" in info - assert "memory" in info - assert "gpu_name" in info or "gpu_cores" in info +def test_experiment_device_info_uses_library_helper(): + import unilab.utils.device as device_module + + assert experiment_module.get_device_info_dict is device_module.get_device_info_dict def test_experiment_tracker_writes_local_run_files(tmp_path, monkeypatch):