Spaces:
Running
Running
File size: 1,841 Bytes
b53ee19 | 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 | import pandas as pd
import json
from pathlib import Path
from datetime import datetime
from typing import Any, Dict, List
import threading
class PredictionLogger:
def __init__(self, log_dir: Path):
self.log_dir = Path(log_dir)
self.log_dir.mkdir(parents=True, exist_ok=True)
self.lock = threading.Lock()
def log_prediction(self,
input_data: Dict[str, Any],
prediction: Any,
model_version: str = "v1",
metadata: Dict[str, Any] = None):
"""Log a single prediction"""
log_entry = {
"timestamp": datetime.now().isoformat(),
"model_version": model_version,
"input": input_data,
"prediction": prediction,
"metadata": metadata or {}
}
log_file = self.log_dir / f"predictions_{datetime.now().strftime('%Y%m%d')}.jsonl"
with self.lock:
with open(log_file, 'a') as f:
f.write(json.dumps(log_entry) + '\n')
def load_predictions(self, date: str = None) -> List[Dict[str, Any]]:
"""Load predictions from log file"""
if date is None:
date = datetime.now().strftime('%Y%m%d')
log_file = self.log_dir / f"predictions_{date}.jsonl"
if not log_file.exists():
return []
predictions = []
with open(log_file, 'r') as f:
for line in f:
predictions.append(json.loads(line))
return predictions
def get_predictions_df(self, date: str = None) -> pd.DataFrame:
"""Get predictions as DataFrame"""
predictions = self.load_predictions(date)
return pd.DataFrame(predictions) if predictions else pd.DataFrame()
|