preprocess/tools/vocal_separation/utils/model_utils.py

# coding: utf-8
__author__ = 'Roman Solovyev (ZFTurbo): https://github.com/ZFTurbo/'

import argparse
import numpy as np
import torch
import torch.nn as nn
from ml_collections import ConfigDict
from torch.optim import Adam, AdamW, SGD, RAdam, RMSprop
from tqdm.auto import tqdm
from typing import Dict, List, Tuple, Any, Union, Optional
import loralib as lora
from .muon import SingleDeviceMuonWithAuxAdam
import torch.distributed as dist

def demix(
    config: ConfigDict,
    model: torch.nn.Module,
    mix: torch.Tensor,
    device: torch.device,
    model_type: str,
    pbar: bool = False
) -> Union[Dict[str, np.ndarray], np.ndarray]:
    """
    Perform audio source separation with a given model.

    Supports both Demucs-specific and generic processing modes, including
    overlapping chunk-based inference with optional progress bar display.
    Handles padding, fading, and batching to reduce artifacts during separation.

    Args:
        config (ConfigDict): Configuration object with audio and inference
            parameters (chunk size, overlap, batch size, etc.).
        model (torch.nn.Module): Source separation model for inference.
        mix (torch.Tensor): Input audio tensor of shape (channels, time).
        device (torch.device): Device on which to run inference (CPU or CUDA).
        model_type (str): Type of model (e.g., 'htdemucs', 'mdx23c') that
            determines processing mode.
        pbar (bool, optional): If True, show a progress bar during chunk
            processing. Defaults to False.

    Returns:
        Union[Dict[str, np.ndarray], np.ndarray]:
            - Dictionary mapping instrument names to separated waveforms if
              multiple instruments are predicted.
            - NumPy array of separated audio if only a single instrument is
              present (Demucs mode).
    """

    should_print = not dist.is_initialized() or dist.get_rank() == 0

    mix = torch.tensor(mix, dtype=torch.float32)

    if model_type == 'htdemucs':
        mode = 'demucs'
    else:
        mode = 'generic'
    # Define processing parameters based on the mode
    if mode == 'demucs':
        chunk_size = config.training.samplerate * config.training.segment
        num_instruments = len(config.training.instruments)
        num_overlap = config.inference.num_overlap
        step = chunk_size // num_overlap
    else:
        if 'chunk_size' in config.inference:
            chunk_size = config.inference.chunk_size
        else:
            chunk_size = config.audio.chunk_size
        num_instruments = len(prefer_target_instrument(config))
        num_overlap = config.inference.num_overlap

        fade_size = chunk_size // 10
        step = chunk_size // num_overlap
        border = chunk_size - step
        length_init = mix.shape[-1]
        windowing_array = _getWindowingArray(chunk_size, fade_size)
        # Add padding for generic mode to handle edge artifacts
        if length_init > 2 * border and border > 0:
            mix = nn.functional.pad(mix, (border, border), mode="reflect")

    batch_size = config.inference.batch_size

    use_amp = getattr(config.training, 'use_amp', True)

    with torch.cuda.amp.autocast(enabled=use_amp):
        with torch.inference_mode():
            # Initialize result and counter tensors
            req_shape = (num_instruments,) + mix.shape
            result = torch.zeros(req_shape, dtype=torch.float32)
            counter = torch.zeros(req_shape, dtype=torch.float32)

            i = 0
            batch_data = []
            batch_locations = []
            if pbar and should_print:
                progress_bar = tqdm(
                    total=mix.shape[1], desc="Processing audio chunks", leave=False
                )
            else:
                progress_bar = None

            while i < mix.shape[1]:
                # Extract chunk and apply padding if necessary
                part = mix[:, i:i + chunk_size].to(device)
                chunk_len = part.shape[-1]
                if mode == "generic" and chunk_len > chunk_size // 2:
                    pad_mode = "reflect"
                else:
                    pad_mode = "constant"
                part = nn.functional.pad(part, (0, chunk_size - chunk_len), mode=pad_mode, value=0)

                batch_data.append(part)
                batch_locations.append((i, chunk_len))
                i += step

                # Process batch if it's full or the end is reached
                if len(batch_data) >= batch_size or i >= mix.shape[1]:
                    arr = torch.stack(batch_data, dim=0)
                    x = model(arr)

                    if mode == "generic":
                        window = windowing_array.clone() # using clone() fixes the clicks at chunk edges when using batch_size=1
                        if i - step == 0:  # First audio chunk, no fadein
                            window[:fade_size] = 1
                        elif i >= mix.shape[1]:  # Last audio chunk, no fadeout
                            window[-fade_size:] = 1

                    for j, (start, seg_len) in enumerate(batch_locations):
                        if mode == "generic":
                            result[..., start:start + seg_len] += x[j, ..., :seg_len].cpu() * window[..., :seg_len]
                            counter[..., start:start + seg_len] += window[..., :seg_len]
                        else:
                            result[..., start:start + seg_len] += x[j, ..., :seg_len].cpu()
                            counter[..., start:start + seg_len] += 1.0

                    batch_data.clear()
                    batch_locations.clear()

                if progress_bar:
                    progress_bar.update(step)

            if progress_bar:
                progress_bar.close()
            

            """
            # mix: B, 2, T
            # req_shape = (num_instruments,) + mix.shape
            req_shape = (num_instruments,) + mix.shape
            result = torch.zeros(req_shape, dtype=torch.float32)
            counter = torch.zeros(req_shape, dtype=torch.float32)

            # prev_i = 0
            i = 0
            batch_data = []
            batch_locations = []

            while i < mix.shape[-1]:
                part = mix[:, :, i:i + chunk_size].to(device)
                chunk_len = part.shape[-1]
                if mode == "generic" and chunk_len > chunk_size // 2:
                    pad_mode = "reflect"
                else:
                    pad_mode = "constant"
                part = nn.functional.pad(part, (0, chunk_size - chunk_len), mode=pad_mode, value=0)
                # batch_locations.append((i, chunk_len))
                # prev_i = i
                batch_location = i, i + chunk_len
                i += step
                
                # print(part.shape)
                x = model(part)
                x = x.transpose(0, 1)
                # print(x.shape)

                if mode == "generic":
                    window = windowing_array.clone() # using clone() fixes the clicks at chunk edges when using batch_size=1
                    if i - step == 0:  # First audio chunk, no fadein
                        window[:fade_size] = 1
                    elif i >= mix.shape[1]:  # Last audio chunk, no fadeout
                        window[-fade_size:] = 1

                # for j, (start, seg_len) in enumerate(batch_locations):
                # l = chunk_len if chunk_len < chunk_size else chunk_size
                # print(l, x.shape, result.shape, counter.shape, window.shape)
                # print(result[..., batch_location[0]: batch_location[1]].shape, x[..., :chunk_len].cpu().shape, window[..., :chunk_len].shape)
                if mode == "generic":
                    result[..., batch_location[0]: batch_location[1]] += x[..., :chunk_len].cpu() * window[..., :chunk_len]
                    counter[..., batch_location[0]: batch_location[1]] += window[..., :chunk_len]
                else:
                    result[..., batch_location[0]: batch_location[1]] += x[..., :chunk_len].cpu()
                    counter[..., batch_location[0]: batch_location[1]] += 1.0

                batch_data.clear()
                batch_locations.clear()
            """
            # Compute final estimated sources
            estimated_sources = result / counter
            estimated_sources = estimated_sources.cpu().numpy()
            np.nan_to_num(estimated_sources, copy=False, nan=0.0)

            # Remove padding for generic mode
            if mode == "generic":
                if length_init > 2 * border and border > 0:
                    estimated_sources = estimated_sources[..., border:-border]

    # Return the result as a dictionary or a single array
    if mode == "demucs":
        instruments = config.training.instruments
    else:
        instruments = prefer_target_instrument(config)

    ret_data = {k: v for k, v in zip(instruments, estimated_sources)}

    if mode == "demucs" and num_instruments <= 1:
        return estimated_sources
    else:
        return ret_data


