scripts/train_one_epoch.py

#!/usr/bin/env python3
"""Train the full official training set for one epoch on a single DCU."""

import argparse
import json
import pickle
import random
import time
from pathlib import Path

from _bootstrap import DATA_DIR, OUTPUT_DIR

import numpy as np
import tensorflow as tf
import tf_keras
from sklearn.metrics import average_precision_score, roc_auc_score


DEFAULT_DATA_DIR = DATA_DIR / "features" / "cdr_kmer3_ag_kmer1"
DEFAULT_OUTPUT_DIR = OUTPUT_DIR / "one_epoch"


def parse_args():
    parser = argparse.ArgumentParser()
    parser.add_argument("--data-dir", type=Path, default=DEFAULT_DATA_DIR)
    parser.add_argument("--output-dir", type=Path, default=DEFAULT_OUTPUT_DIR)
    parser.add_argument("--batch-size", type=int, default=64)
    parser.add_argument("--seed", type=int, default=123)
    return parser.parse_args()


def load_split(data_dir, suffix):
    # Register features.InputFeatures before loading the official pickles.
    import model  # noqa: F401

    with (data_dir / f"cdr_features_{suffix}.pickle").open("rb") as reader:
        cdr_features = pickle.load(reader)
    with (data_dir / f"ag_features_{suffix}.pickle").open("rb") as reader:
        ag_features = pickle.load(reader)

    if not cdr_features:
        raise ValueError(f"The {suffix} split is empty")
    if len(cdr_features) != len(ag_features):
        raise ValueError(
            f"Unpaired {suffix} features: CDR={len(cdr_features)}, "
            f"AG={len(ag_features)}"
        )

    cdr_labels = np.asarray(
        [[item.label_id] for item in cdr_features], dtype=np.float32
    )
    ag_labels = np.asarray(
        [[item.label_id] for item in ag_features], dtype=np.float32
    )
    if not np.array_equal(cdr_labels, ag_labels):
        mismatch_count = int(np.count_nonzero(cdr_labels != ag_labels))
        raise ValueError(f"CDR/AG labels differ for {mismatch_count} {suffix} samples")
    if set(np.unique(cdr_labels).tolist()) != {0.0, 1.0}:
        raise ValueError(f"Expected binary labels 0/1, got {np.unique(cdr_labels)}")

    inputs = [
        np.asarray([item.input_ids for item in cdr_features], dtype=np.int32),
        np.asarray([item.cdr_number_ids for item in cdr_features], dtype=np.int32),
        np.asarray([item.input_ids for item in ag_features], dtype=np.int32),
    ]
    sample_count = len(cdr_features)
    expected_shapes = ((sample_count, 24), (sample_count, 24), (sample_count, 2371))
    actual_shapes = tuple(array.shape for array in inputs)
    if actual_shapes != expected_shapes:
        raise ValueError(f"Unexpected {suffix} shapes: {actual_shapes}, expected {expected_shapes}")

    return inputs, cdr_labels


