skullbite/cartridges
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

🎨 Moved things to managers

Browse Source
This commit is contained in:
GeoffreyCoulaud
2023-05-23 16:49:37 +02:00
parent abe41635fd
commit 95524563bb
5 changed files with 34 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files

63
src/importer/importer.py
View File

@@ -22,33 +22,21 @@ class Importer:
n_games_added = 0
n_source_tasks_created = 0
n_source_tasks_done = 0
n_sgdb_tasks_created = 0
n_sgdb_tasks_done = 0
sgdb_cancellable = None
sgdb_error = None
def __init__(self):
self.sources = set()
@property
def n_tasks_created(self):
return self.n_source_tasks_created + self.n_sgdb_tasks_created
@property
def n_tasks_done(self):
return self.n_source_tasks_done + self.n_sgdb_tasks_done
@property
def progress(self):
try:
progress = self.n_tasks_done / self.n_tasks_created
progress = self.n_source_tasks_done / self.n_source_tasks_created
except ZeroDivisionError:
progress = 1
return progress
@property
def finished(self):
return self.n_tasks_created == self.n_tasks_done
return self.n_source_tasks_created == self.n_source_tasks_done
def add_source(self, source):
self.sources.add(source)
@@ -123,15 +111,6 @@ class Importer:
shared.store.add_game(game)
self.n_games_added += 1
# Start sgdb lookup for game
# TODO move to its own manager
task = Task.new(
None, self.sgdb_cancellable, self.sgdb_task_callback, (game,)
)
self.n_sgdb_tasks_created += 1
task.set_task_data((game,))
task.run_in_thread(self.sgdb_task_thread_func)
def source_task_callback(self, _obj, _result, data):
"""Source import callback"""
source, *_rest = data
@@ -141,38 +120,14 @@ class Importer:
if self.finished:
self.import_callback()
def sgdb_task_thread_func(self, _task, _obj, data, cancellable):
"""SGDB query code"""
game, *_rest = data
game.set_loading(1)
sgdb = SGDBHelper()
try:
sgdb.conditionaly_update_cover(game)
except SGDBAuthError as error:
cancellable.cancel()
self.sgdb_error = error
except (HTTPError, SGDBError) as _error:
# TODO handle other SGDB errors
pass
def sgdb_task_callback(self, _obj, _result, data):
"""SGDB query callback"""
game, *_rest = data
logging.debug("SGDB import done for game %s", game.name)
game.set_loading(-1)
self.n_sgdb_tasks_done += 1
self.update_progressbar()
if self.finished:
self.import_callback()
def import_callback(self):
"""Callback called when importing has finished"""
logging.info("Import done")
self.import_dialog.close()
# TODO replace by summary if necessary
self.create_summary_toast()
# TODO create a summary of errors/warnings/tips popup (eg. SGDB, Steam libraries)
if self.sgdb_error is not None:
self.create_sgdb_error_dialog()
# Get the error data from shared.store.managers)
def create_summary_toast(self):
"""N games imported toast"""
@@ -199,16 +154,6 @@ class Importer:
shared.win.toast_overlay.add_toast(toast)
def create_sgdb_error_dialog(self):
"""SGDB error dialog"""
create_dialog(
shared.win,
_("Couldn't Connect to SteamGridDB"),
str(self.sgdb_error),
"open_preferences",
_("Preferences"),
).connect("response", self.dialog_response_callback, "sgdb")
def open_preferences(self, page=None, expander_row=None):
shared.win.get_application().on_preferences_action(
page_name=page, expander_row=expander_row

12
src/importer/sources/steam_source.py
View File

@@ -96,18 +96,6 @@ class SteamSourceIterator(SourceIterator):
if cover_path.is_file():
save_cover(game.game_id, resize_cover(cover_path))
# Get online metadata
# TODO move to its own manager
try:
online_data = steam.get_api_data(appid=appid)
except (HTTPError, JSONDecodeError):
pass
except (SteamNotAGameError, SteamGameNotFoundError):
game.update_values({"blacklisted": True})
else:
game.update_values(online_data)
return game
class SteamSource(Source):
name = "Steam"

6
src/store/display_manager.py
View File

@@ -1,11 +1,15 @@
import src.shared as shared
from src.store.manager import Manager
from src.game import Game
from src.store.manager import Manager
from src.store.sgdb_manager import SGDBManager
from src.store.steam_api_manager import SteamAPIManager
class DisplayManager(Manager):
"""Manager in charge of adding a game to the UI"""
run_after = set((SteamAPIManager, SGDBManager))
def run(self, game: Game) -> None:
# TODO decouple a game from its widget
shared.win.games[game.game_id] = game

6
src/store/file_manager.py
View File

@@ -1,10 +1,14 @@
from src.store.manager import Manager
from src.game import Game
from src.store.manager import Manager
from src.store.sgdb_manager import SGDBManager
from src.store.steam_api_manager import SteamAPIManager
class FileManager(Manager):
"""Manager in charge of saving a game to a file"""
run_after = set((SteamAPIManager, SGDBManager))
def run(self, game: Game) -> None:
# TODO make game.save (disk) not trigger game.update (UI)
game.save()

24
src/store/steam_api_manager.py
View File

@@ -1,11 +1,27 @@
from src.store.manager import Manager
from requests import HTTPError, JSONDecodeError
from src.game import Game
from src.utils.steam import SteamHelper
from src.store.manager import Manager
from src.utils.steam import SteamGameNotFoundError, SteamHelper, SteamNotAGameError
class SteamAPIManager(Manager):
"""Manager in charge of completing a game's data from the Steam API"""
def run(self, game: Game) -> None:
# TODO
pass
# Skip non-steam games
if not game.source.startswith("steam_"):
return
# Get online metadata
appid = str(game.game_id).split("_")[-1]
steam = SteamHelper()
try:
online_data = steam.get_api_data(appid=appid)
except (HTTPError, JSONDecodeError) as error:
# On minor error, just report it
self.report_error(error)
except (SteamNotAGameError, SteamGameNotFoundError):
game.update_values({"blacklisted": True})
else:
game.update_values(online_data)