v0.2.4: send achievements info to GLX

Author: Mixaill

current_version.json

@@ -1,8 +1,8 @@
 {
-    "tag_name": "v0.2.3",
+    "tag_name": "v0.2.4",
     "assets": [
         {
-            "browser_download_url": "https://github.com/Mixaill/galaxy-integration-gw2/archive/v0.2.3.zip",
+            "browser_download_url": "https://github.com/Mixaill/galaxy-integration-gw2/archive/v0.2.4.zip",
             "name": "windows.zip"
         }
     ]

gw2_api.py

@@ -97,7 +97,10 @@ def do_GET(self):
 class GW2API(object):
 
     API_DOMAIN = 'https://api.guildwars2.com'
+
+    API_URL_ACHIEVEMENTS = '/v2/achievements'
     API_URL_ACCOUNT = '/v2/account'
+    API_URL_ACCOUNT_ACHIVEMENTS = '/v2/account/achievements'
 
     LOCALSERVER_HOST = '127.0.0.1'
     LOCALSERVER_PORT = 13338
@@ -133,6 +136,28 @@ def get_owned_games(self) -> List[str]:
     def get_account_age(self) -> int:
         return self._account_info['age']
 
+    def get_account_achievements(self) -> Dict:
+        result = dict()
+
+        (status, achievements_account) = self.__api_get_account_achievements(self._api_key)
+        if status == 200:
+            #select completed achievements
+            ids_to_request = list()
+            for achievement in achievements_account:
+                if achievement['done'] == True:
+                    ids_to_request.append(achievement['id'])
+                    result[achievement['id']] = None
+            
+            #get additional info
+            (status, achievements_info) = self.__api_get_achievements_info(ids_to_request)
+            if status == 200 or status == 206:
+                for achievement in achievements_info:
+                    result[achievement['id']] = achievement['name']
+
+            
+        return result
+
+
     #
     # Authorization server
     #
@@ -202,16 +227,27 @@ def do_auth_apikey(self, api_key : str) -> GW2AuthorizationResult:
         return GW2AuthorizationResult.FINISHED
 
 
-    def __api_get_account_info(self, api_key):
+
+    def __api_get_response(self, api_key, url, parameters):
         result = None
 
-        resp = requests.get(self.API_DOMAIN+self.API_URL_ACCOUNT, params={'access_token': api_key})
-        
+        resp = requests.get(self.API_DOMAIN+url, params=parameters)  
         try: 
             result = json.loads(resp.text)
         except Exception:
             if resp.status_code != 502:
-                logging.error('gw2api/__api_get_account_info: failed to parse response %s' % resp.text)
-
+                logging.error('gw2_api/__api_get_response: failed to parse response %s' % resp.text)
 
         return (resp.status_code, result)
+
+
+    def __api_get_account_info(self, api_key):
+        return self.__api_get_response(api_key, self.API_URL_ACCOUNT, {'access_token': api_key})
+
+
+    def __api_get_account_achievements(self, api_key):
+        return self.__api_get_response(api_key, self.API_URL_ACCOUNT_ACHIVEMENTS, {'access_token': api_key})
+
+
+    def __api_get_achievements_info(self, ids : List[int]):
+        return self.__api_get_response(self, self.API_URL_ACHIEVEMENTS, {'ids': ','.join(str(i) for i in ids)})

manifest.json

@@ -2,7 +2,7 @@
     "name": "Galaxy Guild Wars 2 plugin",
     "platform": "gw2",
     "guid": "637b1fd4-b1d0-4de8-9bf6-757e639807f3",
-    "version": "0.2.3",
+    "version": "0.2.4",
     "description": "Galaxy Guild Wars 2 plugin",
     "author": "Mikhail Paulyshka",
     "email": "[email protected]",

plugin.py

@@ -4,7 +4,7 @@
 import os
 import sys
 import time
-from typing import List
+from typing import Any, List
 
 #expand sys.path
 thirdparty = os.path.join(os.path.dirname(os.path.realpath(__file__)),'3rdparty\\')
@@ -25,14 +25,19 @@
 from galaxy.api.errors import BackendError, InvalidCredentials
 from galaxy.api.consts import Platform, LicenseType, LocalGameState
 from galaxy.api.plugin import Plugin, create_and_run_plugin
-from galaxy.api.types import Authentication, NextStep, Dlc, LicenseInfo, Game, GameTime, LocalGame
+from galaxy.api.types import Achievement, Authentication, NextStep, Dlc, LicenseInfo, Game, GameTime, LocalGame
 from galaxy.proc_tools import process_iter
 
 from gw2_api import GW2API
 import gw2_localgame
 
 class GuildWars2Plugin(Plugin):
 
+    GAME_ID = 'guild_wars_2'
+    GAME_NAME = 'Guild Wars 2'
+
+    SLEEP_CHECK_ACHIEVEMENTS = 1500
+
     SLEEP_CHECK_INSTANCES = 60
 
     SLEEP_CHECK_RUNNING = 5
@@ -43,9 +48,14 @@ def __init__(self, reader, writer, token):
         super().__init__(Platform(manifest['platform']), manifest['version'], reader, writer, token)
         self._gw2_api = GW2API()
         self._game_instances = None
+
+        self.__task_check_for_achievements = None
         self.__task_check_for_instances = None
         self._task_check_for_running  = None
+
         self._last_state = LocalGameState.None_
+        self.__imported_achievements = None
+
 
     async def authenticate(self, stored_credentials=None):
         if not stored_credentials:
@@ -72,6 +82,7 @@ async def authenticate(self, stored_credentials=None):
 
             return Authentication(self._gw2_api.get_account_id(), self._gw2_api.get_account_name())
 
+
     async def pass_login_credentials(self, step, credentials, cookies):
         self._gw2_api.auth_server_stop()
 
@@ -83,14 +94,16 @@ async def pass_login_credentials(self, step, credentials, cookies):
         self.store_credentials({'api_key': api_key})
         return Authentication(self._gw2_api.get_account_id(), self._gw2_api.get_account_name())
 
+
     async def get_local_games(self):
         self._game_instances = gw2_localgame.get_game_instances()
         if len(self._game_instances) == 0:
             self._last_state = LocalGameState.None_
             return []
 
         self._last_state = LocalGameState.Installed
-        return [ LocalGame(game_id='guild_wars_2', local_game_state = self._last_state)]
+        return [ LocalGame(game_id=self.GAME_ID, local_game_state = self._last_state) ]
+
 
     async def get_owned_games(self):
         free_to_play = False
@@ -116,28 +129,53 @@ async def get_owned_games(self):
         if free_to_play:
             license_type = LicenseType.FreeToPlay
 
-        return [ Game(game_id = 'guild_wars_2', game_title = 'Guild Wars 2', dlcs = dlcs, license_info = LicenseInfo(license_type = license_type)) ]
+        return [ Game(game_id = self.GAME_ID, game_title = self.GAME_NAME, dlcs = dlcs, license_info = LicenseInfo(license_type = license_type)) ]
 
 
     async def get_game_time(self, game_id, context):
+        if game_id != self.GAME_ID:
+            logging.warn('plugin/get_game_time: unknown game_id %s' % game_id)
+            return None
+
         time_played = None
         last_played_time = None
-        if game_id == 'guild_wars_2':
-            time_played = int(self._gw2_api.get_account_age() / 60)
-            last_played_time = self.persistent_cache.get('last_played')
+        
+        time_played = int(self._gw2_api.get_account_age() / 60)
+        last_played_time = self.persistent_cache.get('last_played')
 
         return GameTime(game_id = game_id, time_played = time_played, last_played_time = last_played_time)
 
 
     async def launch_game(self, game_id):
-        if game_id != 'guild_wars_2':
+        if game_id != self.GAME_ID:
+            logging.warn('plugin/launch_game: unknown game_id %s' % game_id)
             return
 
         self._game_instances[0].run_game()
 
 
     async def install_game(self, game_id):
-        pass
+        if game_id != self.GAME_ID:
+            logging.warn('plugin/install_game: unknown game_id %s' % game_id)
+            return
+
+
+    async def get_unlocked_achievements(self, game_id: str, context: Any) -> List[Achievement]:
+        result = list()
+
+        if game_id != self.GAME_ID:
+            logging.warn('plugin/get_unlocked_achievements: unknown game_id %s' % game_id)
+            return result
+
+        if not self.__imported_achievements:
+            self.__imported_achievements = list()
+
+        self.__imported_achievements.clear()
+        for key, value in self._gw2_api.get_account_achievements().items():
+            self.__imported_achievements.append(key)
+            result.append(Achievement(0, key, value))
+
+        return result
 
 
     def tick(self):
@@ -147,6 +185,19 @@ def tick(self):
         if not self.__task_check_for_instances or self.__task_check_for_instances.done():
             self.__task_check_for_instances = self.create_task(self.task_check_for_game_instances(), "task_check_for_instances")
 
+        if not self.__task_check_for_achievements or self.__task_check_for_achievements.done():
+            self.__task_check_for_achievements = self.create_task(self.task_check_for_achievements(), "task_check_for_achievements")
+
+
+    async def task_check_for_achievements(self):
+        if self.__imported_achievements:
+            for key, value in self._gw2_api.get_account_achievements().items():
+                if key not in self.__imported_achievements:
+                    self.__imported_achievements.append(key)
+                    self.unlock_achievement(self.GAME_ID, Achievement(0, key, value))
+
+        await asyncio.sleep(self.SLEEP_CHECK_ACHIEVEMENTS)
+
 
     async def task_check_for_game_instances(self):
         self._game_instances = gw2_localgame.get_game_instances()