def initialize_model_and_device(model: torch.nn.Module, device_ids: List[int]) -> Tuple[Union[torch.device, str], torch.nn.Module]:
    """
    Move a model to the correct computation device and wrap with DataParallel if needed.

    Selects GPU(s) if CUDA is available; otherwise defaults to CPU. If multiple
    GPU IDs are provided, wraps the model with `nn.DataParallel` for multi-GPU
    execution.

    Args:
        model (torch.nn.Module): PyTorch model to be initialized.
        device_ids (List[int]): List of GPU device IDs to use. If length > 1,
            the model will be wrapped with DataParallel.

    Returns:
        Tuple[Union[torch.device, str], torch.nn.Module]: A tuple containing:
            - The computation device (`torch.device` or "cpu").
            - The model moved to that device (wrapped in DataParallel if applicable).
    """

    if torch.cuda.is_available():
        if len(device_ids) <= 1:
            device = torch.device(f'cuda:{device_ids[0]}')
            model = model.to(device)
        else:
            device = torch.device(f'cuda:{device_ids[0]}')
            model = nn.DataParallel(model, device_ids=device_ids).to(device)
    else:
        device = 'cpu'
        model = model.to(device)
        print("CUDA is not available. Running on CPU.")

    return device, model


def get_optimizer(config: ConfigDict, model: torch.nn.Module) -> torch.optim.Optimizer:
    """
    Create and configure an optimizer for training.

    Selects the optimizer type based on `config.training.optimizer` and applies
    the corresponding parameters, including support for advanced optimizers
    such as Muon, Prodigy, and 8-bit AdamW. Handles parameter group separation
    for specialized optimizers (e.g., Muon vs. Adam parameters).

    Args:
        config (ConfigDict): Training configuration containing optimizer type,
            learning rate, and optional optimizer-specific parameters.
        model (torch.nn.Module): Model whose parameters will be optimized.

    Returns:
        torch.optim.Optimizer: Initialized optimizer ready for training.

    Raises:
        ValueError: If required optimizer configuration is missing (e.g., for Muon).
        SystemExit: If an unknown optimizer name is encountered.
    """

    should_print = not dist.is_initialized() or dist.get_rank() == 0
    optim_params = dict()
    if 'optimizer' in config:
        optim_params = dict(config['optimizer'])
        if config.training.optimizer != 'muon' and should_print:
            print(f'Optimizer params from config:\n{optim_params}')

    name_optimizer = getattr(config.training, 'optimizer',
                             'No optimizer in config')

    if name_optimizer == 'adam':
        optimizer = Adam(model.parameters(), lr=config.training.lr, **optim_params)
    elif name_optimizer == 'adamw':
        optimizer = AdamW(model.parameters(), lr=config.training.lr, **optim_params)
    elif name_optimizer == 'radam':
        optimizer = RAdam(model.parameters(), lr=config.training.lr, **optim_params)
    elif name_optimizer == 'rmsprop':
        optimizer = RMSprop(model.parameters(), lr=config.training.lr, **optim_params)
    elif name_optimizer == 'prodigy':
        from prodigyopt import Prodigy
        # you can choose weight decay value based on your problem, 0 by default
        # We recommend using lr=1.0 (default) for all networks.
        optimizer = Prodigy(model.parameters(), lr=config.training.lr, **optim_params)
    elif name_optimizer == 'adamw8bit':
        import bitsandbytes as bnb
        optimizer = bnb.optim.AdamW8bit(model.parameters(), lr=config.training.lr, **optim_params)
    elif name_optimizer == 'muon':
        if should_print:
            print("Using Muon optimizer (Single-Device) with AdamW for auxiliary parameters.")
        
        muon_params = [p for p in model.parameters() if p.ndim >= 2]
        adam_params = [p for p in model.parameters() if p.ndim < 2]

        if not hasattr(config, 'optimizer') or 'muon_group' not in config.optimizer or 'adam_group' not in config.optimizer:
            raise ValueError("For the 'muon' optimizer, the config must have an 'optimizer' section "
                             "with 'muon_group' and 'adam_group' dictionaries.")

        muon_group_config = dict(config.optimizer.muon_group)
        adam_group_config = dict(config.optimizer.adam_group)

        if should_print:
            print(f"Muon group params: {muon_group_config}")
            print(f"Adam group params: {adam_group_config}")

        param_groups = [
            dict(params=muon_params, use_muon=True, **muon_group_config),
            dict(params=adam_params, use_muon=False, **adam_group_config),
        ]
        optimizer = SingleDeviceMuonWithAuxAdam(param_groups)
    elif name_optimizer == 'sgd':
        if should_print:
            print('Use SGD optimizer')
        optimizer = SGD(model.parameters(), lr=config.training.lr, **optim_params)
    else:
        if should_print:
            print(f'Unknown optimizer: {name_optimizer}')
        exit()
    return optimizer


