File size: 1,730 Bytes
387721d |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 |
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)
|