1
0
mirror of synced 2024-12-05 02:57:55 +01:00
bemaniutils/bemani/api/objects/records.py

300 lines
13 KiB
Python
Raw Normal View History

from typing import Any, Dict, List, Set, Tuple
from bemani.api.exceptions import APIException
from bemani.api.objects.base import BaseObject
from bemani.common import GameConstants, VersionConstants, APIConstants, DBConstants
from bemani.data import Score, UserID
class RecordsObject(BaseObject):
def __format_ddr_record(self, record: Score) -> Dict[str, Any]:
halo = {
DBConstants.DDR_HALO_NONE: 'none',
DBConstants.DDR_HALO_GOOD_FULL_COMBO: 'gfc',
DBConstants.DDR_HALO_GREAT_FULL_COMBO: 'fc',
DBConstants.DDR_HALO_PERFECT_FULL_COMBO: 'pfc',
DBConstants.DDR_HALO_MARVELOUS_FULL_COMBO: 'mfc',
}.get(record.data.get_int('halo'), 'none')
rank = {
DBConstants.DDR_RANK_AAA: "AAA",
DBConstants.DDR_RANK_AA_PLUS: "AA+",
DBConstants.DDR_RANK_AA: "AA",
DBConstants.DDR_RANK_AA_MINUS: "AA-",
DBConstants.DDR_RANK_A_PLUS: "A+",
DBConstants.DDR_RANK_A: "A",
DBConstants.DDR_RANK_A_MINUS: "A-",
DBConstants.DDR_RANK_B_PLUS: "B+",
DBConstants.DDR_RANK_B: "B",
DBConstants.DDR_RANK_B_MINUS: "B-",
DBConstants.DDR_RANK_C_PLUS: "C+",
DBConstants.DDR_RANK_C: "C",
DBConstants.DDR_RANK_C_MINUS: "C-",
DBConstants.DDR_RANK_D_PLUS: "D+",
DBConstants.DDR_RANK_D: "D",
DBConstants.DDR_RANK_E: "E",
}.get(record.data.get_int('rank'), 'E')
if self.version == VersionConstants.DDR_ACE:
# DDR Ace is specia
ghost = [int(x) for x in record.data.get_str('ghost')]
else:
if 'trace' not in record.data:
ghost = []
else:
ghost = record.data.get_int_array('trace', len(record.data['trace']))
return {
'rank': rank,
'halo': halo,
'combo': record.data.get_int('combo'),
'ghost': ghost,
}
def __format_iidx_record(self, record: Score) -> Dict[str, Any]:
status = {
DBConstants.IIDX_CLEAR_STATUS_NO_PLAY: 'np',
DBConstants.IIDX_CLEAR_STATUS_FAILED: 'failed',
DBConstants.IIDX_CLEAR_STATUS_ASSIST_CLEAR: 'ac',
DBConstants.IIDX_CLEAR_STATUS_EASY_CLEAR: 'ec',
DBConstants.IIDX_CLEAR_STATUS_CLEAR: 'nc',
DBConstants.IIDX_CLEAR_STATUS_HARD_CLEAR: 'hc',
DBConstants.IIDX_CLEAR_STATUS_EX_HARD_CLEAR: 'exhc',
DBConstants.IIDX_CLEAR_STATUS_FULL_COMBO: 'fc',
}.get(record.data.get_int('clear_status'), 'np')
return {
'status': status,
'miss': record.data.get_int('miss_count', -1),
'ghost': [b for b in record.data.get_bytes('ghost')],
'pgreat': record.data.get_int('pgreats', -1),
'great': record.data.get_int('greats', -1),
}
def __format_jubeat_record(self, record: Score) -> Dict[str, Any]:
status = {
DBConstants.JUBEAT_PLAY_MEDAL_FAILED: 'failed',
DBConstants.JUBEAT_PLAY_MEDAL_CLEARED: 'cleared',
DBConstants.JUBEAT_PLAY_MEDAL_NEARLY_FULL_COMBO: 'nfc',
DBConstants.JUBEAT_PLAY_MEDAL_FULL_COMBO: 'fc',
DBConstants.JUBEAT_PLAY_MEDAL_NEARLY_EXCELLENT: 'nec',
DBConstants.JUBEAT_PLAY_MEDAL_EXCELLENT: 'exc',
}.get(record.data.get_int('medal'), 'failed')
if 'ghost' not in record.data:
ghost: List[int] = []
else:
ghost = record.data.get_int_array('ghost', len(record.data['ghost']))
return {
'status': status,
'combo': record.data.get_int('combo', -1),
'ghost': ghost,
}
def __format_museca_record(self, record: Score) -> Dict[str, Any]:
rank = {
DBConstants.MUSECA_GRADE_DEATH: 'death',
DBConstants.MUSECA_GRADE_POOR: 'poor',
DBConstants.MUSECA_GRADE_MEDIOCRE: 'mediocre',
DBConstants.MUSECA_GRADE_GOOD: 'good',
DBConstants.MUSECA_GRADE_GREAT: 'great',
DBConstants.MUSECA_GRADE_EXCELLENT: 'excellent',
DBConstants.MUSECA_GRADE_SUPERB: 'superb',
DBConstants.MUSECA_GRADE_MASTERPIECE: 'masterpiece',
DBConstants.MUSECA_GRADE_PERFECT: 'perfect'
}.get(record.data.get_int('grade'), 'death')
status = {
DBConstants.MUSECA_CLEAR_TYPE_FAILED: 'failed',
DBConstants.MUSECA_CLEAR_TYPE_CLEARED: 'cleared',
DBConstants.MUSECA_CLEAR_TYPE_FULL_COMBO: 'fc',
}.get(record.data.get_int('clear_type'), 'failed')
return {
'rank': rank,
'status': status,
'combo': record.data.get_int('combo', -1),
'buttonrate': record.data.get_dict('stats').get_int('btn_rate'),
'longrate': record.data.get_dict('stats').get_int('long_rate'),
'volrate': record.data.get_dict('stats').get_int('vol_rate'),
}
def __format_popn_record(self, record: Score) -> Dict[str, Any]:
status = {
DBConstants.POPN_MUSIC_PLAY_MEDAL_CIRCLE_FAILED: 'cf',
DBConstants.POPN_MUSIC_PLAY_MEDAL_DIAMOND_FAILED: 'df',
DBConstants.POPN_MUSIC_PLAY_MEDAL_STAR_FAILED: 'sf',
DBConstants.POPN_MUSIC_PLAY_MEDAL_EASY_CLEAR: 'ec',
DBConstants.POPN_MUSIC_PLAY_MEDAL_CIRCLE_CLEARED: 'cc',
DBConstants.POPN_MUSIC_PLAY_MEDAL_DIAMOND_CLEARED: 'dc',
DBConstants.POPN_MUSIC_PLAY_MEDAL_STAR_CLEARED: 'sc',
DBConstants.POPN_MUSIC_PLAY_MEDAL_CIRCLE_FULL_COMBO: 'cfc',
DBConstants.POPN_MUSIC_PLAY_MEDAL_DIAMOND_FULL_COMBO: 'dfc',
DBConstants.POPN_MUSIC_PLAY_MEDAL_STAR_FULL_COMBO: 'sfc',
DBConstants.POPN_MUSIC_PLAY_MEDAL_PERFECT: 'p',
}.get(record.data.get_int('medal'), 'cf')
return {
'status': status,
'combo': record.data.get_int('combo', -1),
}
def __format_reflec_record(self, record: Score) -> Dict[str, Any]:
status = {
DBConstants.REFLEC_BEAT_CLEAR_TYPE_NO_PLAY: 'np',
DBConstants.REFLEC_BEAT_CLEAR_TYPE_FAILED: 'failed',
DBConstants.REFLEC_BEAT_CLEAR_TYPE_CLEARED: 'cleared',
DBConstants.REFLEC_BEAT_CLEAR_TYPE_HARD_CLEARED: 'hc',
DBConstants.REFLEC_BEAT_CLEAR_TYPE_S_HARD_CLEARED: 'shc',
}.get(record.data.get_int('clear_type'), 'np')
halo = {
DBConstants.REFLEC_BEAT_COMBO_TYPE_NONE: 'none',
DBConstants.REFLEC_BEAT_COMBO_TYPE_ALMOST_COMBO: 'ac',
DBConstants.REFLEC_BEAT_COMBO_TYPE_FULL_COMBO: 'fc',
DBConstants.REFLEC_BEAT_COMBO_TYPE_FULL_COMBO_ALL_JUST: 'fcaj',
}.get(record.data.get_int('combo_type'), 'none')
return {
'rate': record.data.get_int('achievement_rate'),
'status': status,
'halo': halo,
'combo': record.data.get_int('combo', -1),
'miss': record.data.get_int('miss_count', -1),
}
def __format_sdvx_record(self, record: Score) -> Dict[str, Any]:
status = {
DBConstants.SDVX_CLEAR_TYPE_NO_PLAY: 'np',
DBConstants.SDVX_CLEAR_TYPE_FAILED: 'failed',
DBConstants.SDVX_CLEAR_TYPE_CLEAR: 'cleared',
DBConstants.SDVX_CLEAR_TYPE_HARD_CLEAR: 'hc',
DBConstants.SDVX_CLEAR_TYPE_ULTIMATE_CHAIN: 'uc',
DBConstants.SDVX_CLEAR_TYPE_PERFECT_ULTIMATE_CHAIN: 'puc',
}.get(record.data.get_int('clear_type'), 'np')
rank = {
DBConstants.SDVX_GRADE_NO_PLAY: 'E',
DBConstants.SDVX_GRADE_D: 'D',
DBConstants.SDVX_GRADE_C: 'C',
DBConstants.SDVX_GRADE_B: 'B',
DBConstants.SDVX_GRADE_A: 'A',
DBConstants.SDVX_GRADE_A_PLUS: 'A+',
DBConstants.SDVX_GRADE_AA: 'AA',
DBConstants.SDVX_GRADE_AA_PLUS: 'AA+',
DBConstants.SDVX_GRADE_AAA: 'AAA',
DBConstants.SDVX_GRADE_AAA_PLUS: 'AAA+',
DBConstants.SDVX_GRADE_S: 'S',
}.get(record.data.get_int('grade'), 'E')
return {
'status': status,
'rank': rank,
'combo': record.data.get_int('combo', -1),
'buttonrate': record.data.get_dict('stats').get_int('btn_rate'),
'longrate': record.data.get_dict('stats').get_int('long_rate'),
'volrate': record.data.get_dict('stats').get_int('vol_rate'),
}
def __format_record(self, cardids: List[str], record: Score) -> Dict[str, Any]:
base = {
'cards': cardids,
'song': str(record.id),
'chart': str(record.chart),
'points': record.points,
'timestamp': record.timestamp,
'updated': record.update,
}
if self.game == GameConstants.DDR:
base.update(self.__format_ddr_record(record))
if self.game == GameConstants.IIDX:
base.update(self.__format_iidx_record(record))
if self.game == GameConstants.JUBEAT:
base.update(self.__format_jubeat_record(record))
if self.game == GameConstants.MUSECA:
base.update(self.__format_museca_record(record))
if self.game == GameConstants.POPN_MUSIC:
base.update(self.__format_popn_record(record))
if self.game == GameConstants.REFLEC_BEAT:
base.update(self.__format_reflec_record(record))
if self.game == GameConstants.SDVX:
base.update(self.__format_sdvx_record(record))
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
2020-12-06 03:23:20 +01:00
else:
return self.version
def fetch_v1(self, idtype: str, ids: List[str], params: Dict[str, Any]) -> List[Dict[str, Any]]:
since = params.get('since')
until = params.get('until')
# Fetch the scores
records: List[Tuple[UserID, Score]] = []
if idtype == APIConstants.ID_TYPE_SERVER:
# Because of the way this query works, we can't apply since/until to it directly.
# If we did, it would miss higher scores earned before since or after until, and
# incorrectly report records.
records.extend(self.data.local.music.get_all_records(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])
records.extend(self.data.local.music.get_all_scores(self.game, self.music_version, songid=songid, songchart=chart, since=since, until=until))
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:
score = self.data.local.music.get_score(self.game, self.music_version, userid, songid, chart)
if score is not None:
records.append((userid, score))
elif idtype == APIConstants.ID_TYPE_CARD:
users: Set[UserID] = set()
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 users:
continue
users.add(userid)
records.extend([(userid, score) for score in self.data.local.music.get_scores(self.game, self.music_version, userid, since=since, until=until)])
else:
raise APIException('Invalid ID type!')
# Now, fetch the users, and filter out scores belonging to orphaned users
id_to_cards: Dict[UserID, List[str]] = {}
retval: List[Dict[str, Any]] = []
for (userid, record) in records:
# Postfilter for queries that can't filter. This will save on data transferred.
if since is not None:
if record.update < since:
continue
if until is not None:
if record.update >= until:
continue
if userid not in id_to_cards:
cards = self.data.local.user.get_cards(userid)
if len(cards) == 0:
# Can't add this user, skip the score
continue
id_to_cards[userid] = cards
# Format the score and add it
retval.append(self.__format_record(id_to_cards[userid], record))
return retval