From 05d2c7c64fb8fbfc70f246f5de2eb9927ef3dd9f Mon Sep 17 00:00:00 2001 From: strhwste <127418227+strhwste@users.noreply.github.com> Date: Tue, 16 Dec 2025 05:24:31 +0100 Subject: [PATCH] Major stats and reporting enhancements for Spotify data Refactored data loading to use JSON and improved filtering in filemgr/data.py. Added extensive new statistics and reporting features in main.py, including detailed stats, CSV export, and PDF/graph generation. Expanded stats/functions.py with many new analytics functions (artist, album, time, skip, session, and fun stats). Updated .gitignore for new data/report files. Adjusted filemgr/functions.py and types.py for new data structure and fields. Added new stats/graphs.py module for plotting. This update significantly expands the analysis and reporting capabilities for Spotify streaming history. --- .DS_Store | Bin 0 -> 8196 bytes .gitignore | 5 + filemgr/.DS_Store | Bin 0 -> 6148 bytes filemgr/data.py | 36 +- filemgr/functions.py | 6 +- filemgr/types.py | 7 + main.py | 629 ++++++++++++++- stats/.DS_Store | Bin 0 -> 6148 bytes stats/functions.py | 1812 +++++++++++++++++++++++++++++++++++++++++- stats/graphs.py | 449 +++++++++++ 10 files changed, 2927 insertions(+), 17 deletions(-) create mode 100644 .DS_Store create mode 100644 filemgr/.DS_Store create mode 100644 stats/.DS_Store create mode 100644 stats/graphs.py diff --git a/.DS_Store b/.DS_Store new file mode 100644 index 0000000000000000000000000000000000000000..4cf270ab37e100ef59bfe065eac7ac7aaea055fe GIT binary patch literal 8196 zcmeHMzl#$=6n=BbnZPw`A*2v)5la!SQN(6F61egp7K(_MyO>-caYGW1oYh)i<&}aU zc48x9Yc1MY+V}?uVrQd;-UYn>3Y-)Lux4|s&RO@JS4UMq z75Fa|;O9e(&X{=29NI?*2D=3SmSMIH#}OX@V~>f)%pp8z#-##Xs_|P4*oz|q ze%bQNc*%yE%aT~ z0*kDL=<2umk+QCA^}>^>t831@vc|th%X;18pDNVB_3ER_0q&PkSm%xJsPnf7S>LPY zw;%cImW!HcH@begQ5+T4^OE&DM@^`g(gam?pkho7n&J~*QNF9%ur%>`gyR-!?!n%L zZ39PR*i(4+J)RjUQX#!@XR-TV}K z_do_Ntn0>tuK!L39aVv2D^PNoHQxU>=D+_x z_WYWt3aA1nUjb9DH|zY9+pn#;ikD!C(YMgKaa`umreNR@w%hGE@aP|gIF7w9@t8S; R2TflDSQ&Is1^%f5KLH7t7F_@U literal 0 HcmV?d00001 diff --git a/.gitignore b/.gitignore index 0341498..9dd4325 100644 --- a/.gitignore +++ b/.gitignore @@ -132,3 +132,8 @@ dmypy.json *.zip MyData/ /MyData/ +Spotify Extended Streaming History/ +*.pdf +*.csv +stats_report.html +stats_report.md diff --git a/filemgr/.DS_Store b/filemgr/.DS_Store new file mode 100644 index 0000000000000000000000000000000000000000..e042ad8ce728ab31d7e387710e480c20fb3b4ece GIT binary patch literal 6148 zcmeHKu}%Xq47H)dNnJWJW<_G@#u%zAVnpf(9Q2Cl(CSV?Eg&)E0~q-yev9YW6iyr+ z7!X3X6h9}i6W^0AiHV3Oo5h%DL_`IeAdAr>GTb|LV9o;|=QV1&rka*?Lsid0e{o21 zA0pd|T6#q8ANg0S+wrV!)~%*}^8CH$x+%-~s+q&1IKMxBd3ty~+vmO5d0%W#r+0h3 z+vVphE;Sj~&VV!E3^)VMz|R@Loh?!f69XlVUzPpmGTSlxK7n z*iuVKOfZawNf9#;)=;2^vXvOD;TR9*7Y&o5h7()y!M5{z@xpm`tRJ#Faa45e3^)T_ z1`f45lKX#(U#7Rn?}m8K8E^*v83Q~hXXOMRWq0eB&y%}0pk1Mfh+h%|0=x4FKu6Az fqdckeL2UR%!=xyy$aoG1`a>WS;>sEL1qR*#hrKc` literal 0 HcmV?d00001 diff --git a/filemgr/data.py b/filemgr/data.py index c4e768f..3799954 100644 --- a/filemgr/data.py +++ b/filemgr/data.py @@ -1,3 +1,4 @@ +import json from dataclasses import dataclass from datetime import timedelta from os import listdir @@ -51,17 +52,36 @@ def my_library(self): # data loaders (from file) def _load_streaming_history(self): - files = list(filter(lambda x: x.startswith('StreamingHistory'), listdir(self._root))) + files = list(filter(lambda x: x.startswith('Streaming_History'), listdir(self._root))) files.sort() all_ = [] for file in files: with open(self._root + file, encoding='UTF-8') as f: - all_ += eval(f.read()) + data = json.load(f) + for item in data: + # Filter out podcasts and audiobooks + if (item.get('master_metadata_track_name') + and not item.get('episode_name') + and not item.get('audiobook_title')): + + all_.append({ + 'endTime': item['ts'], + 'artistName': item['master_metadata_album_artist_name'], + 'albumName': item['master_metadata_album_album_name'], + 'trackName': item['master_metadata_track_name'], + 'msPlayed': item['ms_played'], + 'platform': item['platform'], + 'connCountry': item['conn_country'], + 'reasonStart': item['reason_start'], + 'reasonEnd': item['reason_end'], + 'shuffle': item['shuffle'], + 'skipped': item['skipped'] + }) # Convert string timestamps to time objects for h in all_: - h['endTime'] = strptime(h['endTime'], '%Y-%m-%d %H:%M') + h['endTime'] = strptime(h['endTime'], '%Y-%m-%dT%H:%M:%SZ') h['msPlayed'] = timedelta(milliseconds=h['msPlayed']) return all_ @@ -84,7 +104,9 @@ def _load_playlists(self): return all_ def _load_my_library(self): - with open(self._root + 'YourLibrary.json', encoding='UTF-8') as f: - all_ = eval(f.read()) - - return all_['tracks'], all_['albums'], all_['shows'], all_['episodes'], all_['artists'] + try: + with open(self._root + 'YourLibrary.json', encoding='UTF-8') as f: + all_ = eval(f.read()) + return all_['tracks'], all_['albums'], all_['shows'], all_['episodes'], all_['artists'] + except FileNotFoundError: + return [], [], [], [], [] diff --git a/filemgr/functions.py b/filemgr/functions.py index 818ecf3..ddb658d 100644 --- a/filemgr/functions.py +++ b/filemgr/functions.py @@ -5,7 +5,7 @@ def _extract_data(path: str) -> None: - if not os.path.exists('MyData'): + if not os.path.exists('Spotify Extended Streaming History'): with ZipFile(path) as zip_: # todo: check extracted location & modifiable root path zip_.extractall() @@ -18,6 +18,4 @@ def load_zipped_data(path: str = 'my_spotify_data.zip') -> MyData: :param path: relative path to zip file ('my_spotify_data.zip' default) """ _extract_data(path) - - # todo: modifiable root path - return MyData() + return MyData(root_path='Spotify Extended Streaming History/') diff --git a/filemgr/types.py b/filemgr/types.py index d1ba351..39e866e 100644 --- a/filemgr/types.py +++ b/filemgr/types.py @@ -28,5 +28,12 @@ class PlayList(TypedDict): class History(TypedDict): endTime: struct_time # str # yyyy-mm-dd HH:MM artistName: str + albumName: str trackName: str msPlayed: timedelta # int + platform: str + connCountry: str + reasonStart: str + reasonEnd: str + shuffle: bool + skipped: bool diff --git a/main.py b/main.py index 2b15cf2..ec1ba36 100644 --- a/main.py +++ b/main.py @@ -1,8 +1,55 @@ import time import traceback +import sys +import csv +from io import StringIO +import matplotlib.pyplot as plt +from matplotlib.backends.backend_pdf import PdfPages from filemgr import load_zipped_data -from stats import history_range, play_time, play_counts +from stats import ( + history_range, play_time, play_counts, play_counts_by_artist, play_counts_by_album, + artist_history_over_time, platform_usage, listening_by_hour, skipped_ratio, + location_counts, most_skipped_artist, most_skipped_track, longest_played_artist, top_artist_per_month, + listening_by_day_of_week, discovery_rate, seasonal_listening, one_hit_wonders, day_night_split, longest_listening_streak, + true_skip_rate, most_consecutive_plays, album_loyalty, longest_artist_relationship, forgotten_favorites, hourly_heatmap_data, + most_musical_day, weekend_vs_weekday, immediate_skips, variety_score, + listening_velocity, the_comeback, clockwork_artists, session_analysis, skipless_albums, sampler_vs_completionist, + commute_heroes, marathon_tracks, one_week_wonders, sound_of_silence, + calendar_heatmap_data, picky_grid_data, device_habits_data, artist_eras_data, + control_freak_data, shuffle_paradox_data, natural_death_data, + active_listening_heatmap_data, active_listening_trend_data, + get_artist_traits, + night_shift_artists, new_years_transitions, consistency_king, alphabet_challenge, longest_played_tracks, + obsession_score, early_bird_artists, nine_to_five_artists, party_animal_tracks, sunday_scaries_tracks, + unskippable_streak, artist_hopper, discovery_peak, comfort_zone, single_day_record, + manual_laborer, shuffle_roulette, session_starter, session_closer, quick_fix, skippers_remorse, + remix_junkie, live_fanatic, short_king, epic_saga, collaborator, alphabet_artists, spelling_bee, same_name_game, + midnight_club, lunch_break, monday_blues, hump_day_hero, quarterly_review, album_purist, + instant_skips, get_full_song_stats +) +from stats.graphs import ( + plot_top_items, plot_artist_trends, plot_platform_usage, plot_listening_by_hour, + plot_location_counts, plot_longest_played_artist, plot_skipped_items, + plot_listening_by_day_of_week, plot_discovery_rate, plot_seasonal_listening, plot_day_night_split, plot_hourly_heatmap, + plot_variety_score, + plot_calendar_heatmap, plot_picky_grid, plot_device_habits, plot_artist_eras, + plot_active_heatmap, plot_active_trend, + plot_artist_radar, + create_text_pages, + plot_longest_played_tracks, + plot_comfort_zone +) + +class Tee(object): + def __init__(self, *files): + self.files = files + def write(self, obj): + for f in self.files: + f.write(obj) + def flush(self): + for f in self.files: + f.flush() if __name__ == '__main__': try: @@ -17,6 +64,11 @@ start = start if temp_start == '' else temp_start end = end if temp_end == '' else temp_end + # Start capturing output + captured_output = StringIO() + original_stdout = sys.stdout + sys.stdout = Tee(sys.stdout, captured_output) + print(f' ===> Statistics from [{time.strftime("%Y-%m-%d %H:%M", start)}] to [{time.strftime("%Y-%m-%d %H:%M", end)}]: \n') total_time = play_time(DATA.streaming_history, start, end) @@ -25,13 +77,582 @@ print(f'{total_time!s} or {int(total_time.total_seconds() / 60)} minutes') print('\n') - play_counts = play_counts(DATA.streaming_history) - + # Tracks + track_counts = play_counts(DATA.streaming_history) print('Top Played Tracks:') - for i, k, v in zip(range(1, 21), play_counts.keys(), play_counts.values()): + for i, k, v in zip(range(1, 101), track_counts.keys(), track_counts.values()): + print(f'#{i:2} - {v:4} : {k}') + print('\n') + + # Artists + artist_counts = play_counts_by_artist(DATA.streaming_history) + print('Top Played Artists:') + for i, k, v in zip(range(1, 101), artist_counts.keys(), artist_counts.values()): + print(f'#{i:2} - {v:4} : {k}') + print('\n') + + # Albums + album_counts = play_counts_by_album(DATA.streaming_history) + print('Top Played Albums:') + for i, k, v in zip(range(1, 101), album_counts.keys(), album_counts.values()): + print(f'#{i:2} - {v:4} : {k}') + print('\n') + + # Location + locations = location_counts(DATA.streaming_history) + print('Top Locations:') + for k, v in locations.items(): + print(f'{k}: {v}') + print('\n') + + # Longest Played Artists + time_artists = longest_played_artist(DATA.streaming_history) + print('Top Artists by Time Played:') + for i, (k, v) in enumerate(list(time_artists.items())[:20], 1): + print(f'#{i:2} - {int(v.total_seconds() // 3600)}h {int((v.total_seconds() % 3600) // 60)}m : {k}') + print('\n') + + # Longest Played Tracks + time_tracks = longest_played_tracks(DATA.streaming_history) + print('Top Tracks by Time Played:') + for i, (k, v) in enumerate(list(time_tracks.items())[:20], 1): + print(f'#{i:2} - {int(v.total_seconds() // 3600)}h {int((v.total_seconds() % 3600) // 60)}m : {k}') + print('\n') + + # Most Skipped + skipped_artists = most_skipped_artist(DATA.streaming_history) + print('Most Skipped Artists:') + for i, (k, v) in enumerate(list(skipped_artists.items())[:100], 1): + print(f'#{i:2} - {v:4} : {k}') + print('\n') + + # Top Artist Per Month + print('Top Artist Per Month:') + monthly_tops = top_artist_per_month(DATA.streaming_history) + for month, (artist, count) in monthly_tops.items(): + print(f'{month}: {artist} ({count} plays)') + print('\n') + + # One Hit Wonders + one_hits = one_hit_wonders(DATA.streaming_history) + print('Top One-Hit Wonders (Artists with > 5 plays but only 1 unique track):') + for i, (artist, (track, count)) in enumerate(list(one_hits.items())[:100], 1): + print(f'#{i:2} - {artist} : {track} ({count} plays)') + print('\n') + + # Longest Streak + streak_start, streak_end, streak_duration = longest_listening_streak(DATA.streaming_history) + if streak_start: + print(f'Longest Listening Streak:') + print(f'Duration: {streak_duration}') + print(f'From: {streak_start}') + print(f'To: {streak_end}') + print('\n') + + # True Skip Rate + print('True Skip Rate (Artists with > 20 plays):') + skip_rates = true_skip_rate(DATA.streaming_history) + for i, (k, v) in enumerate(list(skip_rates.items())[:100], 1): + print(f'#{i:2} - {v:5.1f}% : {k}') + print('\n') + + # Consecutive Plays + streak_track, streak_count = most_consecutive_plays(DATA.streaming_history) + print(f'Most Consecutive Plays: {streak_track} ({streak_count} times in a row)\n') + + # Album Loyalty + print('Top Albums by "Loyalty" (Listening to >3 tracks in a row):') + loyal_albums = album_loyalty(DATA.streaming_history) + for i, (k, v) in enumerate(list(loyal_albums.items())[:100], 1): + print(f'#{i:2} - {v:4} sessions : {k}') + print('\n') + + # Longest Relationship + print('Longest Artist Relationships:') + relationships = longest_artist_relationship(DATA.streaming_history) + for i, (k, v) in enumerate(list(relationships.items())[:100], 1): + days = v.days + print(f'#{i:2} - {days:4} days : {k}') + print('\n') + + # Forgotten Favorites + print('Forgotten Favorites (Popular > 6 months ago, 0 plays recently):') + forgotten = forgotten_favorites(DATA.streaming_history) + for i, (k, v) in enumerate(forgotten[:100], 1): + print(f'#{i:2} - {v:4} past plays : {k}') + print('\n') + + # Most Musical Day + musical_day, musical_time = most_musical_day(DATA.streaming_history) + print(f'Most Musical Day: {musical_day} ({int(musical_time.total_seconds() // 3600)}h {int((musical_time.total_seconds() % 3600) // 60)}m)\n') + + # Weekend vs Weekday + weekday_top, weekend_top = weekend_vs_weekday(DATA.streaming_history) + print('Top 5 Weekday Artists:') + for i, (k, v) in enumerate(list(weekday_top.items())[:10], 1): + print(f'#{i:2} - {v:4} : {k}') + print('Top 5 Weekend Artists:') + for i, (k, v) in enumerate(list(weekend_top.items())[:10], 1): print(f'#{i:2} - {v:4} : {k}') print('\n') + # Immediate Skips + print('Top "Nope" Songs (Skipped < 30s):') + nopes = immediate_skips(DATA.streaming_history) + for i, (k, v) in enumerate(list(nopes.items())[:50], 1): + print(f'#{i:2} - {v:4} skips : {k}') + print('\n') + + # Variety Score + print('Variety Score (Unique Artists / Total Plays):') + variety = variety_score(DATA.streaming_history) + for year, score in variety.items(): + print(f'{year}: {score:.3f}') + print('\n') + + # Listening Velocity + print('Listening Velocity (Fastest to 100 plays):') + velocity = listening_velocity(DATA.streaming_history) + for i, (k, v) in enumerate(list(velocity.items())[:50], 1): + print(f'#{i:2} - {v:4} days : {k}') + print('\n') + + # The Comeback + print('The Comeback (Artists with > 1 year gap):') + comebacks = the_comeback(DATA.streaming_history) + for i, (k, v) in enumerate(list(comebacks.items())[:50], 1): + print(f'#{i:2} - {v:4} days gap : {k}') + print('\n') + + # Clockwork Artists + print('Clockwork Artists (>70% plays in 4h window):') + clockwork = clockwork_artists(DATA.streaming_history) + for i, (k, v) in enumerate(list(clockwork.items())[:30], 1): + print(f'#{i:2} - {v} : {k}') + print('\n') + + # Session Analysis + avg_sess_dur, avg_sess_tracks = session_analysis(DATA.streaming_history) + print(f'Average Session: {avg_sess_dur:.1f} minutes, {avg_sess_tracks:.1f} tracks\n') + + # Skipless Albums + print('Skipless Albums (Lowest skip rate, min 50 plays):') + skipless = skipless_albums(DATA.streaming_history) + for i, (k, v) in enumerate(list(skipless.items())[:50], 1): + print(f'#{i:2} - {v:5.1f}% skips : {k}') + print('\n') + + # Sampler vs Completionist + print('Completionist Score (Unique Tracks / Total Plays):') + completionist = sampler_vs_completionist(DATA.streaming_history) + for i, (k, v) in enumerate(list(completionist.items())[:10], 1): + print(f'#{i:2} - {v:.2f} : {k}') + print('\n') + + # Commute Heroes + print('Commute Heroes (Mon-Fri 7-9am & 5-7pm):') + commuters = commute_heroes(DATA.streaming_history) + for i, (k, v) in enumerate(list(commuters.items())[:10], 1): + print(f'#{i:2} - {v:4} plays : {k}') + print('\n') + + # Marathon Tracks + print('Marathon Tracks (> 5 mins):') + marathons = marathon_tracks(DATA.streaming_history) + for i, (k, v) in enumerate(list(marathons.items())[:100], 1): + print(f'#{i:2} - {v:4} plays : {k}') + print('\n') + + # One Week Wonders + print('One Week Wonders (>50 plays in 1 week, <10 others):') + wonders = one_week_wonders(DATA.streaming_history) + for i, (k, v) in enumerate(list(wonders.items())[:100], 1): + print(f'#{i:2} - {v} : {k}') + print('\n') + + # Sound of Silence + silence_start, silence_end, silence_days = sound_of_silence(DATA.streaming_history) + print(f'Sound of Silence (Longest gap): {silence_days} days') + print(f'From: {silence_start}') + print(f'To: {silence_end}\n') + + # Control Freak + cf_data, (cf_artist, cf_count) = control_freak_data(DATA.streaming_history) + print('Control Freak vs Passenger:') + print(f"Active Listening (You clicked): {cf_data['active']:.1f}%") + print(f"Passive Listening (Autoplay/Queue): {cf_data['passive']:.1f}%") + print(f"Most Clicked Artist: {cf_artist} ({cf_count} active starts)") + print('\n') + + # Shuffle Paradox + shuf_data = shuffle_paradox_data(DATA.streaming_history) + print('The Shuffle Paradox:') + print(f"Skip Rate with Shuffle ON: {shuf_data['shuffle_skip_rate']:.1f}%") + print(f"Skip Rate with Shuffle OFF: {shuf_data['normal_skip_rate']:.1f}%") + print('\n') + + # Natural Death + nat_data, respected_artists = natural_death_data(DATA.streaming_history) + print("The 'Natural Causes' Death Rate:") + print(f"Songs finished naturally: {nat_data['natural']:.1f}%") + print(f"Songs killed by user: {nat_data['killed']:.1f}%") + print('Top "Respected" Artists (Highest completion rate):') + for i, (k, v) in enumerate(respected_artists, 1): + print(f'#{i:2} - {v:.1f}% finished : {k}') + print('\n') + + # Skipped Ratio + skipped_stats = skipped_ratio(DATA.streaming_history) + print('Skipped Ratio:') + for k, v in skipped_stats.items(): + print(f'{k}: {v}') + print('\n') + + # Night Shift + print('The Night Shift (Top Artists 2 AM - 5 AM):') + night_artists = night_shift_artists(DATA.streaming_history) + for i, (k, v) in enumerate(list(night_artists.items())[:20], 1): + print(f'#{i:2} - {v:4} plays : {k}') + print('\n') + + # New Year's Transitions + print("New Year's Transitions:") + transitions = new_years_transitions(DATA.streaming_history) + for year, (last, first) in transitions.items(): + print(f"{year} -> {year+1}:") + print(f" Last: {last}") + print(f" First: {first}") + print('\n') + + # Consistency King + print('Consistency King (Tracks played on most unique days):') + consistent_tracks = consistency_king(DATA.streaming_history) + for i, (k, v) in enumerate(list(consistent_tracks.items())[:20], 1): + print(f'#{i:2} - {v:4} days : {k}') + print('\n') + + # Alphabet Challenge + print('The Alphabet Challenge (Most played track for A-Z):') + alphabet_data = alphabet_challenge(DATA.streaming_history) + for char in "ABCDEFGHIJKLMNOPQRSTUVWXYZ": + if char in alphabet_data: + track, count = alphabet_data[char] + print(f"{char}: {track} ({count} plays)") + else: + print(f"{char}: -") + print('\n') + + # Obsession Score + print('Obsession Score (Plays / Unique Days):') + obsessions = obsession_score(DATA.streaming_history) + for i, (k, v) in enumerate(list(obsessions.items())[:10], 1): + print(f'#{i:2} - {v:.2f} : {k}') + print('\n') + + # Early Bird + print('The Early Bird (Top Artists 5 AM - 9 AM):') + early_birds = early_bird_artists(DATA.streaming_history) + for i, (k, v) in enumerate(list(early_birds.items())[:10], 1): + print(f'#{i:2} - {v:4} plays : {k}') + print('\n') + + # 9-to-5 + print('The 9-to-5 (Top Artists Mon-Fri 9 AM - 5 PM):') + workers = nine_to_five_artists(DATA.streaming_history) + for i, (k, v) in enumerate(list(workers.items())[:10], 1): + print(f'#{i:2} - {v:4} plays : {k}') + print('\n') + + # Party Animal + print('The Party Animal (Top Tracks Fri/Sat 10 PM - 4 AM):') + party_tracks = party_animal_tracks(DATA.streaming_history) + for i, (k, v) in enumerate(list(party_tracks.items())[:20], 1): + print(f'#{i:2} - {v:4} plays : {k}') + print('\n') + + # Sunday Scaries + print('The Sunday Scaries (Top Tracks Sun 6 PM - Midnight):') + scary_tracks = sunday_scaries_tracks(DATA.streaming_history) + for i, (k, v) in enumerate(list(scary_tracks.items())[:20], 1): + print(f'#{i:2} - {v:4} plays : {k}') + print('\n') + + # Unskippable Streak + streak_count, streak_start, streak_end = unskippable_streak(DATA.streaming_history) + print(f'Unskippable Streak: {streak_count} songs in a row') + print(f'From: {streak_start}') + print(f'To: {streak_end}\n') + + # Artist Hopper + hopper_score = artist_hopper(DATA.streaming_history) + print(f'Artist Hopper Score (Avg consecutive plays per artist): {hopper_score:.2f}') + print('(Low = Shuffle lover, High = Album listener)\n') + + # Discovery Peak + peak_month, peak_count = discovery_peak(DATA.streaming_history) + print(f'Discovery Peak: {peak_month} ({peak_count} new artists)\n') + + # Comfort Zone + comfort_pct, comfort_top10 = comfort_zone(DATA.streaming_history) + print(f'The Comfort Zone: {comfort_pct:.1f}% of time spent on Top 10 Artists') + print('Top 10 Contributors:') + for i, (artist, duration) in enumerate(comfort_top10, 1): + print(f'#{i:2} - {int(duration.total_seconds() // 3600)}h : {artist}') + print('\n') + + # Single Day Record + rec_artist, rec_date, rec_time = single_day_record(DATA.streaming_history) + print(f'Single Day Record: {rec_artist} on {rec_date}') + print(f'Duration: {int(rec_time.total_seconds() // 3600)}h {int((rec_time.total_seconds() % 3600) // 60)}m\n') + + # --- NEW FUN STATS --- + + # Manual Laborer + print('The Manual Laborer (Top Clicked Tracks):') + manual = manual_laborer(DATA.streaming_history) + for i, (k, v) in enumerate(list(manual.items())[:100], 1): + print(f'#{i:2} - {v:4} clicks : {k}') + print('\n') + + # Shuffle Roulette + print('The Shuffle Roulette (Top Tracks with Shuffle ON):') + shuffled = shuffle_roulette(DATA.streaming_history) + for i, (k, v) in enumerate(list(shuffled.items())[:50], 1): + print(f'#{i:2} - {v:4} plays : {k}') + print('\n') + + # Session Starter + print('The Session Starter (Most frequent session openers):') + starters = session_starter(DATA.streaming_history) + for i, (k, v) in enumerate(list(starters.items())[:50], 1): + print(f'#{i:2} - {v:4} starts : {k}') + print('\n') + + # Session Closer + print('The Session Closer (Most frequent session enders):') + closers = session_closer(DATA.streaming_history) + for i, (k, v) in enumerate(list(closers.items())[:50], 1): + print(f'#{i:2} - {v:4} ends : {k}') + print('\n') + + # Quick Fix + qf_count = quick_fix(DATA.streaming_history) + print(f'The Quick Fix: {qf_count} "Single Song Sessions" (Opened app, played 1 song, closed)\n') + + # Skipper's Remorse + print("Skipper's Remorse (Skipped >50% but played >20 times):") + remorse = skippers_remorse(DATA.streaming_history) + for i, (k, v) in enumerate(list(remorse.items())[:50], 1): + print(f'#{i:2} - {v:.1f}% skipped : {k}') + print('\n') + + # Remix Junkie + remix_pct, remix_cnt = remix_junkie(DATA.streaming_history) + print(f'The Remix Junkie: {remix_pct:.1f}% of tracks are Remixes/Edits ({remix_cnt} tracks)\n') + + # Live Fanatic + live_pct, live_cnt = live_fanatic(DATA.streaming_history) + print(f'The Live Fanatic: {live_pct:.1f}% of tracks are Live/Concert ({live_cnt} tracks)\n') + + # Short King + print('The Short King (Most played tracks < 2 mins):') + shorts = short_king(DATA.streaming_history) + for i, (k, v) in enumerate(list(shorts.items())[:15], 1): + print(f'#{i:2} - {v:4} plays : {k}') + print('\n') + + # Epic Saga + print('The Epic Saga (Most played tracks > 7 mins):') + epics = epic_saga(DATA.streaming_history) + for i, (k, v) in enumerate(list(epics.items())[:15], 1): + print(f'#{i:2} - {v:4} plays : {k}') + print('\n') + + # Collaborator + print('The Collaborator (Most played tracks featuring others):') + collabs = collaborator(DATA.streaming_history) + for i, (k, v) in enumerate(list(collabs.items())[:5], 1): + print(f'#{i:2} - {v:4} plays : {k}') + print('\n') + + # Alphabet Artists + print('The Alphabet Artists (Top Artist for A-Z):') + alpha_artists = alphabet_artists(DATA.streaming_history) + for char in "ABCDEFGHIJKLMNOPQRSTUVWXYZ": + if char in alpha_artists: + artist, count = alpha_artists[char] + print(f"{char}: {artist} ({count} plays)") + else: + print(f"{char}: -") + print('\n') + + # Spelling Bee + long_art, len_art, long_trk, len_trk = spelling_bee(DATA.streaming_history) + print(f'The Spelling Bee:') + print(f'Longest Artist Name: {long_art} ({len_art} chars)') + print(f'Longest Track Title: {long_trk} ({len_trk} chars)\n') + + # Same Name Game + print('The Same Name Game (Titles played from most different artists):') + same_names = same_name_game(DATA.streaming_history) + for i, (k, v) in enumerate(list(same_names.items())[:5], 1): + print(f'#{i:2} - {v:4} artists : {k}') + print('\n') + + # Midnight Club + print('The Midnight Club (Top Artists 00:00 - 01:00):') + midnighters = midnight_club(DATA.streaming_history) + for i, (k, v) in enumerate(list(midnighters.items())[:15], 1): + print(f'#{i:2} - {v:4} plays : {k}') + print('\n') + + # Lunch Break + print('The Lunch Break (Top Artists 12:00 - 14:00):') + lunchers = lunch_break(DATA.streaming_history) + for i, (k, v) in enumerate(list(lunchers.items())[:15], 1): + print(f'#{i:2} - {v:4} plays : {k}') + print('\n') + + # Monday Blues + print('The Monday Blues (Top Tracks on Mondays):') + mondays = monday_blues(DATA.streaming_history) + for i, (k, v) in enumerate(list(mondays.items())[:15], 1): + print(f'#{i:2} - {v:4} plays : {k}') + print('\n') + + # Hump Day Hero + print('The Hump Day Hero (Top Tracks on Wednesdays):') + wednesdays = hump_day_hero(DATA.streaming_history) + for i, (k, v) in enumerate(list(wednesdays.items())[:15], 1): + print(f'#{i:2} - {v:4} plays : {k}') + print('\n') + + # Quarterly Review + print('The Quarterly Review (Top Track per Quarter):') + quarters = quarterly_review(DATA.streaming_history) + for q, (track, count) in quarters.items(): + print(f'Q{q}: {track} ({count} plays)') + print('\n') + + # Album Purist + purist_streak, purist_album, purist_artist = album_purist(DATA.streaming_history) + print(f'The Album Purist:') + print(f'Longest streak of unique songs from one album: {purist_streak}') + print(f'Album: {purist_album} by {purist_artist}\n') + + # Instant Skips (all removed tracks from spotify because it was started but immediately skipped) + print('Instant Skips Probably removed tracks (Skipped < 1s):') + inst_skips = instant_skips(DATA.streaming_history) + total_inst_skips = sum(inst_skips.values()) + print(f'Total Instant Skips: {total_inst_skips}') + for i, (k, v) in enumerate(list(inst_skips.items())[:20], 1): + print(f'#{i:2} - {v:4} skips : {k}') + print('\n') + + print('--- End of Report ---') + + # Stop capturing output + sys.stdout = original_stdout + report_text = captured_output.getvalue() + + # Export CSV + print('Exporting song stats to song_stats.csv...') + song_stats = get_full_song_stats(DATA.streaming_history) + with open('song_stats.csv', 'w', newline='', encoding='utf-8') as csvfile: + fieldnames = ['Artist', 'Track Name', 'Times Played', 'First Played', 'Last Played', 'Skipped', 'Instant Skips', 'User Started'] + writer = csv.DictWriter(csvfile, fieldnames=fieldnames) + writer.writeheader() + writer.writerows(song_stats) + print('Done! CSV exported.\n') + + # Graphs + print('\n') + graph_mode = input('Do you want to (s)how graphs, (e)xport to PDF, or (n)one? (s/e/n): ').lower() + + if graph_mode in ['s', 'e']: + figures = [] + print('Generating graphs...') + + figures.append(plot_top_items(track_counts, 'Top 10 Played Tracks')) + figures.append(plot_top_items(artist_counts, 'Top 10 Played Artists')) + figures.append(plot_top_items(album_counts, 'Top 10 Played Albums')) + + figures.append(plot_location_counts(locations, 'Plays by Location')) + figures.append(plot_longest_played_artist(time_artists, 'Top 10 Artists by Listening Time')) + figures.append(plot_longest_played_tracks(time_tracks, 'Top 10 Tracks by Listening Time')) + figures.append(plot_skipped_items(skipped_artists, 'Top 10 Skipped Artists')) + figures.append(plot_skipped_items(most_skipped_track(DATA.streaming_history), 'Top 10 Skipped Tracks')) + + figures.append(plot_top_items(night_artists, 'The Night Shift (Top Artists 2AM-5AM)')) + + # Fun Stats Graphs + figures.append(plot_listening_by_day_of_week(listening_by_day_of_week(DATA.streaming_history), 'Listening by Day of Week')) + figures.append(plot_discovery_rate(discovery_rate(DATA.streaming_history), 'Artist Discovery Rate (New Artists per Month)')) + figures.append(plot_seasonal_listening(seasonal_listening(DATA.streaming_history), 'Seasonal Listening Habits')) + figures.append(plot_day_night_split(day_night_split(DATA.streaming_history), 'Day vs Night Listening')) + figures.append(plot_hourly_heatmap(hourly_heatmap_data(DATA.streaming_history), 'Listening Heatmap (Day vs Hour)')) + figures.append(plot_variety_score(variety, 'Variety Score Over Years')) + figures.append(plot_comfort_zone(comfort_pct, 'The Comfort Zone (% Time on Top 10 Artists)')) + + # Heatmaps + cal_matrix, cal_years, cal_months = calendar_heatmap_data(DATA.streaming_history) + figures.append(plot_calendar_heatmap(cal_matrix, cal_years, cal_months, 'Listening Calendar (Year vs Month)')) + + figures.append(plot_picky_grid(picky_grid_data(DATA.streaming_history), 'The Picky Grid (Skip Rate % by Day & Hour)')) + + dev_matrix, dev_platforms = device_habits_data(DATA.streaming_history) + figures.append(plot_device_habits(dev_matrix, dev_platforms, 'Device Habits (Platform vs Hour)')) + + eras_matrix, eras_artists, eras_time = artist_eras_data(DATA.streaming_history, top_n=300) + figures.append(plot_artist_eras(eras_matrix, eras_artists, eras_time, 'Artist Eras (Top 300 Artists vs Time)')) + + # Active Listening Deep Dive + figures.append(plot_active_heatmap(active_listening_heatmap_data(DATA.streaming_history), 'Active Listening Heatmap (When do you click play?)')) + figures.append(plot_active_trend(active_listening_trend_data(DATA.streaming_history), 'Active Listening Trend (% of starts that were clicks)')) + + # Artist Trends + top_50_artists = list(artist_counts.keys())[:50] + artist_trends = artist_history_over_time(DATA.streaming_history, top_50_artists) + figures.append(plot_artist_trends(artist_trends, 'Top 50 Artists Trends Over Years')) + + # Platform Usage + platforms = platform_usage(DATA.streaming_history) + figures.append(plot_platform_usage(platforms, 'Platform Usage')) + + # Listening by Hour + hourly_stats = listening_by_hour(DATA.streaming_history) + figures.append(plot_listening_by_hour(hourly_stats, 'Listening Activity by Hour of Day')) + + # Artist Personality Radar + # print("Calculating Artist Personalities...") + # traits = get_artist_traits(DATA.streaming_history) + # Pick top 5 artists + # top_5_artists = list(artist_counts.keys())[:5] + # for artist in top_5_artists: + # if artist in traits: + # figures.append(plot_artist_radar({artist: traits[artist]}, f'Artist Personality: {artist}')) + + # Filter out None figures + figures = [f for f in figures if f is not None] + + if graph_mode == 'e': + filename = f'spotify_stats_report_{int(time.time())}.pdf' + + # Add text pages + print('Generating text pages...') + text_figures = create_text_pages(report_text) + all_figures = text_figures + figures + + print(f'Saving {len(all_figures)} pages to {filename}...') + with PdfPages(filename) as pdf: + for fig in all_figures: + pdf.savefig(fig) + plt.close(fig) + print(f'Done! Report saved to {filename}') + elif graph_mode == 's': + print('Showing graphs...') + plt.show() + except Exception: traceback.print_exc() diff --git a/stats/.DS_Store b/stats/.DS_Store new file mode 100644 index 0000000000000000000000000000000000000000..d27b099e655549617283adc3381841d79c9dca84 GIT binary patch literal 6148 zcmeHKJ5Iwu5S-3{8;5=n)z2of^3F5XiblMdwt}iY}<^S!f@J zB=-@rEvcbf<1zIS3iNO|*@#KEFV{2&P#9n-`@BC4`u dict[History, int]: sort = {k: result[k] for k in sorted(result, key=result.get, reverse=True)} return sort + + +def play_counts_by_artist(streaming_history: list[History]) -> dict[str, int]: + """ + Generates dictionary with unique artist names and number of times played as values + + :return: descending sorted dictionary by play count + """ + result = {} + for i in streaming_history: + key = i['artistName'] + if key not in result: + result[key] = 0 + result[key] += 1 + + sort = {k: result[k] for k in sorted(result, key=result.get, reverse=True)} + + return sort + + +def play_counts_by_album(streaming_history: list[History]) -> dict[str, int]: + """ + Generates dictionary with unique album names and number of times played as values + + :return: descending sorted dictionary by play count + """ + result = {} + for i in streaming_history: + key = f"{i['artistName']} - {i['albumName']}" + if key not in result: + result[key] = 0 + result[key] += 1 + + sort = {k: result[k] for k in sorted(result, key=result.get, reverse=True)} + + return sort + + +def artist_history_over_time(streaming_history: list[History], top_artists: list[str]) -> dict[str, dict[int, int]]: + """ + Generates a dictionary of artist play counts per year for the specified top artists. + Structure: { 'Artist Name': { 2012: 10, 2013: 50, ... }, ... } + """ + result = {artist: {} for artist in top_artists} + + for item in streaming_history: + artist = item['artistName'] + if artist in result: + year = item['endTime'].tm_year + if year not in result[artist]: + result[artist][year] = 0 + result[artist][year] += 1 + + return result + + +def platform_usage(streaming_history: list[History]) -> dict[str, int]: + """ + Calculates the number of plays per platform. + """ + result = {} + for item in streaming_history: + platform = item['platform'] + if platform not in result: + result[platform] = 0 + result[platform] += 1 + return {k: result[k] for k in sorted(result, key=result.get, reverse=True)} + + +def listening_by_hour(streaming_history: list[History]) -> dict[int, int]: + """ + Calculates the number of plays per hour of the day (0-23). + """ + result = {h: 0 for h in range(24)} + for item in streaming_history: + hour = item['endTime'].tm_hour + result[hour] += 1 + return result + + +def skipped_ratio(streaming_history: list[History]) -> dict[str, int]: + """ + Calculates the number of skipped vs not skipped tracks. + """ + skipped = 0 + not_skipped = 0 + for item in streaming_history: + if item['skipped']: + skipped += 1 + else: + not_skipped += 1 + return {'Skipped': skipped, 'Not Skipped': not_skipped} + +def location_counts(streaming_history: list[History]) -> dict[str, int]: + """ + Calculates the number of plays per country. + """ + result = {} + for item in streaming_history: + country = item['connCountry'] + if country not in result: + result[country] = 0 + result[country] += 1 + return {k: result[k] for k in sorted(result, key=result.get, reverse=True)} + + +def most_skipped_artist(streaming_history: list[History]) -> dict[str, int]: + """ + Calculates the number of skips per artist. + """ + result = {} + for item in streaming_history: + if item['skipped']: + artist = item['artistName'] + if artist not in result: + result[artist] = 0 + result[artist] += 1 + return {k: result[k] for k in sorted(result, key=result.get, reverse=True)} + + +def most_skipped_track(streaming_history: list[History]) -> dict[str, int]: + """ + Calculates the number of skips per track. + """ + result = {} + for item in streaming_history: + if item['skipped']: + track = f"{item['artistName']} - {item['trackName']}" + if track not in result: + result[track] = 0 + result[track] += 1 + return {k: result[k] for k in sorted(result, key=result.get, reverse=True)} + + +def longest_played_artist(streaming_history: list[History]) -> dict[str, timedelta]: + """ + Calculates the total listening time per artist. + """ + result = defaultdict(timedelta) + for item in streaming_history: + result[item['artistName']] += item['msPlayed'] + + return {k: result[k] for k in sorted(result, key=result.get, reverse=True)} + + +def top_artist_per_month(streaming_history: list[History]) -> dict[str, tuple[str, int]]: + """ + Finds the most played artist for each month. + Returns: {'YYYY-MM': ('Artist Name', play_count), ...} + """ + monthly_counts = defaultdict(lambda: defaultdict(int)) + + for item in streaming_history: + month_key = f"{item['endTime'].tm_year}-{item['endTime'].tm_mon:02d}" + monthly_counts[month_key][item['artistName']] += 1 + + result = {} + for month, artists in monthly_counts.items(): + top_artist = max(artists.items(), key=lambda x: x[1]) + result[month] = top_artist + + return dict(sorted(result.items())) + +def listening_by_day_of_week(streaming_history: list[History]) -> dict[str, int]: + """ + Calculates plays by day of the week. + """ + days = {0: 'Monday', 1: 'Tuesday', 2: 'Wednesday', 3: 'Thursday', 4: 'Friday', 5: 'Saturday', 6: 'Sunday'} + result = {day: 0 for day in days.values()} + + for item in streaming_history: + day_idx = item['endTime'].tm_wday + result[days[day_idx]] += 1 + + return result + + +def discovery_rate(streaming_history: list[History]) -> dict[str, int]: + """ + Calculates new artists discovered per month. + """ + seen_artists = set() + result = defaultdict(int) + + # Ensure history is sorted by time + sorted_history = sorted(streaming_history, key=lambda x: x['endTime']) + + for item in sorted_history: + artist = item['artistName'] + if artist not in seen_artists: + seen_artists.add(artist) + month_key = f"{item['endTime'].tm_year}-{item['endTime'].tm_mon:02d}" + result[month_key] += 1 + + return dict(sorted(result.items())) + + +def seasonal_listening(streaming_history: list[History]) -> dict[str, int]: + """ + Calculates plays by season. + """ + seasons = {'Winter': 0, 'Spring': 0, 'Summer': 0, 'Autumn': 0} + + for item in streaming_history: + month = item['endTime'].tm_mon + if month in [12, 1, 2]: + seasons['Winter'] += 1 + elif month in [3, 4, 5]: + seasons['Spring'] += 1 + elif month in [6, 7, 8]: + seasons['Summer'] += 1 + elif month in [9, 10, 11]: + seasons['Autumn'] += 1 + + return seasons + + +def one_hit_wonders(streaming_history: list[History], min_plays: int = 5) -> dict[str, tuple[str, int]]: + """ + Finds artists with > min_plays but only 1 unique track. + Returns {artist: (track_name, play_count)} + """ + artist_tracks = defaultdict(set) + artist_plays = defaultdict(int) + + for item in streaming_history: + artist = item['artistName'] + track = item['trackName'] + artist_tracks[artist].add(track) + artist_plays[artist] += 1 + + result = {} + for artist, tracks in artist_tracks.items(): + if len(tracks) == 1 and artist_plays[artist] >= min_plays: + result[artist] = (list(tracks)[0], artist_plays[artist]) + + return dict(sorted(result.items(), key=lambda x: x[1][1], reverse=True)) + + +def day_night_split(streaming_history: list[History]) -> dict[str, int]: + """ + Calculates plays during Day (6-18) vs Night (18-6). + """ + result = {'Day': 0, 'Night': 0} + for item in streaming_history: + hour = item['endTime'].tm_hour + if 6 <= hour < 18: + result['Day'] += 1 + else: + result['Night'] += 1 + return result + + +def longest_listening_streak(streaming_history: list[History], gap_tolerance_minutes: int = 10) -> tuple[datetime, datetime, timedelta]: + """ + Calculates the longest continuous listening streak. + """ + if not streaming_history: + return None + + sorted_history = sorted(streaming_history, key=lambda x: x['endTime']) + + max_streak_duration = timedelta(0) + max_streak_start = None + max_streak_end = None + + current_streak_start = None + current_streak_end = None + + for item in sorted_history: + # Calculate start time of the track + end_dt = datetime.fromtimestamp(mktime(item['endTime'])) + duration = item['msPlayed'] # timedelta + start_dt = end_dt - duration + + if current_streak_end is None: + current_streak_start = start_dt + current_streak_end = end_dt + continue + + # Check gap + gap = start_dt - current_streak_end + + if gap <= timedelta(minutes=gap_tolerance_minutes): + # Extend streak + current_streak_end = end_dt + else: + # End of streak, check if it's the longest + streak_duration = current_streak_end - current_streak_start + if streak_duration > max_streak_duration: + max_streak_duration = streak_duration + max_streak_start = current_streak_start + max_streak_end = current_streak_end + + # Start new streak + current_streak_start = start_dt + current_streak_end = end_dt + + # Check last streak + if current_streak_end and current_streak_start: + streak_duration = current_streak_end - current_streak_start + if streak_duration > max_streak_duration: + max_streak_duration = streak_duration + max_streak_start = current_streak_start + max_streak_end = current_streak_end + + return max_streak_start, max_streak_end, max_streak_duration + +def true_skip_rate(streaming_history: list[History], min_plays: int = 20) -> dict[str, float]: + """ + Calculates skip percentage per artist (skips / total_starts). + Only considers artists with > min_plays to avoid skewed data. + """ + stats = defaultdict(lambda: {'plays': 0, 'skips': 0}) + for item in streaming_history: + artist = item['artistName'] + stats[artist]['plays'] += 1 + if item['skipped']: + stats[artist]['skips'] += 1 + + results = {} + for artist, data in stats.items(): + if data['plays'] > min_plays: + results[artist] = (data['skips'] / data['plays']) * 100 + + return dict(sorted(results.items(), key=lambda x: x[1], reverse=True)) + + +def most_consecutive_plays(streaming_history: list[History]) -> tuple[str, int]: + """ + Finds the track played the most times consecutively (back-to-back). + """ + if not streaming_history: + return ("None", 0) + + max_streak = 0 + max_track = "" + + current_streak = 1 + # Create a unique identifier for the track + current_track = f"{streaming_history[0]['artistName']} - {streaming_history[0]['trackName']}" + + for i in range(1, len(streaming_history)): + track = f"{streaming_history[i]['artistName']} - {streaming_history[i]['trackName']}" + if track == current_track: + current_streak += 1 + else: + if current_streak > max_streak: + max_streak = current_streak + max_track = current_track + current_track = track + current_streak = 1 + + return (max_track, max_streak) + + +def album_loyalty(streaming_history: list[History]) -> dict[str, int]: + """ + Counts how many times a user listened to at least 3 songs from the same album in a row. + """ + loyalty_counts = defaultdict(int) + if not streaming_history: + return loyalty_counts + + current_album = streaming_history[0]['albumName'] + current_streak = 1 + + for i in range(1, len(streaming_history)): + album = streaming_history[i]['albumName'] + # Ignore singles or unknown albums if necessary, but keeping it simple for now + if album == current_album and album: + current_streak += 1 + else: + if current_streak >= 3: + loyalty_counts[current_album] += 1 + current_album = album + current_streak = 1 + + return dict(sorted(loyalty_counts.items(), key=lambda x: x[1], reverse=True)) + + +def longest_artist_relationship(streaming_history: list[History]) -> dict[str, timedelta]: + """ + Calculates time between first and last listen for each artist. + """ + first_seen = {} + last_seen = {} + + for item in streaming_history: + artist = item['artistName'] + # Convert struct_time to datetime + ts = item['endTime'] + dt = datetime.fromtimestamp(time.mktime(ts)) + + if artist not in first_seen or dt < first_seen[artist]: + first_seen[artist] = dt + if artist not in last_seen or dt > last_seen[artist]: + last_seen[artist] = dt + + durations = {} + for artist in first_seen: + diff = last_seen[artist] - first_seen[artist] + # Filter out artists listened to for less than a day + if diff.total_seconds() > 86400: + durations[artist] = diff + + return dict(sorted(durations.items(), key=lambda x: x[1], reverse=True)) + + +def forgotten_favorites(streaming_history: list[History], months_forgotten: int = 6) -> list[tuple[str, int]]: + """ + Identifies artists that were popular in the past but have 0 plays in the last X months. + """ + if not streaming_history: + return [] + + # Sort history by time + sorted_history = sorted(streaming_history, key=lambda x: x['endTime']) + last_date = datetime.fromtimestamp(mktime(sorted_history[-1]['endTime'])) + cutoff_date = last_date - timedelta(days=30 * months_forgotten) + + recent_artists = set() + past_counts = defaultdict(int) + + for item in sorted_history: + dt = datetime.fromtimestamp(mktime(item['endTime'])) + artist = item['artistName'] + if dt > cutoff_date: + recent_artists.add(artist) + else: + past_counts[artist] += 1 + + forgotten = [] + for artist, count in past_counts.items(): + if artist not in recent_artists and count > 20: # Minimum plays to be considered a "favorite" + forgotten.append((artist, count)) + + return sorted(forgotten, key=lambda x: x[1], reverse=True) + + +def hourly_heatmap_data(streaming_history: list[History]) -> list[list[int]]: + """ + Prepares data for a 2D heatmap: 7 days x 24 hours. + Returns a 7x24 matrix where cell [d][h] is the play count. + """ + # 7 rows (Mon-Sun), 24 columns (0-23 hours) + matrix = [[0 for _ in range(24)] for _ in range(7)] + + for item in streaming_history: + day = item['endTime'].tm_wday # 0=Mon, 6=Sun + hour = item['endTime'].tm_hour + matrix[day][hour] += 1 + + return matrix + +def most_musical_day(streaming_history: list[History]) -> tuple[str, timedelta]: + """ + Finds the single date with the most listening time. + """ + daily_time = defaultdict(timedelta) + + for item in streaming_history: + date_key = time.strftime("%Y-%m-%d", item['endTime']) + daily_time[date_key] += item['msPlayed'] + + if not daily_time: + return ("None", timedelta(0)) + + max_day = max(daily_time.items(), key=lambda x: x[1]) + return max_day + + +def weekend_vs_weekday(streaming_history: list[History]) -> tuple[dict[str, int], dict[str, int]]: + """ + Returns top artists for Weekdays (Mon-Fri) and Weekends (Sat-Sun). + """ + weekday_counts = defaultdict(int) + weekend_counts = defaultdict(int) + + for item in streaming_history: + day = item['endTime'].tm_wday + artist = item['artistName'] + + if day < 5: # 0-4 is Mon-Fri + weekday_counts[artist] += 1 + else: # 5-6 is Sat-Sun + weekend_counts[artist] += 1 + + top_weekday = dict(sorted(weekday_counts.items(), key=lambda x: x[1], reverse=True)) + top_weekend = dict(sorted(weekend_counts.items(), key=lambda x: x[1], reverse=True)) + + return top_weekday, top_weekend + + +def immediate_skips(streaming_history: list[History]) -> dict[str, int]: + """ + Counts tracks skipped within 30 seconds (30000 ms). + """ + skips = defaultdict(int) + + for item in streaming_history: + if item['skipped'] and item['msPlayed'] < timedelta(seconds=30): + track = f"{item['artistName']} - {item['trackName']}" + skips[track] += 1 + + return dict(sorted(skips.items(), key=lambda x: x[1], reverse=True)) + + +def variety_score(streaming_history: list[History]) -> dict[int, float]: + """ + Calculates Diversity Index (Unique Artists / Total Plays) per year. + """ + yearly_stats = defaultdict(lambda: {'plays': 0, 'artists': set()}) + + for item in streaming_history: + year = item['endTime'].tm_year + yearly_stats[year]['plays'] += 1 + yearly_stats[year]['artists'].add(item['artistName']) + + scores = {} + for year, data in yearly_stats.items(): + if data['plays'] > 0: + scores[year] = len(data['artists']) / data['plays'] + + return dict(sorted(scores.items())) + +def listening_velocity(streaming_history: list[History], target_plays: int = 100) -> dict[str, int]: + """ + Calculates days taken to reach X plays for an artist. + """ + artist_plays = defaultdict(list) + + # Collect all timestamps for each artist + for item in streaming_history: + artist_plays[item['artistName']].append(item['endTime']) + + velocity = {} + for artist, timestamps in artist_plays.items(): + if len(timestamps) >= target_plays: + # Sort timestamps just in case + sorted_ts = sorted(timestamps) + first_play = datetime.fromtimestamp(mktime(sorted_ts[0])) + target_play = datetime.fromtimestamp(mktime(sorted_ts[target_plays-1])) + + days_taken = (target_play - first_play).days + velocity[artist] = days_taken + + # Sort by fastest (lowest days) + return dict(sorted(velocity.items(), key=lambda x: x[1])) + + +def the_comeback(streaming_history: list[History], gap_days: int = 365) -> dict[str, int]: + """ + Finds artists with a gap of > X days between plays. + Returns {artist: max_gap_days} + """ + artist_timestamps = defaultdict(list) + for item in streaming_history: + artist_timestamps[item['artistName']].append(datetime.fromtimestamp(mktime(item['endTime']))) + + comebacks = {} + for artist, timestamps in artist_timestamps.items(): + if len(timestamps) < 2: + continue + + sorted_ts = sorted(timestamps) + max_gap = 0 + + for i in range(1, len(sorted_ts)): + gap = (sorted_ts[i] - sorted_ts[i-1]).days + if gap > max_gap: + max_gap = gap + + if max_gap >= gap_days: + comebacks[artist] = max_gap + + return dict(sorted(comebacks.items(), key=lambda x: x[1], reverse=True)) + + +def clockwork_artists(streaming_history: list[History]) -> dict[str, str]: + """ + Identifies artists played mostly (>70%) in specific 4-hour windows. + """ + artist_hours = defaultdict(lambda: defaultdict(int)) + artist_total = defaultdict(int) + + for item in streaming_history: + artist = item['artistName'] + hour = item['endTime'].tm_hour + + # Define windows: 0-4, 4-8, 8-12, 12-16, 16-20, 20-24 + window_start = (hour // 4) * 4 + window_label = f"{window_start:02d}:00-{window_start+4:02d}:00" + + artist_hours[artist][window_label] += 1 + artist_total[artist] += 1 + + clockwork = {} + for artist, total in artist_total.items(): + if total < 50: # Minimum plays + continue + + for window, count in artist_hours[artist].items(): + if count / total >= 0.7: + clockwork[artist] = f"{window} ({int(count/total*100)}%)" + + return clockwork + + +def session_analysis(streaming_history: list[History], session_break_min: int = 20) -> tuple[float, float]: + """ + Calculates average session length (minutes) and tracks per session. + """ + if not streaming_history: + return (0.0, 0.0) + + sorted_history = sorted(streaming_history, key=lambda x: x['endTime']) + + sessions = [] + current_session_start = datetime.fromtimestamp(mktime(sorted_history[0]['endTime'])) + current_session_end = current_session_start + current_session_tracks = 0 + + for i in range(len(sorted_history)): + ts = datetime.fromtimestamp(mktime(sorted_history[i]['endTime'])) + + if i > 0: + prev_ts = datetime.fromtimestamp(mktime(sorted_history[i-1]['endTime'])) + gap = (ts - prev_ts).total_seconds() / 60 + + if gap > session_break_min: + # End previous session + duration = (current_session_end - current_session_start).total_seconds() / 60 + sessions.append({'duration': duration, 'tracks': current_session_tracks}) + + # Start new session + current_session_start = ts + current_session_tracks = 0 + + current_session_end = ts + current_session_tracks += 1 + + # Add last session + duration = (current_session_end - current_session_start).total_seconds() / 60 + sessions.append({'duration': duration, 'tracks': current_session_tracks}) + + avg_duration = sum(s['duration'] for s in sessions) / len(sessions) + avg_tracks = sum(s['tracks'] for s in sessions) / len(sessions) + + return avg_duration, avg_tracks + + +def skipless_albums(streaming_history: list[History], min_plays: int = 50) -> dict[str, float]: + """ + Finds albums with lowest skip rate (min 50 plays). + """ + album_stats = defaultdict(lambda: {'plays': 0, 'skips': 0}) + + for item in streaming_history: + album = item['albumName'] + if not album: continue + + album_stats[album]['plays'] += 1 + if item['skipped']: + album_stats[album]['skips'] += 1 + + results = {} + for album, data in album_stats.items(): + if data['plays'] >= min_plays: + skip_rate = (data['skips'] / data['plays']) * 100 + results[album] = skip_rate + + # Sort by lowest skip rate + return dict(sorted(results.items(), key=lambda x: x[1])) + + +def sampler_vs_completionist(streaming_history: list[History], min_plays: int = 50) -> dict[str, float]: + """ + Calculates Unique Tracks / Total Plays ratio for top artists. + """ + artist_stats = defaultdict(lambda: {'plays': 0, 'tracks': set()}) + + for item in streaming_history: + artist = item['artistName'] + track = item['trackName'] + artist_stats[artist]['plays'] += 1 + artist_stats[artist]['tracks'].add(track) + + ratios = {} + for artist, data in artist_stats.items(): + if data['plays'] >= min_plays: + ratios[artist] = len(data['tracks']) / data['plays'] + + return dict(sorted(ratios.items(), key=lambda x: x[1], reverse=True)) + + +def commute_heroes(streaming_history: list[History]) -> dict[str, int]: + """ + Top artists during commute hours (Mon-Fri, 7-9 & 17-19). + """ + commute_counts = defaultdict(int) + + for item in streaming_history: + day = item['endTime'].tm_wday + hour = item['endTime'].tm_hour + + if day < 5: # Mon-Fri + if (7 <= hour < 9) or (17 <= hour < 19): + commute_counts[item['artistName']] += 1 + + return dict(sorted(commute_counts.items(), key=lambda x: x[1], reverse=True)) + + +def marathon_tracks(streaming_history: list[History]) -> dict[str, int]: + """ + Most played tracks > 5 minutes long. + """ + long_tracks = defaultdict(int) + + for item in streaming_history: + if item['msPlayed'] > timedelta(minutes=5): + track = f"{item['artistName']} - {item['trackName']}" + long_tracks[track] += 1 + + return dict(sorted(long_tracks.items(), key=lambda x: x[1], reverse=True)) + + +def one_week_wonders(streaming_history: list[History]) -> dict[str, str]: + """ + Artists with >50 plays in one week but <10 in all others. + """ + artist_weekly = defaultdict(lambda: defaultdict(int)) + + for item in streaming_history: + # ISO Year and Week + year, week, _ = datetime.fromtimestamp(mktime(item['endTime'])).isocalendar() + week_key = f"{year}-W{week:02d}" + artist_weekly[item['artistName']][week_key] += 1 + + wonders = {} + for artist, weeks in artist_weekly.items(): + max_week = max(weeks.items(), key=lambda x: x[1]) + max_plays = max_week[1] + + if max_plays > 50: + other_plays = sum(count for w, count in weeks.items() if w != max_week[0]) + if other_plays < 10: + wonders[artist] = f"{max_week[0]} ({max_plays} plays)" + + return wonders + + +def sound_of_silence(streaming_history: list[History]) -> tuple[str, str, int]: + """ + Longest gap between any two plays. + """ + if not streaming_history: + return ("None", "None", 0) + + sorted_history = sorted(streaming_history, key=lambda x: x['endTime']) + + max_gap = timedelta(0) + gap_start = None + gap_end = None + + for i in range(1, len(sorted_history)): + curr = datetime.fromtimestamp(mktime(sorted_history[i]['endTime'])) + prev = datetime.fromtimestamp(mktime(sorted_history[i-1]['endTime'])) + + # Subtract duration of previous track to get true silence start + prev_end = prev # endTime is already the end of the track + + gap = curr - prev_end + if gap > max_gap: + max_gap = gap + gap_start = prev_end + gap_end = curr + + return (str(gap_start), str(gap_end), max_gap.days) + +def calendar_heatmap_data(streaming_history: list[History]) -> tuple[list[list[int]], list[int], list[str]]: + """ + Prepares data for Year vs Month heatmap. + Returns (matrix, years, months). + """ + years = sorted(list(set(item['endTime'].tm_year for item in streaming_history))) + months = ['Jan', 'Feb', 'Mar', 'Apr', 'May', 'Jun', 'Jul', 'Aug', 'Sep', 'Oct', 'Nov', 'Dec'] + + # Matrix: Rows = Years, Cols = Months + matrix = [[0 for _ in range(12)] for _ in range(len(years))] + + year_to_idx = {y: i for i, y in enumerate(years)} + + for item in streaming_history: + y = item['endTime'].tm_year + m = item['endTime'].tm_mon - 1 # 0-11 + matrix[year_to_idx[y]][m] += 1 + + return matrix, years, months + + +def picky_grid_data(streaming_history: list[History]) -> list[list[float]]: + """ + Prepares data for Day vs Hour Skip Rate heatmap. + Returns 7x24 matrix of skip percentages. + """ + # 7 rows (Mon-Sun), 24 columns (0-23 hours) + plays = [[0 for _ in range(24)] for _ in range(7)] + skips = [[0 for _ in range(24)] for _ in range(7)] + + for item in streaming_history: + day = item['endTime'].tm_wday + hour = item['endTime'].tm_hour + plays[day][hour] += 1 + if item['skipped']: + skips[day][hour] += 1 + + # Calculate percentages + matrix = [[0.0 for _ in range(24)] for _ in range(7)] + for d in range(7): + for h in range(24): + if plays[d][h] > 0: + matrix[d][h] = (skips[d][h] / plays[d][h]) * 100 + + return matrix + + +def device_habits_data(streaming_history: list[History]) -> tuple[list[list[int]], list[str]]: + """ + Prepares data for Platform vs Hour heatmap. + Returns (matrix, platforms). + """ + platforms = sorted(list(set(item['platform'] for item in streaming_history))) + # Limit to top 10 platforms to keep heatmap readable if there are many + platform_counts = defaultdict(int) + for item in streaming_history: + platform_counts[item['platform']] += 1 + + top_platforms = [p for p, c in sorted(platform_counts.items(), key=lambda x: x[1], reverse=True)[:10]] + platform_to_idx = {p: i for i, p in enumerate(top_platforms)} + + # Matrix: Rows = Platforms, Cols = Hours + matrix = [[0 for _ in range(24)] for _ in range(len(top_platforms))] + + for item in streaming_history: + p = item['platform'] + if p in platform_to_idx: + h = item['endTime'].tm_hour + matrix[platform_to_idx[p]][h] += 1 + + return matrix, top_platforms + + +def artist_eras_data(streaming_history: list[History], top_n: int = 20, normalize: bool = True) -> tuple[list[list[float]], list[str], list[str]]: + """ + Prepares data for Top Artists vs Month (Eras) heatmap. + Returns (matrix, artists, time_labels). + If normalize is True, each artist's row is scaled 0-1 based on their peak month. + """ + # Get top artists + artist_counts = defaultdict(int) + for item in streaming_history: + artist_counts[item['artistName']] += 1 + top_artists = [a for a, c in sorted(artist_counts.items(), key=lambda x: x[1], reverse=True)[:top_n]] + artist_to_idx = {a: i for i, a in enumerate(top_artists)} + + # Get time range (Year-Month) + sorted_history = sorted(streaming_history, key=lambda x: x['endTime']) + start_date = datetime.fromtimestamp(mktime(sorted_history[0]['endTime'])) + end_date = datetime.fromtimestamp(mktime(sorted_history[-1]['endTime'])) + + # Generate month labels + time_labels = [] + curr = start_date.replace(day=1) + while curr <= end_date: + time_labels.append(curr.strftime("%Y-%m")) + if curr.month == 12: + curr = curr.replace(year=curr.year+1, month=1) + else: + curr = curr.replace(month=curr.month+1) + + time_to_idx = {t: i for i, t in enumerate(time_labels)} + + # Matrix: Rows = Artists, Cols = Months + # Using float for potential normalization + matrix = [[0.0 for _ in range(len(time_labels))] for _ in range(len(top_artists))] + + for item in streaming_history: + a = item['artistName'] + if a in artist_to_idx: + t = time.strftime("%Y-%m", item['endTime']) + if t in time_to_idx: + matrix[artist_to_idx[a]][time_to_idx[t]] += 1.0 + + # Normalize per artist (row-wise) + if normalize: + for i in range(len(matrix)): + row_max = max(matrix[i]) + if row_max > 0: + for j in range(len(matrix[i])): + matrix[i][j] = matrix[i][j] / row_max + + return matrix, top_artists, time_labels + +def control_freak_data(streaming_history: list[History]) -> tuple[dict[str, float], tuple[str, int]]: + """ + Analyzes Active (User initiated) vs Passive (Queue/Autoplay) listening. + Returns ({'active': %, 'passive': %}, (most_clicked_artist, count)) + """ + active_reasons = {'clickrow', 'playbtn'} + + active_count = 0 + total_count = 0 + artist_clicks = defaultdict(int) + + for item in streaming_history: + total_count += 1 + if item['reasonStart'] in active_reasons: + active_count += 1 + artist_clicks[item['artistName']] += 1 + + if total_count == 0: + return {'active': 0.0, 'passive': 0.0}, ('None', 0) + + active_pct = (active_count / total_count) * 100 + passive_pct = 100 - active_pct + + most_clicked = sorted(artist_clicks.items(), key=lambda x: x[1], reverse=True) + top_click = most_clicked[0] if most_clicked else ('None', 0) + + return {'active': active_pct, 'passive': passive_pct}, top_click + + +def shuffle_paradox_data(streaming_history: list[History]) -> dict[str, float]: + """ + Compares skip rates when Shuffle is On vs Off. + Returns {'shuffle_skip_rate': %, 'normal_skip_rate': %} + """ + shuffle_plays = 0 + shuffle_skips = 0 + normal_plays = 0 + normal_skips = 0 + + for item in streaming_history: + is_shuffle = item.get('shuffle', False) + + if is_shuffle: + shuffle_plays += 1 + if item['skipped']: + shuffle_skips += 1 + else: + normal_plays += 1 + if item['skipped']: + normal_skips += 1 + + shuffle_rate = (shuffle_skips / shuffle_plays * 100) if shuffle_plays > 0 else 0.0 + normal_rate = (normal_skips / normal_plays * 100) if normal_plays > 0 else 0.0 + + return {'shuffle_skip_rate': shuffle_rate, 'normal_skip_rate': normal_rate} + + +def natural_death_data(streaming_history: list[History], min_plays: int = 50) -> tuple[dict[str, float], list[tuple[str, float]]]: + """ + Analyzes how tracks end (Natural 'trackdone' vs User intervention). + Returns ({'natural': %, 'killed': %}, top_respected_artists) + """ + natural_ends = 0 + total_ends = 0 + + artist_respect = defaultdict(lambda: {'finished': 0, 'total': 0}) + + for item in streaming_history: + total_ends += 1 + is_natural = item['reasonEnd'] == 'trackdone' + + if is_natural: + natural_ends += 1 + + artist_respect[item['artistName']]['total'] += 1 + if is_natural: + artist_respect[item['artistName']]['finished'] += 1 + + if total_ends == 0: + return {'natural': 0.0, 'killed': 0.0}, [] + + natural_pct = (natural_ends / total_ends) * 100 + killed_pct = 100 - natural_pct + + # Calculate respect score for artists + respect_scores = [] + for artist, data in artist_respect.items(): + if data['total'] >= min_plays: + score = (data['finished'] / data['total']) * 100 + respect_scores.append((artist, score)) + + # Sort by highest respect score + top_respected = sorted(respect_scores, key=lambda x: x[1], reverse=True)[:10] + + return {'natural': natural_pct, 'killed': killed_pct}, top_respected + +def active_listening_heatmap_data(streaming_history: list[History]) -> list[list[int]]: + """ + Prepares data for Active Starts (Day vs Hour) heatmap. + Returns 7x24 matrix of active start counts. + """ + active_reasons = {'clickrow', 'playbtn'} + # 7 rows (Mon-Sun), 24 columns (0-23 hours) + matrix = [[0 for _ in range(24)] for _ in range(7)] + + for item in streaming_history: + if item['reasonStart'] in active_reasons: + day = item['endTime'].tm_wday + hour = item['endTime'].tm_hour + matrix[day][hour] += 1 + + return matrix + + +def active_listening_trend_data(streaming_history: list[History]) -> dict[int, float]: + """ + Calculates percentage of active starts per year. + Returns {year: active_percentage} + """ + active_reasons = {'clickrow', 'playbtn'} + yearly_stats = defaultdict(lambda: {'active': 0, 'total': 0}) + + for item in streaming_history: + year = item['endTime'].tm_year + yearly_stats[year]['total'] += 1 + if item['reasonStart'] in active_reasons: + yearly_stats[year]['active'] += 1 + + trends = {} + for year, data in yearly_stats.items(): + if data['total'] > 0: + trends[year] = (data['active'] / data['total']) * 100 + + return dict(sorted(trends.items())) + +def get_artist_traits(streaming_history: list[History], top_n: int = 5) -> dict: + """ + Calculates personality traits for the top N artists. + Traits: Loyalty (1-skip), Discovery (unique/total), Night Owl (night/total), + Weekend Warrior (weekend/total), Active Choice (active/total). + """ + # 1. Identify Top N Artists + artist_counts = defaultdict(int) + for item in streaming_history: + artist_counts[item['artistName']] += 1 + + top_artists = [k for k, v in sorted(artist_counts.items(), key=lambda x: x[1], reverse=True)[:top_n]] + + # 2. Calculate Traits + traits = defaultdict(lambda: {'plays': 0, 'skips': 0, 'unique_tracks': set(), + 'night_plays': 0, 'weekend_plays': 0, 'active_starts': 0}) + + for item in streaming_history: + artist = item['artistName'] + if artist in top_artists: + t = traits[artist] + t['plays'] += 1 + if item['skipped']: t['skips'] += 1 + t['unique_tracks'].add(item['trackName']) + + # Night (6pm - 6am) + if item['endTime'].tm_hour < 6 or item['endTime'].tm_hour >= 18: + t['night_plays'] += 1 + + # Weekend (Sat=5, Sun=6) + if item['endTime'].tm_wday >= 5: + t['weekend_plays'] += 1 + + # Active Start + if item['reasonStart'] in ['clickrow', 'playbtn', 'appload', 'remote']: + t['active_starts'] += 1 + + # 3. Normalize to 0-1 scale + result = {} + for artist in top_artists: + data = traits[artist] + total = data['plays'] + if total == 0: continue + + result[artist] = [ + 1 - (data['skips'] / total), # Loyalty + len(data['unique_tracks']) / total, # Discovery (approx) + data['night_plays'] / total, # Night Owl + data['weekend_plays'] / total, # Weekend Warrior + data['active_starts'] / total # Active Choice + ] + + return result + +def longest_played_tracks(streaming_history: list[History]) -> dict[str, timedelta]: + """Calculates the total listening time per track.""" + result = defaultdict(timedelta) + for item in streaming_history: + track = f"{item['artistName']} - {item['trackName']}" + result[track] += item['msPlayed'] + + return {k: result[k] for k in sorted(result, key=result.get, reverse=True)} + +def night_shift_artists(streaming_history: list[History]) -> dict[str, int]: + """Top artists played between 2 AM and 5 AM.""" + counts = defaultdict(int) + for item in streaming_history: + if 2 <= item['endTime'].tm_hour < 5: + counts[item['artistName']] += 1 + return dict(sorted(counts.items(), key=lambda x: x[1], reverse=True)) + +def new_years_transitions(streaming_history: list[History]) -> dict[int, tuple[str, str]]: + """Returns {year: (last_song_of_year, first_song_of_next_year)}""" + sorted_history = sorted(streaming_history, key=lambda x: x['endTime']) + transitions = {} + + if not sorted_history: + return {} + + by_year = defaultdict(list) + for item in sorted_history: + by_year[item['endTime'].tm_year].append(item) + + years = sorted(by_year.keys()) + for i in range(len(years) - 1): + current_year = years[i] + next_year = years[i+1] + + last_song = by_year[current_year][-1] + first_song = by_year[next_year][0] + + last_str = f"{last_song['artistName']} - {last_song['trackName']} ({time.strftime('%Y-%m-%d %H:%M', last_song['endTime'])})" + first_str = f"{first_song['artistName']} - {first_song['trackName']} ({time.strftime('%Y-%m-%d %H:%M', first_song['endTime'])})" + + transitions[current_year] = (last_str, first_str) + + return transitions + +def consistency_king(streaming_history: list[History]) -> dict[str, int]: + """Track played on the most unique dates.""" + track_dates = defaultdict(set) + for item in streaming_history: + track = f"{item['artistName']} - {item['trackName']}" + date_str = f"{item['endTime'].tm_year}-{item['endTime'].tm_mon}-{item['endTime'].tm_mday}" + track_dates[track].add(date_str) + + result = {k: len(v) for k, v in track_dates.items()} + return dict(sorted(result.items(), key=lambda x: x[1], reverse=True)) + +def alphabet_challenge(streaming_history: list[History]) -> dict[str, tuple[str, int]]: + """Most played track for each letter A-Z.""" + track_counts = defaultdict(int) + for item in streaming_history: + track = f"{item['artistName']} - {item['trackName']}" + track_counts[track] += 1 + + alphabet = "ABCDEFGHIJKLMNOPQRSTUVWXYZ" + best_of_letter = {} + + for char in alphabet: + candidates = {k: v for k, v in track_counts.items() if k.upper().startswith(char)} + if candidates: + best = max(candidates.items(), key=lambda x: x[1]) + best_of_letter[char] = best + + return best_of_letter + +def obsession_score(streaming_history: list[History]) -> dict[str, float]: + """ + Calculates Obsession Score: Total Plays / Unique Days Played. + High score means many plays in few days. + Returns dict {track_name: score} + """ + track_stats = defaultdict(lambda: {'plays': 0, 'days': set()}) + + for item in streaming_history: + track = f"{item['artistName']} - {item['trackName']}" + date_str = time.strftime('%Y-%m-%d', item['endTime']) + track_stats[track]['plays'] += 1 + track_stats[track]['days'].add(date_str) + + scores = {} + for track, stats in track_stats.items(): + if stats['plays'] > 10: # Minimum plays to qualify + scores[track] = stats['plays'] / len(stats['days']) + + return dict(sorted(scores.items(), key=lambda x: x[1], reverse=True)) + +def early_bird_artists(streaming_history: list[History]) -> dict[str, int]: + """ + Top artists played between 5 AM and 9 AM. + """ + counts = defaultdict(int) + for item in streaming_history: + hour = item['endTime'].tm_hour + if 5 <= hour < 9: + counts[item['artistName']] += 1 + return dict(sorted(counts.items(), key=lambda x: x[1], reverse=True)) + +def nine_to_five_artists(streaming_history: list[History]) -> dict[str, int]: + """ + Top artists played Mon-Fri, 9 AM - 5 PM. + """ + counts = defaultdict(int) + for item in streaming_history: + hour = item['endTime'].tm_hour + wday = item['endTime'].tm_wday + if 0 <= wday <= 4 and 9 <= hour < 17: + counts[item['artistName']] += 1 + return dict(sorted(counts.items(), key=lambda x: x[1], reverse=True)) + +def party_animal_tracks(streaming_history: list[History]) -> dict[str, int]: + """ + Top tracks played Fri/Sat nights (10 PM - 4 AM). + """ + counts = defaultdict(int) + for item in streaming_history: + hour = item['endTime'].tm_hour + wday = item['endTime'].tm_wday + + is_fri_night = (wday == 4 and hour >= 22) + is_sat_early = (wday == 5 and hour < 4) + is_sat_night = (wday == 5 and hour >= 22) + is_sun_early = (wday == 6 and hour < 4) + + if is_fri_night or is_sat_early or is_sat_night or is_sun_early: + track = f"{item['artistName']} - {item['trackName']}" + counts[track] += 1 + + return dict(sorted(counts.items(), key=lambda x: x[1], reverse=True)) + +def sunday_scaries_tracks(streaming_history: list[History]) -> dict[str, int]: + """ + Top tracks played Sunday 6 PM - Midnight. + """ + counts = defaultdict(int) + for item in streaming_history: + hour = item['endTime'].tm_hour + wday = item['endTime'].tm_wday + if wday == 6 and 18 <= hour <= 23: + track = f"{item['artistName']} - {item['trackName']}" + counts[track] += 1 + return dict(sorted(counts.items(), key=lambda x: x[1], reverse=True)) + +def unskippable_streak(streaming_history: list[History]) -> tuple[int, str, str]: + """ + Longest sequence of songs played without skipping. + Returns (count, start_date, end_date) + """ + # Sort by time just in case + sorted_history = sorted(streaming_history, key=lambda x: x['endTime']) + + max_streak = 0 + current_streak = 0 + streak_start = None + best_start = None + best_end = None + + for item in sorted_history: + # Check if skipped. Note: 'skipped' might be None in some data exports, + # so we fallback to checking if msPlayed is reasonably long (e.g. > 30s) if skipped is missing + was_skipped = item.get('skipped') + if was_skipped is None: + # Fallback logic: if played < 30s and reasonEnd is not 'trackdone' + was_skipped = (item['msPlayed'].total_seconds() < 30 and item.get('reasonEnd') != 'trackdone') + + if not was_skipped: + if current_streak == 0: + streak_start = item['endTime'] + current_streak += 1 + else: + if current_streak > max_streak: + max_streak = current_streak + best_start = streak_start + best_end = item['endTime'] + current_streak = 0 + + # Check last streak + if current_streak > max_streak: + max_streak = current_streak + best_start = streak_start + best_end = sorted_history[-1]['endTime'] + + start_str = time.strftime('%Y-%m-%d %H:%M', best_start) if best_start else "-" + end_str = time.strftime('%Y-%m-%d %H:%M', best_end) if best_end else "-" + + return max_streak, start_str, end_str + +def artist_hopper(streaming_history: list[History]) -> float: + """ + Average consecutive plays per artist switch. + """ + sorted_history = sorted(streaming_history, key=lambda x: x['endTime']) + if not sorted_history: + return 0.0 + + streaks = [] + current_streak = 0 + last_artist = None + + for item in sorted_history: + artist = item['artistName'] + if artist == last_artist: + current_streak += 1 + else: + if current_streak > 0: + streaks.append(current_streak) + current_streak = 1 + last_artist = artist + + if current_streak > 0: + streaks.append(current_streak) + + return sum(streaks) / len(streaks) if streaks else 0.0 + +def discovery_peak(streaming_history: list[History]) -> tuple[str, int]: + """ + Month with most new artist discoveries. + Returns (month_str, count) + """ + sorted_history = sorted(streaming_history, key=lambda x: x['endTime']) + known_artists = set() + discoveries = defaultdict(int) + + for item in sorted_history: + artist = item['artistName'] + if artist not in known_artists: + month = time.strftime('%Y-%m', item['endTime']) + discoveries[month] += 1 + known_artists.add(artist) + + if not discoveries: + return ("-", 0) + + best_month = max(discoveries.items(), key=lambda x: x[1]) + return best_month + +def comfort_zone(streaming_history: list[History]) -> tuple[float, list[tuple[str, timedelta]]]: + """ + % of total time spent on Top 10 Artists. + Returns (percentage, top_10_list) + """ + artist_time = defaultdict(timedelta) + total_time = timedelta() + + for item in streaming_history: + artist_time[item['artistName']] += item['msPlayed'] + total_time += item['msPlayed'] + + if total_time.total_seconds() == 0: + return 0.0, [] + + sorted_artists = sorted(artist_time.items(), key=lambda x: x[1], reverse=True) + top_10 = sorted_artists[:10] + + top_10_time = sum((t for a, t in top_10), timedelta()) + percentage = (top_10_time.total_seconds() / total_time.total_seconds()) * 100 + + return percentage, top_10 + +def single_day_record(streaming_history: list[History]) -> tuple[str, str, timedelta]: + """ + Most time spent listening to a single artist in one day. + Returns (artist, date_str, duration) + """ + day_artist_time = defaultdict(timedelta) + + for item in streaming_history: + date_str = time.strftime('%Y-%m-%d', item['endTime']) + key = (date_str, item['artistName']) + day_artist_time[key] += item['msPlayed'] + + if not day_artist_time: + return ("-", "-", timedelta(0)) + + (best_date, best_artist), best_time = max(day_artist_time.items(), key=lambda x: x[1]) + + return best_artist, best_date, best_time + +def manual_laborer(streaming_history: list[History]) -> dict[str, int]: + """ + Top tracks played by clicking (reasonStart='clickrow'). + """ + counts = defaultdict(int) + for item in streaming_history: + # reasonStart might vary by platform/year, but 'clickrow' is standard for manual selection + if item.get('reasonStart') == 'clickrow': + track = f"{item['artistName']} - {item['trackName']}" + counts[track] += 1 + return dict(sorted(counts.items(), key=lambda x: x[1], reverse=True)) + +def shuffle_roulette(streaming_history: list[History]) -> dict[str, int]: + """ + Top tracks played when Shuffle was ON. + """ + counts = defaultdict(int) + for item in streaming_history: + if item.get('shuffle') is True: + track = f"{item['artistName']} - {item['trackName']}" + counts[track] += 1 + return dict(sorted(counts.items(), key=lambda x: x[1], reverse=True)) + +def _get_sessions(streaming_history: list[History], gap_minutes: int = 30) -> list[list[History]]: + """Helper to group history into sessions.""" + sorted_history = sorted(streaming_history, key=lambda x: x['endTime']) + sessions = [] + if not sorted_history: + return sessions + + current_session = [sorted_history[0]] + + for i in range(1, len(sorted_history)): + prev_end = sorted_history[i-1]['endTime'] + curr_end = sorted_history[i]['endTime'] + + # Convert struct_time to timestamp for comparison + prev_ts = mktime(prev_end) + curr_ts = mktime(curr_end) + + # Calculate gap. Note: endTime is when the song ENDED. + # So gap = (curr_end - curr_duration) - prev_end + # But we only have endTime and msPlayed. + # Start time of current = curr_end - msPlayed + curr_duration_sec = sorted_history[i]['msPlayed'].total_seconds() + curr_start_ts = curr_ts - curr_duration_sec + + if (curr_start_ts - prev_ts) > (gap_minutes * 60): + sessions.append(current_session) + current_session = [] + + current_session.append(sorted_history[i]) + + if current_session: + sessions.append(current_session) + + return sessions + +def session_starter(streaming_history: list[History]) -> dict[str, int]: + """ + Track that most frequently starts a session. + """ + sessions = _get_sessions(streaming_history) + counts = defaultdict(int) + for sess in sessions: + if sess: + item = sess[0] + track = f"{item['artistName']} - {item['trackName']}" + counts[track] += 1 + return dict(sorted(counts.items(), key=lambda x: x[1], reverse=True)) + +def session_closer(streaming_history: list[History]) -> dict[str, int]: + """ + Track that most frequently ends a session. + """ + sessions = _get_sessions(streaming_history) + counts = defaultdict(int) + for sess in sessions: + if sess: + item = sess[-1] + track = f"{item['artistName']} - {item['trackName']}" + counts[track] += 1 + return dict(sorted(counts.items(), key=lambda x: x[1], reverse=True)) + +def quick_fix(streaming_history: list[History]) -> int: + """ + Count of 'Single Song Sessions'. + """ + sessions = _get_sessions(streaming_history) + return sum(1 for s in sessions if len(s) == 1) + +def skippers_remorse(streaming_history: list[History]) -> dict[str, float]: + """ + Tracks skipped >50% of time, but played >20 times. + Returns {track: skip_rate} + """ + track_stats = defaultdict(lambda: {'plays': 0, 'skips': 0}) + + for item in streaming_history: + track = f"{item['artistName']} - {item['trackName']}" + track_stats[track]['plays'] += 1 + + # Check skip logic + was_skipped = item.get('skipped') + if was_skipped is None: + was_skipped = (item['msPlayed'].total_seconds() < 30 and item.get('reasonEnd') != 'trackdone') + + if was_skipped: + track_stats[track]['skips'] += 1 + + results = {} + for track, stats in track_stats.items(): + if stats['plays'] > 20: + rate = (stats['skips'] / stats['plays']) * 100 + if rate > 50: + results[track] = rate + + return dict(sorted(results.items(), key=lambda x: x[1], reverse=True)) + +def remix_junkie(streaming_history: list[History]) -> tuple[float, int]: + """ + Percentage and count of tracks that are Remix/Mix/Edit. + """ + total = len(streaming_history) + if total == 0: + return 0.0, 0 + + keywords = ['remix', ' mix', ' edit', 'club', 'vip', 'dub'] + count = 0 + for item in streaming_history: + name = item['trackName'].lower() + if any(k in name for k in keywords): + count += 1 + + return (count / total) * 100, count + +def live_fanatic(streaming_history: list[History]) -> tuple[float, int]: + """ + Percentage and count of tracks that are Live/Concert. + """ + total = len(streaming_history) + if total == 0: + return 0.0, 0 + + keywords = ['live', 'concert', 'performance', 'session', 'tour'] + count = 0 + for item in streaming_history: + # Check both track and album + name = (item['trackName'] + " " + item['albumName']).lower() + if any(k in name for k in keywords): + count += 1 + + return (count / total) * 100, count + +def short_king(streaming_history: list[History]) -> dict[str, int]: + """ + Most played tracks under 2 minutes (that were finished). + """ + counts = defaultdict(int) + for item in streaming_history: + # 2 mins = 120 seconds + if item['msPlayed'].total_seconds() < 120 and item.get('reasonEnd') == 'trackdone': + track = f"{item['artistName']} - {item['trackName']}" + counts[track] += 1 + return dict(sorted(counts.items(), key=lambda x: x[1], reverse=True)) + +def epic_saga(streaming_history: list[History]) -> dict[str, int]: + """ + Most played tracks over 7 minutes (that were finished). + """ + counts = defaultdict(int) + for item in streaming_history: + # 7 mins = 420 seconds + if item['msPlayed'].total_seconds() > 420 and item.get('reasonEnd') == 'trackdone': + track = f"{item['artistName']} - {item['trackName']}" + counts[track] += 1 + return dict(sorted(counts.items(), key=lambda x: x[1], reverse=True)) + +def collaborator(streaming_history: list[History]) -> dict[str, int]: + """ + Most played tracks featuring other artists. + """ + counts = defaultdict(int) + keywords = ['feat.', 'ft.', 'with ', 'featuring'] + for item in streaming_history: + name = (item['trackName'] + " " + item['artistName']).lower() + if any(k in name for k in keywords): + track = f"{item['artistName']} - {item['trackName']}" + counts[track] += 1 + return dict(sorted(counts.items(), key=lambda x: x[1], reverse=True)) + +def alphabet_artists(streaming_history: list[History]) -> dict[str, tuple[str, int]]: + """ + Most played Artist for A-Z. + """ + # First count all artists + artist_counts = defaultdict(int) + for item in streaming_history: + artist_counts[item['artistName']] += 1 + + # Group by letter + letter_best = {} + for char in "ABCDEFGHIJKLMNOPQRSTUVWXYZ": + # Filter artists starting with char + candidates = {k: v for k, v in artist_counts.items() if k.upper().startswith(char)} + if candidates: + best_artist = max(candidates.items(), key=lambda x: x[1]) + letter_best[char] = best_artist + + return letter_best + +def spelling_bee(streaming_history: list[History]) -> tuple[str, int, str, int]: + """ + Longest Artist Name and Track Title. + Returns (artist, len, track, len) + """ + max_artist = "" + max_track = "" + + for item in streaming_history: + if len(item['artistName']) > len(max_artist): + max_artist = item['artistName'] + if len(item['trackName']) > len(max_track): + max_track = item['trackName'] + + return max_artist, len(max_artist), max_track, len(max_track) + +def same_name_game(streaming_history: list[History]) -> dict[str, int]: + """ + Track titles listened to from the most DIFFERENT artists. + """ + title_artists = defaultdict(set) + + for item in streaming_history: + # Normalize title slightly to catch "Home" vs "Home " + title = item['trackName'].strip() + # Ignore generic titles like "Intro", "Untitled" + if title.lower() in ['intro', 'untitled', 'track 1', 'outro']: + continue + title_artists[title].add(item['artistName']) + + # Count unique artists per title + counts = {t: len(a) for t, a in title_artists.items() if len(a) > 1} + return dict(sorted(counts.items(), key=lambda x: x[1], reverse=True)) + +def midnight_club(streaming_history: list[History]) -> dict[str, int]: + """ + Top artists played between Midnight and 1 AM. + """ + counts = defaultdict(int) + for item in streaming_history: + hour = item['endTime'].tm_hour + if hour == 0: + counts[item['artistName']] += 1 + return dict(sorted(counts.items(), key=lambda x: x[1], reverse=True)) + +def lunch_break(streaming_history: list[History]) -> dict[str, int]: + """ + Top artists played between 12 PM and 2 PM. + """ + counts = defaultdict(int) + for item in streaming_history: + hour = item['endTime'].tm_hour + if 12 <= hour < 14: + counts[item['artistName']] += 1 + return dict(sorted(counts.items(), key=lambda x: x[1], reverse=True)) + +def monday_blues(streaming_history: list[History]) -> dict[str, int]: + """ + Top tracks played on Mondays. + """ + counts = defaultdict(int) + for item in streaming_history: + if item['endTime'].tm_wday == 0: # Monday is 0 + track = f"{item['artistName']} - {item['trackName']}" + counts[track] += 1 + return dict(sorted(counts.items(), key=lambda x: x[1], reverse=True)) + +def hump_day_hero(streaming_history: list[History]) -> dict[str, int]: + """ + Top tracks played on Wednesdays. + """ + counts = defaultdict(int) + for item in streaming_history: + if item['endTime'].tm_wday == 2: # Wednesday is 2 + track = f"{item['artistName']} - {item['trackName']}" + counts[track] += 1 + return dict(sorted(counts.items(), key=lambda x: x[1], reverse=True)) + +def quarterly_review(streaming_history: list[History]) -> dict[int, tuple[str, int]]: + """ + Top track for each Quarter (Q1-Q4). + """ + q_counts = defaultdict(lambda: defaultdict(int)) + + for item in streaming_history: + month = item['endTime'].tm_mon + quarter = (month - 1) // 3 + 1 + track = f"{item['artistName']} - {item['trackName']}" + q_counts[quarter][track] += 1 + + results = {} + for q in range(1, 5): + if q_counts[q]: + best = max(q_counts[q].items(), key=lambda x: x[1]) + results[q] = best + else: + results[q] = ("-", 0) + + return results + +def album_purist(streaming_history: list[History]) -> tuple[int, str, str]: + """ + Longest streak of unique songs played from the same album in a row. + Returns (streak_length, album_name, artist_name) + """ + sorted_history = sorted(streaming_history, key=lambda x: x['endTime']) + + max_streak = 0 + best_album = "-" + best_artist = "-" + + current_album = None + current_artist = None + current_tracks = set() + + for item in sorted_history: + album = item['albumName'] + artist = item['artistName'] + track = item['trackName'] + + # Ignore empty albums or singles (often album name is same as track name) + if not album or album == track: + current_album = None + current_tracks = set() + continue + + if album == current_album and artist == current_artist: + current_tracks.add(track) + else: + # Check previous streak + if len(current_tracks) > max_streak: + max_streak = len(current_tracks) + best_album = current_album + best_artist = current_artist + + # Start new streak + current_album = album + current_artist = artist + current_tracks = {track} + + # Check last + if len(current_tracks) > max_streak: + max_streak = len(current_tracks) + best_album = current_album + best_artist = current_artist + + return max_streak, best_album, best_artist + +def instant_skips(streaming_history: list[History]) -> dict[str, int]: + """ + Returns a dictionary of songs skipped in less than 1 second (1000ms). + Key: "Artist - Track" + Value: Count of instant skips + """ + skips = {} + for item in streaming_history: + # Check if skipped and duration < 1 second + # msPlayed is a timedelta, so we check total_seconds() + if item['skipped'] and item['msPlayed'].total_seconds() < 1.0: + key = f"{item['artistName']} - {item['trackName']}" + if key not in skips: + skips[key] = 0 + skips[key] += 1 + + return {k: v for k, v in sorted(skips.items(), key=lambda item: item[1], reverse=True)} + + + +def get_full_song_stats(streaming_history: list[History]) -> list[dict]: + """ + Aggregates stats for all songs for CSV export. + Returns a list of dicts with keys: + Artist, Track Name, Times Played, First Played, Last Played, Skipped, Instant Skips, User Started + """ + stats = {} + + for item in streaming_history: + key = (item['artistName'], item['trackName']) + + if key not in stats: + stats[key] = { + 'Artist': item['artistName'], + 'Track Name': item['trackName'], + 'Times Played': 0, + 'First Played': item['endTime'], + 'Last Played': item['endTime'], + 'Skipped': 0, + 'Instant Skips': 0, + 'User Started': 0 + } + + entry = stats[key] + entry['Times Played'] += 1 + + # Update dates + if item['endTime'] < entry['First Played']: + entry['First Played'] = item['endTime'] + if item['endTime'] > entry['Last Played']: + entry['Last Played'] = item['endTime'] + + # Update counts + if item['skipped']: + entry['Skipped'] += 1 + if item['msPlayed'].total_seconds() < 1.0: + entry['Instant Skips'] += 1 + + if item['reasonStart'] in ['clickrow', 'playbtn']: + entry['User Started'] += 1 + + # Format dates + result = [] + for entry in stats.values(): + entry['First Played'] = time.strftime('%Y-%m-%d %H:%M:%S', entry['First Played']) + entry['Last Played'] = time.strftime('%Y-%m-%d %H:%M:%S', entry['Last Played']) + result.append(entry) + + return result diff --git a/stats/graphs.py b/stats/graphs.py new file mode 100644 index 0000000..f79b69d --- /dev/null +++ b/stats/graphs.py @@ -0,0 +1,449 @@ +import matplotlib.pyplot as plt +import numpy as np + +def plot_top_items(data: dict, title: str, n: int = 10): + """ + Plots a bar chart of the top n items from the dictionary. + """ + top_items = list(data.items())[:n] + labels = [item[0] for item in top_items] + values = [item[1] for item in top_items] + + # Reverse for horizontal bar chart to have top item at the top + labels.reverse() + values.reverse() + + fig = plt.figure(figsize=(10, 8)) + plt.barh(labels, values, color='skyblue') + plt.xlabel('Play Count') + plt.title(title) + plt.tight_layout() + return fig + +def plot_artist_trends(data: dict, title: str): + """ + Plots a line chart for artist listening trends over years. + """ + fig = plt.figure(figsize=(12, 8)) + + # Get all unique years across all artists to set x-axis + all_years = set() + for year_data in data.values(): + all_years.update(year_data.keys()) + + if not all_years: + print("No data to plot.") + return fig + + sorted_years = sorted(list(all_years)) + + for artist, year_counts in data.items(): + # Fill in missing years with 0 + counts = [year_counts.get(year, 0) for year in sorted_years] + plt.plot(sorted_years, counts, label=artist, marker='o') + + plt.xlabel('Year') + plt.ylabel('Play Count') + plt.title(title) + plt.legend(bbox_to_anchor=(1.05, 1), loc='upper left') + plt.grid(True) + plt.tight_layout() + return fig + +def plot_platform_usage(data: dict, title: str): + """ + Plots a pie chart for platform usage. + """ + labels = list(data.keys()) + sizes = list(data.values()) + + fig = plt.figure(figsize=(8, 8)) + plt.pie(sizes, labels=labels, autopct='%1.1f%%', startangle=140) + plt.axis('equal') # Equal aspect ratio ensures that pie is drawn as a circle. + plt.title(title) + plt.tight_layout() + return fig + + +def plot_listening_by_hour(data: dict, title: str): + """ + Plots a bar chart for listening activity by hour of the day. + """ + hours = list(data.keys()) + counts = list(data.values()) + + fig = plt.figure(figsize=(10, 6)) + plt.bar(hours, counts, color='lightgreen') + plt.xlabel('Hour of Day (0-23)') + plt.ylabel('Play Count') + plt.title(title) + plt.xticks(hours) + plt.grid(axis='y', linestyle='--', alpha=0.7) + plt.tight_layout() + return fig + +def plot_location_counts(data: dict, title: str): + """ + Plots a bar chart for plays by location (country). + """ + return plot_top_items(data, title, n=len(data)) + + +def plot_longest_played_artist(data: dict, title: str, n: int = 10): + """ + Plots a bar chart for longest played artists (in hours). + """ + top_items = list(data.items())[:n] + labels = [item[0] for item in top_items] + # Convert timedelta to hours for plotting + values = [item[1].total_seconds() / 3600 for item in top_items] + + labels.reverse() + values.reverse() + + fig = plt.figure(figsize=(10, 8)) + plt.barh(labels, values, color='salmon') + plt.xlabel('Hours Played') + plt.title(title) + plt.tight_layout() + return fig + + +def plot_skipped_items(data: dict, title: str, n: int = 10): + """ + Plots a bar chart for most skipped items. + """ + top_items = list(data.items())[:n] + labels = [item[0] for item in top_items] + values = [item[1] for item in top_items] + + labels.reverse() + values.reverse() + + fig = plt.figure(figsize=(10, 8)) + plt.barh(labels, values, color='orange') + plt.xlabel('Skip Count') + plt.title(title) + plt.tight_layout() + return fig + +def plot_listening_by_day_of_week(data: dict, title: str): + """ + Plots a bar chart for listening by day of week. + """ + days = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday'] + counts = [data.get(day, 0) for day in days] + + fig = plt.figure(figsize=(10, 6)) + plt.bar(days, counts, color='mediumpurple') + plt.xlabel('Day of Week') + plt.ylabel('Play Count') + plt.title(title) + plt.tight_layout() + return fig + + +def plot_discovery_rate(data: dict, title: str): + """ + Plots a line chart for new artists discovered per month. + """ + months = list(data.keys()) + counts = list(data.values()) + + fig = plt.figure(figsize=(12, 6)) + plt.plot(months, counts, marker='o', linestyle='-', color='teal') + plt.xlabel('Month') + plt.ylabel('New Artists') + plt.title(title) + plt.xticks(rotation=45) + + # Reduce x-ticks if too many + if len(months) > 20: + plt.xticks(months[::int(len(months)/20)], rotation=45) + + plt.tight_layout() + return fig + + +def plot_seasonal_listening(data: dict, title: str): + """ + Plots a pie chart for seasonal listening. + """ + labels = list(data.keys()) + sizes = list(data.values()) + colors = ['lightblue', 'lightgreen', 'gold', 'orange'] # Winter, Spring, Summer, Autumn + + fig = plt.figure(figsize=(8, 8)) + plt.pie(sizes, labels=labels, autopct='%1.1f%%', startangle=90, colors=colors) + plt.axis('equal') + plt.title(title) + plt.tight_layout() + return fig + + +def plot_day_night_split(data: dict, title: str): + """ + Plots a pie chart for day vs night listening. + """ + labels = list(data.keys()) + sizes = list(data.values()) + colors = ['gold', 'midnightblue'] + + fig = plt.figure(figsize=(8, 8)) + plt.pie(sizes, labels=labels, autopct='%1.1f%%', startangle=90, colors=colors) + plt.axis('equal') + plt.title(title) + plt.tight_layout() + return fig + +def plot_hourly_heatmap(data: list[list[int]], title: str): + """ + Plots a heatmap of listening activity (Day of Week vs Hour of Day). + """ + days = ['Mon', 'Tue', 'Wed', 'Thu', 'Fri', 'Sat', 'Sun'] + hours = [str(i) for i in range(24)] + + fig = plt.figure(figsize=(12, 6)) + plt.imshow(data, cmap='YlGnBu', aspect='auto') + + plt.xticks(range(len(hours)), hours) + plt.yticks(range(len(days)), days) + + plt.xlabel('Hour of Day') + plt.ylabel('Day of Week') + plt.title(title) + plt.colorbar(label='Play Count') + + plt.tight_layout() + return fig + +def plot_variety_score(data: dict, title: str): + """ + Plots a line chart for Variety Score over years. + """ + years = list(data.keys()) + scores = list(data.values()) + + fig = plt.figure(figsize=(10, 6)) + plt.plot(years, scores, marker='o', linestyle='-', color='purple') + plt.xlabel('Year') + plt.ylabel('Variety Score (Unique Artists / Total Plays)') + plt.title(title) + plt.grid(True) + plt.tight_layout() + return fig + +def plot_calendar_heatmap(data: list[list[int]], years: list[int], months: list[str], title: str): + """ + Plots Year vs Month heatmap. + """ + fig = plt.figure(figsize=(10, len(years)*0.8 + 2)) + plt.imshow(data, cmap='Greens', aspect='auto') + + plt.xticks(range(len(months)), months) + plt.yticks(range(len(years)), years) + + plt.xlabel('Month') + plt.ylabel('Year') + plt.title(title) + plt.colorbar(label='Play Count') + + plt.tight_layout() + return fig + + +def plot_picky_grid(data: list[list[float]], title: str): + """ + Plots Day vs Hour Skip Rate heatmap. + """ + days = ['Mon', 'Tue', 'Wed', 'Thu', 'Fri', 'Sat', 'Sun'] + hours = [str(i) for i in range(24)] + + fig = plt.figure(figsize=(12, 6)) + plt.imshow(data, cmap='Reds', aspect='auto') + + plt.xticks(range(len(hours)), hours) + plt.yticks(range(len(days)), days) + + plt.xlabel('Hour of Day') + plt.ylabel('Day of Week') + plt.title(title) + plt.colorbar(label='Skip Rate %') + + plt.tight_layout() + return fig + + +def plot_device_habits(data: list[list[int]], platforms: list[str], title: str): + """ + Plots Platform vs Hour heatmap. + """ + hours = [str(i) for i in range(24)] + + fig = plt.figure(figsize=(12, len(platforms)*0.8 + 2)) + plt.imshow(data, cmap='Blues', aspect='auto') + + plt.xticks(range(len(hours)), hours) + plt.yticks(range(len(platforms)), platforms) + + plt.xlabel('Hour of Day') + plt.ylabel('Platform') + plt.title(title) + plt.colorbar(label='Play Count') + + plt.tight_layout() + return fig + + +def plot_artist_eras(data: list[list[float]], artists: list[str], time_labels: list[str], title: str): + """ + Plots Top Artists vs Time (Eras) heatmap. + """ + # Dynamic height: 0.25 inches per artist, min 6, max 30 (to fit screen better) + height = max(6, min(30, len(artists) * 0.25)) + fig = plt.figure(figsize=(14, height)) + + plt.imshow(data, cmap='magma', aspect='auto', interpolation='nearest') + + # Reduce x-ticks + step = max(1, len(time_labels) // 20) + plt.xticks(range(0, len(time_labels), step), [time_labels[i] for i in range(0, len(time_labels), step)], rotation=45) + + # Smaller font for y-ticks if many artists + fontsize = 10 if len(artists) < 30 else 6 + plt.yticks(range(len(artists)), artists, fontsize=fontsize) + + plt.xlabel('Time') + plt.ylabel('Artist') + plt.title(title) + plt.colorbar(label='Relative Listening Intensity (Normalized per Artist)') + + plt.tight_layout() + return fig + +def plot_active_heatmap(data: list[list[int]], title: str): + """ + Plots Active Starts (Day vs Hour) heatmap. + """ + days = ['Mon', 'Tue', 'Wed', 'Thu', 'Fri', 'Sat', 'Sun'] + hours = [str(i) for i in range(24)] + + fig = plt.figure(figsize=(12, 6)) + plt.imshow(data, cmap='Greens', aspect='auto') + + plt.xticks(range(len(hours)), hours) + plt.yticks(range(len(days)), days) + + plt.xlabel('Hour of Day') + plt.ylabel('Day of Week') + plt.title(title) + plt.colorbar(label='Active Starts (Clicks)') + + plt.tight_layout() + return fig + + +def plot_active_trend(data: dict[int, float], title: str): + """ + Plots Active Listening % trend over years. + """ + years = list(data.keys()) + values = list(data.values()) + + fig = plt.figure(figsize=(10, 6)) + plt.plot(years, values, marker='o', linestyle='-', color='forestgreen', linewidth=2) + + plt.xlabel('Year') + plt.ylabel('Active Listening %') + plt.title(title) + plt.grid(True, linestyle='--', alpha=0.7) + plt.tight_layout() + return fig + +def plot_artist_radar(data: dict, title: str): + """ + Plots a radar chart comparing artist traits. + """ + labels = ['Loyalty', 'Discovery', 'Night Owl', 'Weekend', 'Active Choice'] + num_vars = len(labels) + + # Compute angle for each axis + angles = np.linspace(0, 2 * np.pi, num_vars, endpoint=False).tolist() + angles += angles[:1] # Close the loop + + fig, ax = plt.subplots(figsize=(8, 8), subplot_kw=dict(polar=True)) + + colors = ['#FF5733', '#33FF57', '#3357FF', '#F333FF', '#FF33A8'] + + for i, (artist, stats) in enumerate(data.items()): + values = list(stats) + values += values[:1] # Close the loop + ax.plot(angles, values, linewidth=2, label=artist, color=colors[i % len(colors)]) + ax.fill(angles, values, alpha=0.1, color=colors[i % len(colors)]) + + ax.set_theta_offset(np.pi / 2) + ax.set_theta_direction(-1) + + ax.set_xticks(angles[:-1]) + ax.set_xticklabels(labels) + + ax.set_title(title, y=1.1) + ax.legend(loc='upper right', bbox_to_anchor=(1.1, 1.1)) + + return fig + +def create_text_pages(text: str, lines_per_page: int = 60): + """ + Converts a long string of text into a list of matplotlib figures, + each representing a page of text. + """ + lines = text.split('\n') + pages = [] + + for i in range(0, len(lines), lines_per_page): + chunk = lines[i:i + lines_per_page] + page_text = '\n'.join(chunk) + + fig = plt.figure(figsize=(8.27, 11.69)) # A4 size + plt.axis('off') + # Use a monospace font to preserve alignment + plt.text(0.05, 0.95, page_text, transform=fig.transFigure, + fontsize=8, family='monospace', verticalalignment='top') + pages.append(fig) + + return pages + +def plot_longest_played_tracks(data: dict, title: str, n: int = 10): + """ + Plots a bar chart for longest played tracks (in hours). + """ + top_items = list(data.items())[:n] + labels = [item[0] for item in top_items] + values = [item[1].total_seconds() / 3600 for item in top_items] + + labels.reverse() + values.reverse() + + fig = plt.figure(figsize=(10, 8)) + plt.barh(labels, values, color='mediumpurple') + plt.xlabel('Hours Played') + plt.title(title) + plt.tight_layout() + return fig + +def plot_comfort_zone(percentage: float, title: str): + """ + Plots a pie chart showing the Comfort Zone percentage. + """ + labels = ['Top 10 Artists', 'Others'] + sizes = [percentage, 100 - percentage] + colors = ['#1DB954', '#191414'] # Spotify Green and Black + explode = (0.1, 0) # explode the 1st slice + + fig = plt.figure(figsize=(8, 8)) + plt.pie(sizes, explode=explode, labels=labels, colors=colors, + autopct='%1.1f%%', shadow=True, startangle=140) + plt.axis('equal') # Equal aspect ratio ensures that pie is drawn as a circle. + plt.title(title) + plt.tight_layout() + return fig