def normalize_batch(x: torch.Tensor, y: torch.Tensor) -> Tuple[torch.Tensor, torch.Tensor]:
    """
    Apply mean-variance normalization to a pair of tensors.

    Computes the mean and standard deviation from `x` and normalizes both `x`
    and `y` using those statistics. This ensures the two tensors are scaled
    consistently.

    Args:
        x (torch.Tensor): Input tensor used to compute normalization statistics.
        y (torch.Tensor): Input tensor normalized using the same statistics as `x`.

    Returns:
        Tuple[torch.Tensor, torch.Tensor]: Normalized tensors `(x, y)`.
    """

    mean = x.mean()
    std = x.std()
    if std != 0:
        x = (x - mean) / std
        y = (y - mean) / std
    return x, y


def apply_tta(
    config,
    model: torch.nn.Module,
    mix: torch.Tensor,
    waveforms_orig: Dict[str, torch.Tensor],
    device: torch.device,
    model_type: str
) -> Dict[str, torch.Tensor]:
    """
    Enhance source separation results using Test-Time Augmentation (TTA).

    Applies augmentations such as channel reversal and polarity inversion to
    the input mixture, reprocesses with the model, and combines the results
    with the original predictions by averaging.

    Args:
        config: Configuration object with model and inference parameters.
        model (torch.nn.Module): Trained source separation model.
        mix (torch.Tensor): Input mixture tensor of shape (channels, time).
        waveforms_orig (Dict[str, torch.Tensor]): Dictionary of separated
            sources before augmentation.
        device (torch.device): Computation device (CPU or CUDA).
        model_type (str): Model type identifier used for demixing.

    Returns:
        Dict[str, torch.Tensor]: Dictionary of separated sources after applying TTA.
    """

    # Create augmentations: channel inversion and polarity inversion
    track_proc_list = [mix[::-1].copy(), -1.0 * mix.copy()]

    # Process each augmented mixture
    for i, augmented_mix in enumerate(track_proc_list):
        waveforms = demix(config, model, augmented_mix, device, model_type=model_type)
        for el in waveforms:
            if i == 0:
                waveforms_orig[el] += waveforms[el][::-1].copy()
            else:
                waveforms_orig[el] -= waveforms[el]

    # Average the results across augmentations
    for el in waveforms_orig:
        waveforms_orig[el] /= len(track_proc_list) + 1

    return waveforms_orig


