Untitled
unknown
plain_text
10 months ago
3.4 kB
5
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)Editor is loading...
Leave a Comment