def main():
    args = parse_args()
    if args.batch_size < 2:
        raise ValueError(f"--batch-size must be at least 2, got {args.batch_size}")
    if not args.data_dir.is_dir():
        raise FileNotFoundError(f"Data directory not found: {args.data_dir}")

    # Never reuse an existing directory: this prevents accidental checkpoint overwrite.
    args.output_dir.mkdir(parents=True, exist_ok=False)

    random.seed(args.seed)
    np.random.seed(args.seed)
    tf.random.set_seed(args.seed)

    gpus = tf.config.list_physical_devices("GPU")
    if not gpus:
        raise RuntimeError(
            "TensorFlow cannot see a DCU. Set HIP_VISIBLE_DEVICES and "
            "CUDA_VISIBLE_DEVICES before starting Python."
        )
    for gpu in gpus:
        tf.config.experimental.set_memory_growth(gpu, True)

    train_inputs, train_labels = load_split(args.data_dir, "tr")
    val_inputs, val_labels = load_split(args.data_dir, "val")

    # Import after configuring the accelerator to avoid initializing it early.
    from model import get_model

    with tf.device("/GPU:0"):
        model = get_model()
        initial_weights = [
            variable.numpy().copy() for variable in model.trainable_variables
        ]

        start_time = time.perf_counter()
        history = model.fit(
            train_inputs,
            train_labels,
            validation_data=(val_inputs, val_labels),
            epochs=1,
            batch_size=args.batch_size,
            shuffle=True,
            verbose=1,
        )
        training_seconds = time.perf_counter() - start_time

        val_predictions = model.predict(
            val_inputs,
            batch_size=args.batch_size,
            verbose=1,
        ).reshape(-1)

    if not np.all(np.isfinite(val_predictions)):
        raise FloatingPointError("Validation predictions contain NaN or Inf")
    if np.any((val_predictions < 0.0) | (val_predictions > 1.0)):
        raise ValueError("Validation predictions are outside [0, 1]")

    training_loss = float(history.history["loss"][-1])
    validation_loss = float(history.history["val_loss"][-1])
    if not np.isfinite(training_loss) or not np.isfinite(validation_loss):
        raise FloatingPointError(
            f"Non-finite loss: train={training_loss}, val={validation_loss}"
        )

    changed_count = 0
    max_weight_change = 0.0
    for before, variable in zip(initial_weights, model.trainable_variables):
        after = variable.numpy()
        if not np.all(np.isfinite(after)):
            raise FloatingPointError(f"Trainable variable contains NaN/Inf: {variable.name}")
        change = float(np.max(np.abs(after - before)))
        max_weight_change = max(max_weight_change, change)
        if change > 0.0:
            changed_count += 1

    labels_flat = val_labels.reshape(-1).astype(np.int32)
    val_auroc = float(roc_auc_score(labels_flat, val_predictions))
    val_aupr = float(average_precision_score(labels_flat, val_predictions))
    dcu_variable_count = sum("GPU:0" in variable.device for variable in model.variables)

    if changed_count == 0:
        raise AssertionError("No trainable weights changed during the epoch")
    if dcu_variable_count == 0:
        raise RuntimeError("No model variables were placed on GPU:0")

    checkpoint_path = args.output_dir / "model_after_one_epoch.weights.h5"
    predictions_path = args.output_dir / "validation_predictions.npz"
    metrics_path = args.output_dir / "metrics.json"

    model.save_weights(str(checkpoint_path))
    np.savez_compressed(
        predictions_path,
        labels=labels_flat,
        predictions=val_predictions,
    )

    # Verify that the newly written checkpoint can be loaded and reproduces output.
    with tf.device("/GPU:0"):
        reloaded_model = get_model()
        reloaded_model.load_weights(str(checkpoint_path))
        reload_count = min(args.batch_size, len(labels_flat))
        reloaded_predictions = reloaded_model(
            [array[:reload_count] for array in val_inputs],
            training=False,
        ).numpy().reshape(-1)
    reload_max_error = float(
        np.max(np.abs(reloaded_predictions - val_predictions[:reload_count]))
    )
    if reload_max_error > 1e-6:
        raise AssertionError(
            f"Reloaded checkpoint max error {reload_max_error:.10g} exceeds 1e-6"
        )

    metrics = {
        "tensorflow_version": tf.__version__,
        "tf_keras_version": tf_keras.__version__,
        "seed": args.seed,
        "batch_size": args.batch_size,
        "epochs": 1,
        "initialization": "random",
        "train_samples": int(train_labels.size),
        "validation_samples": int(val_labels.size),
        "training_loss": training_loss,
        "validation_loss": validation_loss,
        "validation_auroc": val_auroc,
        "validation_aupr": val_aupr,
        "training_seconds": training_seconds,
        "throughput_samples_per_second": float(train_labels.size / training_seconds),
        "trainable_tensors_changed": changed_count,
        "trainable_tensor_count": len(initial_weights),
        "max_absolute_weight_change": max_weight_change,
        "variables_on_gpu0": dcu_variable_count,
        "model_variable_count": len(model.variables),
        "checkpoint_reload_max_error": reload_max_error,
    }
    metrics_path.write_text(
        json.dumps(metrics, indent=2, ensure_ascii=False) + "\n",
        encoding="utf-8",
    )

    print("\nFull training-set one-epoch validation")
    print(f"TensorFlow: {tf.__version__}")
    print(f"tf_keras: {tf_keras.__version__}")
    print(f"DCU: {gpus[0]}")
    print("Initialization: random (official Model99.h5 was not loaded)")
    print(f"Train/validation samples: {train_labels.size}/{val_labels.size}")
    print(f"Batch size/epochs: {args.batch_size}/1")
    print(f"Training loss: {training_loss:.10g}")
    print(f"Validation loss: {validation_loss:.10g}")
    print(f"Validation AUROC: {val_auroc:.10f}")
    print(f"Validation AUPR: {val_aupr:.10f}")
    print(f"Training seconds: {training_seconds:.3f}")
    print(f"Training throughput samples/s: {train_labels.size / training_seconds:.3f}")
    print(f"Trainable tensors changed: {changed_count}/{len(initial_weights)}")
    print(f"Max absolute weight change: {max_weight_change:.10g}")
    print(f"Variables placed on GPU:0: {dcu_variable_count}/{len(model.variables)}")
    print(f"Checkpoint reload max error: {reload_max_error:.10g}")
    print(f"Checkpoint: {checkpoint_path}")
    print(f"Metrics: {metrics_path}")
    print(f"Validation predictions: {predictions_path}")
    print("Full training-set one epoch: PASS")
    print("Official Model99.h5 was not loaded or overwritten.")


if __name__ == "__main__":
    main()