Humble App support

requirements-dev.txt

@@ -10,4 +10,5 @@ pytest-flakes==4.0.0
 pytest-mock==1.10.4
 freezegun==1.2.1
 PyGithub==1.53
-fog.buildtools~=1.0
+fog.buildtools~=1.0
+types-toml==0.10.7

src/humbleapp/humbleapp.py

@@ -0,0 +1,178 @@
+import json
+import sys
+import enum
+import os
+import pathlib
+import logging
+import typing as t
+import webbrowser
+from dataclasses import dataclass
+
+IS_WINDOWS = sys.platform == 'win32'
+
+if IS_WINDOWS:
+    import winreg
+
+
+logger = logging.getLogger(__name__)
+
+
+Json = t.Dict[str, t.Any]
+GameMachineName = str
+Timestamp = int
+
+
+class GameStatus(enum.Enum):
+    AVAILABLE = "available"
+    DOWNLOADED = "downloaded"
+    INSTALLED = "installed"
+
+
+class TroveCategory(enum.Enum):
+    PREMIUM = "premium"
+    GENERAL = "general"
+
+
+@dataclass
+class VaultGame:
+    machine_name: GameMachineName
+    game_name: str
+    date_added: Timestamp
+    date_ended: t.Optional[Timestamp]
+    is_available: bool
+    last_played: Timestamp
+    file_size: int
+    status: GameStatus
+    trove_category: TroveCategory
+    executable_path: t.Optional[str]
+    file_path: t.Optional[str]
+
+    @property
+    def full_executable_path(self) -> t.Optional[pathlib.Path]:
+        if self.file_path and self.executable_path:
+            return pathlib.Path(self.file_path) / self.executable_path
+        return None
+
+
+@dataclass
+class UserInfo:
+    is_paused: bool
+    owns_active_content: bool
+    can_resubscribe: bool
+    user_id: int
+    has_ever_subscribed: bool
+    has_perks: bool
+    user_key: str
+    has_beta_access: bool
+    will_receive_future_months: bool
+
+
+@dataclass
+class  Settings:
+    download_location: pathlib.Path
+
+
+@dataclass
+class HumbleAppConfig:
+    settings: Settings
+    game_collection: t.List[VaultGame]
+
+
+class FileWatcher:
+    def __init__(self, path: pathlib.PurePath) -> None:
+        self._path = path
+        self._prev_mtime: float = 0.0
+
+    @property
+    def path(self) -> pathlib.PurePath:
+        return self._path
+
+    def has_changed(self) -> t.Optional[bool]:
+        try:
+            last_mtime = os.stat(self._path).st_mtime
+        except OSError:
+            self._prev_mtime = 0.0
+            return None
+        changed = last_mtime != self._prev_mtime
+        self._prev_mtime = last_mtime
+        return changed
+
+
+def parse_humble_app_config(path: pathlib.PurePath)  -> HumbleAppConfig:
+    def parse_game(raw):
+        return VaultGame(
+            machine_name=raw["machineName"],
+            game_name=raw["gameName"],
+            status=GameStatus(raw["status"]),
+            is_available=raw["isAvailable"],
+            last_played=Timestamp(raw["lastPlayed"]),
+            file_size=raw["fileSize"],
+            date_added=raw["dateAdded"],
+            date_ended=raw["dateEnded"],
+            trove_category=TroveCategory(raw["troveCategory"]),
+            file_path=raw.get("filePath"),
+            executable_path=raw.get("executablePath"),
+        )
+
+    with open(path, encoding="utf-8") as f:
+        content = json.load(f)
+
+    games = [parse_game(g) for g in content['game-collection-4']]
+
+    return HumbleAppConfig(
+        settings=Settings(
+            pathlib.Path(content['settings']['downloadLocation']),
+        ),
+        game_collection=games,
+    )
+
+
+def get_app_path_for_uri_handler(protocol: str) -> t.Optional[str]:
+    """Source: https://github.com/FriendsOfGalaxy/galaxy-integration-origin/blob/master/src/uri_scheme_handler.py"""
+
+    if not IS_WINDOWS:
+        return None
+
+    def _get_path_from_cmd_template(cmd_template: str) -> str:
+        return cmd_template.replace("\"", "").partition("%")[0].strip()
+
+    try:
+        with winreg.OpenKey(
+            winreg.HKEY_CLASSES_ROOT, r"{}\shell\open\command".format(protocol)
+        ) as key:
+            executable_template = winreg.QueryValue(key, None)
+            return _get_path_from_cmd_template(executable_template)
+    except OSError:
+        return None
+
+
+class HumbleAppClient:
+    PROTOCOL = "humble"
+
+    @classmethod
+    def _open(cls, cmd: str, arg: str):
+        cmd = f"{cls.PROTOCOL}://{cmd}/{arg}"
+        logger.info(f"Opening {cmd}")
+        webbrowser.open(cmd)
+
+    def get_exe_path(self) -> t.Optional[str]:
+        return get_app_path_for_uri_handler(self.PROTOCOL)
+
+    def is_installed(self):
+        path = self.get_exe_path()
+        if path:
+            if os.path.exists(path):
+                return True
+            else:
+                logger.debug(f"{path} does not exists")
+        return False
+
+    def launch(self, game_id: GameMachineName):
+        self._open("launch", game_id)
+
+    def download(self, game_id: GameMachineName):
+        self._open("download", game_id)
+
+    def uninstall(self, game_id: GameMachineName):
+        self._open("uninstall", game_id)
+

