| """
|
| 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"))
|
|
|
| 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"
|
|
|
