← back to Watches
analytics/utils.py
60 lines
#!/usr/bin/env python3
"""
Utility functions for analytics modules
"""
import json
import numpy as np
import pandas as pd
class NumpyEncoder(json.JSONEncoder):
"""JSON encoder that handles numpy types"""
def default(self, obj):
if isinstance(obj, np.integer):
return int(obj)
elif isinstance(obj, np.floating):
return float(obj)
elif isinstance(obj, np.ndarray):
return obj.tolist()
elif isinstance(obj, np.bool_):
return bool(obj)
elif pd.isna(obj):
return None
return super(NumpyEncoder, self).default(obj)
def save_json(data, filepath):
"""Save data to JSON with numpy type handling"""
with open(filepath, 'w') as f:
json.dump(data, f, indent=2, cls=NumpyEncoder)
def convert_numpy_types(obj):
"""Recursively convert numpy types to Python native types"""
if isinstance(obj, dict):
return {k: convert_numpy_types(v) for k, v in obj.items()}
elif isinstance(obj, list):
return [convert_numpy_types(item) for item in obj]
elif isinstance(obj, np.integer):
return int(obj)
elif isinstance(obj, np.floating):
# Handle NaN and Infinity
val = float(obj)
if np.isnan(val) or np.isinf(val):
return None
return val
elif isinstance(obj, np.bool_):
return bool(obj)
elif isinstance(obj, np.ndarray):
return obj.tolist()
elif isinstance(obj, float):
# Handle regular float NaN/Inf
if np.isnan(obj) or np.isinf(obj):
return None
return obj
elif pd.isna(obj):
return None
else:
return obj