Feet saber album process!

README.md

@@ -59,11 +59,7 @@ Generate full DD Beat Map from json files in repository (no --song-id does all)
 Generate full DD Beat Map from Feet Saber directory (WIP)
 
 ```bash
-# WIP when I have time.  https://beatsaver.com/maps/229ed is what i've been using to test.
-.venv/Scripts/python fs2dd.py --fs-map-dir "229ed (Yell! (DJ Shimamura Remix) [feat. Moimoi] [Feet saber] - KikaeAeon)"
+.venv/Scripts/python fs2dd.py --fs-map-dir "path/to/map/folder"
+.venv/Scripts/python fs2dd.py --fs-map-id 229ed
+.venv/Scripts/python fs2dd.py --fs-playlist-id 3474
 ```
-
-### Feet Saber TODO:
-
-- Map 1-9 x position of notes (model/feetsaber.py:213)
-- Map timings and logic for line notes (fs2dd.py:74)

fs2dd.py

@@ -1,12 +1,16 @@
 from __future__ import annotations
 
 import argparse
+import concurrent
 import os
 import shutil
+import zipfile
+from concurrent.futures import ThreadPoolExecutor
 from datetime import datetime
 
 from model.dancedash import DD_LINE_LEFT
 from model.dancedash import DD_LINE_RIGHT
+from model.dancedash import DDAlbumInfo
 from model.dancedash import DDBeatMap
 from model.dancedash import DDBeatMapData
 from model.dancedash import DDBeatMapInfoFile
@@ -25,9 +29,14 @@ from model.feetsaber import FS_TO_DD_NOTE_TYPE
 from model.feetsaber import FSBeatMapFile
 from model.feetsaber import FSInfoDat
 from util import convert_egg_to_ogg_and_get_length
+from util import download_and_extract_zip
+from util import download_image_from_url
+from util import get_feet_saber_map_from_id
+from util import get_feet_saber_maps_from_playlist
 from util import get_first_image_file_in_folder
 from util import random_9_digit_int
 from util import yyyymmdd_to_ticks
