game.py

import json
import os
import platform
from dataclasses import dataclass
from datetime import datetime
from enum import IntEnum
from logging import getLogger
from typing import Dict, Iterator, Callable, Tuple
from typing import List, Optional

from PyQt5.QtCore import QObject, pyqtSignal, QRunnable, QSettings, pyqtSlot, QThreadPool
from PyQt5.QtGui import QPixmap
from legendary.models.game import Game, InstalledGame, SaveGameFile
from requests.exceptions import ConnectionError, HTTPError

from rare.shared import LegendaryCoreSingleton, ArgumentsSingleton
from rare.utils.image_manager import ImageManagerSingleton
from rare.utils.paths import data_dir

logger = getLogger("RareGame")


class RareGame(QObject):
    @dataclass
    class Metadata:
        queued: bool = False
        queue_pos: Optional[int] = None
        last_played: Optional[datetime] = None

        @classmethod
        def from_json(cls, data):
            return cls(
                queued=data.get("queued", False),
                queue_pos=data.get("queue_pos", None),
                last_played=datetime.strptime(data.get("last_played", "None"), "%Y-%m-%dT%H:%M:%S.%f"),
            )

        def __dict__(self):
            return dict(
                queued=self.queued,
                queue_pos=self.queue_pos,
                last_played=self.last_played.strftime("%Y-%m-%dT%H:%M:%S.%f"),
            )

        def __bool__(self):
            return self.queued or self.queue_pos is not None or self.last_played is not None

    def __init__(self, game: Game):
        super(RareGame, self).__init__()
        self.core = LegendaryCoreSingleton()
        self.image_manager = ImageManagerSingleton()

        # Update names for Unreal Engine
        if game.app_title == "Unreal Engine":
            game.app_title += f" {game.app_name.split('_')[-1]}"
            logger.info(self.core.get_game(game.app_name))
            self.core.lgd.set_game_meta(game.app_name, game)
        self.game: Game = game

        # None if origin or not installed
        self.igame: Optional[InstalledGame] = self.core.get_installed_game(game.app_name)
        # Null QPixmap if image is not found, will queue image for download
        self.image: QPixmap = self.image_manager.get_pixmap(self.app_name, self.is_installed)

        self.metadata: RareGame.Metadata = RareGame.Metadata()
        self.dlcs: List[RareGame] = list()
        self.saves: List[SaveGameFile] = list()

        # self.saves = self.core.get_save_games(game.app_name)

    @property
    def app_name(self) -> str:
        return self.game.app_name

    @property
    def app_title(self) -> str:
        return self.game.app_title

    @property
    def developer(self) -> str:
        """!
        @brief Property to report the developer of a Game

        @return str
        """
        return self.game.metadata["developer"]

    @property
    def install_size(self) -> int:
        """!
        @brief Property to report the installation size of an InstalledGame

        @return int The size of the installation
        """
        if self.igame is not None:
            return self.igame.install_size
        else:
            return 0

    @property
    def version(self) -> str:
        """!
        @brief Reports the currently installed version of the Game

        For InstalledGame reports the currently installed version, which might be
        different from the remote version available from EGS. For not installed Games
        it reports the already known version.

        @return str The current version of the game
        """
        if self.igame is not None:
            return self.igame.version
        else:
            return self.game.app_version()

    @property
    def remote_version(self) -> str:
        """!
        @brief Property to report the remote version of an InstalledGame

        If the Game is installed, requests the latest version string from EGS,
        otherwise it reports the already known version of the Game for Windows.

        @return str The current version from EGS
        """
        if self.igame is not None:
            return self.core.get_asset(
                self.game.app_name, platform=self.igame.platform, update=False
            ).build_version
        else:
            return self.game.app_version()

    @property
    def has_update(self) -> bool:
        """!
        @brief Property to report if an InstalledGame has updates available

        Games have to be installed and have assets available to have
        updates

        @return bool If there is an update available
        """
        if self.igame is not None and self.core.lgd.assets is not None:
            try:
                if self.remote_version != self.igame.version:
                    return True
            except ValueError:
                logger.error(f"Asset error for {self.game.app_title}")
                return False
        return False

    @property
    def is_installed(self) -> bool:
        """!
        @brief Property to report if a game is installed

        This returns True if InstalledGame data have been loaded for the game
        or if the game is a game without assets, for example an Origin game.

        @return bool If the game should be considered installed
        """
        return (self.igame is not None) or self.is_non_asset

    @is_installed.setter
    def is_installed(self, installed: bool) -> None:
        """!
        @brief Sets the installation status of a game

        If this is set to True the InstalledGame data is fetched
        for the game, if set to False the igame attribute is cleared.

        @param installed The installation status of the game
        @return None
        """
        if installed:
            self.igame = self.core.get_installed_game(self.game.app_name)
        else:
            self.igame = None
        self.setStatedPixmap(installed)

    @property
    def is_offline(self) -> bool:
        """!
        @brief Property to report if a game can run offline

        Checks if the game can run without connectin the internet.
        It's a simple wrapper around legendary provided information,
        with handling of not installed games.

        @return bool If the games can run without network
        """
        # lk: I used `is_installed` here, but since non_asset games
        # lk: are treated as installed, the base check should be used instead
        if self.igame is not None:
            return self.igame.can_run_offline
        else:
            return False

    @property
    def is_foreign(self) -> bool:
        """!
        @brief Property to report if a game doesn't belong to the current account

        Checks if a game belongs to the currently logged in account. Games that require
        a network connection or remote authentication will fail to run from another account
        despite being installed. On the other hand, games that do not require network,
        can be executed, facilitating a rudementary game sharing option on the same computer.

        @return bool If the game belongs to another count or not
        """
        ret = True
        try:
            if self.igame is not None:
                _ = self.core.get_asset(self.game.app_name, platform=self.igame.platform).build_version
                ret = False
        except ValueError:
            logger.warning(f"Game {self.game.app_title} has no metadata. Set offline true")
        except AttributeError:
            ret = False
        return ret

    @property
    def needs_verification(self) -> bool:
        """!
        @brief Property to report if a games requires to be verified

        Simple wrapper around legendary's attribute with installation
        status check

        @return bool If the games needs to be verified
        """
        if self.igame is not None:
            return self.igame.needs_verification
        else:
            return False

    @needs_verification.setter
    def needs_verification(self, not_update: bool) -> None:
        """!
        @brief Sets the verification status of a game.

        The operation here is reversed. since the property is
        named like this. After the verification, set this to 'False'
        to update the InstalledGame in the widget.

        @param not_update If the game requires verification
        @return None
        """
        if not not_update:
            self.igame = self.core.get_installed_game(self.game.app_name)

    @property
    def is_dlc(self) -> bool:
        """!
        @brief Property to report if Game is a dlc

        @return bool
        """
        return self.game.is_dlc

    @property
    def is_mac(self) -> bool:
        """!
        @brief Property to report if Game has a mac version

        @return bool
        """
        return "Mac" in self.game.asset_infos.keys()

    @property
    def is_win32(self) -> bool:
        """!
        @brief Property to report if Game is 32bit game

        @return bool
        """
        return "Win32" in self.game.asset_infos.keys()

    @property
    def is_unreal(self) -> bool:
        """!
        @brief Property to report if a Game is an Unreal Engine bundle

        @return bool
        """
        if not self.is_non_asset:
            return self.game.asset_infos["Windows"].namespace == "ue"
        else:
            return False

    @property
    def is_non_asset(self) -> bool:
        """!
        @brief Property to report if a Game doesn't have assets

        Typically, games have assets, however some games that require
        other launchers do not have them. Rare treats these games as installed
        offering to execute their launcher.

        @return bool If the game doesn't have assets
        """
        return not self.game.asset_infos

    def refresh_image(self) -> None:
        self.image.reload()

    @property
    def can_run(self) -> bool:
        if self.is_installed:
            if self.is_non_asset:
                return True
            elif self.game_running or self.needs_verification:
                return False
            else:
                return True
        else:
            return False


