bemaniutils/bemani/backend/museca/base.py

# vim: set fileencoding=utf-8
from typing import Dict, Optional
from typing_extensions import Final

from bemani.backend.base import Base
from bemani.backend.core import CoreHandler, CardManagerHandler, PASELIHandler
from bemani.common import (
    Profile,
    ValidatedDict,
    GameConstants,
    DBConstants,
    Parallel,
    Model,
)
from bemani.data import UserID, Config, Data
from bemani.protocol import Node


class MusecaBase(CoreHandler, CardManagerHandler, PASELIHandler, Base):
    """
    Base game class for all Museca version that we support.
    """

    game: GameConstants = GameConstants.MUSECA

    CHART_TYPE_GREEN: Final[int] = 0
    CHART_TYPE_ORANGE: Final[int] = 1
    CHART_TYPE_RED: Final[int] = 2

    GRADE_DEATH: Final[int] = DBConstants.MUSECA_GRADE_DEATH
    GRADE_POOR: Final[int] = DBConstants.MUSECA_GRADE_POOR
    GRADE_MEDIOCRE: Final[int] = DBConstants.MUSECA_GRADE_MEDIOCRE
    GRADE_GOOD: Final[int] = DBConstants.MUSECA_GRADE_GOOD
    GRADE_GREAT: Final[int] = DBConstants.MUSECA_GRADE_GREAT
    GRADE_EXCELLENT: Final[int] = DBConstants.MUSECA_GRADE_EXCELLENT
    GRADE_SUPERB: Final[int] = DBConstants.MUSECA_GRADE_SUPERB
    GRADE_MASTERPIECE: Final[int] = DBConstants.MUSECA_GRADE_MASTERPIECE
    GRADE_PERFECT: Final[int] = DBConstants.MUSECA_GRADE_PERFECT

    CLEAR_TYPE_FAILED: Final[int] = DBConstants.MUSECA_CLEAR_TYPE_FAILED
    CLEAR_TYPE_CLEARED: Final[int] = DBConstants.MUSECA_CLEAR_TYPE_CLEARED
    CLEAR_TYPE_FULL_COMBO: Final[int] = DBConstants.MUSECA_CLEAR_TYPE_FULL_COMBO

    def __init__(self, data: Data, config: Config, model: Model) -> None:
        super().__init__(data, config, model)
        if model.rev == "X":
            self.omnimix = True
        else:
            self.omnimix = False

    @property
    def music_version(self) -> int:
        if self.omnimix:
            return DBConstants.OMNIMIX_VERSION_BUMP + self.version
        return self.version

    def previous_version(self) -> Optional["MusecaBase"]:
        """
        Returns the previous version of the game, based on this game. Should
        be overridden.
        """
        return None

    def game_to_db_clear_type(self, clear_type: int) -> int:
        # Given a game clear type, return the canonical database identifier.
        raise Exception("Implement in subclass!")

    def db_to_game_clear_type(self, clear_type: int) -> int:
        # Given a database clear type, return the game's identifier.
        raise Exception("Implement in subclass!")

    def game_to_db_grade(self, grade: int) -> int:
        # Given a game grade, return the canonical database identifier.
        raise Exception("Implement in subclass!")

    def db_to_game_grade(self, grade: int) -> int:
        # Given a database grade, return the game's identifier.
        raise Exception("Implement in subclass!")

    def get_profile_by_refid(self, refid: Optional[str]) -> Optional[Node]:
        """
        Given a RefID, return a formatted profile node. Basically every game
        needs a profile lookup, even if it handles where that happens in
        a different request. This is provided for code deduplication.
        """
        if refid is None:
            return None

        # First try to load the actual profile
        userid = self.data.remote.user.from_refid(self.game, self.version, refid)
        profile = self.get_profile(userid)
        if profile is None:
            return None

        # Now, return it
        return self.format_profile(userid, profile)

    def new_profile_by_refid(
        self, refid: Optional[str], name: Optional[str], locid: Optional[int]
    ) -> Node:
        """
        Given a RefID and an optional name, create a profile and then return
        a formatted profile node. Similar rationale to get_profile_by_refid.
        """
        if refid is None:
            return None

        if name is None:
            name = "NONAME"

        # First, create and save the default profile
        userid = self.data.remote.user.from_refid(self.game, self.version, refid)
        profile = Profile(
            self.game,
            self.version,
            refid,
            0,
            {
                "name": name,
                "loc": locid,
            },
        )
        self.put_profile(userid, profile)
        return self.format_profile(userid, profile)

    def format_profile(self, userid: UserID, profile: Profile) -> Node:
        """
        Base handler for a profile. Given a userid and a profile dictionary,
        return a Node representing a profile. Should be overridden.
        """
        return Node.void("game")

    def unformat_profile(
        self, userid: UserID, request: Node, oldprofile: Profile
    ) -> Profile:
        """
        Base handler for profile parsing. Given a request and an old profile,
        return a new profile that's been updated with the contents of the request.
        Should be overridden.
        """
        return oldprofile

    def get_clear_rates(self) -> Dict[int, Dict[int, Dict[str, int]]]:
        """
        Returns a dictionary similar to the following:

        {
            musicid: {
                chart: {
                    total: total plays,
                    clears: total clears,
                },
            },
        }
        """
        all_attempts, remote_attempts = Parallel.execute(
            [
                lambda: self.data.local.music.get_all_attempts(
                    game=self.game,
                    version=self.music_version,
                ),
                lambda: self.data.remote.music.get_clear_rates(
                    game=self.game,
                    version=self.music_version,
                ),
            ]
        )
        attempts: Dict[int, Dict[int, Dict[str, int]]] = {}
        for _, attempt in all_attempts:
            # Terrible temporary structure is terrible.
            if attempt.id not in attempts:
                attempts[attempt.id] = {}
            if attempt.chart not in attempts[attempt.id]:
                attempts[attempt.id][attempt.chart] = {
                    "total": 0,
                    "clears": 0,
                }

            # We saw an attempt, keep the total attempts in sync.
            attempts[attempt.id][attempt.chart]["total"] = (
                attempts[attempt.id][attempt.chart]["total"] + 1
            )

            if (
                attempt.data.get_int("clear_type", self.CLEAR_TYPE_FAILED)
                != self.CLEAR_TYPE_FAILED
            ):
                # This attempt was a failure, so don't count it against clears of full combos
                continue

            # It was at least a clear
            attempts[attempt.id][attempt.chart]["clears"] = (
                attempts[attempt.id][attempt.chart]["clears"] + 1
            )

        # Merge in remote attempts
        for songid in remote_attempts:
            if songid not in attempts:
                attempts[songid] = {}

            for songchart in remote_attempts[songid]:
                if songchart not in attempts[songid]:
                    attempts[songid][songchart] = {
                        "total": 0,
                        "clears": 0,
                    }

                attempts[songid][songchart]["total"] += remote_attempts[songid][
                    songchart
                ]["plays"]
                attempts[songid][songchart]["clears"] += remote_attempts[songid][
                    songchart
                ]["clears"]

        return attempts

    def update_score(
        self,
        userid: Optional[UserID],
        songid: int,
        chart: int,
        points: int,
        clear_type: int,
        grade: int,
        combo: int,
        stats: Optional[Dict[str, int]] = None,
    ) -> None:
        """
        Given various pieces of a score, update the user's high score and score
        history in a controlled manner, so all games in SDVX series can expect
        the same attributes in a score.
        """
        # Range check clear type
        if clear_type not in [
            self.CLEAR_TYPE_FAILED,
            self.CLEAR_TYPE_CLEARED,
            self.CLEAR_TYPE_FULL_COMBO,
        ]:
            raise Exception(f"Invalid clear type value {clear_type}")

        # Range check grade
        if grade not in [
            self.GRADE_DEATH,
            self.GRADE_POOR,
            self.GRADE_MEDIOCRE,
            self.GRADE_GOOD,
            self.GRADE_GREAT,
            self.GRADE_EXCELLENT,
            self.GRADE_SUPERB,
            self.GRADE_MASTERPIECE,
            self.GRADE_PERFECT,
        ]:
            raise Exception(f"Invalid grade value {grade}")

        if userid is not None:
            oldscore = self.data.local.music.get_score(
                self.game,
                self.music_version,
                userid,
                songid,
                chart,
            )
        else:
            oldscore = None

        # Score history is verbatum, instead of highest score
        history = ValidatedDict({})
        oldpoints = points

        if oldscore is None:
            # If it is a new score, create a new dictionary to add to
            scoredata = ValidatedDict({})
            raised = True
            highscore = True
        else:
            # Set the score to any new record achieved
            raised = points > oldscore.points
            highscore = points >= oldscore.points
            points = max(oldscore.points, points)
            scoredata = oldscore.data

        # Replace grade and clear type
        scoredata.replace_int(
            "clear_type", max(scoredata.get_int("clear_type"), clear_type)
        )
        history.replace_int("clear_type", clear_type)
        scoredata.replace_int("grade", max(scoredata.get_int("grade"), grade))
        history.replace_int("grade", grade)

        # If we have a combo, replace it
        scoredata.replace_int("combo", max(scoredata.get_int("combo"), combo))
        history.replace_int("combo", combo)

        # If we have play stats, replace it
        if stats is not None:
            if raised:
                # We have stats, and there's a new high score, update the stats
                scoredata.replace_dict("stats", stats)
            history.replace_dict("stats", stats)

        # Look up where this score was earned
        lid = self.get_machine_id()

        if userid is not None:
            # Write the new score back
            self.data.local.music.put_score(
                self.game,
                self.music_version,
                userid,
                songid,
                chart,
                lid,
                points,
                scoredata,
                highscore,
            )

        # Save the history of this score too
        self.data.local.music.put_attempt(
            self.game,
            self.music_version,
            userid,
            songid,
            chart,
            lid,
            oldpoints,
            history,
            raised,
        )