#!/usr/bin/env python3 """Petit exercice CPU original : un checkpoint de reprise, pas seulement des poids.""" import argparse import csv import hashlib import json import math from pathlib import Path import platform import random import sys import numpy as np import torch from torch import nn VERSION = "1.0.0" ROOT = Path(__file__).resolve().parent SEED = 240926 BATCH_SIZE = 4 MAX_CHECKPOINT_BYTES = 2_000_000 def sha256(path): return hashlib.sha256(Path(path).read_bytes()).hexdigest() def environment(): # Liste explicite : aucun chemin, nom de machine, compte ou environnement exporté. return {"python": platform.python_version(), "pytorch": str(torch.__version__), "numpy": str(np.__version__), "device": "cpu", "dtype": "float64", "threads": 1, "gpu_tested": False, "amp": False, "data_workers": 0} def source_identity(): return {"project_version": VERSION, "train_sha256": sha256(ROOT / "train.py"), "data_sha256": sha256(ROOT / "data.csv")} def read_data(): with (ROOT / "data.csv").open(encoding="utf-8", newline="") as handle: rows = list(csv.DictReader(handle)) if len(rows) != 24 or [int(row["row_id"]) for row in rows] != list(range(24)): raise ValueError("Le CSV doit contenir les 24 lignes numérotées de l’exercice.") features = torch.tensor([[float(row["x1"]), float(row["x2"])] for row in rows], dtype=torch.float64, device="cpu") targets = torch.tensor([[float(row["target"])] for row in rows], dtype=torch.float64, device="cpu") if not torch.isfinite(features).all() or not torch.isfinite(targets).all(): raise ValueError("Les données doivent être finies.") return features, targets def capture_rng(generator): # PCG64 stocke ici des chaînes et des entiers Python, pas un ndarray NumPy. return {"python": random.getstate(), "numpy_pcg64": generator.bit_generator.state, "torch_cpu": torch.get_rng_state()} def restore_rng(state, generator): random.setstate(state["python"]) generator.bit_generator.state = state["numpy_pcg64"] torch.set_rng_state(state["torch_cpu"]) def next_draws(generator): """Observer le prochain tirage de chaque RNG sans changer la suite du travail.""" before = capture_rng(generator) values = {"python": random.random(), "numpy": float(generator.random()), "torch_cpu": torch.rand(3, dtype=torch.float64).tolist()} restore_rng(before, generator) return values def load_checkpoint(path): """Charger uniquement le checkpoint créé dans un répertoire maîtrisé de cet exercice.""" path = Path(path) sidecar = path.with_suffix(path.suffix + ".sha256") if not path.is_file() or not sidecar.is_file(): raise ValueError("Checkpoint ou empreinte manquant. Utilisez la sortie de cet exercice.") if path.stat().st_size > MAX_CHECKPOINT_BYTES: raise ValueError("Le checkpoint dépasse la taille prévue pour cet exercice.") digest = sidecar.read_text(encoding="ascii").strip() if len(digest) != 64 or digest != sha256(path): raise ValueError("Empreinte du checkpoint différente : chargement refusé.") # Aucun fallback weights_only=False, aucun safe_global ou module arbitraire ajouté. try: state = torch.load(path, map_location="cpu", weights_only=True) except Exception: raise ValueError("Checkpoint non compatible avec le chargement restreint. Aucun chargement alternatif n’est tenté.") from None required = {"schema", "identity", "environment", "model", "optimizer", "scheduler", "progress", "rng", "history", "eval_mse", "next_rng_draws", "next_lr"} if not isinstance(state, dict) or set(state) != required or state["schema"] != 1: raise ValueError("Schéma de checkpoint inattendu.") if state["identity"] != source_identity(): raise ValueError("Le code ou les données diffèrent de ceux du checkpoint.") if state["environment"] != environment(): raise ValueError("Les versions ou paramètres CPU diffèrent. Refaites le protocole complet.") progress = state["progress"] if (set(progress) != {"step", "epoch", "cursor", "permutation", "samples_seen"} or sorted(progress["permutation"]) != list(range(24)) or progress["cursor"] not in range(0, 25, BATCH_SIZE) or progress["samples_seen"] != progress["step"] * BATCH_SIZE or progress["samples_seen"] != progress["epoch"] * 24 + progress["cursor"] or len(state["history"]) != progress["step"]): raise ValueError("Progression des données incohérente.") return state def train(steps, output, resume=None, omit_rng_restore=False): if steps < 1 or steps > 1000: raise ValueError("Choisissez de 1 à 1 000 étapes supplémentaires.") if omit_rng_restore and resume is None: raise ValueError("Le contrôle négatif nécessite un checkpoint de reprise.") if output.exists(): raise ValueError("Le répertoire de sortie existe déjà. Choisissez un nouveau nom.") torch.set_num_threads(1) torch.set_num_interop_threads(1) torch.use_deterministic_algorithms(True) random.seed(SEED) generator = np.random.Generator(np.random.PCG64(SEED)) torch.manual_seed(SEED) features, targets = read_data() model = nn.Sequential(nn.Linear(2, 8), nn.Tanh(), nn.Dropout(0.25), nn.Linear(8, 1)) model = model.to(device="cpu", dtype=torch.float64) optimizer = torch.optim.Adam(model.parameters(), lr=0.03) # Construire l’ordonnanceur AVANT optimizer.load_state_dict. scheduler = torch.optim.lr_scheduler.StepLR(optimizer, step_size=3, gamma=0.5) progress = {"step": 0, "epoch": 0, "cursor": 0, "permutation": generator.permutation(len(features)).tolist(), "samples_seen": 0} history = [] resumed_next_draws = None resumed_next_lr = None if resume is not None: state = load_checkpoint(resume) model.load_state_dict(state["model"]) scheduler.load_state_dict(state["scheduler"]) optimizer.load_state_dict(state["optimizer"]) progress = state["progress"] history = state["history"] # Dernière restauration : les constructions précédentes ont consommé de l’aléatoire. if not omit_rng_restore: restore_rng(state["rng"], generator) resumed_next_draws = next_draws(generator) resumed_next_lr = float(optimizer.param_groups[0]["lr"]) model.train() for _ in range(steps): if progress["cursor"] == len(features): progress["epoch"] += 1 progress["cursor"] = 0 progress["permutation"] = generator.permutation(len(features)).tolist() cursor = progress["cursor"] indices = progress["permutation"][cursor:cursor + BATCH_SIZE] # Trois RNG sont effectivement utilisés : Python (gain), NumPy (bruit), torch (dropout). gain = 1.0 + 0.02 * (random.random() - 0.5) noise = torch.tensor(generator.normal(0.0, 0.01, (BATCH_SIZE, 2)), dtype=torch.float64) batch = features[indices] * gain + noise lr_used = float(optimizer.param_groups[0]["lr"]) optimizer.zero_grad(set_to_none=True) loss = nn.functional.mse_loss(model(batch), targets[indices]) if not math.isfinite(float(loss.detach())): raise ValueError("Perte non finie : essai interrompu.") loss.backward() optimizer.step() scheduler.step() progress["step"] += 1 progress["cursor"] += BATCH_SIZE progress["samples_seen"] += BATCH_SIZE history.append({"step": progress["step"], "batch_rows": indices, "loss": float(loss.detach()), "lr_used": lr_used}) # Évaluation sans dropout, puis retour explicite au mode entraînement. model.eval() with torch.no_grad(): eval_mse = float(nn.functional.mse_loss(model(features), targets)) model.train() next_rng = next_draws(generator) next_lr = float(optimizer.param_groups[0]["lr"]) state = {"schema": 1, "identity": source_identity(), "environment": environment(), "model": model.state_dict(), "optimizer": optimizer.state_dict(), "scheduler": scheduler.state_dict(), "progress": progress, "rng": capture_rng(generator), "history": history, "eval_mse": eval_mse, "next_rng_draws": next_rng, "next_lr": next_lr} output.mkdir(parents=True, exist_ok=False) temporary = output / "checkpoint.tmp" destination = output / "checkpoint.pt" torch.save(state, temporary) temporary.replace(destination) destination.with_suffix(".pt.sha256").write_text(sha256(destination) + "\n", encoding="ascii") report = {"identity": source_identity(), "environment": environment(), "steps_in_this_process": steps, "resumed": resume is not None, "rng_restored": resume is not None and not omit_rng_restore, "progress": progress, "eval_mse": eval_mse, "next_lr": next_lr, "next_rng_draws": next_rng, "resumed_next_rng_draws": resumed_next_draws, "resumed_next_lr": resumed_next_lr, "history": history, "checkpoint_sha256": sha256(destination)} (output / "summary.json").write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8") return {"status": "ok", "device": "cpu", "total_steps": progress["step"], "samples_seen": progress["samples_seen"], "outputs": ["checkpoint.pt", "checkpoint.pt.sha256", "summary.json"]} def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--steps", required=True, type=int, help="Nombre d’étapes supplémentaires.") parser.add_argument("--output", required=True, type=Path, help="Nouveau répertoire de sortie.") parser.add_argument("--resume", type=Path, help="Checkpoint de cet exercice, créé par vous et resté sous votre contrôle.") parser.add_argument("--omit-rng-restore", action="store_true", help="Contrôle négatif : ne pas restaurer les RNG.") args = parser.parse_args() try: result = train(args.steps, args.output, args.resume, args.omit_rng_restore) except (ValueError, OSError, RuntimeError, KeyError, TypeError) as error: # Les erreurs métier ci-dessus ne contiennent pas les chemins du poste. message = str(error) if isinstance(error, ValueError) else "Échec de l’exercice. Vérifiez le répertoire, les dépendances et la provenance du checkpoint." print(json.dumps({"status": "error", "error_type": type(error).__name__, "message": message}, ensure_ascii=False)) return 2 print(json.dumps(result, ensure_ascii=False)) return 0 if __name__ == "__main__": sys.exit(main())