modules/part_synthesis/utils/general_utils.py

"""
General utility functions for the TRELLIS project.

This module provides a collection of utility functions organized into several categories:
- Dictionary utilities: Operations for merging, traversal, reduction, and flattening of dictionaries
- Context management: Tools for managing nested context managers
- Image utilities: Functions for creating image grids and annotating images
- Debug utilities: Tools for numerical comparison and tolerance checking
- Print utilities: Helpers for text formatting and display
"""

import re
import numpy as np
import cv2
import torch
import contextlib


# Dictionary utils
def _dict_merge(dicta, dictb, prefix=''):
    """
    Merge two dictionaries recursively with conflict detection.
    
    Args:
        dicta: First dictionary to merge
        dictb: Second dictionary to merge
        prefix: Used for error reporting to track nested keys
        
    Returns:
        A new merged dictionary
        
    Raises:
        ValueError: If the same key exists in both dictionaries but has different types
    """
    assert isinstance(dicta, dict), 'input must be a dictionary'
    assert isinstance(dictb, dict), 'input must be a dictionary'
    dict_ = {}
    # Get all unique keys from both dictionaries
    all_keys = set(dicta.keys()).union(set(dictb.keys()))
    for key in all_keys:
        if key in dicta.keys() and key in dictb.keys():
            # If key exists in both, recursively merge if both are dictionaries
            if isinstance(dicta[key], dict) and isinstance(dictb[key], dict):
                dict_[key] = _dict_merge(dicta[key], dictb[key], prefix=f'{prefix}.{key}')
            else:
                # Raise error for conflicting non-dictionary values
                raise ValueError(f'Duplicate key {prefix}.{key} found in both dictionaries. Types: {type(dicta[key])}, {type(dictb[key])}')
        elif key in dicta.keys():
            # Copy values from first dictionary
            dict_[key] = dicta[key]
        else:
            # Copy values from second dictionary
            dict_[key] = dictb[key]
    return dict_


def dict_merge(dicta, dictb):
    """
    Merge two dictionaries.
    
    This is the public interface that wraps _dict_merge with default prefix.
    """
    return _dict_merge(dicta, dictb, prefix='')


def dict_foreach(dic, func, special_func={}):
    """
    Recursively apply a function to all non-dictionary leaf values in a dictionary.
    
    Args:
        dic: Dictionary to process
        func: Default function to apply to each leaf value
        special_func: Dictionary mapping keys to special functions for specific keys
        
    Returns:
        Transformed dictionary with function applied to all leaf values
    """
    assert isinstance(dic, dict), 'input must be a dictionary'
    for key in dic.keys():
        if isinstance(dic[key], dict):
            # Recursively process nested dictionaries
            dic[key] = dict_foreach(dic[key], func)
        else:
            # Apply special function if key is in special_func, otherwise use default
            if key in special_func.keys():
                dic[key] = special_func[key](dic[key])
            else:
                dic[key] = func(dic[key])
    return dic


def dict_reduce(dicts, func, special_func={}):
    """
    Reduce a list of dictionaries. Leaf values must be scalars.
    
    Args:
        dicts: List of dictionaries to reduce
        func: Default reduction function (takes a list of values, returns single value)
        special_func: Dictionary mapping keys to special reduction functions
        
    Returns:
        A single merged dictionary with values reduced according to the provided functions
    """
    assert isinstance(dicts, list), 'input must be a list of dictionaries'
    assert all([isinstance(d, dict) for d in dicts]), 'input must be a list of dictionaries'
    assert len(dicts) > 0, 'input must be a non-empty list of dictionaries'
    # Collect all unique keys across all dictionaries
    all_keys = set([key for dict_ in dicts for key in dict_.keys()])
    reduced_dict = {}
    for key in all_keys:
        # Extract values for this key from all dictionaries
        vlist = [dict_[key] for dict_ in dicts if key in dict_.keys()]
        if isinstance(vlist[0], dict):
            # Recursively reduce nested dictionaries
            reduced_dict[key] = dict_reduce(vlist, func, special_func)
        else:
            # Apply special function if key is in special_func, otherwise use default
            if key in special_func.keys():
                reduced_dict[key] = special_func[key](vlist)
            else:
                reduced_dict[key] = func(vlist)
    return reduced_dict


