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

🎨 Black codestyle

Browse Source
This commit is contained in:
GeoffreyCoulaud
2023-05-05 03:04:44 +02:00
parent f432c41843
commit 8b7b23379f
4 changed files with 167 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
src/importer/decorators.py
View File

@@ -17,25 +17,38 @@ from pathlib import Path
from os import PathLike
from functools import wraps
def replaced_by_path(path: PathLike): # Decorator builder
"""Replace the method's returned path with the override if the override exists on disk"""
def decorator(original_function): # Built decorator (closure)
@wraps(original_function)
def wrapper(*args, **kwargs): # func's override
p = Path(path).expanduser()
if p.exists(): return p
else: return original_function(*args, **kwargs)
if p.exists():
return p
else:
return original_function(*args, **kwargs)
return wrapper
return decorator
def replaced_by_schema_key(key: str): # Decorator builder
"""Replace the method's returned path with the path pointed by the key if it exists on disk"""
def decorator(original_function): # Built decorator (closure)
@wraps(original_function)
def wrapper(*args, **kwargs): # func's override
schema = args[0].win.schema
try: override = schema.get_string(key)
except Exception: return original_function(*args, **kwargs)
else: return replaced_by_path(override)(*args, **kwargs)
try:
override = schema.get_string(key)
except Exception:
return original_function(*args, **kwargs)
else:
return replaced_by_path(override)(*args, **kwargs)
return wrapper
return decorator

127
src/importer/importer.py
View File

@@ -1,21 +1,45 @@
from threading import Thread, Lock
from gi.repository import Adw, Gtk, Gio
class Importer():
from .game import Game
from .steamgriddb import SGDBSave
# Display values
class Importer:
win = None
progressbar = None
import_statuspage = None
import_dialog = None
sources = None
# Importer values
count_total = 0
count_done = 0
sources = list()
progress_lock = None
counts = None
games_lock = None
games = None
def __init__(self, win) -> None:
self.games = set()
self.sources = list()
self.counts = dict()
self.games_lock = Lock()
self.progress_lock = Lock()
self.win = win
@property
def progress(self):
# Compute overall values
done = 0
total = 0
for source in self.sources:
done += self.counts[source.id]["done"]
total += self.counts[source.id]["total"]
# Compute progress
progress = 1
if total > 0:
progress = 1 - done / total
return progress
def create_dialog(self):
"""Create the import dialog"""
self.progressbar = Gtk.ProgressBar(margin_start=12, margin_end=12)
@@ -37,44 +61,83 @@ class Importer():
"""Close the import dialog"""
self.import_dialog.close()
def get_progress(self):
"""Get the current progression as a number between 0 and 1"""
progress = 1
if self.total_queue > 0:
progress = 1 - self.queue / self.total_queue
return progress
def update_progressbar(self):
"""Update the progress bar"""
progress = self.get_progress()
progress = self.progress()
self.progressbar.set_fraction(progress)
def add_source(self, source):
"""Add a source to import games from"""
self.sources.append(source)
self.counts[source.id] = {"done": 0, "total": 0}
def import_games(self):
"""Import games from the specified sources"""
self.create_dialog()
# TODO make that async, you doofus
# Every source does its job on the side, informing of the amount of work and when a game is done.
# At the end of the task, it returns the games.
# Idea 1 - Work stealing queue
# 1. Sources added to the queue
# 2. Worker A takes source X and advances it
# 3. Worker A puts back source X to the queue
# 4. Worker B takes source X, that has ended
# 5. Worker B doesn't add source X back to the queue
# Idea 2 - Gio.Task
# 1. A task is created for every source
# 2. Source X finishes
# 3. Importer adds the games
# Scan all sources
threads = []
for source in self.sources:
for game in source:
game.save()
t = Thread(
None,
self.__import_from_source,
args=tuple(
source,
),
)
threads.append(t)
t.start()
# Wait for all of them to finish
for t in threads:
t.join()
self.close_dialog()
def __import_from_source(self, *args, **kwargs):
"""Source import thread entry point"""
# TODO just get Game objects from the sources
source, *rest = args
iterator = source.__iter__()
for game_values in iterator:
game = Game(self.win, game_values)
self.games_lock.acquire()
self.games.add(game)
self.games_lock.release()
self.progress_lock.acquire()
self.counts[source.id]["total"] = len(iterator)
if not game.blacklisted:
self.counts[source.id]["done"] += 1
self.update_progressbar()
self.progress_lock.release()
# TODO remove after not needed
def save_game(self, values=None, cover_path=None):
if values:
game = Game(self.win, values)
if save_cover:
save_cover(self.win, game.game_id, resize_cover(self.win, cover_path))
self.games.add(game)
self.games_no += 1
if game.blacklisted:
self.games_no -= 1
self.queue -= 1
self.update_progressbar()
if self.queue == 0 and not self.blocker:
if self.games:
self.total_queue = len(self.games)
self.queue = len(self.games)
self.import_statuspage.set_title(_("Importing Covers…"))
self.update_progressbar()
SGDBSave(self.win, self.games, self)
else:
self.done()

