src/audiointeraction/utils.py

"""Training / inference helpers."""

import inspect
import json
import os
import re
import shutil
import subprocess
from pathlib import Path
from typing import Any, Iterable, List, Literal, Optional, Union

import lightning as L
import torch
import torch.nn as nn
from lightning.fabric.loggers import CSVLogger, TensorBoardLogger
from lightning.fabric.strategies import FSDPStrategy
from lightning.fabric.utilities.load import _lazy_load as lazy_load
from lightning.pytorch.cli import instantiate_class
from lightning.pytorch.loggers import WandbLogger
from typing_extensions import Self


# === Output dir / checkpoint resume ===

def init_out_dir(out_dir: Path) -> Path:
    if not isinstance(out_dir, Path):
        out_dir = Path(out_dir)
    if not out_dir.is_absolute() and "LIGHTNING_ARTIFACTS_DIR" in os.environ:
        return Path(os.getenv("LIGHTNING_ARTIFACTS_DIR")) / out_dir
    return out_dir


def find_resume_path(resume: Union[bool, Literal["auto"], Path], out_dir: Path) -> Optional[Path]:
    if not resume or isinstance(resume, Path):
        return resume or None
    # `resume` is True or "auto"; find the newest lit_model.pth under out_dir.
    candidates = list(out_dir.rglob("lit_model.pth"))
    if not candidates:
        if resume == "auto":
            return None
        raise FileNotFoundError(
            f"resume=True but no `lit_model.pth` under {out_dir}; set resume='auto' to skip or False to start fresh."
        )
    return max(candidates, key=lambda p: p.stat().st_mtime)


# === Misc math / introspection ===

def find_multiple(n: int, k: int) -> int:
    if n % k == 0:
        return n
    return n + k - (n % k)


def num_parameters(module: nn.Module, requires_grad: Optional[bool] = None) -> int:
    total = 0
    for p in module.parameters():
        if requires_grad is None or p.requires_grad == requires_grad:
            total += p.numel()
    return total


# === Cross-entropy ===

def chunked_cross_entropy(
    logits: Union[torch.Tensor, List[torch.Tensor]],
    targets: torch.Tensor,
    chunk_size: int = 128,
    ignore_index: int = -100,
) -> torch.Tensor:
    """Memory-friendly cross-entropy: optionally chunk along the seq axis to bound the activation peak."""
    if isinstance(logits, list):
        if chunk_size == 0:
            logits = torch.cat(logits, dim=1).reshape(-1, logits[0].size(-1))
            return torch.nn.functional.cross_entropy(logits, targets.reshape(-1), ignore_index=ignore_index)
        logit_chunks = [c.reshape(-1, c.size(-1)) for c in logits]
        target_chunks = [c.reshape(-1) for c in targets.split(logits[0].size(1), dim=1)]
        loss_chunks = [
            torch.nn.functional.cross_entropy(lc, tc, ignore_index=ignore_index, reduction="none")
            for lc, tc in zip(logit_chunks, target_chunks)
        ]
        non_masked = (targets != ignore_index).sum()
        return torch.cat(loss_chunks).sum() / non_masked.maximum(torch.ones_like(non_masked))

    logits = logits.reshape(-1, logits.size(-1))
    targets = targets.reshape(-1)
    if chunk_size == 0:
        return torch.nn.functional.cross_entropy(logits, targets, ignore_index=ignore_index)
    logit_chunks = logits.split(chunk_size)
    target_chunks = targets.split(chunk_size)
    loss_chunks = [
        torch.nn.functional.cross_entropy(lc, tc, ignore_index=ignore_index, reduction="none")
        for lc, tc in zip(logit_chunks, target_chunks)
    ]
    non_masked = (targets != ignore_index).sum()
    return torch.cat(loss_chunks).sum() / non_masked.maximum(torch.ones_like(non_masked))


# === Precision / checkpoint IO ===

