Untitled

 avatar
unknown
plain_text
a month ago
3.4 kB
3
Indexable
    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