18
src/importer/source.py
View File

@@ -42,17 +42,25 @@ class Source(Iterable):
"""The source's full name"""
s = self.name
if self.variant is not None:
s += " (%s)" % self.variant
s += f" ({self.variant})"
return s
@property
def id(self):
"""The source's identifier"""
s = self.name.lower()
if self.variant is not None:
s += f"_{self.variant.lower()}"
return s
@property
def game_id_format(self):
"""The string format used to construct game IDs"""
_format = self.name.lower()
f = self.name.lower()
if self.variant is not None:
_format += "_" + self.variant.lower()
_format += "_{game_id}_{game_internal_id}"
return _format
f += f"_{self.variant.lower()}"
f += "_{game_id}"
return f
@property
@abstractmethod

41
src/importer/sources/lutris_source.py
View File

@@ -1,21 +1,20 @@
from functools import cached_property
from sqlite3 import connect
from src.game2 import Game
from src.utils.save_cover import resize_cover, save_cover
from src.importer.source import Source, SourceIterator
from src.importer.decorators import replaced_by_schema_key, replaced_by_path
class LutrisSourceIterator(SourceIterator):
ignore_steam_games = False
ignore_steam_games = False # TODO get that value
db_connection = None
db_cursor = None
db_location = None
db_request = None
def __init__(self, ignore_steam_games) -> None:
def __init__(self, ignore_steam_games):
super().__init__()
self.ignore_steam_games = ignore_steam_games
self.db_connection = None
@@ -43,8 +42,8 @@ class LutrisSourceIterator(SourceIterator):
self.db_cursor = self.db_connection.execute(self.db_request)
self.state = self.States.READY
# Get next DB value
while True:
# Get next DB value
try:
row = self.db_cursor.__next__()
except StopIteration as e:
@@ -57,26 +56,38 @@ class LutrisSourceIterator(SourceIterator):
# Build basic game
values = {
"name" : row[1],
"hidden" : row[4],
"source" : self.source.full_name,
"game_id" : self.source.game_id_format.format(game_id=row[2]),
"hidden": row[4],
"name": row[1],
"source": f"{self.source.id}_{row[3]}",
"game_id": self.source.game_id_format.format(
game_id=row[2], game_internal_id=row[0]
),
"executable": self.source.executable_format.format(game_id=row[2]),
"developer" : None, # TODO get developer metadata on Lutris
"developer": None, # TODO get developer metadata on Lutris
}
# TODO Add official image
# TODO Add SGDB image
# Save official image
image_path = self.source.cache_location / "coverart" / f"{row[2]}.jpg"
if image_path.exists():
resized = resize_cover(self.win, image_path)
save_cover(self.win, values["game_id"], resized)
# Save SGDB
return values
class LutrisSource(Source):
class LutrisSource(Source):
name = "Lutris"
executable_format = "xdg-open lutris:rungameid/{game_id}"
location = None
cache_location = None
def __init__(self, win) -> None:
@property
def game_id_format(self):
return super().game_id_format + "_{game_internal_id}"
def __init__(self, win):
super().__init__(win)
def __iter__(self):