Untitled
import json from datetime import datetime from typing import Dict, List, Set import os class MetricsHandler: def __init__(self, metrics_file='metrics.json'): self.metrics_file = metrics_file self.metrics = self._load_metrics() def _load_metrics(self) -> Dict: """Load metrics from JSON file or create default structure""" if os.path.exists(self.metrics_file): try: with open(self.metrics_file, 'r') as f: return json.load(f) except json.JSONDecodeError: return self._create_default_metrics() return self._create_default_metrics() def _create_default_metrics(self) -> Dict: """Create default metrics structure""" return { 'basic_metrics': { 'total_items_sorted': 0, 'items_sorted_per_bin': {}, 'classification_results': [], # List of all classification results with details 'sort_timestamps': [], 'fill_levels_per_bin': {}, 'bin_emptying_counts': {}, 'api_vs_local_usage': {'api': 0, 'local': 0} }, 'time_metrics': { 'daily_usage_counts': {}, 'weekly_usage_counts': {}, 'monthly_usage_counts': {}, 'time_between_sorts': [], 'time_of_day_patterns': [0] * 24, # 24 hours 'daily_weekly_monthly_streaks': { 'daily': 0, 'weekly': 0, 'monthly': 0 }, 'time_to_empty_from_90': [], 'fill_rate_per_bin': {} }, 'short_term_tracking': { 'items_last_5_minutes': { 'count': 0, 'timestamps': [] }, 'current_sorting_streak': { 'start_date': None, 'days': 0, 'last_sort_date': None } }, 'hour_coverage': { 'hours_sorted': [], 'weekend_streak': { 'count': 0, 'dates': [] } }, 'seasonal_tracking': { 'seasonal_counts': { 'summer': 0, 'winter': 0, 'current_season': None, 'season_start': None } }, 'environmental_impact': { 'co2_saved': 0.0, 'trees_saved': 0.0, 'paper_weight_recycled': 0.0, 'plastic_weight_recycled': 0.0, 'organic_weight_processed': 0.0 }, 'fill_level_history': {}, 'achievements': {}, 'bin_specialization': {} } def save_metrics(self): """Save metrics to JSON file""" with open(self.metrics_file, 'w') as f: json.dump(self.metrics, f, indent=2) def record_sort(self, bin_id: str, classification_result: str, fill_level: float, classification_method: str): """Record a sorting event""" timestamp = datetime.now().isoformat() # Update basic metrics self.metrics['basic_metrics']['total_items_sorted'] += 1 self.metrics['basic_metrics']['items_sorted_per_bin'][bin_id] = \ self.metrics['basic_metrics']['items_sorted_per_bin'].get(bin_id, 0) + 1 # Record classification result with details sort_record = { 'timestamp': timestamp, 'bin_id': bin_id, 'classification_result': classification_result, 'fill_level': fill_level, 'method': classification_method } self.metrics['basic_metrics']['classification_results'].append(sort_record) self.metrics['basic_metrics']['sort_timestamps'].append(timestamp) self.metrics['basic_metrics']['fill_levels_per_bin'][bin_id] = fill_level self.metrics['basic_metrics']['api_vs_local_usage'][classification_method] += 1 # Update time metrics hour = datetime.now().hour self.metrics['time_metrics']['time_of_day_patterns'][hour] += 1 # Update short-term tracking self._update_short_term_tracking(timestamp) # Update hour coverage if hour not in self.metrics['hour_coverage']['hours_sorted']: self.metrics['hour_coverage']['hours_sorted'].append(hour) # Save updated metrics self.save_metrics() def record_bin_empty(self, bin_id: str): """Record when a bin is emptied""" timestamp = datetime.now().isoformat() self.metrics['basic_metrics']['bin_emptying_counts'][bin_id] = \ self.metrics['basic_metrics']['bin_emptying_counts'].get(bin_id, 0) + 1 if bin_id not in self.metrics['fill_level_history']: self.metrics['fill_level_history'][bin_id] = { 'emptying_timestamps': [] } self.metrics['fill_level_history'][bin_id]['emptying_timestamps'].append(timestamp) self.save_metrics() def _update_short_term_tracking(self, timestamp: str): """Update short-term tracking metrics""" current_time = datetime.now() # Clean up old timestamps five_minutes_ago = current_time.timestamp() - 300 # 5 minutes in seconds self.metrics['short_term_tracking']['items_last_5_minutes']['timestamps'] = [ ts for ts in self.metrics['short_term_tracking']['items_last_5_minutes']['timestamps'] if datetime.fromisoformat(ts).timestamp() > five_minutes_ago ] # Add new timestamp self.metrics['short_term_tracking']['items_last_5_minutes']['timestamps'].append(timestamp) self.metrics['short_term_tracking']['items_last_5_minutes']['count'] = \ len(self.metrics['short_term_tracking']['items_last_5_minutes']['timestamps']) def get_metrics(self) -> Dict: """Get all metrics""" return self.metrics def get_metric(self, category: str, metric: str) -> any: """Get a specific metric""" return self.metrics.get(category, {}).get(metric, None)
Leave a Comment