From a5cfedbe17c0b46f3f48310b753151ac7db928c1 Mon Sep 17 00:00:00 2001 From: Robert Kaye Date: Thu, 11 Apr 2024 16:56:06 +0200 Subject: [PATCH] Remove two outdated patches. A better world trip patch is coming soon. --- tests/test_utils.py | 2 - troi/patches/periodic_jams.py | 2 +- troi/patches/weekly_flashback_jams.py | 124 ------------------ troi/patches/world_trip.py | 174 -------------------------- 4 files changed, 1 insertion(+), 301 deletions(-) delete mode 100755 troi/patches/weekly_flashback_jams.py delete mode 100755 troi/patches/world_trip.py diff --git a/tests/test_utils.py b/tests/test_utils.py index 2504502..38d2c4a 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -10,9 +10,7 @@ def test_discover_patches(self): patches = discover_patches() assert "periodic-jams" in patches - assert "weekly-flashback-jams" in patches assert "playlist-from-mbids" in patches - assert "world-trip" in patches assert "recs-to-playlist" in patches assert "transfer-playlist" in patches diff --git a/troi/patches/periodic_jams.py b/troi/patches/periodic_jams.py index b9973d5..b85f78a 100755 --- a/troi/patches/periodic_jams.py +++ b/troi/patches/periodic_jams.py @@ -100,7 +100,7 @@ def create(self, inputs): else: jam_types = jam_type.lower() if jam_type not in self.JAM_TYPES: - raise RuntimeError("Jam type must be one of %s" % ", ".join(jam_types)) + raise RuntimeError("Jam type must be one of %s" % ", ".join(self.JAM_TYPES)) recs = troi.listenbrainz.recs.UserRecordingRecommendationsElement(user_name, "raw", diff --git a/troi/patches/weekly_flashback_jams.py b/troi/patches/weekly_flashback_jams.py deleted file mode 100755 index 2c9a55f..0000000 --- a/troi/patches/weekly_flashback_jams.py +++ /dev/null @@ -1,124 +0,0 @@ -import random -from collections import defaultdict - -import troi.filters -import troi.listenbrainz.recs -import troi.musicbrainz.mbid_mapping -import troi.musicbrainz.recording_lookup -import troi.playlist -import troi.sorts -from troi import Element, Recording, Playlist, PipelineError - - -class DecadePlaylistSplitterElement(Element): - ''' - Take a list of recordings that have their year attribute filled out - and output N playlists broken down by decade of the Recordings. Recordings - with no year set will be ignored and playlists will only be generated - for decades that have at least minimum_count recordings. - ''' - - def __init__(self, minimum_count=20): - Element.__init__(self) - self.minimum_count = minimum_count - - @staticmethod - def inputs(): - return [Recording] - - @staticmethod - def outputs(): - return [Playlist] - - def read(self, inputs): - """ - Sort the recordings into decades and return playlists for decades that have the minimum number of tracks. - """ - recordings = inputs[0] - if not recordings or len(recordings) == 0: - return inputs[0] - - decades = defaultdict(list) - for r in recordings: - if not r.year: - continue - - decade = (r.year // 10) * 10 - decades[decade].append(r) - - playlists = [] - for decade in decades: - if len(decades[decade]) < self.minimum_count: - continue - - random.shuffle(decades[decade]) - playlists.append(Playlist("%ss flashback jams" % str(decade), filename="%ss_flashback_jams.jspf" % str(decade), - recordings=decades[decade])) - - return playlists - - -class WeeklyFlashbackJams(troi.patch.Patch): - """ - See below for description - """ - - def __init__(self, args): - troi.patch.Patch.__init__(self, args) - - @staticmethod - def inputs(): - """ - Generate weekly flashback playlists from the ListenBrainz recommended recordings. - - \b - USER_NAME: is a MusicBrainz user name that has an account on ListenBrainz. - TYPE: is The type of daily jam. Must be 'top', 'similar' or 'raw'. - """ - return [ - {"type": "argument", "args": ["user_name"]}, - {"type": "argument", "args": ["type"]} - ] - - @staticmethod - def outputs(): - return [Recording] - - @staticmethod - def slug(): - return "weekly-flashback-jams" - - @staticmethod - def description(): - return "Generate weekly flashback playlists from the ListenBrainz recommended recordings." - - def create(self, inputs): - user_name = inputs['user_name'] - type = inputs['type'] - - if type not in ("top", "similar", "raw"): - raise PipelineError("type must be either 'top', 'similar' or 'raw'") - - recs = troi.listenbrainz.recs.UserRecordingRecommendationsElement(user_name=user_name, - artist_type=type, - count=-1) - r_lookup = troi.musicbrainz.recording_lookup.RecordingLookupElement(skip_not_found=True) - r_lookup.set_sources(recs) - - y_lookup = troi.musicbrainz.mbid_mapping.MBIDMappingLookupElement() - y_lookup.set_sources(r_lookup) - - # Filter out tracks that do not fit into the given year range - year_sort = troi.sorts.YearSortElement() - year_sort.set_sources(y_lookup) - - decade_splitter = DecadePlaylistSplitterElement() - decade_splitter.set_sources(year_sort) - - shaper = troi.playlist.PlaylistRedundancyReducerElement(max_artist_occurrence=3) - shaper.set_sources(decade_splitter) - - shuffle = troi.playlist.PlaylistShuffleElement() - shuffle.set_sources(shaper) - - return shuffle diff --git a/troi/patches/world_trip.py b/troi/patches/world_trip.py deleted file mode 100755 index 8908f9c..0000000 --- a/troi/patches/world_trip.py +++ /dev/null @@ -1,174 +0,0 @@ -import logging -from collections import defaultdict -from urllib.parse import quote - -import countryinfo -import requests - -import troi.patch -from troi import Element, Artist, Recording, Playlist, PipelineError, DEVELOPMENT_SERVER_URL - -logger = logging.getLogger(__name__) - - -def recording_from_row(row): - if row['recording_mbid'] is None: - return None - - r = Recording(mbid=row['recording_mbid']) - if 'artist_credit_name' in row: - r.artist = Artist(name=row['artist_credit_name']) - - if 'recording_name' in row: - r.name = row['recording_name'] - - if 'year' in row: - r.year = row['year'] - - if 'listen_count' in row: - r.listenbrainz={"listen_count": row["listen_count"]} - - return r - - -class WorldTripElement(Element): - ''' - Given a continent (africa, americas, asia, europe, oceania) and - a sort order latitude or longitude, pick random tracks from each country and - add tracks to a playlist sorted by latitude or longitude. - - Arguments: - continent: one of: (Africa, North America, South America, Asia, Europe, Oceania) - latitude: boolean. Sort by latiduded if True, otherwise sort by longitude - ''' - - def __init__(self, continent, latitude): - super().__init__() - self.continent = continent - self.latitude = latitude - - @staticmethod - def inputs(): - return [] - - @staticmethod - def outputs(): - return [Recording] - - def read(self, inputs): - - countries = countryinfo.CountryInfo().all() - - continents = defaultdict(list) - for country in countries.values(): - if 'region' not in country: - continue - - continents[country['region'].lower()].append({'name': country['name'], - 'code': country['ISO']['alpha2'], - 'latlng': country['latlng'] }) - - if self.continent not in continents: - names = continents.keys() - raise RuntimeError(f"Cannot find continent {self.continent}. Must be one of {names}") - - logger.info("Fetch tracks from countries:") - if self.latitude: - continent = sorted(continents[self.continent], key=lambda c: c['latlng'][0], reverse=True) - else: - continent = sorted(continents[self.continent], key=lambda c: c['latlng'][1]) - - for i, country in enumerate(continent): - r = requests.get("http://musicbrainz.org/ws/2/area?query=%s&fmt=json" % country['name']) - if r.status_code != 200: - raise PipelineError("Cannot fetch country code from MusicBrainz. HTTP code %s" % r.status_code) - - country_code = r.json()['areas'][0]['id'] - r = requests.post(DEVELOPMENT_SERVER_URL + "/area-random-recordings/json", json=[{ "area_mbid": country_code, - "start_year": 2012, - "end_year": 2021 }]) - if r.status_code != 200: - raise PipelineError("Cannot fetch first dataset recordings from ListenBrainz. HTTP code %s (%s)" % (r.status_code, r.text)) - - recordings = [] - for row in r.json(): - recordings.append(recording_from_row(row)) - - country["recordings"] = recordings - - - recordings = [] - for i in range(3): - for country in continent: - try: - recordings.append(country["recordings"][i]) - except KeyError: - logger.error("Found no tracks for %s" % country["name"]) - except IndexError: - logger.error("Found too few tracks for %s" % country["name"]) - - return recordings - - -class WorldTripPatch(troi.patch.Patch): - """ - See below for description - """ - - NAME = "Three trips across %s with ListenBrainz (%s)" - DESC = """

- This playlist contains randomly selected tracks from the last few years released in - the countries of %s, arranged from %s. Once the playlist reaches the southernmost - country, the list repeats twice more, for a total of three trips across the continent. -

- """ - - def __init__(self): - troi.patch.Patch.__init__(self) - - @staticmethod - def inputs(): - """ - Generate a playlist that picks tracks for each country in a continent. - - \b - CONTINENT: A name of a continent, all lower case. - SORT: Must be longitude or latitude - """ - return [ - {"type": "argument", "args": ["continent"]}, - {"type": "argument", "args": ["sort"]} - ] - - @staticmethod - def outputs(): - return [Playlist] - - @staticmethod - def slug(): - return "world-trip" - - @staticmethod - def description(): - return "Generate a playlist for a given continent" - - def create(self, inputs): - - if inputs["sort"] not in ("longitude", "latitude"): - raise PipelineError("Argument sort must be either 'longitude' or 'latitude'.") - - if inputs["sort"] == "longitude": - latitude = False - lat_string = "North -> South" - elif inputs["sort"] == "latitude": - latitude = True - lat_string = "West -> East" - else: - raise RuntimeError("sort must be longitude or latitude") - - trip = WorldTripElement(inputs['continent'], latitude) - pl_maker = troi.playlist.PlaylistMakerElement(self.NAME % (inputs["continent"], lat_string), - self.DESC % (quote(inputs["continent"]), lat_string)) - pl_maker.set_sources(trip) - - return pl_maker