from typing import List, Dict, Tuple, Any from bemani.api.exceptions import APIException from bemani.api.objects.base import BaseObject from bemani.common import APIConstants, DBConstants, GameConstants from bemani.data import Attempt, UserID class StatisticsObject(BaseObject): def __format_statistics(self, stats: Dict[str, Any]) -> Dict[str, Any]: return { 'cards': [], 'song': str(stats['id']), 'chart': str(stats['chart']), 'plays': stats.get('plays', -1), 'clears': stats.get('clears', -1), 'combos': stats.get('combos', -1), } def __format_user_statistics(self, cardids: List[str], stats: Dict[str, Any]) -> Dict[str, Any]: base = self.__format_statistics(stats) base['cards'] = cardids return base @property def music_version(self) -> int: if self.game in [GameConstants.IIDX, GameConstants.MUSECA]: if self.omnimix: return self.version + DBConstants.OMNIMIX_VERSION_BUMP else: return self.version else: return self.version def __is_play(self, attempt: Attempt) -> bool: if self.game in [ GameConstants.DDR, GameConstants.JUBEAT, GameConstants.MUSECA, GameConstants.POPN_MUSIC, ]: return True if self.game == GameConstants.IIDX: return attempt.data.get_int('clear_status') != DBConstants.IIDX_CLEAR_STATUS_NO_PLAY if self.game == GameConstants.REFLEC_BEAT: return attempt.data.get_int('clear_type') != DBConstants.REFLEC_BEAT_CLEAR_TYPE_NO_PLAY if self.game == GameConstants.SDVX: return attempt.data.get_int('clear_type') != DBConstants.SDVX_CLEAR_TYPE_NO_PLAY return False def __is_clear(self, attempt: Attempt) -> bool: if not self.__is_play(attempt): return False if self.game == GameConstants.DDR: return attempt.data.get_int('rank') != DBConstants.DDR_RANK_E if self.game == GameConstants.IIDX: return attempt.data.get_int('clear_status') != DBConstants.IIDX_CLEAR_STATUS_FAILED if self.game == GameConstants.JUBEAT: return attempt.data.get_int('medal') != DBConstants.JUBEAT_PLAY_MEDAL_FAILED if self.game == GameConstants.MUSECA: return attempt.data.get_int('clear_type') != DBConstants.MUSECA_CLEAR_TYPE_FAILED if self.game == GameConstants.POPN_MUSIC: return attempt.data.get_int('medal') not in [ DBConstants.POPN_MUSIC_PLAY_MEDAL_CIRCLE_FAILED, DBConstants.POPN_MUSIC_PLAY_MEDAL_DIAMOND_FAILED, DBConstants.POPN_MUSIC_PLAY_MEDAL_STAR_FAILED, ] if self.game == GameConstants.REFLEC_BEAT: return attempt.data.get_int('clear_type') != DBConstants.REFLEC_BEAT_CLEAR_TYPE_FAILED if self.game == GameConstants.SDVX: return ( attempt.data.get_int('grade') != DBConstants.SDVX_GRADE_NO_PLAY and attempt.data.get_int('clear_type') not in [ DBConstants.SDVX_CLEAR_TYPE_NO_PLAY, DBConstants.SDVX_CLEAR_TYPE_FAILED, ] ) return False def __is_combo(self, attempt: Attempt) -> bool: if not self.__is_play(attempt): return False if self.game == GameConstants.DDR: return attempt.data.get_int('halo') != DBConstants.DDR_HALO_NONE if self.game == GameConstants.IIDX: return attempt.data.get_int('clear_status') == DBConstants.IIDX_CLEAR_STATUS_FULL_COMBO if self.game == GameConstants.JUBEAT: return attempt.data.get_int('medal') in [ DBConstants.JUBEAT_PLAY_MEDAL_FULL_COMBO, DBConstants.JUBEAT_PLAY_MEDAL_NEARLY_EXCELLENT, DBConstants.JUBEAT_PLAY_MEDAL_EXCELLENT, ] if self.game == GameConstants.MUSECA: return attempt.data.get_int('clear_type') == DBConstants.MUSECA_CLEAR_TYPE_FULL_COMBO if self.game == GameConstants.POPN_MUSIC: return attempt.data.get_int('medal') in [ DBConstants.POPN_MUSIC_PLAY_MEDAL_CIRCLE_FULL_COMBO, DBConstants.POPN_MUSIC_PLAY_MEDAL_DIAMOND_FULL_COMBO, DBConstants.POPN_MUSIC_PLAY_MEDAL_STAR_FULL_COMBO, DBConstants.POPN_MUSIC_PLAY_MEDAL_PERFECT, ] if self.game == GameConstants.REFLEC_BEAT: return attempt.data.get_int('combo_type') in [ DBConstants.REFLEC_BEAT_COMBO_TYPE_FULL_COMBO, DBConstants.REFLEC_BEAT_COMBO_TYPE_FULL_COMBO_ALL_JUST, ] if self.game == GameConstants.SDVX: return attempt.data.get_int('clear_type') in [ DBConstants.SDVX_CLEAR_TYPE_ULTIMATE_CHAIN, DBConstants.SDVX_CLEAR_TYPE_PERFECT_ULTIMATE_CHAIN, ] return False def __aggregate_global(self, attempts: List[Attempt]) -> List[Dict[str, Any]]: stats: Dict[int, Dict[int, Dict[str, int]]] = {} for attempt in attempts: if attempt.id not in stats: stats[attempt.id] = {} if attempt.chart not in stats[attempt.id]: stats[attempt.id][attempt.chart] = { 'plays': 0, 'clears': 0, 'combos': 0, } if self.__is_play(attempt): stats[attempt.id][attempt.chart]['plays'] += 1 if self.__is_clear(attempt): stats[attempt.id][attempt.chart]['clears'] += 1 if self.__is_combo(attempt): stats[attempt.id][attempt.chart]['combos'] += 1 retval = [] for songid in stats: for songchart in stats[songid]: stat = stats[songid][songchart] stat['id'] = songid stat['chart'] = songchart retval.append(self.__format_statistics(stat)) return retval def __aggregate_local(self, cards: Dict[int, List[str]], attempts: List[Tuple[UserID, Attempt]]) -> List[Dict[str, Any]]: stats: Dict[UserID, Dict[int, Dict[int, Dict[str, int]]]] = {} for (userid, attempt) in attempts: if userid not in stats: stats[userid] = {} if attempt.id not in stats[userid]: stats[userid][attempt.id] = {} if attempt.chart not in stats[userid][attempt.id]: stats[userid][attempt.id][attempt.chart] = { 'plays': 0, 'clears': 0, 'combos': 0, } if self.__is_play(attempt): stats[userid][attempt.id][attempt.chart]['plays'] += 1 if self.__is_clear(attempt): stats[userid][attempt.id][attempt.chart]['clears'] += 1 if self.__is_combo(attempt): stats[userid][attempt.id][attempt.chart]['combos'] += 1 retval = [] for userid in stats: for songid in stats[userid]: for songchart in stats[userid][songid]: stat = stats[userid][songid][songchart] stat['id'] = songid stat['chart'] = songchart retval.append(self.__format_user_statistics(cards[userid], stat)) return retval def fetch_v1(self, idtype: str, ids: List[str], params: Dict[str, Any]) -> List[Dict[str, Any]]: retval: List[Dict[str, Any]] = [] # Fetch the attempts if idtype == APIConstants.ID_TYPE_SERVER: retval = self.__aggregate_global( [attempt[1] for attempt in self.data.local.music.get_all_attempts(self.game, self.music_version)] ) elif idtype == APIConstants.ID_TYPE_SONG: if len(ids) == 1: songid = int(ids[0]) chart = None else: songid = int(ids[0]) chart = int(ids[1]) retval = self.__aggregate_global( [attempt[1] for attempt in self.data.local.music.get_all_attempts(self.game, self.music_version, songid=songid, songchart=chart)] ) elif idtype == APIConstants.ID_TYPE_INSTANCE: songid = int(ids[0]) chart = int(ids[1]) cardid = ids[2] userid = self.data.local.user.from_cardid(cardid) if userid is not None: retval = self.__aggregate_local( {userid: self.data.local.user.get_cards(userid)}, self.data.local.music.get_all_attempts(self.game, self.music_version, songid=songid, songchart=chart, userid=userid) ) elif idtype == APIConstants.ID_TYPE_CARD: id_to_cards: Dict[int, List[str]] = {} attempts: List[Tuple[UserID, Attempt]] = [] for cardid in ids: userid = self.data.local.user.from_cardid(cardid) if userid is not None: # Don't duplicate loads for users with multiple card IDs if multiples # of those IDs are requested. if userid in id_to_cards: continue id_to_cards[userid] = self.data.local.user.get_cards(userid) attempts.extend( self.data.local.music.get_all_attempts(self.game, self.music_version, userid=userid) ) retval = self.__aggregate_local(id_to_cards, attempts) else: raise APIException('Invalid ID type!') return retval