File size: 2,152 Bytes
5863f1d | 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 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 | """
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"
|