def dict_any(dic, func):
    """
    Check if any value in the dictionary satisfies the given predicate function.
    
    Args:
        dic: Dictionary to check
        func: Predicate function that returns True/False for each leaf value
        
    Returns:
        True if any leaf value satisfies the predicate, False otherwise
    
    dict any time: {'step': 16.795613527297974, 'elapsed': 16.795613527297974}
    dict any step: 16.795613527297974
    dict any elapsed: 16.795613527297974
    dict any loss: {'bin_3': {'mse': nan}, 'bin_5': {'mse': nan}, 'mse': nan, 'loss': nan, 'bin_4': {'mse': nan}, 'bin_7': {'mse': nan}, 'bin_8': {'mse': nan}}
    dict any bin_3: {'mse': nan}
    dict any mse: nan
    """
    assert isinstance(dic, dict), 'input must be a dictionary'
    for key in dic.keys():
        # print(f"dict any {key}: {dic[key]}")
        if isinstance(dic[key], dict):
            # Recursively check nested dictionaries
            if dict_any(dic[key], func):
                return True
        else:
            # Check current value against predicate
            if func(dic[key]):
                return True
    return False


def dict_all(dic, func):
    """
    Check if all values in the dictionary satisfy the given predicate function.
    
    Args:
        dic: Dictionary to check
        func: Predicate function that returns True/False for each leaf value
        
    Returns:
        True if all leaf values satisfy the predicate, False otherwise
    """
    assert isinstance(dic, dict), 'input must be a dictionary'
    for key in dic.keys():
        if isinstance(dic[key], dict):
            # Recursively check nested dictionaries
            if not dict_all(dic[key], func):
                return False
        else:
            # Check current value against predicate
            if not func(dic[key]):
                return False
    return True


def dict_flatten(dic, sep='.'):
    """
    Flatten a nested dictionary into a dictionary with no nested dictionaries.
    
    Args:
        dic: Dictionary to flatten
        sep: Separator string used to join key levels in the flattened dictionary
        
    Returns:
        A flattened dictionary with compound keys joined by the separator
    """
    assert isinstance(dic, dict), 'input must be a dictionary'
    flat_dict = {}
    for key in dic.keys():
        if isinstance(dic[key], dict):
            # Recursively flatten nested dictionaries and prefix with current key
            sub_dict = dict_flatten(dic[key], sep=sep)
            for sub_key in sub_dict.keys():
                flat_dict[str(key) + sep + str(sub_key)] = sub_dict[sub_key]
        else:
            # Copy leaf values directly
            flat_dict[key] = dic[key]
    return flat_dict


# Context utils
@contextlib.contextmanager
def nested_contexts(*contexts):
    """
    Create a single context manager from multiple context manager factories.
    
    This utility allows combining multiple context managers into a single
    context manager, simplifying code that needs to use multiple contexts.
    
    Args:
        *contexts: Context manager factory functions (functions that return context managers)
        
    Yields:
        Control to the inner block when all contexts are entered
    """
    with contextlib.ExitStack() as stack:
        for ctx in contexts:
            stack.enter_context(ctx())
        yield