+from util import zipdir
 
 
 def map_sphere_nodes(
@@ -38,7 +47,7 @@ def map_sphere_nodes(
 
     def process_notes(note_type: FS_LEFT_NOTE | FS_RIGHT_NOTE, line_type: DD_LINE_LEFT | DD_LINE_RIGHT):
         all_lines_of_type = [
-            o for o in fs_beat_map.obstacles if o.customData.dd_note_type == line_type
+            o for o in fs_beat_map.obstacles if o.customData and o.customData.dd_note_type == line_type
         ]
         all_notes = [
             n for n in fs_beat_map.notes if n.type == note_type
@@ -70,7 +79,7 @@ def map_line_nodes(
     lines = []
 
     line_obstacles = [
-        o for o in fs_beat_map.obstacles if o.customData.is_fs
+        o for o in fs_beat_map.obstacles if o.customData and o.customData.is_fs
     ]
 
     obstacles_by_type = {
@@ -182,7 +191,7 @@ def map_down_and_jump_notes(
     return sorted(downs + ups, key=lambda r: r.noteOrder)
 
 
-def create_dd_tracks_from_fs(fs_map_dir: str) -> DDBeatMapInfoFile:
+def create_dd_tracks_from_fs(fs_map_dir: str, prefix_dir: str = '', ost_id: int | None = None) -> DDBeatMapInfoFile:
     info_file_path = os.path.join(fs_map_dir, 'Info.dat')
     fs_info = FSInfoDat.from_json_file(info_file_path)
 
@@ -191,7 +200,7 @@ def create_dd_tracks_from_fs(fs_map_dir: str) -> DDBeatMapInfoFile:
             fs_map_dir,
         ) if f.endswith('.egg')
     ][0]
-    target_dir = f'{fs_info.songName} - {fs_info.levelAuthorName}'
+    target_dir = f'{prefix_dir}/{fs_info.songName} - {fs_info.levelAuthorName}'
     if not os.path.exists(target_dir):
         os.makedirs(target_dir)
 
@@ -252,7 +261,7 @@ def create_dd_tracks_from_fs(fs_map_dir: str) -> DDBeatMapInfoFile:
 
     create_ticks = yyyymmdd_to_ticks(datetime.now().strftime('%Y%m%d'))
     dd_beat_map_info = DDBeatMapInfoFile(
-        OstId=0,  # default
+        OstId=ost_id,
         CreateTicks=create_ticks,
         CreateTime=str(create_ticks),
         BeatMapId=random_9_digit_int(),
@@ -280,13 +289,78 @@ if __name__ == '__main__':
         description='Create Dance Dash tracks from Feet Saber maps',
     )
     parser.add_argument(
-        '--fs-map-dir',
-        type=str,
-        help='The directory containing the Feet Saber map files',
-        required=True,
+        '--fs-map-dir', type=str, help='The directory containing the Feet Saber map files', required=False,
+    )
+    parser.add_argument(
+        '--fs-map-id', type=str,
+        help='The ID to fetch the Feet Saber map zip', required=False,
+    )
+    parser.add_argument(
+        '--fs-playlist-id', type=str,
+        help='The ID to fetch the Feet Saber playlist', required=False,
     )
 
     args = parser.parse_args()
+
+    if args.fs_playlist_id:
+        ost_id = int(args.fs_playlist_id) if args.fs_playlist_id.isdigit(
+        ) else random_9_digit_int()
+        target_dir = f'bin/{ost_id}'
+
+        if not os.path.exists(target_dir):
+            os.makedirs(target_dir)
+
+        fs_map_urls, album_cover_url, author = get_feet_saber_maps_from_playlist(
+            args.fs_playlist_id,
+        )
+        album_cover_path = None
+        if album_cover_url:
+            album_cover_path = download_image_from_url(
+                album_cover_url, f'{target_dir}/{ost_id}',
+            )
+
+        def download_and_process_track(url, target_dir, ost_id):
+            fs_map_dir = download_and_extract_zip(url)
+            track = create_dd_tracks_from_fs(fs_map_dir, target_dir, ost_id)
+            return track
+
+        tracks = []
+        with ThreadPoolExecutor() as executor:
+            future_to_url = {
+                executor.submit(download_and_process_track, url, target_dir, ost_id): url for url in
+                fs_map_urls
+            }
+            for future in concurrent.futures.as_completed(future_to_url):
+                track = future.result()
+                tracks.append(track)
+
+        album_info = DDAlbumInfo(
+            OstName=f'Feet Saber - {args.fs_playlist_id} by {author}',
+            BeatMapIdList=sorted([track.BeatMapId for track in tracks]),
+            OstId=ost_id,
+            CoverPath=os.path.basename(album_cover_path),
+            CreateTime=yyyymmdd_to_ticks(datetime.now().strftime('%Y%m%d')),
+        ).save_to_file(target_dir)
+        print(f'Created album info file: {album_info}')
+
+        print('Zipping tracks...')
+        with zipfile.ZipFile(f'bin/{ost_id}.zip', 'w', zipfile.ZIP_DEFLATED) as zipf:
+            zipdir(
+                target_dir, zipf,
+                f'Dance Dash_Data/StreamingAssets/NewDLC/{ost_id}',
+            )
+
+        print(f'Created bin/{ost_id}.zip')
+        raise SystemExit(0)
+
+    if not args.fs_map_dir and not args.fs_map_id:
+        print('Either --fs-map-dir or --fs-map-id must be provided.')
+        raise SystemExit(1)
+
+    if args.fs_map_id:
+        url = get_feet_saber_map_from_id(args.fs_map_id)
+        args.fs_map_dir = download_and_extract_zip(url)
+
     if not os.path.exists(args.fs_map_dir):
         print(f'Invalid Feet Saber map directory: {args.fs_map_dir}')
         raise SystemExit(1)

model/feetsaber.py

@@ -127,10 +127,12 @@ class FSInfoDat:
                     noteJumpMovementSpeed=beatmap['_noteJumpMovementSpeed'],
                     noteJumpStartBeatOffset=beatmap['_noteJumpStartBeatOffset'],
                     customData=FSBeatMapCustomData(
-                        difficultyLabel=beatmap['_customData']['_difficultyLabel'],
+                        difficultyLabel=beatmap['_customData'].get(
+                            '_difficultyLabel',
+                        ),
                         editorOffset=beatmap['_customData']['_editorOffset'],
                         editorOldOffset=beatmap['_customData']['_editorOldOffset'],
-                        suggestions=beatmap['_customData']['_suggestions'],
+                        suggestions=beatmap['_customData'].get('_suggestions'),
                         requirements=beatmap['_customData']['_requirements'],
                     ),
                 ) for beatmap in dbset['_difficultyBeatmaps']
@@ -211,9 +213,18 @@ class FSBeatMapFileNote:
 
     @property
     def to_dd_x(self) -> 1 | 2 | 3 | 4 | 5 | 6 | 7 | 8 | 9:
+        if not self.customData:
+            if self.lineIndex == 0:
+                return 2
+            elif self.lineIndex == 1:
+                return 4
+            elif self.lineIndex == 2:
+                return 6
+            elif self.lineIndex == 3:
+                return 8
         x = self.customData.position[0]
         x_min, x_max = -2, 1
-        y_min, y_max = 1, 9
+        y_min, y_max = 2, 8
         y = (x - x_min) * (y_max - y_min) / (x_max - x_min) + y_min
         return round(y)
 
@@ -271,7 +282,7 @@ class FSBeatMapFileObstacle:
 
     def to_dd_x(self, x: float = None) -> 1 | 2 | 3 | 4 | 5 | 6 | 7 | 8 | 9:
         x_min, x_max = -2, 1
-        y_min, y_max = 1, 9
+        y_min, y_max = 2, 8
         if not x:
             x = self.customData.position[0]
         if x < x_min:
@@ -301,12 +312,16 @@ class FSBeatMapFileObstacle:
 
     @property
     def is_down(self) -> bool:
+        if not self.customData:
+            return False
         if not self.customData.track:
             return False
         return self.customData.track.casefold() == 'DownArch'.casefold()
 
     @property
     def is_up(self) -> bool:
+        if not self.customData:
+            return False
         if not self.customData.track:
             return False
         return self.customData.track.casefold() == 'JumpBar'.casefold()
@@ -335,7 +350,7 @@ class FSBeatMapFile:
                     time=bm['_time'],
                     name=bm['_name'],
                     color=bm['_color'],
-                ) for bm in json_dict['_customData']['_bookmarks']
+                ) for bm in json_dict['_customData'].get('_bookmarks') or []
             ],
         )
 
