import pytz from pkg.tool import get_tweets_since_last_friday, format_time_str, get_time_since_last_tweet, get_hourly_weighted_array from pkg.dash.app_init import app from dash.dependencies import Input, Output from datetime import timedelta from datetime import datetime from dash import html @app.callback( [Output('info-tooltip', 'children')], [Input('clock-interval', 'n_intervals'), Input('target-input', 'value'), Input('predict-tweets-input', 'value'), Input('hour-input', 'value')] ) def update_info(n, target_value, increment_value, hour_value): pace = calculate_tweet_pace() decline_rates = calculate_pace_decline_rate() pace_increases = calculate_pace_increase_in_hour(increment_value, hour_value) tweet_count, days_to_next_friday = get_pace_params() remain_hours = days_to_next_friday * 24 now = tweet_count table_style_border = {'textAlign': 'center', 'border': '1px solid white'} table_style_c = {'textAlign': 'center'} table_style_l = {'textAlign': 'left'} target = int(target_value) if target_value is not None else None avg_tweets_per_day = calculate_avg_tweets_per_day(target, now, remain_hours) if target is not None else "TBD" days_passed = 7 - days_to_next_friday avg_tweets = round(tweet_count / days_passed, 2) if days_passed > 0 else 0 custom_increment = pace_increases.get('custom_increment') if custom_increment is None: custom_header = "[TBD]" custom_value = "TBD" else: increment = pace_increases['custom_increment_key'] hours = int(hour_value) custom_header = f"{increment}({hours}h)" custom_value = f"{custom_increment:.2f}" pace_table_rows = [ html.Tr([ html.Th('Pace', style=table_style_border), html.Th('DCL', style=table_style_border), html.Th('0(1h)', style=table_style_border), html.Th('1(1h)', style=table_style_border), html.Th('5(1h)', style=table_style_border), html.Th('10(1h)', style=table_style_border), html.Th('20(1h)', style=table_style_border), html.Th(custom_header, style=table_style_border) ]), html.Tr([ html.Td(f"{pace:.2f}", style=table_style_border), html.Td(decline_rates, style=table_style_border), html.Td(f"{pace_increases['increase_0']:.2f}", style=table_style_border), html.Td(f"{pace_increases['increase_1']:.2f}", style=table_style_border), html.Td(f"{pace_increases['increase_5']:.2f}", style=table_style_border), html.Td(f"{pace_increases['increase_10']:.2f}", style=table_style_border), html.Td(f"{pace_increases['increase_20']:.2f}", style=table_style_border), html.Td(custom_value, style=table_style_border) ]), html.Tr([ html.Td(f"Remain", colSpan=4, style=table_style_c), html.Td(format_time_str(days_to_next_friday), colSpan=4, style=table_style_l) ]), html.Tr([ html.Td(f"Last Tweet", colSpan=4, style=table_style_c), html.Td(format_time_str(get_time_since_last_tweet()), colSpan=4, style=table_style_l) ]), html.Tr([ html.Th("Target", colSpan=2, style=table_style_border), html.Th("Need's Avg Tweets Per Day", colSpan=4, style=table_style_border), html.Th("Avg Tweets", colSpan=2, style=table_style_border), ]), html.Tr([ html.Td(f"{target if target else '[TBD]'}", colSpan=2, style=table_style_border), html.Td(f"{avg_tweets_per_day}", colSpan=4, style=table_style_border), html.Td(f"{avg_tweets}", colSpan=2, style=table_style_border), ]) ] pace_table = html.Table(pace_table_rows, style={ 'width': '100%', 'textAlign': 'left', 'borderCollapse': 'collapse' }) return [pace_table] def get_pace_params(): est = pytz.timezone('US/Eastern') now = datetime.now(est) today = now.date() days_to_next_friday = (4 - today.weekday()) % 7 next_friday = (now.replace(hour=12, minute=0, second=0, microsecond=0) + timedelta(days=days_to_next_friday)) if now > next_friday: next_friday += timedelta(days=7) days_to_next_friday = (next_friday - now).total_seconds() / (24 * 60 * 60) tweet_count = get_tweets_since_last_friday() return tweet_count, days_to_next_friday def calculate_tweet_pace(): tweet_count, days_to_next_friday = get_pace_params() pace = (tweet_count / (7 - days_to_next_friday)) * days_to_next_friday + tweet_count return round(pace, 6) if pace > 0 else float(tweet_count) def calculate_pace_decline_rate(): tweet_count, days_to_next_friday = get_pace_params() T = 7 decline_per_day = -(tweet_count * T) / ((T - days_to_next_friday) ** 2) decline_per_hour = decline_per_day / 24 return round(decline_per_hour, 2) def calculate_pace_for_increment(increment, hours, tweet_count, days_to_next_friday, current_pace): future_days = days_to_next_friday - (hours / 24) new_tweet_count = tweet_count + increment if future_days <= 0: return round(new_tweet_count, 2) new_pace = (new_tweet_count / (7 - future_days)) * future_days + new_tweet_count return round(max(new_pace, new_tweet_count), 2) def calculate_pace_increase_in_hour(increment_value, hour_value): tweet_count, days_to_next_friday = get_pace_params() current_pace = (tweet_count / (7 - days_to_next_friday)) * days_to_next_friday + tweet_count increments = [0, 1, 5, 10, 20] pace_increases = {} for inc in increments: pace_increases[f'increase_{inc}'] = calculate_pace_for_increment( inc, 1, tweet_count, days_to_next_friday, current_pace ) if increment_value is None or hour_value is None: pace_increases['custom_increment'] = None else: increment = int(increment_value) hours = int(hour_value) pace_increases['custom_increment'] = calculate_pace_for_increment( increment, hours, tweet_count, days_to_next_friday, current_pace ) pace_increases['custom_increment_key'] = increment return pace_increases def calculate_avg_tweets_per_day(target, now, remain): Xi = get_hourly_weighted_array() if remain <= 0: return "remain<=0" if target <= now: return "Already reach" fx = max(remain - 12, 0) if remain > 12: fy = sum(Xi[0:12]) * 24 else: full_hours = int(remain) fractional_hour = remain - full_hours if full_hours >= 24: full_hours = 23 fractional_hour = 0 if full_hours < 0: full_hours = 0 if full_hours > 0: fy = sum(Xi[0:full_hours]) + Xi[full_hours] * fractional_hour else: fy = Xi[0] * fractional_hour fy *= 24 if fx + fy == 0: return "fx + fy = 0" result = (target - now) / ((fx + fy) / 24) return round(result, 2)