def _getWindowingArray(window_size: int, fade_size: int) -> torch.Tensor:
    """
    Generate a windowing array with a linear fade-in at the beginning and a fade-out at the end.

    This function creates a window of size `window_size` where the first `fade_size` elements
    linearly increase from 0 to 1 (fade-in) and the last `fade_size` elements linearly decrease
    from 1 to 0 (fade-out). The middle part of the window is filled with ones.

    Parameters:
    ----------
    window_size : int
        The total size of the window.
    fade_size : int
        The size of the fade-in and fade-out regions.

    Returns:
    -------
    torch.Tensor
        A tensor of shape (window_size,) containing the generated windowing array.

    Example:
    -------
    If `window_size=10` and `fade_size=3`, the output will be:
    tensor([0.0000, 0.5000, 1.0000, 1.0000, 1.0000, 1.0000, 1.0000, 1.0000, 0.5000, 0.0000])
    """

    fadein = torch.linspace(0, 1, fade_size)
    fadeout = torch.linspace(1, 0, fade_size)

    window = torch.ones(window_size)
    window[-fade_size:] = fadeout
    window[:fade_size] = fadein
    return window


def prefer_target_instrument(config: ConfigDict) -> List[str]:
    """
        Return the list of target instruments based on the configuration.
        If a specific target instrument is specified in the configuration,
        it returns a list with that instrument. Otherwise, it returns the list of instruments.

        Parameters:
        ----------
        config : ConfigDict
            Configuration object containing the list of instruments or the target instrument.

        Returns:
        -------
        List[str]
            A list of target instruments.
        """
    if getattr(config.training, 'target_instrument', None):
        return [config.training.target_instrument]
    else:
        return config.training.instruments


