"""Per-job system manifest — runtime environment snapshot for ``run_job``.
A ``run_job(output="x")`` call writes ``x.out`` (the human-readable text
log) and, since v0.5.1, ``x.system`` (a TOML manifest pinning *which
machine produced these numbers*). The two files are siblings: the
``.out`` carries the chemistry, the ``.system`` carries the hardware,
linked-library, and runtime context needed to interpret a wall-time
figure or reproduce a calculation on a different box.
Without the manifest, an ``.out`` says ``SCF total: 0.015 s`` with no
indication whether that's an Apple M2 Pro at 12 OMP threads or an
8-year-old Xeon at 1 thread — making bundled reference outputs much
less useful for newcomers comparing their own runs.
Public API
----------
``system_info(*, record_hostname=True)``
Collect the runtime environment as a nested dict shaped like the
TOML output. Pure-Python; never raises (every probe falls back to
``"unknown"`` on failure). The hostname can be redacted via the
``record_hostname=False`` kwarg or the ``VIBEQC_NO_HOSTNAME=1``
environment variable — engineering's bundled docs runs use the
latter so machine names don't leak into the public bundle.
``write_system_manifest(out_path, wall_seconds, basename, *, record_hostname=True)``
Render ``system_info()`` plus per-run fields (``wall_seconds``,
``basename``, ISO timestamp) as TOML and write to
``out_path.with_suffix('.system')``. Returns the path written.
The TOML shape is fixed and machine-readable — every documented section
+ key is always present, even when the underlying probe couldn't
resolve a value (you get ``"unknown"`` rather than a missing key). This
keeps downstream parsers simple.
"""
from __future__ import annotations
import datetime as _dt
import hashlib
import json
import os
import platform
import socket
import subprocess
import sys
from pathlib import Path
from typing import Any, Optional
from .banner import VIBEQC_VERSION, build_info, codename_for_version
__all__ = [
"system_info",
"write_system_manifest",
"run_fingerprint",
]
_UNKNOWN = "unknown"
def _safe(callable_, default=_UNKNOWN):
"""Run ``callable_()`` and return its result, swallowing any
exception and returning ``default`` instead. Probes must never
raise — a missing CPU model shouldn't abort a calculation."""
try:
result = callable_()
except Exception:
return default
return result if result else default
def _cpu_model() -> str:
"""Best-effort CPU model string. On macOS, ``platform.processor()``
returns generic strings like ``"arm"`` or ``"i386"`` — useless for
distinguishing M1 vs M2 vs M3 Pro/Max — so we prefer the
``sysctl machdep.cpu.brand_string`` probe there. On Linux,
``/proc/cpuinfo`` is the authority. ``platform.processor()`` is
only used as a last-resort fallback (and only when it returns
something more specific than the platform-generic value).
Falls back to ``"unknown"``."""
if sys.platform == "darwin":
def _sysctl():
return subprocess.check_output(
["sysctl", "-n", "machdep.cpu.brand_string"],
stderr=subprocess.DEVNULL,
text=True,
timeout=2.0,
).strip()
out = _safe(_sysctl, default="")
if out:
return out
if sys.platform.startswith("linux"):
def _cpuinfo():
with open("/proc/cpuinfo", "r", encoding="ascii",
errors="replace") as f:
for line in f:
if line.startswith("model name"):
return line.split(":", 1)[1].strip()
return ""
out = _safe(_cpuinfo, default="")
if out:
return out
proc = _safe(platform.processor, default="")
if proc and proc != "unknown":
return proc
return _UNKNOWN
def _os_pretty() -> str:
"""Human-friendly OS name + version (``"macOS 14.4"``,
``"Ubuntu 22.04"``). Falls back to ``"<system> <release>"``."""
if sys.platform == "darwin":
ver = _safe(lambda: platform.mac_ver()[0], default="")
if ver:
return f"macOS {ver}"
if sys.platform.startswith("linux"):
def _osrelease():
with open("/etc/os-release", "r", encoding="utf-8") as f:
fields: dict[str, str] = {}
for line in f:
if "=" in line:
k, v = line.rstrip("\n").split("=", 1)
fields[k] = v.strip().strip('"')
return fields.get("PRETTY_NAME") or fields.get("NAME") or ""
out = _safe(_osrelease, default="")
if out:
return out
sysname = _safe(platform.system, default="")
rel = _safe(platform.release, default="")
return (f"{sysname} {rel}".strip()) or _UNKNOWN
def _omp_threads() -> int:
"""OpenMP thread count vibe-qc would actually use for this job —
queried via the C++ ``get_num_threads()`` so it agrees with the
figure the SCF driver actually saw. Falls back to 0 (sentinel:
"unknown") if the native module isn't importable."""
try:
from ._vibeqc_core import get_num_threads
return int(get_num_threads())
except Exception:
return 0
def _libecpint_version() -> str:
try:
from ._vibeqc_core import libecpint_version
return str(libecpint_version())
except Exception:
return _UNKNOWN
_GIB = 1024 ** 3
def _total_memory_bytes() -> int:
"""Best-effort total RAM probe. psutil first (when available),
then platform-specific (sysctl on macOS, /proc/meminfo on Linux,
sysconf cross-platform). Returns 0 when no probe works."""
try:
import psutil # type: ignore[import-not-found]
return int(psutil.virtual_memory().total)
except ImportError:
pass
if sys.platform == "darwin":
try:
out = subprocess.check_output(
["sysctl", "-n", "hw.memsize"],
stderr=subprocess.DEVNULL,
text=True,
timeout=2.0,
).strip()
return int(out)
except (subprocess.CalledProcessError, FileNotFoundError,
subprocess.TimeoutExpired, OSError, ValueError):
pass
if sys.platform.startswith("linux"):
try:
with open("/proc/meminfo", "r", encoding="ascii") as f:
for line in f:
if line.startswith("MemTotal:"):
return int(line.split()[1]) * 1024
except OSError:
pass
try:
if hasattr(os, "sysconf"):
page = os.sysconf("SC_PAGE_SIZE")
n_pages = os.sysconf("SC_PHYS_PAGES")
if page > 0 and n_pages > 0:
return int(page) * int(n_pages)
except (ValueError, OSError):
pass
return 0
def _memory_gb() -> tuple[float, float]:
"""``(total_gb, available_gb)``. Reuses the existing best-effort
probe in :mod:`vibeqc.memory` for *available* memory so we don't
duplicate platform handling. Total is its own probe (psutil →
``sysctl hw.memsize`` on macOS → ``/proc/meminfo`` on Linux →
``sysconf`` fallback). Returns ``(0.0, 0.0)`` if nothing succeeds —
the field stays present in the manifest, the value just signals
"could not probe" the same way the memory-estimator block does."""
total = _total_memory_bytes()
try:
from .memory import available_memory_bytes
avail = available_memory_bytes()
except Exception:
avail = 0
return (total / _GIB, avail / _GIB)
def _cpu_cores() -> tuple[int, int]:
"""``(physical_cores, logical_cores)``. Logical from
:func:`os.cpu_count`; physical from ``sysctl hw.physicalcpu`` on
macOS, ``/proc/cpuinfo`` core-id deduplication on Linux, or
``logical`` (== physical with HT off) as a last resort."""
logical = int(_safe(lambda: os.cpu_count() or 0, default=0))
physical = 0
if sys.platform == "darwin":
try:
out = subprocess.check_output(
["sysctl", "-n", "hw.physicalcpu"],
stderr=subprocess.DEVNULL,
text=True,
timeout=2.0,
).strip()
physical = int(out)
except (subprocess.CalledProcessError, FileNotFoundError,
subprocess.TimeoutExpired, OSError, ValueError):
physical = 0
elif sys.platform.startswith("linux"):
try:
with open("/proc/cpuinfo", "r", encoding="ascii",
errors="replace") as f:
# Each (physical id, core id) pair counts a physical
# core. Hyper-threaded siblings share both fields.
seen: set[tuple[str, str]] = set()
phys_id = ""
core_id = ""
for line in f:
if line.startswith("physical id"):
phys_id = line.split(":", 1)[1].strip()
elif line.startswith("core id"):
core_id = line.split(":", 1)[1].strip()
elif line.strip() == "":
if phys_id or core_id:
seen.add((phys_id, core_id))
phys_id = core_id = ""
if phys_id or core_id:
seen.add((phys_id, core_id))
physical = len(seen)
except OSError:
physical = 0
if physical <= 0:
physical = logical
return (physical, logical)
def _hostname_redacted() -> bool:
"""Honor ``VIBEQC_NO_HOSTNAME=1`` (or any non-empty, non-"0"
value) as a global opt-out — orthogonal to the function-call
kwarg. Either lever produces ``hostname = "<redacted>"``."""
val = os.environ.get("VIBEQC_NO_HOSTNAME", "").strip().lower()
return val not in ("", "0", "false", "no")
[docs]
def system_info(*, record_hostname: bool = True) -> dict[str, Any]:
"""Collect the runtime environment as a nested dict.
The returned dict has the same shape as the on-disk ``.system``
TOML — top-level keys are sections (``vibeqc``, ``host``, ``cpu``,
``memory``, ``python``, ``libraries``), each mapping to the keys
documented in :mod:`vibeqc.system_info`.
Parameters
----------
record_hostname
If ``False`` (or if the ``VIBEQC_NO_HOSTNAME=1`` env var is
set), the ``host.hostname`` field is set to ``"<redacted>"``.
The field is always present so the TOML shape stays stable
for parsers; we never *omit* it.
"""
info = build_info()
git_sha = info.get("sha") or _UNKNOWN if info else _UNKNOWN
git_branch = info.get("branch") or _UNKNOWN if info else _UNKNOWN
is_release = bool(info.get("is_release")) if info else False
if not record_hostname or _hostname_redacted():
hostname = "<redacted>"
else:
hostname = _safe(socket.gethostname)
arch = _safe(platform.machine, default=_UNKNOWN)
os_name = _safe(platform.system, default=_UNKNOWN)
os_release = _safe(platform.release, default=_UNKNOWN)
physical, logical = _cpu_cores()
total_gb, avail_gb = _memory_gb()
py_version = _safe(platform.python_version, default=_UNKNOWN)
py_impl = _safe(platform.python_implementation, default=_UNKNOWN)
py_exe = _safe(lambda: sys.executable, default=_UNKNOWN)
# Linked-library versions: source from banner.library_versions so
# the manifest matches the banner's "linked:" line exactly. Add
# libecpint (banner doesn't include it, but it's part of the
# linked-native-deps story).
from .banner import library_versions
lv = library_versions()
libs = {
"libint": lv.get("libint", _UNKNOWN),
"libxc": lv.get("libxc", _UNKNOWN),
"spglib": lv.get("spglib", _UNKNOWN),
"libecpint": _libecpint_version(),
}
return {
"vibeqc": {
"version": VIBEQC_VERSION,
"codename": codename_for_version(VIBEQC_VERSION) or "",
"git_sha": git_sha,
"git_branch": git_branch,
"is_release": is_release,
},
"host": {
"hostname": hostname,
"os": os_name,
"os_release": os_release,
"os_pretty": _os_pretty(),
"arch": arch,
},
"cpu": {
"model": _cpu_model(),
"physical_cores": physical,
"logical_cores": logical,
"omp_threads_used": _omp_threads(),
},
"memory": {
"total_gb": round(total_gb, 2),
"available_gb": round(avail_gb, 2),
},
"python": {
"version": py_version,
"implementation": py_impl,
"executable": py_exe,
},
"libraries": libs,
}
# ---------------------------------------------------------------------------
# Hand-rolled TOML emitter
# ---------------------------------------------------------------------------
#
# The manifest is fixed-shape (keys and types are known statically), so a
# general TOML emitter would be overkill. We hand-format with a tiny
# value-quoting helper. Round-tripping is verified in
# tests/test_system_manifest.py via stdlib ``tomllib.loads``.
def _toml_str(s: str) -> str:
"""Quote a string as a TOML basic string. Escapes ``\\`` and ``"``
plus control chars per the TOML 1.0 spec."""
out = []
for ch in s:
if ch == "\\":
out.append("\\\\")
elif ch == '"':
out.append('\\"')
elif ch == "\n":
out.append("\\n")
elif ch == "\r":
out.append("\\r")
elif ch == "\t":
out.append("\\t")
elif ord(ch) < 0x20:
out.append(f"\\u{ord(ch):04X}")
else:
out.append(ch)
return '"' + "".join(out) + '"'
def _toml_value(v: Any) -> str:
if isinstance(v, bool):
return "true" if v else "false"
if isinstance(v, int):
return str(v)
if isinstance(v, float):
# Always emit a decimal point so the value parses as float, not int.
s = repr(v)
if "." not in s and "e" not in s and "E" not in s:
s += ".0"
return s
return _toml_str(str(v))
# Section ordering for the emitted file — fixed so diffs are stable
# across runs and across machines. Within a section, keys retain the
# dict-insertion order from system_info() above.
_SECTION_ORDER = (
"vibeqc", "host", "cpu", "memory", "python", "libraries", "run",
)
# ---------------------------------------------------------------------------
# Run fingerprint — deterministic identity hash for a calculation
# ---------------------------------------------------------------------------
#
# Goal: a short hex string that answers "is this the same calculation
# as that one?" without diffing files. Two runs with the same molecule,
# basis, method, functional, charge, multiplicity, and explicitly-passed
# options must produce the same fingerprint, on any machine, in any
# Python version. Two runs that differ in any of those inputs must
# produce different fingerprints.
#
# We hash a *canonicalised* JSON representation rather than the raw
# objects so the result is independent of: dict iteration order, float
# repr() drift, integer vs float coordinate types, and BLAS / NumPy
# build-specific str() formatting. Truncate to 16 hex chars (64 bits)
# — collision probability is negligible for the population a single
# user / project produces (millions of jobs over a career), and a
# short string fits one TOML line plus a "look at the .system file"
# bug report URL without wrapping.
_GEOM_QUANT = 1.0e-10 # bohr — tighter than any meaningful XYZ precision
def _round_coord(x: float) -> float:
"""Round to ``_GEOM_QUANT`` and normalize -0.0 → 0.0 so two
geometries that are arithmetically identical hash identically
even when one came in as a negative-zero float."""
q = round(float(x) / _GEOM_QUANT) * _GEOM_QUANT
return 0.0 if q == 0.0 else q
def _canonical_options(options: Optional[dict]) -> dict:
"""Filter ``options`` to a JSON-serialisable canonical dict.
We keep public, scalar values (bool / int / float / str) and
flat sequences thereof. Anything else (NumPy arrays, callables,
pybind11 result handles) is dropped — the fingerprint is
documenting *intent*, not internal scratch state, and a callback
function would defeat determinism by design.
Sorted by key so dict iteration order doesn't change the hash.
"""
if not options:
return {}
canon: dict[str, Any] = {}
for key in sorted(options):
if key.startswith("_"):
continue
v = options[key]
if isinstance(v, bool) or isinstance(v, (int, float, str)):
canon[key] = v
elif v is None:
canon[key] = None
elif isinstance(v, (list, tuple)):
try:
canon[key] = [
x if isinstance(x, (bool, int, float, str, type(None)))
else str(x)
for x in v
]
except Exception:
continue
return canon
[docs]
def run_fingerprint(
molecule,
*,
basis: str,
method: str,
functional: Optional[str] = None,
charge: Optional[int] = None,
multiplicity: Optional[int] = None,
options: Optional[dict] = None,
) -> str:
"""Deterministic SHA-256 (truncated to 16 hex chars) over the
canonicalised inputs that define a calculation.
Stable across runs, machines, and Python versions — the same
inputs always produce the same string, and any change to geometry,
basis, method, functional, charge, multiplicity, or recognized
options changes it.
Use as a cache key, a bug-report identifier, or to answer
"is the run on disk the same calculation as the one I'm about to
launch?" without diffing files.
Parameters
----------
molecule
A :class:`Molecule` (or any object exposing the same
``atoms`` / ``charge`` / ``multiplicity`` interface). Atomic
coordinates are rounded to ``1e-10`` bohr before hashing — far
below any meaningful geometric precision, but enough to
absorb pure float-repr drift.
basis
Basis-set name. Lowercased before hashing so ``"6-31G*"`` and
``"6-31g*"`` collide.
method
``"rhf"`` / ``"uhf"`` / ``"rks"`` / ``"uks"`` / ``"mp2"`` /
``"ump2"`` / a periodic-method label. Lowercased before
hashing.
functional
XC functional for KS methods. ``None`` means "not applicable
/ not specified" — both encoded the same way so a HF + DFT
comparison swings the fingerprint.
charge, multiplicity
Optional explicit overrides; default to the values on
``molecule``. Pass these only when the calculation will
actually run on a different charge/multiplicity than the
molecule object reports (rare; the molecule normally carries
the right values).
options
Free-form dict of public attributes describing the calculation
(DIIS settings, max iterations, …). Filtered to JSON-scalar
keys via :func:`_canonical_options`. Pass ``None`` if no
options matter for identity.
Returns
-------
str
16-hex-character lowercase string (a 64-bit prefix of the
SHA-256). Stable surface — downstream consumers compare for
equality, not for length.
"""
atoms = []
for atom in getattr(molecule, "atoms", []):
z = int(atom.Z)
xyz = [_round_coord(c) for c in atom.xyz]
atoms.append([z, xyz])
payload = {
"geometry": atoms,
"basis": str(basis).strip().lower(),
"method": str(method).strip().lower(),
"functional": (
str(functional).strip().lower() if functional else None
),
"charge": int(
charge if charge is not None else getattr(molecule, "charge", 0)
),
"multiplicity": int(
multiplicity if multiplicity is not None
else getattr(molecule, "multiplicity", 1)
),
"options": _canonical_options(options),
}
# ``sort_keys=True`` so dict iteration order on the top-level
# payload (and any nested options dict) doesn't affect the hash.
# ``separators=`` strips whitespace so the byte-stream is the same
# under any future Python json formatting tweaks.
blob = json.dumps(
payload, sort_keys=True, separators=(",", ":"),
ensure_ascii=True,
).encode("ascii")
return hashlib.sha256(blob).hexdigest()[:16]
def _format_manifest(info: dict[str, Any]) -> str:
"""Render the info dict as the on-disk TOML manifest."""
lines = [
"# vibe-qc system manifest — written alongside output-<job>.out by run_job(...).",
"# Captures the runtime environment so bundled reference outputs are",
"# reproducible and wall-time numbers are interpretable.",
"",
]
for section in _SECTION_ORDER:
if section not in info:
continue
lines.append(f"[{section}]")
for key, value in info[section].items():
lines.append(f"{key:<14s} = {_toml_value(value)}")
lines.append("")
return "\n".join(lines)
[docs]
def write_system_manifest(
out_path: os.PathLike | str,
wall_seconds: float,
basename: str,
*,
record_hostname: bool = True,
fingerprint: Optional[str] = None,
) -> Path:
"""Write the per-job system manifest next to ``out_path``.
The manifest path is ``out_path.with_suffix('.system')`` — pass
either the ``.out`` file path or the bare stem; both produce the
same target.
Parameters
----------
out_path
The text-output path (or stem) the calculation produced.
``write_system_manifest`` writes to the sibling ``.system``
path. The parent directory must already exist (``run_job``
creates it).
wall_seconds
Total job wall-clock time in seconds. Recorded in the
``[run]`` section so a reader can pair the wall-time with the
host hardware.
basename
Path stem identifying the job (e.g. ``"input-h2o-rhf"``).
Recorded as ``run.basename`` so the manifest is self-
identifying without needing to read its filename.
record_hostname
Forwarded to :func:`system_info`. ``False`` (or
``VIBEQC_NO_HOSTNAME=1`` in the env) writes
``hostname = "<redacted>"``.
fingerprint
Optional :func:`run_fingerprint` value for this job. When
passed, it is recorded as ``[run].fingerprint`` so a reader
can answer "is this the same calculation?" without diffing
files. ``None`` (default) omits the field — back-compatible
with manifests written by v0.5.1 / v0.5.2 / v0.5.3.
Returns
-------
pathlib.Path
The path of the written ``.system`` file.
"""
info = system_info(record_hostname=record_hostname)
run_section: dict[str, Any] = {
"timestamp_iso": _dt.datetime.now().astimezone().isoformat(
timespec="seconds",
),
"wall_seconds": float(wall_seconds),
"basename": str(basename),
}
if fingerprint is not None:
run_section["fingerprint"] = str(fingerprint)
info["run"] = run_section
target = Path(os.fspath(out_path)).with_suffix(".system")
target.write_text(_format_manifest(info), encoding="utf-8")
return target