def get_default_supported_precision(training: bool) -> str:
    """Return bf16 if the GPU supports it, otherwise 16-bit. Picks `-mixed` for training, `-true` for inference."""
    if torch.cuda.is_available() and torch.cuda.is_bf16_supported():
        return "bf16-mixed" if training else "bf16-true"
    return "16-mixed" if training else "16-true"


def load_checkpoint(fabric: L.Fabric, model: nn.Module, checkpoint_path: Path, strict: bool = True) -> None:
    if isinstance(fabric.strategy, FSDPStrategy):
        fabric.load_raw(checkpoint_path, model, strict=strict)
    else:
        state_dict = lazy_load(checkpoint_path)
        state_dict = state_dict.get("model", state_dict)
        model.load_state_dict(state_dict, strict=strict)


# === Iteration ===

class CycleIterator:
    """Indefinitely cycles through an iterable, exposing the current epoch count."""

    def __init__(self, iterable: Iterable) -> None:
        self.iterable = iterable
        self.epoch = 0
        self._iterator = None

    def __next__(self) -> Any:
        if self._iterator is None:
            self._iterator = iter(self.iterable)
        try:
            return next(self._iterator)
        except StopIteration:
            self._iterator = iter(self.iterable)
            self.epoch += 1
            return next(self._iterator)

    def __iter__(self) -> Self:
        return self


# === Setup helpers (out dir, logger, devices, optimizer) ===

def copy_config_files(source_dir: Path, out_dir: Path) -> None:
    """Copy model_config.yaml and tokenizer files (when present) into the output directory."""
    for name in ("config.json", "generation_config.json", "model_config.yaml",
                 "tokenizer.json", "tokenizer.model", "tokenizer_config.json"):
        src = source_dir / name
        if src.exists():
            shutil.copy(src, out_dir)


def parse_devices(devices: Union[str, int]) -> int:
    if devices in (-1, "auto"):
        return torch.cuda.device_count() or 1
    if isinstance(devices, int) and devices > 0:
        return devices
    raise ValueError(f"Devices must be 'auto' or a positive integer, got: {devices!r}")


def choose_logger(
    logger_name: Literal["csv", "tensorboard", "wandb"],
    out_dir: Path,
    name: str,
    log_interval: int = 1,
    resume: Optional[bool] = None,
    **kwargs: Any,
):
    if logger_name == "csv":
        return CSVLogger(root_dir=(out_dir / "logs"), name="csv", flush_logs_every_n_steps=log_interval, **kwargs)
    if logger_name == "tensorboard":
        return TensorBoardLogger(root_dir=(out_dir / "logs"), name="tensorboard", **kwargs)
    if logger_name == "wandb":
        return WandbLogger(project=name, resume=resume, **kwargs)
    raise ValueError(f"`--logger_name={logger_name}` is not a valid option. Choose from 'csv', 'tensorboard', 'wandb'.")


def instantiate_torch_optimizer(optimizer, model_parameters, **kwargs):
    """Resolve `optimizer` (str name like 'AdamW' or a dict with class_path/init_args) to an actual optimizer."""
    if isinstance(optimizer, str):
        class_module, class_name = optimizer.rsplit(".", 1) if "." in optimizer else ("torch.optim", optimizer)
        module = __import__(class_module, fromlist=[class_name])
        optimizer_cls = getattr(module, class_name)
        valid = set(inspect.signature(optimizer_cls).parameters)
        kwargs = {k: v for k, v in kwargs.items() if k in valid}
        return optimizer_cls(model_parameters, **kwargs)
    if isinstance(optimizer, dict):
        optimizer = dict(optimizer)
        class_module, class_name = optimizer["class_path"].rsplit(".", 1)
        module = __import__(class_module, fromlist=[class_name])
        valid = set(inspect.signature(getattr(module, class_name)).parameters)
        optimizer["init_args"].update({k: v for k, v in kwargs.items() if k in valid})
        return instantiate_class(model_parameters, optimizer)
    raise ValueError(f'Unrecognized "optimizer" value: {optimizer}')


# === GPU topology (nvlink / xgmi) ===