class ApiResults(QObject):

    class Worker(QRunnable):
        class Result(IntEnum):
            GAMES = 1
            NON_ASSET = 2
            WIN32 = 3
            MACOS = 4

        class Signals(QObject):
            progress = pyqtSignal(int, str)
            result = pyqtSignal(object, int)

        def __init__(self, request: Result):
            super(ApiResults.Worker, self).__init__()
            self.setAutoDelete(True)

            self.signals = ApiResults.Worker.Signals()
            self.core = LegendaryCoreSingleton()
            self.args = ArgumentsSingleton()

            self.run: Callable
            if request == ApiResults.Worker.Result.GAMES:
                self.run = self.request_games
            elif request == ApiResults.Worker.Result.NON_ASSET:
                self.run = self.request_non_asset
            elif request == ApiResults.Worker.Result.WIN32:
                self.run = self.request_win32
            elif request == ApiResults.Worker.Result.MACOS:
                self.run = self.request_macos
            else:
                self.run = self.no_run

        def no_run(self):
            pass

        def request_games(self):
            start_time = time.time()
            self.signals.progress.emit(25, "Loading games for Windows")
            result = self.core.get_game_and_dlc_list(
                update_assets=not self.args.offline,
                platform='Windows',
                skip_ue=False,
            )
            self.signals.result.emit(result, ApiResults.Worker.Result.GAMES)
            print(f"Games: {len(result[0])}, DLCs {len(result[1])}")
            print(f"Request Games: {time.time() - start_time} seconds")

        def request_non_asset(self):
            start_time = time.time()
            self.signals.progress.emit(40, "Loading games without assets")
            try:
                result = self.core.get_non_asset_library_items(force_refresh=False, skip_ue=False)
            except (HTTPError, ConnectionError) as e:
                logger.warning(f"Exception while fetching EGS data {e}")
                result = ([], {})
            self.signals.result.emit(result, ApiResults.Worker.Result.NON_ASSET)
            print(f"Non asset: {len(result[0])}, DLCs {len(result[1])}")
            print(f"Request Non Asset: {time.time() - start_time} seconds")

        def request_win32(self):
            start_time = time.time()
            self.signals.progress.emit(30, "Loading games for Windows (32bit)")
            result = self.core.get_game_and_dlc_list(
                update_assets=not self.args.offline, platform="Win32", skip_ue=False
            )
            self.signals.result.emit(([], {}), ApiResults.Worker.Result.WIN32)
            print(f"Win32: {len(result[0])}, DLCs {len(result[1])}")
            print(f"Request Win32: {time.time() - start_time} seconds")

        def request_macos(self):
            start_time = time.time()
            self.signals.progress.emit(35, "Loading games for MacOS")
            result = self.core.get_game_and_dlc_list(
                update_assets=not self.args.offline, platform="Mac", skip_ue=False
            )
            self.signals.result.emit(([], {}), ApiResults.Worker.Result.MACOS)
            print(f"MacOS: {len(result[0])}, DLCs {len(result[1])}")
            print(f"Request MacOS: {time.time() - start_time} seconds")

    __games: Dict[str, RareGame] = dict()
    __games_fetched: bool = False
    __non_asset_fetched: bool = False
    __win32_fetched: bool = False
    __macos_fetched: bool = False
    __saves_fetched: bool = True

    completed = pyqtSignal()
    progress = pyqtSignal(int, str)

    def __init__(self):
        super(ApiResults, self).__init__()
        self.core = LegendaryCoreSingleton()
        self.args = ArgumentsSingleton()
        self.settings = QSettings()
        self.thread_pool = QThreadPool().globalInstance()

    def load_metadata(self):
        metadata_json = os.path.join(data_dir, "game_meta.json")
        if os.path.exists(metadata_json):
            try:
                metadata = json.load(open(metadata_json))
            except json.JSONDecodeError:
                logger.warning("Game metadata json file is corrupt")
            else:
                for app_name, data in metadata.items():
                    self.__games[app_name].metadata = RareGame.Metadata.from_json(data)

    def save_metadata(self):
        json.dump(
            {app_name: game.metadata.__dict__() for app_name, game in self.__games.items() if game.metadata},
            open(os.path.join(data_dir, "game_meta.json"), "w"),
            indent=2,
        )

    def save_file(self):
        self.save_metadata()

    def get_game(self, app_name: str) -> RareGame:
        return self.__games[app_name]

    def add_game(self, game: Game) -> None:
        if game.app_name not in self.__games:
            self.__games[game.app_name] = RareGame(game)
        else:
            logger.warning(f"{game.app_name} already present in {type(self).__name__}")

    def __add_games_and_dlcs(self, games_list: List[Game], dlcs_dict: Dict[str, List]) -> None:
        for game in games_list:
            self.add_game(game)
        for catalog_item_id, dlcs in dlcs_dict.items():
            rare_game_dlcs = [RareGame(dlc) for dlc in dlcs]
            for dlc in rare_game_dlcs:
                self.__games[dlc.app_name] = dlc
            # FIXME: find a better way to do this maybe?
            main_game = [game for game in self.__filter_games(lambda g: g.game.catalog_item_id == catalog_item_id)]
            if len(main_game) > 1:
                raise RuntimeError("Why are there more results than one?")
            main_game[0].dlcs.extend(rare_game_dlcs)

    @pyqtSlot(object, int)
    def handle_result(self, result: Tuple, res_type: int):
        print("handle_result")
        logger.debug(f"Api Request got from {res_type}")
        if res_type == ApiResults.Worker.Result.GAMES:
            print("GAMELIST")
            games, dlc_dict = result
            self.__add_games_and_dlcs(games, dlc_dict)
            self.__games_fetched = True
        if res_type == ApiResults.Worker.Result.NON_ASSET:
            print("NON_ASSET")
            games, dlc_dict = result
            self.__add_games_and_dlcs(games, dlc_dict)
            self.__non_asset_fetched = True
        if res_type == ApiResults.Worker.Result.WIN32:
            print("WIN32")
            self.__win32_fetched = True
        if res_type == ApiResults.Worker.Result.MACOS:
            print("MACOS")
            self.__macos_fetched = True

        if (
            self.__games_fetched
            and self.__non_asset_fetched
            and self.__win32_fetched
            and self.__macos_fetched
            and self.__saves_fetched
        ):
            self.progress.emit(90, "Loading Rare game metadata")
            self.load_metadata()
            self.progress.emit(100, "Loading completed, launching Rare")
            print("Fetch time", time.time() - self.start_time, "seconds")
            self.completed.emit()

    def fetch(self):
        self.start_time = time.time()
        games_worker = ApiResults.Worker(request=ApiResults.Worker.Result.GAMES)
        games_worker.signals.result.connect(self.handle_result)
        games_worker.signals.progress.connect(self.progress)
        self.thread_pool.start(games_worker)

        non_asset_worker = ApiResults.Worker(request=ApiResults.Worker.Result.NON_ASSET)
        non_asset_worker.signals.result.connect(self.handle_result)
        non_asset_worker.signals.progress.connect(self.progress)
        self.thread_pool.start(non_asset_worker)

        if self.settings.value("win32_meta", False, bool):
            win32_worker = ApiResults.Worker(request=ApiResults.Worker.Result.WIN32)
            win32_worker.signals.result.connect(self.handle_result)
            win32_worker.signals.progress.connect(self.progress)
            self.thread_pool.start(win32_worker)
        else:
            self.__win32_fetched = True

        if self.settings.value("mac_meta", platform.system() == "Darwin", bool):
            macos_worker = ApiResults.Worker(request=ApiResults.Worker.Result.MACOS)
            macos_worker.signals.result.connect(self.handle_result)
            macos_worker.signals.progress.connect(self.progress)
            self.thread_pool.start(macos_worker)
        else:
            self.__macos_fetched = True

    def __filter_games(self, condition: Callable[[RareGame], bool]) -> Iterator[RareGame]:
        return filter(condition, self.__games.values())

    @property
    def games(self) -> Iterator[RareGame]:
        return self.__filter_games(lambda game: not game.is_dlc)

    @property
    def dlcs(self) -> Iterator[RareGame]:
        """!
        RareGames that ARE DLCs themselves
        """
        return self.__filter_games(lambda game: game.is_dlc)

    @property
    def has_dlcs(self) -> Iterator[RareGame]:
        """!
        RareGames that HAVE DLCs associated with them
        """
        return self.__filter_games(lambda game: bool(game.dlcs))

    @property
    def bit32_games(self) -> Iterator[RareGame]:
        return self.__filter_games(lambda game: game.is_win32)

    @property
    def mac_games(self) -> Iterator[RareGame]:
        return self.__filter_games(lambda game: game.is_mac)

    @property
    def no_asset_games(self) -> Iterator[RareGame]:
        return self.__filter_games(lambda game: game.is_non_asset)

    @property
    def unreal_engine(self) -> Iterator[RareGame]:
        return self.__filter_games(lambda game: game.is_unreal)

    @property
    def updates(self) -> Iterator[RareGame]:
        return self.__filter_games(lambda game: game.has_update)

    @property
    def saves(self) -> Iterator[SaveGameFile]:
        """!
        SaveGameFiles across games
        """
        for game in self.__games.values():
            if game.saves:
                yield game.saves

    @property
    def has_saves(self) -> Iterator[RareGame]:
        """!
        RareGames that have SaveGameFiles associated with them
        """
        return self.__filter_games(lambda game: bool(game.saves))