def load_not_compatible_weights(model: torch.nn.Module, old_model: dict, verbose: bool = False) -> None:
    """
    Load a possibly incompatible state dict into `model` with best-effort matching.

    Accepts either a raw state_dict or a checkpoint dict with weights under "state" or "state_dict".
    For each param/buffer in `model`: if the name exists and shapes match → copy;
    if ndim matches but shapes differ → zero-pad/crop the source to fit the target;
    if the name is missing or ndim differs → skip. Optional logging on rank 0 when `verbose=True`.

    Args:
        model: Target PyTorch module.
        old_model: Source weights (state_dict or checkpoint dict).
        verbose: Print brief load decisions.

    Returns:
        None
    """

    should_print = verbose and (not dist.is_initialized() or dist.get_rank() == 0)

    new_model = model.state_dict()

    if 'state' in old_model:
        # Fix for htdemucs weights loading
        old_model = old_model['state']
    if 'state_dict' in old_model:
        # Fix for apollo weights loading
        old_model = old_model['state_dict']
    if 'model_state_dict' in old_model:
        # Fix for full_check_point
        old_model = old_model['model_state_dict']

    for el in new_model:
        if el in old_model:
            if should_print:
                print(f'Match found for {el}!')
            if new_model[el].shape == old_model[el].shape:
                if should_print:
                    print('Action: Just copy weights!')
                new_model[el] = old_model[el]
            else:
                if len(new_model[el].shape) != len(old_model[el].shape) and should_print:
                    print('Action: Different dimension! Too lazy to write the code... Skip it')
                else:
                    if should_print:
                        print(f'Shape is different: {tuple(new_model[el].shape)} != {tuple(old_model[el].shape)}')
                    ln = len(new_model[el].shape)
                    max_shape = []
                    slices_old = []
                    slices_new = []
                    for i in range(ln):
                        max_shape.append(max(new_model[el].shape[i], old_model[el].shape[i]))
                        slices_old.append(slice(0, old_model[el].shape[i]))
                        slices_new.append(slice(0, new_model[el].shape[i]))
                    # print(max_shape)
                    # print(slices_old, slices_new)
                    slices_old = tuple(slices_old)
                    slices_new = tuple(slices_new)
                    max_matrix = np.zeros(max_shape, dtype=np.float32)
                    for i in range(ln):
                        max_matrix[slices_old] = old_model[el].cpu().numpy()
                    max_matrix = torch.from_numpy(max_matrix)
                    new_model[el] = max_matrix[slices_new]
        else:
            if should_print:
                print(f'Match not found for {el}!')
    model.load_state_dict(
        new_model
    )