src/library.py

@@ -1,6 +1,5 @@
 from functools import reduce
 from math import ceil
-import time
 import logging
 import asyncio
 from typing import Callable, Dict, List, Sequence, Set, Iterable, Any, Coroutine, NamedTuple, TypeVar, Generator

src/local/humbleapp_adapter.py

@@ -0,0 +1,102 @@
+import enum
+import pathlib
+import os
+import typing as t
+
+from galaxy.api.types import SubscriptionGame, LocalGame, LocalGameState
+from galaxy.api.consts import OSCompatibility
+from humbleapp.humbleapp import FileWatcher, GameStatus, TroveCategory, VaultGame, GameMachineName, HumbleAppConfig, parse_humble_app_config
+from humbleapp.humbleapp import HumbleAppClient as _HumbleAppClient
+
+
+class HumbleAppGameCategory(enum.Enum):
+    HUMBLE_GAMES_COLLECTION = "Humble Games Collection"
+    HUMBLE_VAULT = "Humble Vault"
+
+
+SUBSCRIPTION_NAME_TO_TROVE_CATEGORY = {
+    HumbleAppGameCategory.HUMBLE_GAMES_COLLECTION: TroveCategory.PREMIUM,
+    HumbleAppGameCategory.HUMBLE_VAULT: TroveCategory.GENERAL
+}
+
+
+def _vault_to_galaxy_subscription_game(vault_game: VaultGame) -> SubscriptionGame:
+    return SubscriptionGame(
+        game_title=vault_game.game_name,
+        game_id=vault_game.machine_name,
+        start_time=vault_game.date_added,
+        end_time=vault_game.date_ended
+    )
+
+
+def _vault_to_galaxy_local_game(vault_game: VaultGame) -> LocalGame:
+    local_game_state_map = {
+        GameStatus.AVAILABLE: LocalGameState.None_,
+        GameStatus.DOWNLOADED: LocalGameState.None_,
+        GameStatus.INSTALLED: LocalGameState.Installed,
+    }
+    return LocalGame(
+        vault_game.machine_name,
+        local_game_state_map[vault_game.status]
+    )
+
+
+class HumbleAppClient:
+    CONFIG_PATH = pathlib.PurePath(os.path.expandvars(r"%appdata%")) / "Humble App" / "config.json"
+
+    def __init__(self) -> None:
+        self._client = _HumbleAppClient()
+        self._config = FileWatcher(self.CONFIG_PATH)
+        self._games: t.Dict[GameMachineName, VaultGame] = {} 
+
+    def __contains__(self, game_id: str) -> bool:
+        return game_id in self._games
+
+    def get_subscription_games(self, subscription_name: HumbleAppGameCategory) -> t.List[SubscriptionGame]:
+        category = SUBSCRIPTION_NAME_TO_TROVE_CATEGORY[subscription_name]
+        return [
+            _vault_to_galaxy_subscription_game(vg)
+            for vg in self._games.values()
+            if vg.trove_category is category
+        ]
+
+    def get_local_games(self) -> t.List[LocalGame]:
+        return [
+            _vault_to_galaxy_local_game(vg)
+            for vg in self._games.values()
+        ]
+
+    @property
+    def os_compatibility(self):
+        return OSCompatibility.Windows
+
+    def refresh_game_list(self) -> None:
+        config = self._parse_config()
+        if config is not None:
+            self._games = {vg.machine_name: vg for vg in config.game_collection}
+
+    def _parse_config(self) -> t.Optional[HumbleAppConfig]:
+        if self._config.has_changed():
+            return parse_humble_app_config(self.CONFIG_PATH)
+        return None
+
+    def get_local_size(self, game_id: str) -> t.Optional[int]:
+        game = self._games.get(game_id)
+        if game is None:
+            return None
+        if game.file_path and not os.path.exists(game.file_path):
+            return 0
+        else:
+            return game.file_size
+
+    def is_installed(self) -> bool:
+        return self._client.is_installed()
+
+    def install(self,  game_id: str) -> None:
+        self._client.download(game_id)
+
+    def uninstall(self, game_id: str) -> None:
+        self._client.uninstall(game_id)
+
+    def launch(self, game_id: str) -> None:
+        self._client.launch(game_id)

src/model/subscription.py

@@ -221,7 +221,7 @@ def __init__(self, data: dict):
     @property
     def content_choices_made(self) -> t.List[str]:
         try:
-            return self._content_choices_made['initial']['choices_made']
+            return self._content_choices_made['initial']['choices_made']  # type: ignore
         except (KeyError, TypeError):
             return []