Untitled
def _update_time_metrics(self, timestamp: str, bin_id: str): """Update time-based metrics with fixed streak tracking""" current_time = datetime.fromisoformat(timestamp) # Update daily/weekly/monthly counts date_str = current_time.date().isoformat() week_str = f"{current_time.year}-W{current_time.isocalendar()[1]}" month_str = f"{current_time.year}-{current_time.month:02d}" # Update usage counts self.metrics['time_metrics']['daily_usage_counts'][date_str] = \ self.metrics['time_metrics']['daily_usage_counts'].get(date_str, 0) + 1 self.metrics['time_metrics']['weekly_usage_counts'][week_str] = \ self.metrics['time_metrics']['weekly_usage_counts'].get(week_str, 0) + 1 self.metrics['time_metrics']['monthly_usage_counts'][month_str] = \ self.metrics['time_metrics']['monthly_usage_counts'].get(month_str, 0) + 1 # Update streaks self._update_streaks(current_time) # Update time of day patterns hour = current_time.hour self.metrics['time_metrics']['time_of_day_patterns'][hour] += 1 # Update hour coverage if hour not in self.metrics['hour_coverage']['hours_sorted']: self.metrics['hour_coverage']['hours_sorted'].append(hour) # Update weekend tracking with improved consecutive weekend detection if current_time.weekday() >= 5: # Saturday = 5, Sunday = 6 weekend_dates = self.metrics['hour_coverage']['weekend_streak']['dates'] # Get the weekend date (Saturday) for current sort current_weekend = (current_time.date() - timedelta(days=current_time.weekday() - 5)) # Only add if it's a new weekend if not weekend_dates or \ (current_weekend != (datetime.fromisoformat(weekend_dates[-1]).date() - timedelta(days=datetime.fromisoformat(weekend_dates[-1]).weekday() - 5))): weekend_dates.append(timestamp) # Check for consecutive weekends if len(weekend_dates) >= 2: prev_weekend = datetime.fromisoformat(weekend_dates[-2]).date() prev_weekend = prev_weekend - timedelta(days=prev_weekend.weekday() - 5) weeks_diff = (current_weekend - prev_weekend).days / 7 if weeks_diff == 1: # Consecutive weekends self.metrics['hour_coverage']['weekend_streak']['count'] += 1 elif weeks_diff > 1: # Streak broken self.metrics['hour_coverage']['weekend_streak']['count'] = 1 # Track weekly bin usage if 'weekly_bin_usage' not in self.metrics['time_metrics']: self.metrics['time_metrics']['weekly_bin_usage'] = {} if week_str not in self.metrics['time_metrics']['weekly_bin_usage']: self.metrics['time_metrics']['weekly_bin_usage'][week_str] = set() # Convert set to list for JSON serialization current_bins = set(self.metrics['time_metrics']['weekly_bin_usage'][week_str]) current_bins.add(bin_id) self.metrics['time_metrics']['weekly_bin_usage'][week_str] = list(current_bins)
Leave a Comment