# Image utils
def make_grid(images, nrow=None, ncol=None, aspect_ratio=None):
    """
    Arrange multiple images into a grid.
    
    Args:
        images: List of images to arrange
        nrow: Number of rows (calculated if not provided)
        ncol: Number of columns (calculated if not provided)
        aspect_ratio: Desired width/height ratio for the grid layout
        
    Returns:
        A single image containing the grid of input images
    """
    num_images = len(images)
    # Calculate grid dimensions if not explicitly provided
    if nrow is None and ncol is None:
        if aspect_ratio is not None:
            # Calculate rows to achieve desired aspect ratio
            nrow = int(np.round(np.sqrt(num_images / aspect_ratio)))
        else:
            # Default to a roughly square grid
            nrow = int(np.sqrt(num_images))
        ncol = (num_images + nrow - 1) // nrow
    elif nrow is None and ncol is not None:
        # Calculate rows based on fixed columns
        nrow = (num_images + ncol - 1) // ncol
    elif nrow is not None and ncol is None:
        # Calculate columns based on fixed rows
        ncol = (num_images + nrow - 1) // nrow
    else:
        assert nrow * ncol >= num_images, 'nrow * ncol must be greater than or equal to the number of images'
    
    # Create empty grid with appropriate dimensions
    if images[0].ndim == 2:
        # Grayscale images
        grid = np.zeros((nrow * images[0].shape[0], ncol * images[0].shape[1]), dtype=images[0].dtype)
    else:
        # Color images
        grid = np.zeros((nrow * images[0].shape[0], ncol * images[0].shape[1], images[0].shape[2]), dtype=images[0].dtype)
    
    # Place each image in the grid
    for i, img in enumerate(images):
        row = i // ncol
        col = i % ncol
        grid[row * img.shape[0]:(row + 1) * img.shape[0], col * img.shape[1]:(col + 1) * img.shape[1]] = img
    return grid


def notes_on_image(img, notes=None):
    """
    Add text notes to an image by padding the bottom and adding text.
    
    Args:
        img: Input image
        notes: Text to add at the bottom of the image
        
    Returns:
        Image with notes added
    """
    # Add padding at the bottom for the notes
    img = np.pad(img, ((0, 32), (0, 0), (0, 0)), 'constant', constant_values=0)
    img = cv2.cvtColor(img, cv2.COLOR_RGB2BGR)
    if notes is not None:
        # Add text to the padded area
        img = cv2.putText(img, notes, (0, img.shape[0] - 4), cv2.FONT_HERSHEY_SIMPLEX, 1, (255, 255, 255), 1)
    img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)
    return img


def save_image_with_notes(img, path, notes=None):
    """
    Save an image with optional text notes at the bottom.
    
    Args:
        img: Input image (numpy array or PyTorch tensor)
        path: File path to save the image
        notes: Optional text to add at the bottom of the image
    """
    # Convert PyTorch tensor to numpy if needed
    if isinstance(img, torch.Tensor):
        img = img.cpu().numpy().transpose(1, 2, 0)
    # Scale floating point images to 0-255 range
    if img.dtype == np.float32 or img.dtype == np.float64:
        img = np.clip(img * 255, 0, 255).astype(np.uint8)
    # Add notes to the image
    img = notes_on_image(img, notes)
    # Save with proper color conversion for OpenCV
    cv2.imwrite(path, cv2.cvtColor(img, cv2.COLOR_RGB2BGR))


# debug utils
def atol(x, y):
    """
    Absolute tolerance - computes absolute difference between x and y.
    
    Useful for numerical comparisons when absolute error matters.
    
    Args:
        x, y: Tensors to compare
        
    Returns:
        Absolute difference |x - y|
    """
    return torch.abs(x - y)


def rtol(x, y):
    """
    Relative tolerance - computes relative difference between x and y.
    
    Useful for numerical comparisons when relative error matters,
    especially when comparing values of different magnitudes.
    
    Args:
        x, y: Tensors to compare
        
    Returns:
        Relative difference |x - y| / max(|x|, |y|)
    """
    return torch.abs(x - y) / torch.clamp_min(torch.maximum(torch.abs(x), torch.abs(y)), 1e-12)


# print utils
def indent(s, n=4):
    """
    Indent a multi-line string.
    
    Args:
        s: Input string to indent
        n: Number of spaces to add before each line (except the first)
        
    Returns:
        Indented string with all lines except the first indented by n spaces
    """
    lines = s.split('\n')
    for i in range(1, len(lines)):
        lines[i] = ' ' * n + lines[i]
    return '\n'.join(lines)