420 lines
15 KiB
Python
420 lines
15 KiB
Python
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
|
|
from model.dancedash import DDDownPos
|
|
from model.dancedash import DDDownPos2D
|
|
from model.dancedash import DDJumpPos
|
|
from model.dancedash import DDJumpPos2D
|
|
from model.dancedash import DDLineNode
|
|
from model.dancedash import DDRoadBlockNode
|
|
from model.dancedash import DDSphereNode
|
|
from model.dancedash import ORDER_COUNT_PER_BEAT
|
|
from model.dancedash import X_Y
|
|
from model.feetsaber import FS_LEFT_NOTE
|
|
from model.feetsaber import FS_RIGHT_NOTE
|
|
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(
|
|
fs_beat_map: FSBeatMapFile,
|
|
fs_info_dat: FSInfoDat,
|
|
total_time_seconds: float,
|
|
) -> list[DDSphereNode]:
|
|
|
|
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 and o.customData.dd_note_type == line_type
|
|
]
|
|
all_notes = [
|
|
n for n in fs_beat_map.notes if n.type == note_type
|
|
]
|
|
non_overlapping_notes = [
|
|
note for note in all_notes
|
|
if not any(obstacle.time <= note.time < obstacle.end_time for obstacle in all_lines_of_type)
|
|
]
|
|
return [
|
|
DDSphereNode(
|
|
noteOrder=round(note.time * ORDER_COUNT_PER_BEAT),
|
|
time=(fs_info_dat.bps * note.time) / total_time_seconds,
|
|
position=X_Y(x=note.to_dd_x, y=0),
|
|
noteType=FS_TO_DD_NOTE_TYPE[note.type],
|
|
)
|
|
for note in non_overlapping_notes
|
|
]
|
|
|
|
spheres = process_notes(FS_LEFT_NOTE, DD_LINE_LEFT) + \
|
|
process_notes(FS_RIGHT_NOTE, DD_LINE_RIGHT)
|
|
return sorted(spheres, key=lambda s: s.noteOrder)
|
|
|
|
|
|
def map_line_nodes(
|
|
fs_beat_map: FSBeatMapFile,
|
|
fs_info_dat: FSInfoDat,
|
|
total_time_seconds: float,
|
|
) -> list[DDLineNode]:
|
|
lines = []
|
|
|
|
line_obstacles = [
|
|
o for o in fs_beat_map.obstacles if o.customData and o.customData.is_fs
|
|
]
|
|
|
|
obstacles_by_type = {
|
|
DD_LINE_LEFT: [
|
|
o for o in line_obstacles if o.customData.dd_note_type == DD_LINE_LEFT
|
|
],
|
|
DD_LINE_RIGHT: [
|
|
o for o in line_obstacles if o.customData.dd_note_type == DD_LINE_RIGHT
|
|
],
|
|
}
|
|
|
|
index_in_line = 0
|
|
line_group_id = 1
|
|
last_obstacle = None
|
|
for note_type, obstacles in obstacles_by_type.items():
|
|
for obstacle in obstacles:
|
|
if obstacle.is_part_of_last_obstacle(last_obstacle) and not obstacle.customData.is_fs_slider:
|
|
index_in_line += 1
|
|
is_left = obstacle.customData.position[0] < last_obstacle.customData.position[0]
|
|
last_obstacle_x = lines[-1].position.x
|
|
lines.append(
|
|
DDLineNode(
|
|
lineGroupId=line_group_id,
|
|
indexInLine=index_in_line,
|
|
noteOrder=round(obstacle.time * ORDER_COUNT_PER_BEAT),
|
|
time=(fs_info_dat.bps * obstacle.time) /
|
|
total_time_seconds,
|
|
position=X_Y(
|
|
last_obstacle_x +
|
|
(-1 if is_left else 1), y=0,
|
|
),
|
|
noteType=note_type,
|
|
),
|
|
)
|
|
index_in_line += 1
|
|
lines.append(
|
|
DDLineNode(
|
|
lineGroupId=line_group_id,
|
|
indexInLine=index_in_line,
|
|
noteOrder=round(
|
|
obstacle.time * ORDER_COUNT_PER_BEAT,
|
|
) + ORDER_COUNT_PER_BEAT / 4,
|
|
time=(fs_info_dat.bps * obstacle.time) /
|
|
total_time_seconds,
|
|
position=X_Y(
|
|
last_obstacle_x +
|
|
(-1 if is_left else 1), y=0,
|
|
),
|
|
noteType=note_type,
|
|
),
|
|
)
|
|
last_obstacle = obstacle
|
|
continue
|
|
|
|
if not obstacle.is_part_of_last_obstacle(last_obstacle):
|
|
index_in_line = 0
|
|
line_group_id += 1
|
|
|
|
lines.append(
|
|
DDLineNode(
|
|
lineGroupId=line_group_id,
|
|
indexInLine=index_in_line,
|
|
noteOrder=round(obstacle.time * ORDER_COUNT_PER_BEAT),
|
|
time=(fs_info_dat.bps * obstacle.time) /
|
|
total_time_seconds,
|
|
position=X_Y(x=obstacle.to_dd_x(), y=0),
|
|
noteType=note_type,
|
|
),
|
|
)
|
|
index_in_line += 1
|
|
lines.append(
|
|
DDLineNode(
|
|
lineGroupId=line_group_id,
|
|
indexInLine=index_in_line,
|
|
noteOrder=round(obstacle.end_time * ORDER_COUNT_PER_BEAT),
|
|
time=(fs_info_dat.bps * obstacle.time) /
|
|
total_time_seconds,
|
|
position=X_Y(x=obstacle.end_to_dd_x, y=0),
|
|
noteType=note_type,
|
|
),
|
|
)
|
|
last_obstacle = obstacle
|
|
|
|
return lines
|
|
|
|
|
|
def map_down_and_jump_notes(
|
|
fs_beat_map: FSBeatMapFile,
|
|
fs_info_dat: FSInfoDat,
|
|
total_time_seconds: float,
|
|
) -> list[DDRoadBlockNode]:
|
|
def create_block(obstacle, position, position2D):
|
|
return DDRoadBlockNode(
|
|
noteOrder=round(obstacle.time * ORDER_COUNT_PER_BEAT) +
|
|
(ORDER_COUNT_PER_BEAT / 16),
|
|
time=(fs_info_dat.bps * obstacle.time) / total_time_seconds,
|
|
position=position,
|
|
position2D=position2D,
|
|
)
|
|
|
|
downs = [
|
|
create_block(o, DDDownPos, DDDownPos2D)
|
|
for o in fs_beat_map.obstacles if o.is_down
|
|
]
|
|
ups = [
|
|
create_block(o, DDJumpPos, DDJumpPos2D)
|
|
for o in fs_beat_map.obstacles if o.is_up
|
|
]
|
|
return sorted(downs + ups, key=lambda r: r.noteOrder)
|
|
|
|
|
|
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)
|
|
|
|
egg_song_path = [
|
|
os.path.join(fs_map_dir, f) for f in os.listdir(
|
|
fs_map_dir,
|
|
) if f.endswith('.egg')
|
|
][0]
|
|
dir_name = f'{fs_info.songName} - {fs_info.levelAuthorName}'.replace('/', '-').replace('?', '').replace('\\', '-').replace( # noqa
|
|
':', '-',
|
|
).replace('*', '-').replace("\"", '-').replace('<', '-').replace('>', '-').replace('|', '-').replace('.', '-').replace(' ', '-') # noqa
|
|
target_dir = f'{prefix_dir}/{dir_name}'
|
|
if not os.path.exists(target_dir):
|
|
os.makedirs(target_dir)
|
|
|
|
ogg_song_name, song_length = convert_egg_to_ogg_and_get_length(
|
|
egg_song_path,
|
|
)
|
|
|
|
new_song_path = os.path.join(target_dir, ogg_song_name)
|
|
shutil.copy(egg_song_path, new_song_path)
|
|
song_path = new_song_path
|
|
if song_cover_path := get_first_image_file_in_folder(fs_map_dir):
|
|
new_song_cover_path = os.path.join(
|
|
target_dir, os.path.basename(song_cover_path),
|
|
)
|
|
shutil.copy(song_cover_path, new_song_cover_path)
|
|
song_cover_path = new_song_cover_path
|
|
|
|
song_paths = []
|
|
for difficulty_set in fs_info.difficultyBeatmapSets:
|
|
for difficulty in difficulty_set.difficultyBeatmaps:
|
|
beat_map: FSBeatMapFile = difficulty.get_beatmap(fs_map_dir)
|
|
|
|
sphere_notes = map_sphere_nodes(beat_map, fs_info, song_length)
|
|
line_notes = map_line_nodes(beat_map, fs_info, song_length)
|
|
road_block_notes = map_down_and_jump_notes(
|
|
beat_map, fs_info, song_length,
|
|
)
|
|
|
|
total_note_count = len(sphere_notes + line_notes + road_block_notes) # noqa
|
|
dd_beat_map = DDBeatMap(
|
|
data=DDBeatMapData(
|
|
name=fs_info.songName,
|
|
sphereNodes=sphere_notes,
|
|
lineNodes=line_notes,
|
|
roadBlockNodes=road_block_notes,
|
|
),
|
|
BPM=fs_info.beatsPerMinute,
|
|
NPS=str(round(total_note_count / song_length, 2)),
|
|
)
|
|
|
|
difficulty_name = f'{difficulty_set.beatmapCharacteristicName}_{difficulty.difficulty}'
|
|
song_paths.append(
|
|
dd_beat_map.save_to_file(
|
|
target_dir, f'{difficulty_name}.json',
|
|
),
|
|
)
|
|
|
|
available_difficulties = [
|
|
'Easy', 'Normal',
|
|
'Hard', 'Expert', 'Master', 'ACE',
|
|
]
|
|
song_path_difficulties = zip(song_paths, available_difficulties)
|
|
difficulties = {
|
|
f'DRS_{d}': os.path.basename(
|
|
sp,
|
|
) for sp, d in song_path_difficulties
|
|
}
|
|
|
|
create_ticks = yyyymmdd_to_ticks(datetime.now().strftime('%Y%m%d'))
|
|
dd_beat_map_info = DDBeatMapInfoFile(
|
|
OstId=ost_id,
|
|
CreateTicks=create_ticks,
|
|
CreateTime=str(create_ticks),
|
|
BeatMapId=random_9_digit_int(),
|
|
SongName=fs_info.songName,
|
|
SongLength=str(song_length),
|
|
SongAuthorName=fs_info.songAuthorName,
|
|
LevelAuthorName=fs_info.levelAuthorName,
|
|
Bpm=str(fs_info.beatsPerMinute),
|
|
SongPath=os.path.basename(song_path),
|
|
CoverPath=os.path.basename(song_cover_path or '') or None,
|
|
SongPreviewSection=fs_info.previewStartTime,
|
|
**difficulties,
|
|
)
|
|
|
|
dd_beat_map_info.save_to_file(target_dir)
|
|
|
|
print(
|
|
f'Created {dd_beat_map_info.SongName} ({dd_beat_map_info.BeatMapId})',
|
|
)
|
|
return dd_beat_map_info
|
|
|
|
|
|
if __name__ == '__main__':
|
|
parser = argparse.ArgumentParser(
|
|
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=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-map-ids', type=str,
|
|
help='IDs to fetch the Feet Saber map zip, separated by comma', 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 args.fs_map_ids:
|
|
ost_id = random_9_digit_int()
|
|
fs_map_ids = args.fs_map_ids.split(',')
|
|
target_dir = f'bin/{ost_id}'
|
|
|
|
fs_map_urls = [
|
|
get_feet_saber_map_from_id(fs_map_id) for fs_map_id in fs_map_ids
|
|
]
|
|
|
|
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 - {ost_id}',
|
|
BeatMapIdList=sorted([track.BeatMapId for track in tracks]),
|
|
OstId=ost_id,
|
|
CoverPath='',
|
|
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)
|
|
|
|
create_dd_tracks_from_fs(args.fs_map_dir)
|
|
raise SystemExit(0)
|