# RareGameMeta is being reimplemented in ApiResults. Left in for my reference
class RareGameMeta:
    _games: Dict[str, RareGame] = {}

    def __init__(self):
        meta_data = {}
        if os.path.exists(p := os.path.join(data_dir, "game_meta.json")):
            try:
                meta_data = json.load(open(p))
            except json.JSONDecodeError:
                logger.warning("Game meta json file corrupt")
        else:
            with open(p, "w") as file:
                file.write("{}")

        for app_name, data in meta_data.items():
            self._games[app_name].metadata = RareGame.Metadata.from_json(data)

    def get_games(self):
        return list(self._games.values())

    def get_game(self, app_name):
        return self._games.get(app_name, RareGame(app_name))

    def set_game(self, app_name: str, game: RareGame):
        self._games[app_name] = game
        self.save_file()

    def save_file(self):
        json.dump(
            {app_name: game.metadata.__dict__() for app_name, game in self._games.items() if game.metadata},
            open(os.path.join(data_dir, "game_meta.json"), "w"),
            indent=4,
        )


if __name__ == "__main__":
    import sys
    import time
    from PyQt5.QtWidgets import (
        QApplication,
        QDialog,
        QVBoxLayout,
        QProgressBar,
        QLabel,
        QPushButton,
        QScrollArea,
        QWidget
    )
    from rare.shared import GlobalSignalsSingleton
    from rare.utils.extra_widgets import LibraryLayout
    from rare.components.tabs.games.game_widgets import IconGameWidget, ListGameWidget


    class Namespace:
        def __init__(self, **kwargs):
            self.__dict__.update(kwargs)

    class RareGameTestDialog(QDialog):
        def __init__(self):
            super(RareGameTestDialog, self).__init__()
            self.setFixedSize(1280, 800)
            self.progbar = QProgressBar(self)
            self.label = QLabel(self)
            self.button = QPushButton("Fetch", self)
            self.button.clicked.connect(self.do_work)

            self.icon_scroll_area = QScrollArea(self)
            self.icon_scroll_area.setWidgetResizable(True)
            self.icon_scroll_widget = QWidget(self.icon_scroll_area)
            self.icon_scroll_widget.setLayout(LibraryLayout(self.icon_scroll_widget))
            self.icon_scroll_area.setWidget(self.icon_scroll_widget)

            self.list_scroll_area = QScrollArea(self)
            self.list_scroll_area.setWidgetResizable(True)
            self.list_scroll_widget = QWidget(self.list_scroll_area)
            self.list_scroll_widget.setLayout(QVBoxLayout(self.list_scroll_widget))
            self.list_scroll_area.setWidget(self.list_scroll_widget)

            layout = QVBoxLayout(self)
            layout.addWidget(self.progbar)
            layout.addWidget(self.label)
            layout.addWidget(self.button)
            layout.addWidget(self.icon_scroll_area)
            layout.addWidget(self.list_scroll_area)

            core = LegendaryCoreSingleton(init=True)
            arguments = Namespace(offline=False)
            args = ArgumentsSingleton(arguments)
            signals = GlobalSignalsSingleton(init=True)
            im = ImageManagerSingleton(self)

            self.api_results = ApiResults()
            self.api_results.progress.connect(self.on_progress)
            self.api_results.completed.connect(self.on_completed)

            core.login()

        def do_work(self):
            start_time = time.time()
            self.api_results.fetch()
            print(time.time() - start_time, "seconds")

        @pyqtSlot(int, str)
        def on_progress(self, prog: int, stat: str):
            self.progbar.setValue(prog)
            self.label.setText(stat)

        @pyqtSlot()
        def on_completed(self):
            start_time = time.time()
            games = [x for x in self.api_results.games]
            dlcs = [x for x in self.api_results.dlcs]
            win32 = [x for x in self.api_results.bit32_games]
            mac = [x for x in self.api_results.mac_games]
            unreal = [x for x in self.api_results.unreal_engine]
            updates = [x for x in self.api_results.updates]
            no_assets = [x for x in self.api_results.no_asset_games]
            saves = [x for x in self.api_results.saves]
            metadata = [x for x in self.api_results.games if x.metadata]

            games.sort(key=lambda g: g.is_installed, reverse=True)

            for game in games:
                # image = LibraryImageWidget(game.app_name)
                # image = LibraryProgressWidget(game.app_name)
                # image.setFixedSize(game.image.size())

                # image = QLabel()

                icon_widg = IconGameWidget(game.game, None)
                icon_widg.setPixmap(game.image)
                list_widg = ListGameWidget(game.game, None)
                list_widg.setPixmap(game.image)

                self.icon_scroll_widget.layout().addWidget(icon_widg)
                self.list_scroll_widget.layout().addWidget(list_widg)

            print(time.time() - start_time, "seconds")
            pass

    app = QApplication(sys.argv)
    app.setApplicationName("Rare")
    app.setOrganizationName("Rare")
    dialog = RareGameTestDialog()
    dialog.show()
    sys.exit(app.exec_())