def load_lora_weights(model: torch.nn.Module, lora_path: str, device: str = 'cpu') -> None:
    """
    Load LoRA weights into a model.
    This function updates the given model with LoRA-specific weights from the specified checkpoint file.
    It does not require the checkpoint to match the model's full state dictionary, as only LoRA layers are updated.

    Parameters:
    ----------
    model : Module
        The PyTorch model into which the LoRA weights will be loaded.
    lora_path : str
        Path to the LoRA checkpoint file.
    device : str, optional
        The device to load the weights onto, by default 'cpu'. Common values are 'cpu' or 'cuda'.

    Returns:
    -------
    None
        The model is updated in place.
    """
    lora_state_dict = torch.load(lora_path, map_location=device)
    model.load_state_dict(lora_state_dict, strict=False)


def load_start_checkpoint(args: argparse.Namespace,
                          model: torch.nn.Module,
                          old_model: None,
                          type_: str = 'train') -> None:
    """
    Load an initial checkpoint into `model`.

    For `type_ == "train"`, performs a tolerant load using `old_model` (a state dict or a
    checkpoint dict) via `load_not_compatible_weights`, allowing partial shape mismatches.
    For other modes, loads a strict state dict from `args.start_check_point`, with special
    handling for HTDemucs/Apollo checkpoints (keys under "state"/"state_dict"). If
    `args.lora_checkpoint` is set, LoRA weights are applied after the base load.

    Args:
        args: Namespace with at least `start_check_point`, `model_type`, and optionally `lora_checkpoint`.
        model: Target PyTorch module to receive weights.
        old_model: Source weights for tolerant loading in train mode (state dict or checkpoint dict).
        type_: Loading strategy; "train" uses tolerant loading, otherwise strict loading from path.

    Returns:
        None
    """
    should_print = not dist.is_initialized() or dist.get_rank() == 0

    if should_print:
        print(f'Start from checkpoint: {args.start_check_point}')
    if type_ in ['train']:
        if 1:
            load_not_compatible_weights(model, old_model, verbose=False)
        else:
            model.load_state_dict(torch.load(args.start_check_point))
    else:
        device='cpu'
        if args.model_type in ['htdemucs', 'apollo']:
            state_dict = torch.load(args.start_check_point, map_location=device, weights_only=False)
            # Fix for htdemucs pretrained models
            if 'state' in state_dict:
                state_dict = state_dict['state']
            # Fix for apollo pretrained models
            if 'state_dict' in state_dict:
                state_dict = state_dict['state_dict']
        else:
            state_dict = torch.load(args.start_check_point, map_location=device, weights_only=True)
        model.load_state_dict(state_dict)

    if args.lora_checkpoint:
        if should_print:
            print(f"Loading LoRA weights from: {args.lora_checkpoint}")
        load_lora_weights(model, args.lora_checkpoint)


def bind_lora_to_model(config: Dict[str, Any], model: nn.Module) -> nn.Module:
    """
    Replaces specific layers in the model with LoRA-extended versions.

    Parameters:
    ----------
    config : Dict[str, Any]
        Configuration containing parameters for LoRA. It should include a 'lora' key with parameters for `MergedLinear`.
    model : nn.Module
        The original model in which the layers will be replaced.

    Returns:
    -------
    nn.Module
        The modified model with the replaced layers.
    """

    if 'lora' not in config:
        raise ValueError("Configuration must contain the 'lora' key with parameters for LoRA.")

    replaced_layers = 0  # Counter for replaced layers
    should_print = not dist.is_initialized() or dist.get_rank() == 0

    for name, module in model.named_modules():
        hierarchy = name.split('.')
        layer_name = hierarchy[-1]

        # Check if this is the target layer to replace (and layer_name == 'to_qkv')
        if isinstance(module, nn.Linear):
            try:
                # Get the parent module
                parent_module = model
                for submodule_name in hierarchy[:-1]:
                    parent_module = getattr(parent_module, submodule_name)

                # Replace the module with LoRA-enabled layer
                setattr(
                    parent_module,
                    layer_name,
                    lora.MergedLinear(
                        in_features=module.in_features,
                        out_features=module.out_features,
                        bias=module.bias is not None,
                        **config['lora']
                    )
                )
                replaced_layers += 1  # Increment the counter

            except Exception as e:
                if should_print:
                    print(f"Error replacing layer {name}: {e}")

    if replaced_layers == 0 and should_print:
        print("Warning: No layers were replaced. Check the model structure and configuration.")
    elif should_print:
        print(f"Number of layers replaced with LoRA: {replaced_layers}")

    return model


