json_utils.py

import numpy as np
import pandas as pd
import logging

logger = logging.getLogger(__name__)

def convert_numpy_types(obj):
    """
    Recursively convert NumPy types to standard Python types for JSON serialization.
    """
    if isinstance(obj, (np.int_, np.intc, np.intp, np.int8,
                        np.int16, np.int32, np.int64, np.uint8,
                        np.uint16, np.uint32, np.uint64)):
        return int(obj)
    elif isinstance(obj, (np.float_, np.float16, np.float32, np.float64)):
        return float(obj)
    elif isinstance(obj, (np.ndarray,)):
        return obj.tolist()
    elif isinstance(obj, dict):
        return {k: convert_numpy_types(v) for k, v in obj.items()}
    elif isinstance(obj, list):
        return [convert_numpy_types(i) for i in obj]
    elif isinstance(obj, (pd.Timestamp, pd.DatetimeIndex)):
        return str(obj)
    return obj

def log_object_types(obj, prefix="", depth=0, max_depth=3):
    """
    Log the types of keys and values in a dictionary or list to help debug serialization issues.
    """
    if depth > max_depth:
        return

    indent = "  " * depth
    if isinstance(obj, dict):
        for k, v in obj.items():
            logger.error(f"{indent}{prefix}Key: {k} ({type(k)}), Value Type: {type(v)}")
            if isinstance(v, (dict, list)):
                log_object_types(v, prefix=f"{k}.", depth=depth+1, max_depth=max_depth)
    elif isinstance(obj, list):
        for i, v in enumerate(obj[:5]): # Only check first 5 to avoid spam
            logger.error(f"{indent}{prefix}Index: {i}, Value Type: {type(v)}")
            if isinstance(v, (dict, list)):
                log_object_types(v, prefix=f"[{i}].", depth=depth+1, max_depth=max_depth)