@@ -356,7 +371,7 @@ class FSBeatMapFile:
                 cutDirection=note['_cutDirection'],
                 customData=FSBeatMapFileNoteCustomData(
                     position=tuple(note['_customData']['_position']),
-                ),
+                ) if '_customData' in note else None,
             ) for note in json_dict['_notes']
         ]
 
@@ -381,7 +396,7 @@ class FSBeatMapFile:
                         float(c)
                         for c in obstacle['_customData']['_color']
                     ),
-                ),
+                ) if '_customData' in obstacle else None,
             ) for obstacle in json_dict['_obstacles']
         ]
 

util.py

@@ -3,9 +3,14 @@ from __future__ import annotations
 import contextlib
 import os
 import random
+import shutil
 import subprocess
+import tempfile
+import zipfile
 from datetime import datetime
 
+import requests
+
 
 def get_drs_ogg_and_duration(folder_path) -> tuple[str, str] | tuple[None, None]:
     ogg_files = [f for f in os.listdir(folder_path) if f.endswith('.ogg')]
@@ -81,7 +86,7 @@ def zipdir(path, ziph, archiveroot):
 
 
 def random_9_digit_int():
-    return random.randint(10**8, 10**9 - 1)
+    return random.randint(10 ** 8, 10 ** 9 - 1)
 
 
 def convert_egg_to_ogg_and_get_length(egg_path: str) -> tuple[str, float]:
@@ -101,3 +106,67 @@ def convert_egg_to_ogg_and_get_length(egg_path: str) -> tuple[str, float]:
         ffprobe_cmd, text=True,
     ).strip()
     return os.path.basename(ogg_path), float(duration_str)
+
+
+def get_feet_saber_map_from_id(beat_saber_map_id: str) -> str:
+    response = requests.get(
+        f'https://beatsaver.com/api/maps/id/{beat_saber_map_id}',
+    )
+    response.raise_for_status()
+    response_dict = response.json()
+    return response_dict['versions'][0]['downloadURL']
+
+
+def get_feet_saber_maps_from_playlist(playlist_id: str) -> tuple[list[str], str, str]:
+    response = requests.get(
+        f'https://beatsaver.com/api/playlists/id/{playlist_id}/0',
+    )
+    response.raise_for_status()
+    response_dict = response.json()
+    download_urls = [
+        song['map']['versions'][0]['downloadURL']
+        for song in response_dict['maps']
+    ]
+    image_url = response_dict['playlist']['playlistImage']
+    author = response_dict['playlist']['owner']['name']
+    return download_urls, image_url, author
+
+
+def download_image_from_url(url, target_path):
+    response = requests.get(url, stream=True)
+    response.raise_for_status()  # Raise exception for bad responses
+
+    content_type = response.headers['content-type']
+    if 'jpeg' in content_type or 'jpg' in content_type:
+        ext = '.jpg'
+    elif 'png' in content_type:
+        ext = '.png'
+    else:
+        ext = '.jpg'
+
+    with open(f'{target_path}{ext}', 'wb') as file:
+        for chunk in response.iter_content(chunk_size=8192):
+            file.write(chunk)
+
+    print(f'Downloaded image to {target_path}{ext}')
+
+    return f'{target_path}{ext}'
+
+
+def download_and_extract_zip(url):
+    tmpdirname = tempfile.mkdtemp()  # Manually creating a temporary directory
+
+    try:
+        response = requests.get(url, stream=True)
+        zip_path = os.path.join(tmpdirname, 'downloaded.zip')
+        with open(zip_path, 'wb') as out_file:
+            for chunk in response.iter_content(chunk_size=8192):
+                out_file.write(chunk)
+
+        with zipfile.ZipFile(zip_path, 'r') as zip_ref:
+            zip_ref.extractall(tmpdirname)
+
+        return tmpdirname
+    except Exception as e:
+        shutil.rmtree(tmpdirname)
+        raise e