bemaniutils/bemani/api/objects/statistics.py

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,
            GameConstants.JUBEAT,
        }:
            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: APIConstants, 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