#!/usr/bin/env python3 """Kernodeck diagnostic v1.0.0 — original code, MIT license. Default: require a usable CUDA/ROCm GPU. Use --device cpu deliberately for CPU. No network call, environment dump, host name, user path, or raw exception output. The small, fixed calculation checks execution and gradients; it is not a benchmark. """ from __future__ import annotations import argparse import contextlib import io import json import re import subprocess import sys from pathlib import Path VERSION = "1.0.0" MESSAGES = { "CPU_CHECK_PASSED": "Calcul et gradient vérifiés sur CPU uniquement.", "GPU_CHECK_PASSED": "Calcul et gradient vérifiés sur le GPU sélectionné.", "CLI_ARGUMENTS_INVALID": "Arguments invalides. Consultez --help ; aucune valeur saisie n’est reproduite.", "TORCH_MISSING": "PyTorch est absent du Python utilisé pour lancer ce script.", "TORCH_IMPORT_FAILED": "PyTorch a été trouvé mais son import a échoué. Vérifiez le paquet et ses dépendances.", "GPU_BACKEND_ABSENT": "Ce paquet PyTorch ne déclare ni backend CUDA ni backend HIP/ROCm.", "GPU_UNAVAILABLE": "Le paquet déclare un backend GPU, mais aucun GPU utilisable n’est visible dans ce processus.", "DEVICE_INDEX_INVALID": "L’index demandé ne correspond à aucun périphérique visible.", "CHECK_FAILED": "Le petit calcul ou son gradient ne correspond pas au résultat attendu.", "OUT_OF_MEMORY": "Une allocation a manqué de mémoire pendant ce petit contrôle.", "RUNTIME_ERROR": "Une opération du backend a échoué à l’étape indiquée. Le détail brut n’est pas exporté.", "TIMEOUT": "Le processus de contrôle a dépassé le délai autorisé et a été arrêté.", "WORKER_FAILED": "Le processus de contrôle s’est arrêté sans rapport JSON valide.", "OUTPUT_WRITE_FAILED": "Le fichier demandé n’a pas été écrit : il existe déjà ou n’est pas accessible.", "INTERRUPTED": "Le contrôle a été interrompu.", } def report_base(requested: str) -> dict: """Only fixed, documented fields belong in the report.""" return { "schema": "kernova-diagnostic-v1", "script_version": VERSION, "requested_device": requested, "status": "failed", "code": "WORKER_FAILED", "exit_code": 11, "message": MESSAGES["WORKER_FAILED"], "stage": "startup", "runtime": { "python": ".".join(str(part) for part in sys.version_info[:3]), "os_family": {"win32": "Windows", "linux": "Linux", "darwin": "macOS"}.get(sys.platform, "other"), }, "pytorch": None, "execution": None, } def finish(report: dict, code: str, exit_code: int, stage: str) -> dict: report.update(status="passed" if exit_code == 0 else "failed", code=code, exit_code=exit_code, message=MESSAGES[code], stage=stage) return report def version_text(value) -> str | None: """Keep release numbers, not arbitrary package strings or local build labels.""" if value is None: return None text = str(value) if re.fullmatch(r"[0-9]+(?:\.[0-9]+){1,3}(?:(?:a|b|rc)[0-9]+)?(?:\+(?:cpu|cu[0-9]+|rocm[0-9.]+))?", text): return text return "custom_or_unrecognized" def model_text(value) -> str: text = str(value) if len(text) <= 100 and re.fullmatch(r"(?:NVIDIA|AMD|GeForce|Tesla|Quadro|Radeon|Instinct)[A-Za-z0-9 ()_.+-]*", text): return text return "model_name_not_exported" def backend_kind(cuda_version, hip_version) -> str: # ROCm intentionally shares torch.cuda and the device name "cuda". if hip_version is not None: return "rocm" if cuda_version is not None: return "cuda" return "cpu" def probe(torch, requested: str, index: int) -> dict: """Run the fixed check. The injected module parameter also allows honest unit tests.""" report = report_base(requested) stage = "backend_detection" try: cuda_version = torch.version.cuda hip_version = torch.version.hip backend = backend_kind(cuda_version, hip_version) report["pytorch"] = { "version": version_text(torch.__version__), "cuda_build": version_text(cuda_version), "hip_build": version_text(hip_version), "backend": backend, "gpu_available": None, "visible_device_count": None, } # CPU mode deliberately does not initialize/query the GPU driver. if requested == "gpu": if backend == "cpu": report["pytorch"].update(gpu_available=False, visible_device_count=0) return finish(report, "GPU_BACKEND_ABSENT", 5, stage) stage = "device_visibility" available = bool(torch.cuda.is_available()) count = int(torch.cuda.device_count()) report["pytorch"].update(gpu_available=available, visible_device_count=count) if not available or count == 0: return finish(report, "GPU_UNAVAILABLE", 6, stage) if index >= count: return finish(report, "DEVICE_INDEX_INVALID", 7, stage) stage = "device_properties" properties = torch.cuda.get_device_properties(index) device = torch.device("cuda", index) execution = {"device": "cuda:" + str(index), "backend": backend, "model": model_text(properties.name), "total_memory_bytes": int(properties.total_memory)} else: device = torch.device("cpu") execution = {"device": "cpu", "backend": "cpu", "model": None, "total_memory_bytes": None} report["execution"] = execution execution.update(dtype="float32", matrix_shape=[2, 2], forward_verified=False, backward_verified=False, expected_loss=196.0, observed_loss=None) stage = "allocation" left = torch.tensor([[1.0, 2.0], [3.0, 4.0]], device=device, dtype=torch.float32, requires_grad=True) right = torch.tensor([[2.0, 0.0], [1.0, 2.0]], device=device, dtype=torch.float32) stage = "forward" product = left @ right loss = product.square().sum() stage = "backward" loss.backward() stage = "synchronization" if requested == "gpu": torch.cuda.synchronize(device) stage = "validation" expected_product = torch.tensor([[4.0, 4.0], [10.0, 8.0]], dtype=torch.float32) expected_gradient = torch.tensor([[16.0, 24.0], [40.0, 52.0]], dtype=torch.float32) forward_ok = bool(torch.equal(product.detach().cpu(), expected_product)) backward_ok = left.grad is not None and bool(torch.equal(left.grad.detach().cpu(), expected_gradient)) observed = float(loss.detach().cpu().item()) execution.update(forward_verified=forward_ok, backward_verified=backward_ok, observed_loss=observed if observed == 196.0 else None) # These small integers are exactly representable here; this is not a # promise of bitwise reproducibility for arbitrary floating-point models. if not forward_ok or not backward_ok or observed != 196.0: return finish(report, "CHECK_FAILED", 8, stage) return finish(report, "GPU_CHECK_PASSED" if requested == "gpu" else "CPU_CHECK_PASSED", 0, "complete") except Exception as error: oom_class = getattr(torch, "OutOfMemoryError", ()) if isinstance(oom_class, type) and isinstance(error, oom_class): return finish(report, "OUT_OF_MEMORY", 9, stage) return finish(report, "RUNTIME_ERROR", 9, stage) def worker(requested: str, index: int) -> dict: report = report_base(requested) # The parent captures native library output as well. None is copied to JSON. with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()): try: import torch except ModuleNotFoundError as error: code = "TORCH_MISSING" if error.name == "torch" else "TORCH_IMPORT_FAILED" return finish(report, code, 3 if code == "TORCH_MISSING" else 4, "import") except Exception: return finish(report, "TORCH_IMPORT_FAILED", 4, "import") return probe(torch, requested, index) def parse_driver_rows(raw: str) -> list[dict] | None: """Accept only two numeric fields; no serial, UUID, process, model, or path.""" rows = [] for line in raw.splitlines(): parts = [part.strip() for part in line.split(",")] if len(parts) != 2 or not re.fullmatch(r"[0-9]+(?:\.[0-9]+){1,3}", parts[0]): return None if not re.fullmatch(r"[0-9]{1,9}", parts[1]): return None rows.append({"driver_version": parts[0], "memory_total_mib": int(parts[1])}) if len(rows) > 64: return None return rows or None def host_check(runner=subprocess.run) -> dict: result = {"tool": "nvidia-smi", "scope": "NVIDIA driver visibility only", "status": "unavailable", "devices": []} try: completed = runner(["nvidia-smi", "--query-gpu=driver_version,memory.total", "--format=csv,noheader,nounits"], capture_output=True, text=True, timeout=3, check=False) if completed.returncode != 0: return result rows = parse_driver_rows(completed.stdout) if rows is None: result["status"] = "unrecognized_output" else: result.update(status="reported", devices=rows) except subprocess.TimeoutExpired: result["status"] = "timeout" except (OSError, UnicodeError): pass return result def run_worker(requested: str, index: int, timeout: int, runner=subprocess.run) -> dict: report = report_base(requested) try: completed = runner([sys.executable, str(Path(__file__).resolve()), "--worker", "--device", requested, "--device-index", str(index)], capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=timeout, check=False) decoded = json.loads(completed.stdout) if not isinstance(decoded, dict) or decoded.get("schema") != "kernova-diagnostic-v1": raise ValueError if decoded.get("exit_code") != completed.returncode: raise ValueError return decoded except subprocess.TimeoutExpired: return finish(report, "TIMEOUT", 10, "worker") except (OSError, ValueError, TypeError): return finish(report, "WORKER_FAILED", 11, "worker") class SafeParser(argparse.ArgumentParser): def error(self, message): # argparse normally echoes invalid supplied values; keep them out. report = finish(report_base("unspecified"), "CLI_ARGUMENTS_INVALID", 2, "arguments") print(json.dumps(report, ensure_ascii=True, indent=2)) raise SystemExit(2) def main() -> int: parser = SafeParser(prog="kernodeck-diagnostic-v1.py", description=__doc__) parser.add_argument("--device", choices=("gpu", "cpu"), default="gpu", help="gpu par défaut ; cpu demande un contrôle CPU explicite") parser.add_argument("--device-index", type=int, default=0, help="index visible du GPU, de 0 à 63") parser.add_argument("--timeout", type=int, default=30, help="délai du contrôle Python, de 5 à 120 secondes") parser.add_argument("--host-check", action="store_true", help="lecture NVIDIA facultative, limitée à 3 secondes") parser.add_argument("--output", help="nouveau fichier JSON ; un fichier existant n’est jamais écrasé") parser.add_argument("--worker", action="store_true", help=argparse.SUPPRESS) args = parser.parse_args() if not 0 <= args.device_index <= 63 or not 5 <= args.timeout <= 120: parser.error("out of range") if args.worker: report = worker(args.device, args.device_index) else: report = run_worker(args.device, args.device_index, args.timeout) if args.host_check: report["host_check"] = host_check() if args.output: try: with open(args.output, "x", encoding="utf-8") as handle: handle.write(json.dumps(report, ensure_ascii=False, indent=2) + "\n") except (OSError, ValueError): report = finish(report, "OUTPUT_WRITE_FAILED", 12, "output") # ASCII escapes make the report portable even in older Windows consoles. print(json.dumps(report, ensure_ascii=True, indent=2)) return report["exit_code"] if __name__ == "__main__": try: raise SystemExit(main()) except KeyboardInterrupt: print(json.dumps(finish(report_base("unspecified"), "INTERRUPTED", 130, "worker"), ensure_ascii=True, indent=2)) raise SystemExit(130)