79 lines
2.6 KiB
Python
79 lines
2.6 KiB
Python
|
|
import os
|
|
import pickle
|
|
import logging
|
|
from datetime import datetime
|
|
from typing import Dict, Any
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
HISTORY_FILE = "rider_history.pkl"
|
|
|
|
class RiderHistoryService:
|
|
def __init__(self, history_file: str = HISTORY_FILE):
|
|
self.history_file = history_file
|
|
self.history = self._load_history()
|
|
|
|
def _load_history(self) -> Dict[int, Dict[str, float]]:
|
|
"""Load history from pickle file."""
|
|
if not os.path.exists(self.history_file):
|
|
return {}
|
|
|
|
try:
|
|
with open(self.history_file, 'rb') as f:
|
|
return pickle.load(f)
|
|
except Exception as e:
|
|
logger.error(f"Failed to load rider history: {e}")
|
|
return {}
|
|
|
|
def _save_history(self):
|
|
"""Save history to pickle file."""
|
|
try:
|
|
with open(self.history_file, 'wb') as f:
|
|
pickle.dump(self.history, f)
|
|
except Exception as e:
|
|
logger.error(f"Failed to save rider history: {e}")
|
|
|
|
def update_rider_stats(self, rider_id: int, distance_km: float, order_count: int):
|
|
"""Update cumulative stats for a rider."""
|
|
rider_id = int(rider_id)
|
|
if rider_id not in self.history:
|
|
self.history[rider_id] = {
|
|
"total_km": 0.0,
|
|
"total_orders": 0,
|
|
"last_updated": datetime.now().isoformat()
|
|
}
|
|
|
|
self.history[rider_id]["total_km"] += distance_km
|
|
self.history[rider_id]["total_orders"] += order_count
|
|
self.history[rider_id]["last_updated"] = datetime.now().isoformat()
|
|
|
|
# Auto-save on update
|
|
self._save_history()
|
|
|
|
def get_rider_score(self, rider_id: int) -> float:
|
|
"""
|
|
Get a score representing the rider's historical 'load' (KMs).
|
|
Higher Score = More KMs driven recently.
|
|
"""
|
|
rider_id = int(rider_id)
|
|
stats = self.history.get(rider_id, {})
|
|
return stats.get("total_km", 0.0)
|
|
|
|
def get_preferred_assignment_type(self, rider_id: int, all_rider_scores: Dict[int, float]) -> str:
|
|
"""
|
|
Determine if rider should get 'Long' or 'Short' routes based on population average.
|
|
"""
|
|
score = self.get_rider_score(rider_id)
|
|
if not all_rider_scores:
|
|
return "ANY"
|
|
|
|
avg_score = sum(all_rider_scores.values()) / len(all_rider_scores)
|
|
|
|
# If rider has driven LESS than average, prefer LONG routes (Risky)
|
|
if score < avg_score:
|
|
return "LONG"
|
|
# If rider has driven MORE than average, prefer SHORT routes (Economy)
|
|
else:
|
|
return "SHORT"
|