def check_nvlink_connectivity(fabric=None):
    """Best-effort check that the multi-GPU interconnect is fast. Prints a warning if not."""
    custom_print = fabric.print if fabric is not None else print
    if os.getenv("RANK", "0") != "0":
        return
    try:
        if not torch.cuda.is_available():
            custom_print("No GPUs available")
            return
        gpu_name = torch.cuda.get_device_properties(0).name.lower()
        if "nvidia" in gpu_name:
            _check_nvidia_connectivity(custom_print)
        elif "advanced micro devices" in gpu_name or "amd" in gpu_name:
            _check_amd_connectivity(custom_print)
        else:
            custom_print(f"Unrecognized GPU vendor: {gpu_name}")
    except Exception as e:
        custom_print(f"An error occurred while checking GPU connectivity: {e}")


def _check_nvidia_connectivity(custom_print):
    result = subprocess.run(["nvidia-smi", "topo", "-m"], stdout=subprocess.PIPE, text=True)
    if result.returncode != 0:
        custom_print("Failed to run nvidia-smi"); return
    lines = result.stdout.strip().split("\n")
    start = next((i for i, line in enumerate(lines) if "GPU0" in line), None)
    if start is None:
        custom_print("Failed to parse nvidia-smi output"); return
    headers = lines[start].split()
    gpu_count = len([h for h in headers if re.match(r"^GPU\d+$", h)])
    all_nvlink = all(
        all("NV" in conn for conn in row.split()[1:1 + gpu_count] if conn != "X")
        for row in lines[start + 1: start + 1 + gpu_count]
    )
    if all_nvlink:
        custom_print("All GPUs are fully connected via NVLink.")
    else:
        custom_print("Warning: not all GPUs are fully connected via NVLink — multi-GPU throughput may be suboptimal.")


def _check_amd_connectivity(custom_print):
    result = subprocess.run(["rocm-smi", "--showtopotype"], stdout=subprocess.PIPE, text=True)
    if result.returncode != 0:
        custom_print("Failed to run rocm-smi"); return
    lines = result.stdout.strip().split("\n")
    gpu_idx = next((i for i, line in enumerate(lines) if re.match(r"^\s*GPU0", line)), None)
    if gpu_idx is None or gpu_idx == 0:
        custom_print("Failed to parse rocm-smi output (no GPU headers found)"); return
    headers = lines[gpu_idx - 1].strip().split()
    gpu_count = len([h for h in headers if re.match(r"^GPU\d+$", h)])
    gpu_lines = [l.strip() for l in lines[gpu_idx: gpu_idx + gpu_count] if re.match(r"^\s*GPU\d+", l)]
    if len(gpu_lines) != gpu_count:
        custom_print("Mismatch in GPU count when parsing rocm-smi output"); return
    all_xgmi = all(
        all(conn in ("XGMI", "0") for conn in row.split()[1:1 + gpu_count])
        for row in gpu_lines
    )
    if all_xgmi:
        custom_print("All GPUs are fully connected via XGMI.")
    else:
        custom_print("Warning: not all GPUs are fully connected via XGMI — multi-GPU throughput may be suboptimal.")


# === Misc ===

def fix_and_load_json(s: str):
    """Tolerant JSON parser: strips trailing commas and inserts missing ones before retrying."""
    s = re.sub(r',(\s*[}\]])', r'\1', s)
    s = re.sub(r'(?<=[}\]0-9truefalsenull"])\s*(\n\s*)"', r',\1"', s)
    try:
        return json.loads(s)
    except json.JSONDecodeError as e:
        raise ValueError(f"Failed to parse JSON after fixing: {e}")


def create_finetuning_performance_report(training_time: float, device_type: str) -> str:
    lines = [
        "",
        "| ------------------------------------------------------",
        f"| Training Time : {training_time:.2f} s",
    ]
    if device_type == "cuda":
        memory_used = torch.cuda.max_memory_allocated() / 1e9
        lines.append(f"| Peak Memory   : {memory_used:.2f} GB")
    lines.append("| ------------------------------------------------------")
    return "\n".join(lines)