chemgraph-loop / src /ui /system_info.py
rockyaaos's picture
ChemGraph Loop: guarded real-agent API (EMT/TBLite single-point energy)
c509967 verified
Raw
History Blame Contribute Delete
7.17 kB
"""Host and build metadata collection for the ChemGraph UI sidebar.
All system-introspection helpers are grouped here so they can be tested
independently of the Streamlit rendering layer.
"""
import os
import platform
import socket
import subprocess
from pathlib import Path
from typing import Dict, Optional
import streamlit as st
import chemgraph as chemgraph_pkg
from chemgraph import __version__ as chemgraph_version
# ---------------------------------------------------------------------------
# Internal helpers
# ---------------------------------------------------------------------------
def _run_command(cmd: list[str], cwd: Optional[Path] = None, timeout: int = 2) -> str:
"""Run a shell command and return stripped stdout.
Parameters
----------
cmd : list[str]
Command and arguments to execute.
cwd : pathlib.Path, optional
Working directory for the command.
timeout : int, optional
Timeout in seconds.
Returns
-------
str
Stripped stdout, or an empty string on failure.
"""
try:
completed = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=timeout,
check=True,
cwd=str(cwd) if cwd else None,
)
except Exception:
return ""
return completed.stdout.strip()
def _find_repo_root(start: Path) -> Optional[Path]:
"""Find git repo root by walking up parents from a starting path.
Parameters
----------
start : pathlib.Path
Starting path for the search.
Returns
-------
pathlib.Path or None
Repository root, or ``None`` when not found.
"""
start = start.resolve()
candidates = [start] + list(start.parents)
for candidate in candidates:
if (candidate / ".git").exists():
return candidate
return None
def _format_bytes(num_bytes: int) -> str:
"""Format a byte count with binary-size units.
Parameters
----------
num_bytes : int
Byte count to format.
Returns
-------
str
Human-readable size, or ``"Unknown"`` for non-positive values.
"""
if num_bytes <= 0:
return "Unknown"
units = ["B", "KB", "MB", "GB", "TB", "PB"]
size = float(num_bytes)
for unit in units:
if size < 1024.0 or unit == units[-1]:
return f"{size:.1f} {unit}"
size /= 1024.0
return "Unknown"
def _get_total_memory_bytes() -> int:
"""Return total system memory in bytes when available."""
try:
page_size = os.sysconf("SC_PAGE_SIZE")
phys_pages = os.sysconf("SC_PHYS_PAGES")
total = int(page_size) * int(phys_pages)
if total > 0:
return total
except Exception:
pass
meminfo = Path("/proc/meminfo")
if meminfo.exists():
try:
for line in meminfo.read_text().splitlines():
if line.startswith("MemTotal:"):
kb = int(line.split()[1])
return kb * 1024
except Exception:
return 0
return 0
def _get_cpu_model() -> str:
"""Try to get a human-readable CPU model name."""
cpuinfo = Path("/proc/cpuinfo")
if cpuinfo.exists():
try:
for line in cpuinfo.read_text().splitlines():
if line.lower().startswith("model name"):
parts = line.split(":", 1)
if len(parts) == 2:
return parts[1].strip()
except Exception:
pass
cpu_name = platform.processor().strip()
if cpu_name:
return cpu_name
return platform.machine()
def _get_gpu_summary() -> str:
"""Return GPU summary from nvidia-smi when available."""
output = _run_command(
[
"nvidia-smi",
"--query-gpu=name,memory.total",
"--format=csv,noheader,nounits",
]
)
if not output:
return "No GPU detected"
entries = []
for line in output.splitlines():
parts = [part.strip() for part in line.split(",")]
if len(parts) >= 2:
name, mem_mib = parts[0], parts[1]
entries.append(f"{name} ({mem_mib} MiB)")
elif parts:
entries.append(parts[0])
return "; ".join(entries) if entries else "No GPU detected"
# ---------------------------------------------------------------------------
# Public API (cached)
# ---------------------------------------------------------------------------
@st.cache_data(ttl=60)
def get_host_info() -> Dict[str, str]:
"""Collect host metadata for sidebar display."""
return {
"hostname": socket.gethostname(),
"platform": f"{platform.system()} {platform.release()}",
"cpu_model": _get_cpu_model(),
"cpu_cores": str(os.cpu_count() or "Unknown"),
"memory_total": _format_bytes(_get_total_memory_bytes()),
"gpu": _get_gpu_summary(),
}
@st.cache_data(ttl=60)
def get_build_info() -> Dict[str, str]:
"""Collect app and repository metadata for sidebar display."""
app_file = Path(__file__).resolve()
chemgraph_file = Path(chemgraph_pkg.__file__).resolve()
repo_root = _find_repo_root(app_file) or _find_repo_root(chemgraph_file)
commit = "Unknown"
commit_date = "Unknown"
branch = "Unknown"
if repo_root:
commit = (
_run_command(["git", "rev-parse", "--short", "HEAD"], cwd=repo_root)
or "Unknown"
)
commit_date = (
_run_command(
["git", "show", "-s", "--format=%cd", "--date=iso", "HEAD"],
cwd=repo_root,
)
or "Unknown"
)
branch = (
_run_command(
["git", "rev-parse", "--abbrev-ref", "HEAD"], cwd=repo_root
)
or "Unknown"
)
return {
"chemgraph_version": str(chemgraph_version),
"commit": commit,
"commit_date": commit_date,
"branch": branch,
"chemgraph_file": str(chemgraph_file),
}
def render_sidebar_host_and_build_info() -> None:
"""Render host and build metadata blocks in the left sidebar."""
host_info = get_host_info()
build_info = get_build_info()
with st.sidebar.expander("\U0001f5a5\ufe0f Host Info", expanded=False):
st.markdown(f"**Hostname:** `{host_info['hostname']}`")
st.markdown(f"**OS:** `{host_info['platform']}`")
st.markdown(f"**CPU:** `{host_info['cpu_model']}`")
st.markdown(f"**CPU Cores:** `{host_info['cpu_cores']}`")
st.markdown(f"**Memory:** `{host_info['memory_total']}`")
st.markdown(f"**GPU:** `{host_info['gpu']}`")
with st.sidebar.expander("\U0001f4e6 Build Info", expanded=False):
st.markdown(f"**ChemGraph Version:** `{build_info['chemgraph_version']}`")
st.markdown(f"**Branch:** `{build_info['branch']}`")
st.markdown(f"**Commit:** `{build_info['commit']}`")
st.markdown(f"**Commit Date:** `{build_info['commit_date']}`")
st.markdown(f"**ChemGraph File:** `{build_info['chemgraph_file']}`")