317 lines
11 KiB
Python
317 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import os
|
|
import shutil
|
|
import zipfile
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from datetime import datetime
|
|
|
|
from drsxml2json import ALL_TRACK_PATHS
|
|
from drsxml2json import get_songdata_from_track_id
|
|
from drsxml2json import TRACK_ID_TO_PATH
|
|
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 DRS2DD_MAP_PREFIX
|
|
from model.dancedash import DRS_TO_DD_LINE_NOTE_TYPE
|
|
from model.dancedash import DRS_TO_DD_NOTE_TYPE
|
|
from model.dancedash import ORDER_COUNT_PER_BEAT
|
|
from model.dancedash import X_Y
|
|
from model.dancerush import ALBUM_NAME
|
|
from model.dancerush import DEFAULT_TRACK_DIR
|
|
from model.dancerush import DRS_DOWN
|
|
from model.dancerush import DRS_JUMP
|
|
from model.dancerush import DRSSongData
|
|
from model.dancerush import DRSSongDifficulty
|
|
from model.dancerush import DRSTrackPoint
|
|
from model.dancerush import OUTPUT_ZIP_NAME
|
|
from util import create_valid_filename
|
|
from util import get_drs_ogg_and_duration
|
|
from util import get_song_cover_path
|
|
from util import yyyymmdd_to_ticks
|
|
from util import zipdir
|
|
|
|
|
|
def map_sphere_nodes(
|
|
difficulty: DRSSongDifficulty,
|
|
total_time_seconds: float,
|
|
bps: float,
|
|
) -> list[DDSphereNode]:
|
|
spheres = []
|
|
for track_step in difficulty.track.sequence_data:
|
|
if track_step.long_point or track_step.is_down_or_up:
|
|
continue
|
|
|
|
if track_step.tick_info.end_tick:
|
|
seconds = difficulty.track.info.determine_seconds_of_tick(
|
|
track_step.tick_info.end_tick,
|
|
)
|
|
else:
|
|
seconds = track_step.tick_info.stime_ms / 1000
|
|
|
|
spheres.append(
|
|
DDSphereNode(
|
|
noteOrder=round(bps * seconds * ORDER_COUNT_PER_BEAT),
|
|
time=seconds / total_time_seconds,
|
|
position=X_Y(x=track_step.position_info.to_dance_dash_x, y=0),
|
|
noteType=DRS_TO_DD_NOTE_TYPE[track_step.kind],
|
|
),
|
|
)
|
|
return spheres
|
|
|
|
|
|
def map_line_nodes(
|
|
difficulty: DRSSongDifficulty,
|
|
total_time_seconds: float,
|
|
bps: float,
|
|
) -> list[DDLineNode]:
|
|
lines = []
|
|
for line_group_id, track_step in enumerate(difficulty.track.sequence_data, start=1):
|
|
if not track_step.long_point or track_step.is_down_or_up:
|
|
continue
|
|
|
|
track_step.long_point.insert(
|
|
0, DRSTrackPoint(
|
|
tick=track_step.tick_info.start_tick,
|
|
left_pos=track_step.position_info.left_pos,
|
|
right_pos=track_step.position_info.right_pos,
|
|
point_time=track_step.tick_info.stime_ms,
|
|
),
|
|
)
|
|
|
|
last_was_shuffle_end = False
|
|
index_in_line = 0
|
|
for drs_track_point in track_step.long_point:
|
|
last_was_shuffle_end = False
|
|
index_in_line += 1
|
|
|
|
if drs_track_point.tick:
|
|
seconds = difficulty.track.info.determine_seconds_of_tick(
|
|
drs_track_point.tick,
|
|
)
|
|
else:
|
|
seconds = drs_track_point.point_time / 1000
|
|
|
|
lines.append(
|
|
DDLineNode(
|
|
lineGroupId=line_group_id,
|
|
indexInLine=index_in_line,
|
|
noteOrder=round(bps * seconds * ORDER_COUNT_PER_BEAT),
|
|
time=seconds / total_time_seconds,
|
|
position=X_Y(x=drs_track_point.to_dance_dash_x, y=0),
|
|
noteType=DRS_TO_DD_LINE_NOTE_TYPE[track_step.kind],
|
|
),
|
|
)
|
|
|
|
if drs_track_point.left_end_pos and drs_track_point.right_end_pos:
|
|
index_in_line += 1
|
|
lines.append(
|
|
lines[-1].line_end(drs_track_point, index_in_line),
|
|
)
|
|
last_was_shuffle_end = True
|
|
|
|
if last_was_shuffle_end:
|
|
lines.append(lines[-1].tail)
|
|
return lines
|
|
|
|
|
|
def map_down_and_jump_notes(
|
|
difficulty: DRSSongDifficulty,
|
|
total_time_seconds: float,
|
|
bps: float,
|
|
) -> list[DDRoadBlockNode]:
|
|
road_blocks = []
|
|
for track_step in difficulty.track.sequence_data:
|
|
if not track_step.is_down_or_up:
|
|
continue
|
|
|
|
if track_step.tick_info.end_tick:
|
|
seconds = difficulty.track.info.determine_seconds_of_tick(
|
|
track_step.tick_info.end_tick,
|
|
)
|
|
else:
|
|
seconds = track_step.tick_info.stime_ms / 1000
|
|
|
|
note_order = round(bps * seconds * ORDER_COUNT_PER_BEAT)
|
|
if track_step.kind == DRS_JUMP:
|
|
# user jumps OVER in DD, not ON like in DRS
|
|
note_order += ORDER_COUNT_PER_BEAT / 2
|
|
elif track_step.kind == DRS_DOWN:
|
|
# user needs to duck UNDER in DD, not go down ON like in DRS
|
|
note_order += ORDER_COUNT_PER_BEAT / 16
|
|
|
|
road_blocks.append(
|
|
DDRoadBlockNode(
|
|
noteOrder=note_order,
|
|
time=seconds / total_time_seconds,
|
|
position=DDJumpPos if track_step.kind == DRS_JUMP else DDDownPos,
|
|
position2D=DDJumpPos2D if track_step.kind == DRS_JUMP else DDDownPos2D,
|
|
),
|
|
)
|
|
return road_blocks
|
|
|
|
|
|
def create_dd_tracks_from_DRSSongData(drs_song_data: DRSSongData, target_dir: str = None) -> DDBeatMapInfoFile:
|
|
if not target_dir:
|
|
target_dir = f'{DEFAULT_TRACK_DIR}/{create_valid_filename(drs_song_data.info.title_name)}/'
|
|
|
|
dd_bpm = int(
|
|
drs_song_data.difficulties.difficulty_1a.track.info.highest_bpm / 100,
|
|
)
|
|
dd_bps = dd_bpm / 60
|
|
|
|
folder_path = TRACK_ID_TO_PATH.get(drs_song_data.song_id)
|
|
song_path, song_length = get_drs_ogg_and_duration(folder_path)
|
|
if not song_path or not song_length:
|
|
print(
|
|
f'No song found for {drs_song_data.info.title_name} ({drs_song_data.song_id})',
|
|
)
|
|
return None
|
|
|
|
if not os.path.exists(target_dir):
|
|
os.makedirs(target_dir)
|
|
|
|
new_song_path = os.path.join(target_dir, drs_song_data.ogg)
|
|
shutil.copy(song_path, new_song_path)
|
|
song_path = new_song_path
|
|
if song_cover_path := get_song_cover_path(folder_path):
|
|
new_song_cover_path = os.path.join(target_dir, drs_song_data.png)
|
|
shutil.copy(song_cover_path, new_song_cover_path)
|
|
song_cover_path = new_song_cover_path
|
|
|
|
song_paths = []
|
|
for attr, difficulty in drs_song_data.difficulties.with_attrs_as_str.items():
|
|
if attr in ('difficulty_2a', 'difficulty_2b'): # 2 player difficulties
|
|
continue
|
|
|
|
song_length_f = float(song_length)
|
|
sphere_notes = map_sphere_nodes(
|
|
difficulty, song_length_f, dd_bps,
|
|
)
|
|
line_notes = map_line_nodes(
|
|
difficulty, song_length_f, dd_bps,
|
|
)
|
|
road_block_notes = map_down_and_jump_notes(
|
|
difficulty, song_length_f, dd_bps,
|
|
)
|
|
|
|
total_note_count = len(sphere_notes + line_notes + road_block_notes) # noqa
|
|
dd_beat_map = DDBeatMap(
|
|
data=DDBeatMapData(
|
|
name=f'{drs_song_data.info.title_name} {attr}',
|
|
sphereNodes=sphere_notes,
|
|
lineNodes=line_notes,
|
|
roadBlockNodes=road_block_notes,
|
|
),
|
|
BPM=dd_bpm,
|
|
NPS=str(round(total_note_count / song_length_f, 2)),
|
|
)
|
|
|
|
song_paths.append(
|
|
dd_beat_map.save_to_file(
|
|
target_dir, f'{attr}.json',
|
|
),
|
|
)
|
|
song_paths.append(
|
|
dd_beat_map.block_less.save_to_file(
|
|
target_dir, f'{attr}_blockless.json',
|
|
),
|
|
)
|
|
|
|
normal, normal_no_blocks, easy, easy_no_blocks = song_paths
|
|
create_ticks = yyyymmdd_to_ticks(str(drs_song_data.info.distribution_date))
|
|
drs_beat_map_info = DDBeatMapInfoFile(
|
|
CreateTicks=create_ticks,
|
|
CreateTime=str(create_ticks),
|
|
OstId=DRS2DD_MAP_PREFIX,
|
|
BeatMapId=DRS2DD_MAP_PREFIX + drs_song_data.song_id,
|
|
SongName=drs_song_data.info.title_name,
|
|
SongLength=song_length,
|
|
SongAuthorName=drs_song_data.info.artist_name,
|
|
Bpm=str(dd_bpm),
|
|
SongPath=os.path.basename(song_path),
|
|
CoverPath=os.path.basename(song_cover_path or '') or None,
|
|
DRS_Easy=os.path.basename(easy_no_blocks),
|
|
DRS_Normal=os.path.basename(normal_no_blocks),
|
|
DRS_Hard=os.path.basename(easy),
|
|
DRS_Expert=os.path.basename(normal),
|
|
)
|
|
drs_beat_map_info.save_to_file(target_dir)
|
|
|
|
print(f'Created {drs_song_data.info.title_name} ({drs_song_data.song_id})')
|
|
return drs_beat_map_info
|
|
|
|
|
|
if __name__ == '__main__':
|
|
parser = argparse.ArgumentParser(
|
|
description='Create Dance Dash tracks from DANCERUSH STARDOM Song Data',
|
|
)
|
|
parser.add_argument(
|
|
'--song-id',
|
|
type=int,
|
|
help='ID of the song to process (number of folder)',
|
|
)
|
|
|
|
args = parser.parse_args()
|
|
if args.song_id:
|
|
song_data = get_songdata_from_track_id(args.song_id)
|
|
if not song_data:
|
|
print(f'No song data found for song id {args.song_id}')
|
|
raise SystemExit(1)
|
|
|
|
target_directory = create_valid_filename(song_data.info.title_name)
|
|
info_file = create_dd_tracks_from_DRSSongData(
|
|
song_data, target_directory,
|
|
)
|
|
print(f'Song {args.song_id} created?: {bool(info_file)}')
|
|
raise SystemExit(0)
|
|
|
|
if not os.path.exists(DEFAULT_TRACK_DIR):
|
|
os.makedirs(DEFAULT_TRACK_DIR)
|
|
|
|
tracks = []
|
|
|
|
def _process_track(song_id):
|
|
if _song_data := get_songdata_from_track_id(int(song_id)):
|
|
try:
|
|
if _info_file := create_dd_tracks_from_DRSSongData(_song_data):
|
|
tracks.append(_info_file)
|
|
except Exception as e:
|
|
print(f'Error creating song {song_id}: {e}')
|
|
|
|
with ThreadPoolExecutor() as executor:
|
|
executor.map(_process_track, ALL_TRACK_PATHS)
|
|
print(f'Created {len(tracks)} tracks')
|
|
|
|
album_info = DDAlbumInfo(
|
|
OstName='DANCERUSH STARDOM',
|
|
BeatMapIdList=sorted([track.BeatMapId for track in tracks]),
|
|
OstId=DRS2DD_MAP_PREFIX,
|
|
CoverPath='DANCERUSH_STARDOM.jpg',
|
|
CreateTime=yyyymmdd_to_ticks(datetime.now().strftime('%Y%m%d')),
|
|
).save_to_file(DEFAULT_TRACK_DIR)
|
|
print(f'Created album info file: {album_info}')
|
|
|
|
shutil.copy('resources/drs/DANCERUSH_STARDOM.jpg', DEFAULT_TRACK_DIR)
|
|
if not os.path.exists('bin'):
|
|
os.makedirs('bin')
|
|
|
|
print('Zipping tracks...')
|
|
with zipfile.ZipFile(f'bin/{OUTPUT_ZIP_NAME}', 'w', zipfile.ZIP_DEFLATED) as zipf:
|
|
zipdir(
|
|
DEFAULT_TRACK_DIR, zipf,
|
|
f'Dance Dash_Data/StreamingAssets/NewDLC/{ALBUM_NAME}',
|
|
)
|
|
|
|
print(f'Created bin/{OUTPUT_ZIP_NAME}')
|
|
raise SystemExit(0)
|