def save_weights(
    store_path: str,
    model: nn.Module,
    device_ids: List[int],
    optimizer: torch.optim.Optimizer,
    epoch: int,
    all_time_all_metrics,
    best_metric: float,
    scheduler: Optional[torch.optim.lr_scheduler.ReduceLROnPlateau] = None,
    train_lora: bool = False
) -> None:
    """
    Save a training checkpoint containing model weights, optimizer/scheduler states, and metadata.

    Behavior:
    - In Distributed Data Parallel (DDP), only rank 0 writes the file to avoid conflicts.
    - If `train_lora` is True, saves only LoRA adapter weights (`lora_state_dict`); otherwise saves the full model.
    - Uses `model.module.state_dict()` when the model is wrapped by DDP/DataParallel.
    - Stores `epoch` and `best_metric` alongside optimizer/scheduler states.

    Args:
        store_path: Destination file path for the checkpoint (will be overwritten).
        model: The model whose weights are being saved (may be wrapped by DDP/DataParallel).
        device_ids: List of GPU device IDs used during training (used to detect DP wrapping in non-DDP runs).
        optimizer: Optimizer whose state will be saved.
        epoch: Current training epoch to record in the checkpoint.
        all_time_all_metrics:
        best_metric: Best validation metric achieved so far.
        scheduler: Optional learning rate scheduler; its state is saved if provided.
        train_lora: If True, save only LoRA adapter weights instead of the full model.

    Returns:
        None
    """

    checkpoint: Dict[str, Any] = {
        "epoch": epoch,
        "optimizer_name": optimizer.__class__.__name__,
        "optimizer_state_dict": optimizer.state_dict(),
        "scheduler_state_dict": scheduler.state_dict() if scheduler else None,
        "best_metric": best_metric,
        "all_metrics": all_time_all_metrics
    }

    # Save model weights
    if train_lora:
        checkpoint["model_state_dict"] = lora.lora_state_dict(model)
    else:
        if dist.is_initialized():
            # In DDP, use .module
            checkpoint["model_state_dict"] = model.module.state_dict()
        else:
            checkpoint["model_state_dict"] = (
                model.state_dict() if len(device_ids) <= 1 else model.module.state_dict()
            )

    # Save only on rank 0 (or if not using DDP)
    if not dist.is_initialized() or dist.get_rank() == 0:
        torch.save(checkpoint, store_path)


def save_last_weights(
    args: argparse.Namespace,
    model: nn.Module,
    device_ids: List[int],
    optimizer: torch.optim.Optimizer,
    epoch: int,
    all_time_all_metrics,
    best_metric: float,
    scheduler: Optional[torch.optim.lr_scheduler.ReduceLROnPlateau] = None,
) -> None:
    """
    Save the latest training checkpoint for continuation or recovery.

    The checkpoint is always written to:
        {args.results_path}/last_{args.model_type}.ckpt

    This wraps `save_weights` and ensures the latest model/optimizer/scheduler
    states are recorded, along with the current epoch and best metric. In DDP,
    only rank 0 performs the save. Supports both standard and LoRA training.

    Args:
        all_time_all_metrics:
        args: Training arguments. Must define `results_path`, `model_type`,
              and `train_lora`.
        model: Model instance (may be wrapped by DDP/DataParallel).
        device_ids: List of GPU IDs used for training.
        optimizer: Optimizer whose state will be saved.
        epoch: Current training epoch.
        best_metric: Current best validation metric.
        scheduler: Optional learning rate scheduler to save state for.

    Returns:
        None
    """
    store_path = f"{args.results_path}/last_{args.model_type}.ckpt"
    save_weights(
        store_path,
        model,
        device_ids,
        optimizer,
        epoch,
        all_time_all_metrics,
        best_metric,
        scheduler,
        args.train_lora,
    )