# vim: set fileencoding=utf-8 from typing import Any, Dict, List, Optional from bemani.backend.ess import EventLogHandler from bemani.backend.sdvx.base import SoundVoltexBase from bemani.backend.sdvx.infiniteinfection import SoundVoltexInfiniteInfection from bemani.common import ID, VersionConstants from bemani.protocol import Node class SoundVoltexGravityWars( EventLogHandler, SoundVoltexBase, ): name = 'SOUND VOLTEX III GRAVITY WARS' version = VersionConstants.SDVX_GRAVITY_WARS GAME_LIMITED_LOCKED = 1 GAME_LIMITED_UNLOCKABLE = 2 GAME_LIMITED_UNLOCKED = 3 GAME_CURRENCY_PACKETS = 0 GAME_CURRENCY_BLOCKS = 1 GAME_CLEAR_TYPE_NO_CLEAR = 1 GAME_CLEAR_TYPE_CLEAR = 2 GAME_CLEAR_TYPE_HARD_CLEAR = 3 GAME_CLEAR_TYPE_ULTIMATE_CHAIN = 4 GAME_CLEAR_TYPE_PERFECT_ULTIMATE_CHAIN = 5 GAME_GRADE_NO_PLAY = 0 GAME_GRADE_D = 1 GAME_GRADE_C = 2 GAME_GRADE_B = 3 GAME_GRADE_A = 4 GAME_GRADE_AA = 5 GAME_GRADE_AAA = 6 GAME_CATALOG_TYPE_SONG = 0 GAME_CATALOG_TYPE_APPEAL_CARD = 1 GAME_CATALOG_TYPE_CREW = 4 GAME_GAUGE_TYPE_SKILL = 1 @classmethod def get_settings(cls) -> Dict[str, Any]: """ Return all of our front-end modifiably settings. """ return { 'bools': [ { 'name': 'Disable Online Matching', 'tip': 'Disable online matching between games.', 'category': 'game_config', 'setting': 'disable_matching', }, { 'name': 'Force Song Unlock', 'tip': 'Force unlock all songs.', 'category': 'game_config', 'setting': 'force_unlock_songs', }, { 'name': 'Force Appeal Card Unlock', 'tip': 'Force unlock all appeal cards.', 'category': 'game_config', 'setting': 'force_unlock_cards', }, { 'name': 'Force Crew Card Unlock', 'tip': 'Force unlock all crew and subcrew cards.', 'category': 'game_config', 'setting': 'force_unlock_crew', }, ], } def previous_version(self) -> Optional[SoundVoltexBase]: return SoundVoltexInfiniteInfection(self.data, self.config, self.model) def _get_skill_analyzer_courses(self) -> List[Dict[str, Any]]: # This is overridden in S1/S2 code. return [] def _get_skill_analyzer_seasons(self) -> Dict[int, str]: # This is overridden in S1/S2 code. return {} def _get_extra_events(self) -> List[int]: # This is overridden in S1/S2 code. return [] def __game_to_db_clear_type(self, clear_type: int) -> int: return { self.GAME_CLEAR_TYPE_NO_CLEAR: self.CLEAR_TYPE_FAILED, self.GAME_CLEAR_TYPE_CLEAR: self.CLEAR_TYPE_CLEAR, self.GAME_CLEAR_TYPE_HARD_CLEAR: self.CLEAR_TYPE_HARD_CLEAR, self.GAME_CLEAR_TYPE_ULTIMATE_CHAIN: self.CLEAR_TYPE_ULTIMATE_CHAIN, self.GAME_CLEAR_TYPE_PERFECT_ULTIMATE_CHAIN: self.CLEAR_TYPE_PERFECT_ULTIMATE_CHAIN, }[clear_type] def __db_to_game_clear_type(self, clear_type: int) -> int: return { self.CLEAR_TYPE_NO_PLAY: self.GAME_CLEAR_TYPE_NO_CLEAR, self.CLEAR_TYPE_FAILED: self.GAME_CLEAR_TYPE_NO_CLEAR, self.CLEAR_TYPE_CLEAR: self.GAME_CLEAR_TYPE_CLEAR, self.CLEAR_TYPE_HARD_CLEAR: self.GAME_CLEAR_TYPE_HARD_CLEAR, self.CLEAR_TYPE_ULTIMATE_CHAIN: self.GAME_CLEAR_TYPE_ULTIMATE_CHAIN, self.CLEAR_TYPE_PERFECT_ULTIMATE_CHAIN: self.GAME_CLEAR_TYPE_PERFECT_ULTIMATE_CHAIN, }[clear_type] def __game_to_db_grade(self, grade: int) -> int: return { self.GAME_GRADE_NO_PLAY: self.GRADE_NO_PLAY, self.GAME_GRADE_D: self.GRADE_D, self.GAME_GRADE_C: self.GRADE_C, self.GAME_GRADE_B: self.GRADE_B, self.GAME_GRADE_A: self.GRADE_A, self.GAME_GRADE_AA: self.GRADE_AA, self.GAME_GRADE_AAA: self.GRADE_AAA, }[grade] def __db_to_game_grade(self, grade: int) -> int: return { self.GRADE_NO_PLAY: self.GAME_GRADE_NO_PLAY, self.GRADE_D: self.GAME_GRADE_D, self.GRADE_C: self.GAME_GRADE_C, self.GRADE_B: self.GAME_GRADE_B, self.GRADE_A: self.GAME_GRADE_A, self.GRADE_A_PLUS: self.GAME_GRADE_A, self.GRADE_AA: self.GAME_GRADE_AA, self.GRADE_AA_PLUS: self.GAME_GRADE_AA, self.GRADE_AAA: self.GAME_GRADE_AAA, self.GRADE_AAA_PLUS: self.GAME_GRADE_AAA, self.GRADE_S: self.GAME_GRADE_AAA, }[grade] def __get_skill_analyzer_skill_levels(self) -> Dict[int, str]: return { 0: 'Skill LEVEL 01 岳翔', 1: 'Skill LEVEL 02 流星', 2: 'Skill LEVEL 03 月衝', 3: 'Skill LEVEL 04 瞬光', 4: 'Skill LEVEL 05 天極', 5: 'Skill LEVEL 06 烈風', 6: 'Skill LEVEL 07 雷電', 7: 'Skill LEVEL 08 麗華', 8: 'Skill LEVEL 09 魔騎士', 9: 'Skill LEVEL 10 剛力羅', 10: 'Skill LEVEL 11 或帝滅斗', 11: 'Skill LEVEL ∞(12) 暴龍天', } def handle_game_3_common_request(self, request: Node) -> Node: game = Node.void('game_3') limited = Node.void('music_limited') game.add_child(limited) # Song unlock config game_config = self.get_game_config() if game_config.get_bool('force_unlock_songs'): ids = set() songs = self.data.local.music.get_all_songs(self.game, self.version) for song in songs: if song.data.get_int('limited') in (self.GAME_LIMITED_LOCKED, self.GAME_LIMITED_UNLOCKABLE): ids.add((song.id, song.chart)) for (songid, chart) in ids: info = Node.void('info') limited.add_child(info) info.add_child(Node.s32('music_id', songid)) info.add_child(Node.u8('music_type', chart)) info.add_child(Node.u8('limited', self.GAME_LIMITED_UNLOCKED)) # Event config event = Node.void('event') game.add_child(event) def enable_event(eid: int) -> None: evt = Node.void('info') event.add_child(evt) evt.add_child(Node.u32('event_id', eid)) if not game_config.get_bool('disable_matching'): enable_event(1) # Matching enabled enable_event(2) # Floor Infection enable_event(3) # Policy Break enable_event(60) # BEMANI Summer Diary for eventid in self._get_extra_events(): enable_event(eventid) # Skill Analyzer config skill_course = Node.void('skill_course') game.add_child(skill_course) seasons = self._get_skill_analyzer_seasons() skillnames = self.__get_skill_analyzer_skill_levels() courses = self._get_skill_analyzer_courses() max_level: Dict[int, int] = {} for course in courses: max_level[course['level']] = max(course['season_id'], max_level.get(course['level'], -1)) for course in courses: info = Node.void('info') skill_course.add_child(info) info.add_child(Node.s16('course_id', course.get('id', course['level']))) info.add_child(Node.s16('level', course['level'])) info.add_child(Node.s32('season_id', course['season_id'])) info.add_child(Node.string('season_name', seasons[course['season_id']])) info.add_child(Node.bool('season_new_flg', max_level[course['level']] == course['season_id'])) info.add_child(Node.string('course_name', course.get('skill_name', skillnames.get(course['level'], '')))) info.add_child(Node.s16('course_type', 0)) info.add_child(Node.s16('skill_name_id', course.get('skill_name_id', course['level']))) info.add_child(Node.bool('matching_assist', course['level'] >= 0 and course['level'] <= 6)) info.add_child(Node.s16('gauge_type', self.GAME_GAUGE_TYPE_SKILL)) info.add_child(Node.s16('paseli_type', 0)) for trackno, trackdata in enumerate(course['tracks']): track = Node.void('track') info.add_child(track) track.add_child(Node.s16('track_no', trackno)) track.add_child(Node.s32('music_id', trackdata['id'])) track.add_child(Node.s8('music_type', trackdata['type'])) return game def handle_game_3_exception_request(self, request: Node) -> Node: return Node.void('game_3') def handle_game_3_shop_request(self, request: Node) -> Node: self.update_machine_name(request.child_value('shopname')) # Respond with number of milliseconds until next request game = Node.void('game_3') game.add_child(Node.u32('nxt_time', 1000 * 5 * 60)) return game def handle_game_3_lounge_request(self, request: Node) -> Node: game = Node.void('game_3') # Refresh interval in seconds. game.add_child(Node.u32('interval', 10)) return game def handle_game_3_entry_s_request(self, request: Node) -> Node: game = Node.void('game_3') # This should be created on the fly for a lobby that we're in. game.add_child(Node.u32('entry_id', 1)) return game def handle_game_3_entry_e_request(self, request: Node) -> Node: # Lobby destroy method, eid node (u32) should be used # to destroy any open lobbies. return Node.void('game_3') def handle_game_3_frozen_request(self, request: Node) -> Node: game = Node.void('game_3') game.add_child(Node.u8('result', 0)) return game def handle_game_3_save_e_request(self, request: Node) -> Node: # This has to do with Policy Break against ReflecBeat and # floor infection, but we don't implement multi-game support so meh. return Node.void('game_3') def handle_game_3_play_e_request(self, request: Node) -> Node: return Node.void('game_3') def handle_game_3_buy_request(self, request: Node) -> Node: refid = request.child_value('refid') if refid is not None: userid = self.data.remote.user.from_refid(self.game, self.version, refid) else: userid = None if userid is not None: profile = self.get_profile(userid) else: profile = None if userid is not None and profile is not None: # Look up packets and blocks packet = profile.get_int('packet') block = profile.get_int('block') # Add on any additional we earned this round packet = packet + (request.child_value('earned_gamecoin_packet') or 0) block = block + (request.child_value('earned_gamecoin_block') or 0) currency_type = request.child_value('currency_type') price = request.child_value('item/price') if isinstance(price, list): # Sometimes we end up buying more than one item at once price = sum(price) if currency_type == self.GAME_CURRENCY_PACKETS: # This is a valid purchase newpacket = packet - price if newpacket < 0: result = 1 else: packet = newpacket result = 0 elif currency_type == self.GAME_CURRENCY_BLOCKS: # This is a valid purchase newblock = block - price if newblock < 0: result = 1 else: block = newblock result = 0 else: # Bad currency type result = 1 if result == 0: # Transaction is valid, update the profile with new packets and blocks profile.replace_int('packet', packet) profile.replace_int('block', block) self.put_profile(userid, profile) # If this was a song unlock, we should mark it as unlocked item_type = request.child_value('item/item_type') item_id = request.child_value('item/item_id') param = request.child_value('item/param') if not isinstance(item_type, list): # Sometimes we buy multiple things at once. Make it easier by always assuming this. item_type = [item_type] item_id = [item_id] param = [param] for i in range(len(item_type)): self.data.local.user.put_achievement( self.game, self.version, userid, item_id[i], f'item_{item_type[i]}', { 'param': param[i], }, ) else: # Unclear what to do here, return a bad response packet = 0 block = 0 result = 1 game = Node.void('game_3') game.add_child(Node.u32('gamecoin_packet', packet)) game.add_child(Node.u32('gamecoin_block', block)) game.add_child(Node.s8('result', result)) return game def handle_game_3_new_request(self, request: Node) -> Node: refid = request.child_value('refid') name = request.child_value('name') loc = ID.parse_machine_id(request.child_value('locid')) self.new_profile_by_refid(refid, name, loc) root = Node.void('game_3') return root def handle_game_3_load_request(self, request: Node) -> Node: refid = request.child_value('refid') root = self.get_profile_by_refid(refid) if root is not None: return root # Figure out if this user has an older profile or not userid = self.data.remote.user.from_refid(self.game, self.version, refid) if userid is not None: previous_game = self.previous_version() else: previous_game = None if previous_game is not None: profile = previous_game.get_profile(userid) else: profile = None if profile is not None: root = Node.void('game_3') root.add_child(Node.u8('result', 2)) root.add_child(Node.string('name', profile.get_str('name'))) return root else: root = Node.void('game_3') root.add_child(Node.u8('result', 1)) return root def handle_game_3_save_request(self, request: Node) -> Node: refid = request.child_value('refid') if refid is not None: userid = self.data.remote.user.from_refid(self.game, self.version, refid) else: userid = None if userid is not None: oldprofile = self.get_profile(userid) newprofile = self.unformat_profile(userid, request, oldprofile) else: newprofile = None if userid is not None and newprofile is not None: self.put_profile(userid, newprofile) return Node.void('game_3') def handle_game_3_load_m_request(self, request: Node) -> Node: refid = request.child_value('dataid') if refid is not None: userid = self.data.remote.user.from_refid(self.game, self.version, refid) else: userid = None if userid is not None: scores = self.data.remote.music.get_scores(self.game, self.version, userid) else: scores = [] # Output to the game game = Node.void('game_3') new = Node.void('new') game.add_child(new) for score in scores: music = Node.void('music') new.add_child(music) music.add_child(Node.u32('music_id', score.id)) music.add_child(Node.u32('music_type', score.chart)) music.add_child(Node.u32('score', score.points)) music.add_child(Node.u32('cnt', score.plays)) music.add_child(Node.u32('clear_type', self.__db_to_game_clear_type(score.data.get_int('clear_type')))) music.add_child(Node.u32('score_grade', self.__db_to_game_grade(score.data.get_int('grade')))) stats = score.data.get_dict('stats') music.add_child(Node.u32('btn_rate', stats.get_int('btn_rate'))) music.add_child(Node.u32('long_rate', stats.get_int('long_rate'))) music.add_child(Node.u32('vol_rate', stats.get_int('vol_rate'))) return game def handle_game_3_save_m_request(self, request: Node) -> Node: refid = request.child_value('refid') if refid is not None: userid = self.data.remote.user.from_refid(self.game, self.version, refid) else: userid = None # Doesn't matter if userid is None here, that's an anonymous score musicid = request.child_value('music_id') chart = request.child_value('music_type') points = request.child_value('score') combo = request.child_value('max_chain') clear_type = self.__game_to_db_clear_type(request.child_value('clear_type')) grade = self.__game_to_db_grade(request.child_value('score_grade')) stats = { 'btn_rate': request.child_value('btn_rate'), 'long_rate': request.child_value('long_rate'), 'vol_rate': request.child_value('vol_rate'), 'critical': request.child_value('critical'), 'near': request.child_value('near'), 'error': request.child_value('error'), } # Save the score self.update_score( userid, musicid, chart, points, clear_type, grade, combo, stats, ) # Return a blank response return Node.void('game_3') def handle_game_3_save_c_request(self, request: Node) -> Node: refid = request.child_value('dataid') if refid is not None: userid = self.data.remote.user.from_refid(self.game, self.version, refid) else: userid = None if userid is not None: course_id = request.child_value('crsid') clear_type = request.child_value('ct') achievement_rate = request.child_value('ar') season_id = request.child_value('ssnid') self.data.local.user.put_achievement( self.game, self.version, userid, (season_id * 100) + course_id, 'course', { 'clear_type': clear_type, 'achievement_rate': achievement_rate, }, ) # Return a blank response return Node.void('game_3')