""" Utility helpers for the Eskom app. Provides: - safe CSV load - model metadata loader (from data/model_metadata.json) - saving predictions - byte-size formatter """ from typing import Optional, Dict, Any from pathlib import Path import pandas as pd import json import logging logger = logging.getLogger(__name__) def safe_read_csv(path: str, nrows: Optional[int] = None) -> Optional[pd.DataFrame]: p = Path(path) if not p.exists(): logger.debug("CSV not found: %s", path) return None try: return pd.read_csv(p, nrows=nrows) except Exception as e: logger.warning("Failed to read CSV %s: %s", path, e) return None def load_model_metadata(path: str = "data/model_metadata.json") -> Dict[str, Any]: """ Loads optional model metadata (training date, metrics, author, notes). If the file is absent or malformed, returns a sensible default dict. """ p = Path(path) default = { "model_type": "unknown", "training_date": None, "training_rows": None, "metrics": {}, "version": "1.0" } if not p.exists(): return default try: data = json.loads(p.read_text(encoding="utf-8")) # merge defaults for k, v in default.items(): if k not in data: data[k] = v return data except Exception as e: logger.warning("Failed to parse model metadata: %s", e) return default def save_predictions(df, out_path: str = "data/predictions.csv") -> bool: try: Path(out_path).parent.mkdir(parents=True, exist_ok=True) df.to_csv(out_path, index=False) return True except Exception as e: logger.exception("Failed to save predictions to %s: %s", out_path, e) return False def human_readable_bytes(num: int) -> str: """ Convert bytes to human-readable string. """ for unit in ["B", "KB", "MB", "GB", "TB"]: if num < 1024.0: return f"{num:.1f}{unit}" num /= 1024.0 return f"{num:.1f}PB"