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

Merge branch 'main' into retroarch-make-exec

Browse Source
This commit is contained in:
GeoffreyCoulaud
2023-08-17 14:09:03 +02:00
parent 195f7dbb7e 191beed12e
commit a39303603c
38 changed files with 763 additions and 535 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
data/gtk/details-window.blp
View File

@@ -116,6 +116,7 @@ template $DetailsWindow : Adw.Window {
tooltip-text: _("More Info");
popover: Popover exec_info_popover {
focusable: true;
Label exec_info_label {
use-markup: true;
@@ -127,7 +128,6 @@ template $DetailsWindow : Adw.Window {
margin-bottom: 6;
margin-start: 6;
margin-end: 6;
selectable: true;
}
};

13
data/gtk/preferences.blp
View File

@@ -85,6 +85,19 @@ template $PreferencesWindow : Adw.PreferencesWindow {
title: _("Import");
icon-name: "document-save-symbolic";
Adw.PreferencesGroup import_behavior_group {
title: _("Behavior");
Adw.ActionRow {
title: _("Remove Uninstalled Games");
activatable-widget: remove_missing_switch;
Switch remove_missing_switch {
valign: center;
}
}
}
Adw.PreferencesGroup sources_group {
title: _("Sources");

3
data/hu.kramo.Cartridges.gschema.xml.in
View File

@@ -10,6 +10,9 @@
<key name="high-quality-images" type="b">
<default>false</default>
</key>
<key name="remove-missing" type="b">
<default>true</default>
</key>
<key name="steam" type="b">
<default>true</default>
</key>

1
po/POTFILES
View File

@@ -15,6 +15,7 @@ src/game.py
src/preferences.py
src/utils/create_dialog.py
src/importer/importer.py
src/importer/sources/source.py
src/importer/sources/location.py
src/store/managers/sgdb_manager.py

127
po/cartridges.pot
View File

@@ -8,13 +8,13 @@ msgid ""
msgstr ""
"Project-Id-Version: Cartridges\n"
"Report-Msgid-Bugs-To: \n"
"POT-Creation-Date: 2023-08-13 15:20+0200\n"
"POT-Creation-Date: 2023-08-16 11:06+0200\n"
"PO-Revision-Date: YEAR-MO-DA HO:MI+ZONE\n"
"Last-Translator: FULL NAME <EMAIL@ADDRESS>\n"
"Language-Team: LANGUAGE <LL@li.org>\n"
"Language: \n"
"MIME-Version: 1.0\n"
"Content-Type: text/plain; charset=CHARSET\n"
"Content-Type: text/plain; charset=UTF-8\n"
"Content-Transfer-Encoding: 8bit\n"
#: data/hu.kramo.Cartridges.desktop.in:3
@@ -59,7 +59,8 @@ msgid "Game Details"
msgstr ""
#: data/hu.kramo.Cartridges.metainfo.xml.in:42 data/gtk/window.blp:418
#: src/details_window.py:241
#: src/details_window.py:241 src/importer/importer.py:292
#: src/importer/importer.py:342
msgid "Preferences"
msgstr ""
@@ -129,7 +130,8 @@ msgstr ""
msgid "Shortcuts"
msgstr ""
#: data/gtk/help-overlay.blp:34 src/game.py:102 src/preferences.py:118
#: data/gtk/help-overlay.blp:34 src/game.py:103 src/preferences.py:120
#: src/importer/importer.py:366
msgid "Undo"
msgstr ""
@@ -157,7 +159,8 @@ msgstr ""
msgid "Remove game"
msgstr ""
#: data/gtk/preferences.blp:13 data/gtk/preferences.blp:291
#: data/gtk/preferences.blp:13 data/gtk/preferences.blp:89
#: data/gtk/preferences.blp:304
msgid "Behavior"
msgstr ""
@@ -197,106 +200,110 @@ msgstr ""
msgid "Import"
msgstr ""
#: data/gtk/preferences.blp:89
#: data/gtk/preferences.blp:92
msgid "Remove Uninstalled Games"
msgstr ""
#: data/gtk/preferences.blp:102
msgid "Sources"
msgstr ""
#: data/gtk/preferences.blp:92
#: data/gtk/preferences.blp:105
msgid "Steam"
msgstr ""
#: data/gtk/preferences.blp:96 data/gtk/preferences.blp:110
#: data/gtk/preferences.blp:151 data/gtk/preferences.blp:201
#: data/gtk/preferences.blp:215 data/gtk/preferences.blp:229
#: data/gtk/preferences.blp:243 data/gtk/preferences.blp:257
#: data/gtk/preferences.blp:109 data/gtk/preferences.blp:123
#: data/gtk/preferences.blp:164 data/gtk/preferences.blp:214
#: data/gtk/preferences.blp:228 data/gtk/preferences.blp:242
#: data/gtk/preferences.blp:256 data/gtk/preferences.blp:270
msgid "Install Location"
msgstr ""
#: data/gtk/preferences.blp:106
#: data/gtk/preferences.blp:119
msgid "Lutris"
msgstr ""
#: data/gtk/preferences.blp:119
#: data/gtk/preferences.blp:132
msgid "Cache Location"
msgstr ""
#: data/gtk/preferences.blp:128
#: data/gtk/preferences.blp:141
msgid "Import Steam Games"
msgstr ""
#: data/gtk/preferences.blp:137
#: data/gtk/preferences.blp:150
msgid "Import Flatpak Games"
msgstr ""
#: data/gtk/preferences.blp:147
#: data/gtk/preferences.blp:160
msgid "Heroic"
msgstr ""
#: data/gtk/preferences.blp:160
#: data/gtk/preferences.blp:173
msgid "Import Epic Games"
msgstr ""
#: data/gtk/preferences.blp:169
#: data/gtk/preferences.blp:182
msgid "Import GOG Games"
msgstr ""
#: data/gtk/preferences.blp:178
#: data/gtk/preferences.blp:191
msgid "Import Amazon Games"
msgstr ""
#: data/gtk/preferences.blp:187
#: data/gtk/preferences.blp:200
msgid "Import Sideloaded Games"
msgstr ""
#: data/gtk/preferences.blp:197
#: data/gtk/preferences.blp:210
msgid "Bottles"
msgstr ""
#: data/gtk/preferences.blp:211
#: data/gtk/preferences.blp:224
msgid "itch"
msgstr ""
#: data/gtk/preferences.blp:225
#: data/gtk/preferences.blp:238
msgid "Legendary"
msgstr ""
#: data/gtk/preferences.blp:239
#: data/gtk/preferences.blp:252
msgid "RetroArch"
msgstr ""
#: data/gtk/preferences.blp:253
#: data/gtk/preferences.blp:266
msgid "Flatpak"
msgstr ""
#: data/gtk/preferences.blp:266
#: data/gtk/preferences.blp:279
msgid "Import Game Launchers"
msgstr ""
#: data/gtk/preferences.blp:279
#: data/gtk/preferences.blp:292
msgid "SteamGridDB"
msgstr ""
#: data/gtk/preferences.blp:283
#: data/gtk/preferences.blp:296
msgid "Authentication"
msgstr ""
#: data/gtk/preferences.blp:286
#: data/gtk/preferences.blp:299
msgid "API Key"
msgstr ""
#: data/gtk/preferences.blp:294
#: data/gtk/preferences.blp:307
msgid "Use SteamGridDB"
msgstr ""
#: data/gtk/preferences.blp:295
#: data/gtk/preferences.blp:308
msgid "Download images when adding or importing games"
msgstr ""
#: data/gtk/preferences.blp:304
#: data/gtk/preferences.blp:317
msgid "Prefer Over Official Images"
msgstr ""
#: data/gtk/preferences.blp:313
#: data/gtk/preferences.blp:326
msgid "Prefer Animated Images"
msgstr ""
@@ -473,52 +480,84 @@ msgid "Couldn't Apply Preferences"
msgstr ""
#. The variable is the title of the game
#: src/game.py:138
#: src/game.py:139
msgid "{} launched"
msgstr ""
#. The variable is the title of the game
#: src/game.py:152
#: src/game.py:153
msgid "{} hidden"
msgstr ""
#: src/game.py:152
#: src/game.py:153
msgid "{} unhidden"
msgstr ""
#: src/game.py:169
#. The variable is the title of the game
#. The variable is the number of games removed
#: src/game.py:170 src/importer/importer.py:363
msgid "{} removed"
msgstr ""
#: src/preferences.py:117
#: src/preferences.py:119
msgid "All games removed"
msgstr ""
#: src/preferences.py:166
#: src/preferences.py:168
msgid ""
"An API key is required to use SteamGridDB. You can generate one {}here{}."
msgstr ""
#: src/preferences.py:295
#: src/preferences.py:294
msgid "Installation Not Found"
msgstr ""
#: src/preferences.py:297
#: src/preferences.py:296
msgid "Select a valid directory."
msgstr ""
#: src/preferences.py:363
#: src/preferences.py:351
msgid "Invalid Directory"
msgstr ""
#: src/preferences.py:369
#: src/preferences.py:357
msgid "Set Location"
msgstr ""
#: src/utils/create_dialog.py:25
#: src/utils/create_dialog.py:25 src/importer/importer.py:291
msgid "Dismiss"
msgstr ""
#: src/importer/importer.py:128
msgid "Importing Games…"
msgstr ""
#: src/importer/importer.py:290
msgid "Warning"
msgstr ""
#: src/importer/importer.py:311
msgid "The following errors occured during import:"
msgstr ""
#: src/importer/importer.py:339
msgid "No new games found"
msgstr ""
#: src/importer/importer.py:351
msgid "1 game imported"
msgstr ""
#. The variable is the number of games
#: src/importer/importer.py:355
msgid "{} games imported"
msgstr ""
#. A single game removed
#: src/importer/importer.py:359
msgid "1 removed"
msgstr ""
#. The variable is the name of the source
#: src/importer/sources/location.py:33
msgid "Select the {} cache directory."

51
src/details_window.py
View File

@@ -19,6 +19,7 @@
import os
from time import time
from typing import Any, Optional
from gi.repository import Adw, Gio, GLib, Gtk
from PIL import Image
@@ -52,16 +53,15 @@ class DetailsWindow(Adw.Window):
apply_button = Gtk.Template.Child()
cover_changed = False
cover_changed: bool = False
def __init__(self, game=None, **kwargs):
def __init__(self, game: Optional[Game] = None, **kwargs: Any):
super().__init__(**kwargs)
self.win = shared.win
self.game = game
self.game_cover = GameCover({self.cover})
self.game: Game = game
self.game_cover: GameCover = GameCover({self.cover})
self.set_transient_for(self.win)
self.set_transient_for(shared.win)
if self.game:
self.set_title(_("Game Details"))
@@ -114,10 +114,17 @@ class DetailsWindow(Adw.Window):
self.exec_info_label.set_label(exec_info_text)
def clear_info_selection(*_args):
self.exec_info_label.select_region(-1, -1)
self.exec_info_popover.update_property(
(Gtk.AccessibleProperty.LABEL,),
(
exec_info_text.replace("<tt>", "").replace("</tt>", ""),
), # Remove formatting, else the screen reader reads it
)
self.exec_info_popover.connect("show", clear_info_selection)
def set_exec_info_a11y_label(*_args: Any) -> None:
self.set_focus(self.exec_info_popover)
self.exec_info_popover.connect("show", set_exec_info_a11y_label)
self.cover_button_delete.connect("clicked", self.delete_pixbuf)
self.cover_button_edit.connect("clicked", self.choose_cover)
@@ -130,13 +137,13 @@ class DetailsWindow(Adw.Window):
self.set_focus(self.name)
self.present()
def delete_pixbuf(self, *_args):
def delete_pixbuf(self, *_args: Any) -> None:
self.game_cover.new_cover()
self.cover_button_delete_revealer.set_reveal_child(False)
self.cover_changed = True
def apply_preferences(self, *_args):
def apply_preferences(self, *_args: Any) -> None:
final_name = self.name.get_text()
final_developer = self.developer.get_text()
final_executable = self.executable.get_text()
@@ -196,10 +203,10 @@ class DetailsWindow(Adw.Window):
self.game.developer = final_developer or None
self.game.executable = final_executable
if self.game.game_id in self.win.game_covers.keys():
self.win.game_covers[self.game.game_id].animation = None
if self.game.game_id in shared.win.game_covers.keys():
shared.win.game_covers[self.game.game_id].animation = None
self.win.game_covers[self.game.game_id] = self.game_cover
shared.win.game_covers[self.game.game_id] = self.game_cover
if self.cover_changed:
save_cover(
@@ -222,9 +229,9 @@ class DetailsWindow(Adw.Window):
self.game_cover.pictures.remove(self.cover)
self.close()
self.win.show_details_view(self.game)
shared.win.show_details_view(self.game)
def update_cover_callback(self, manager: SGDBManager):
def update_cover_callback(self, manager: SGDBManager) -> None:
# Set the game as not loading
self.game.set_loading(-1)
self.game.update()
@@ -241,25 +248,25 @@ class DetailsWindow(Adw.Window):
_("Preferences"),
).connect("response", self.update_cover_error_response)
def update_cover_error_response(self, _widget, response):
def update_cover_error_response(self, _widget: Any, response: str) -> None:
if response == "open_preferences":
shared.win.get_application().on_preferences_action(page_name="sgdb")
def focus_executable(self, *_args):
def focus_executable(self, *_args: Any) -> None:
self.set_focus(self.executable)
def toggle_loading(self):
def toggle_loading(self) -> None:
self.apply_button.set_sensitive(not self.apply_button.get_sensitive())
self.spinner.set_spinning(not self.spinner.get_spinning())
self.cover_overlay.set_opacity(not self.cover_overlay.get_opacity())
def set_cover(self, _source, result, *_args):
def set_cover(self, _source: Any, result: Gio.Task, *_args: Any) -> None:
try:
path = self.file_dialog.open_finish(result).get_path()
except GLib.GError:
return
def resize():
def resize() -> None:
if cover := resize_cover(path):
self.game_cover.new_cover(cover)
self.cover_button_delete_revealer.set_reveal_child(True)
@@ -269,5 +276,5 @@ class DetailsWindow(Adw.Window):
self.toggle_loading()
GLib.Thread.new(None, resize)
def choose_cover(self, *_args):
def choose_cover(self, *_args: Any) -> None:
self.file_dialog.open(self, None, self.set_cover)

6
src/errors/error_producer.py
View File

@@ -8,14 +8,14 @@ class ErrorProducer:
Specifies the report_error and collect_errors methods in a thread-safe manner.
"""
errors: list[Exception] = None
errors_lock: Lock = None
errors: list[Exception]
errors_lock: Lock
def __init__(self) -> None:
self.errors = []
self.errors_lock = Lock()
def report_error(self, error: Exception):
def report_error(self, error: Exception) -> None:
"""Report an error"""
with self.errors_lock:
self.errors.append(error)

6
src/errors/friendly_error.py
View File

@@ -1,4 +1,4 @@
from typing import Iterable
from typing import Iterable, Optional
class FriendlyError(Exception):
@@ -27,8 +27,8 @@ class FriendlyError(Exception):
self,
title: str,
subtitle: str,
title_args: Iterable[str] = None,
subtitle_args: Iterable[str] = None,
title_args: Optional[Iterable[str]] = None,
subtitle_args: Optional[Iterable[str]] = None,
) -> None:
"""Create a friendly error

78
src/game.py
View File

@@ -23,10 +23,12 @@ import shlex
import subprocess
from pathlib import Path
from time import time
from typing import Any, Optional
from gi.repository import Adw, GLib, GObject, Gtk
from src import shared
from src.game_cover import GameCover
# pylint: disable=too-many-instance-attributes
@@ -45,23 +47,23 @@ class Game(Gtk.Box):
game_options = Gtk.Template.Child()
hidden_game_options = Gtk.Template.Child()
loading = 0
filtered = False
loading: int = 0
filtered: bool = False
added = None
executable = None
game_id = None
source = None
hidden = False
last_played = 0
name = None
developer = None
removed = False
blacklisted = False
game_cover = None
version = 0
added: int
executable: str
game_id: str
source: str
hidden: bool = False
last_played: int = 0
name: str
developer: Optional[str] = None
removed: bool = False
blacklisted: bool = False
game_cover: GameCover = None
version: int = 0
def __init__(self, data, **kwargs):
def __init__(self, data: dict[str, Any], **kwargs: Any) -> None:
super().__init__(**kwargs)
self.win = shared.win
@@ -69,6 +71,7 @@ class Game(Gtk.Box):
self.version = shared.SPEC_VERSION
self.update_values(data)
self.base_source = self.source.split("_")[0]
self.set_play_icon()
@@ -81,20 +84,20 @@ class Game(Gtk.Box):
shared.schema.connect("changed", self.schema_changed)
def update_values(self, data):
def update_values(self, data: dict[str, Any]) -> None:
for key, value in data.items():
# Convert executables to strings
if key == "executable" and isinstance(value, list):
value = shlex.join(value)
setattr(self, key, value)
def update(self):
def update(self) -> None:
self.emit("update-ready", {})
def save(self):
def save(self) -> None:
self.emit("save-ready", {})
def create_toast(self, title, action=None):
def create_toast(self, title: str, action: Optional[str] = None) -> None:
toast = Adw.Toast.new(title.format(self.name))
toast.set_priority(Adw.ToastPriority.HIGH)
@@ -110,7 +113,7 @@ class Game(Gtk.Box):
self.win.toast_overlay.add_toast(toast)
def launch(self):
def launch(self) -> None:
self.last_played = int(time())
self.save()
self.update()
@@ -125,10 +128,10 @@ class Game(Gtk.Box):
# pylint: disable=consider-using-with
subprocess.Popen(
args,
cwd=Path.home(),
cwd=shared.home,
shell=True,
start_new_session=True,
creationflags=subprocess.CREATE_NEW_PROCESS_GROUP if os.name == "nt" else 0,
creationflags=subprocess.CREATE_NEW_PROCESS_GROUP if os.name == "nt" else 0, # type: ignore
)
if shared.schema.get_boolean("exit-after-launch"):
@@ -137,7 +140,7 @@ class Game(Gtk.Box):
# The variable is the title of the game
self.create_toast(_("{} launched"))
def toggle_hidden(self, toast=True):
def toggle_hidden(self, toast: bool = True) -> None:
self.hidden = not self.hidden
self.save()
@@ -155,7 +158,7 @@ class Game(Gtk.Box):
"hide",
)
def remove_game(self):
def remove_game(self) -> None:
# Add "removed=True" to the game properties so it can be deleted on next init
self.removed = True
self.save()
@@ -164,55 +167,58 @@ class Game(Gtk.Box):
if self.win.stack.get_visible_child() == self.win.details_view:
self.win.on_go_back_action()
# The variable is the title of the game
self.create_toast(
_("{} removed").format(GLib.markup_escape_text(self.name)), "remove"
# The variable is the title of the game
_("{} removed").format(GLib.markup_escape_text(self.name)),
"remove",
)
def set_loading(self, state):
def set_loading(self, state: int) -> None:
self.loading += state
loading = self.loading > 0
self.cover.set_opacity(int(not loading))
self.spinner.set_spinning(loading)
def get_cover_path(self):
def get_cover_path(self) -> Optional[Path]:
cover_path = shared.covers_dir / f"{self.game_id}.gif"
if cover_path.is_file():
return cover_path
return cover_path # type: ignore
cover_path = shared.covers_dir / f"{self.game_id}.tiff"
if cover_path.is_file():
return cover_path
return cover_path # type: ignore
return None
def toggle_play(self, _widget, _prop1, _prop2, state=True):
def toggle_play(
self, _widget: Any, _prop1: Any, _prop2: Any, state: bool = True
) -> None:
if not self.menu_button.get_active():
self.play_revealer.set_reveal_child(not state)
self.menu_revealer.set_reveal_child(not state)
def main_button_clicked(self, _widget, button):
def main_button_clicked(self, _widget: Any, button: bool) -> None:
if shared.schema.get_boolean("cover-launches-game") ^ button:
self.launch()
else:
self.win.show_details_view(self)
def set_play_icon(self):
def set_play_icon(self) -> None:
self.play_button.set_icon_name(
"help-about-symbolic"
if shared.schema.get_boolean("cover-launches-game")
else "media-playback-start-symbolic"
)
def schema_changed(self, _settings, key):
def schema_changed(self, _settings: Any, key: str) -> None:
if key == "cover-launches-game":
self.set_play_icon()
@GObject.Signal(name="update-ready", arg_types=[object])
def update_ready(self, _additional_data) -> None:
def update_ready(self, _additional_data): # type: ignore
"""Signal emitted when the game needs updating"""
@GObject.Signal(name="save-ready", arg_types=[object])
def save_ready(self, _additional_data) -> None:
def save_ready(self, _additional_data): # type: ignore
"""Signal emitted when the game needs saving"""

45
src/game_cover.py
View File

@@ -17,19 +17,22 @@
#
# SPDX-License-Identifier: GPL-3.0-or-later
from gi.repository import Gdk, GdkPixbuf, Gio, GLib
from pathlib import Path
from typing import Any, Callable, Optional
from gi.repository import Gdk, GdkPixbuf, Gio, GLib, Gtk
from PIL import Image, ImageFilter, ImageStat
from src import shared
class GameCover:
texture = None
blurred = None
luminance = None
path = None
animation = None
anim_iter = None
texture: Optional[Gdk.Texture]
blurred: Optional[Gdk.Texture]
luminance: Optional[tuple[float, float]]
path: Optional[Path]
animation: Optional[GdkPixbuf.PixbufAnimation]
anim_iter: Optional[GdkPixbuf.PixbufAnimationIter]
placeholder = Gdk.Texture.new_from_resource(
shared.PREFIX + "/library_placeholder.svg"
@@ -38,21 +41,21 @@ class GameCover:
shared.PREFIX + "/library_placeholder_small.svg"
)
def __init__(self, pictures, path=None):
def __init__(self, pictures: set[Gtk.Picture], path: Optional[Path] = None) -> None:
self.pictures = pictures
self.new_cover(path)
# Wrap the function in another one as Gio.Task.run_in_thread does not allow for passing args
def create_func(self, path):
def create_func(self, path: Optional[Path]) -> Callable:
self.animation = GdkPixbuf.PixbufAnimation.new_from_file(str(path))
self.anim_iter = self.animation.get_iter()
def wrapper(task, *_args):
def wrapper(task: Gio.Task, *_args: Any) -> None:
self.update_animation((task, self.animation))
return wrapper
def new_cover(self, path=None):
def new_cover(self, path: Optional[Path] = None) -> None:
self.animation = None
self.texture = None
self.blurred = None
@@ -69,14 +72,14 @@ class GameCover:
if not self.animation:
self.set_texture(self.texture)
def get_texture(self):
def get_texture(self) -> Gdk.Texture:
return (
Gdk.Texture.new_for_pixbuf(self.animation.get_static_image())
if self.animation
else self.texture
)
def get_blurred(self):
def get_blurred(self) -> Gdk.Texture:
if not self.blurred:
if self.path:
with Image.open(self.path) as image:
@@ -94,24 +97,24 @@ class GameCover:
stat = ImageStat.Stat(image.convert("L"))
# Luminance values for light and dark mode
self.luminance = [
self.luminance = (
min((stat.mean[0] + stat.extrema[0][0]) / 510, 0.7),
max((stat.mean[0] + stat.extrema[0][1]) / 510, 0.3),
]
)
else:
self.blurred = self.placeholder_small
self.luminance = (0.3, 0.5)
return self.blurred
def add_picture(self, picture):
def add_picture(self, picture: Gtk.Picture) -> None:
self.pictures.add(picture)
if not self.animation:
self.set_texture(self.texture)
else:
self.update_animation((self.task, self.animation))
def set_texture(self, texture):
def set_texture(self, texture: Gdk.Texture) -> None:
self.pictures.discard(
picture for picture in self.pictures if not picture.is_visible()
)
@@ -121,13 +124,13 @@ class GameCover:
for picture in self.pictures:
picture.set_paintable(texture or self.placeholder)
def update_animation(self, data):
def update_animation(self, data: GdkPixbuf.PixbufAnimation) -> None:
if self.animation == data[1]:
self.anim_iter.advance()
self.anim_iter.advance() # type: ignore
self.set_texture(Gdk.Texture.new_for_pixbuf(self.anim_iter.get_pixbuf()))
self.set_texture(Gdk.Texture.new_for_pixbuf(self.anim_iter.get_pixbuf())) # type: ignore
delay_time = self.anim_iter.get_delay_time()
delay_time = self.anim_iter.get_delay_time() # type: ignore
GLib.timeout_add(
20 if delay_time < 20 else delay_time,
self.update_animation,

156
src/importer/importer.py
View File

@@ -19,6 +19,7 @@
# SPDX-License-Identifier: GPL-3.0-or-later
import logging
from typing import Any, Optional
from gi.repository import Adw, GLib, Gtk
@@ -37,41 +38,49 @@ from src.utils.task import Task
class Importer(ErrorProducer):
"""A class in charge of scanning sources for games"""
progressbar = None
import_statuspage = None
import_dialog = None
summary_toast = None
progressbar: Gtk.ProgressBar
import_statuspage: Adw.StatusPage
import_dialog: Adw.MessageDialog
summary_toast: Adw.Toast
sources: set[Source] = None
sources: set[Source]
n_source_tasks_created: int = 0
n_source_tasks_done: int = 0
n_pipelines_done: int = 0
game_pipelines: set[Pipeline] = None
game_pipelines: set[Pipeline]
def __init__(self):
removed_game_ids: set[str] = set()
imported_game_ids: set[str] = set()
def __init__(self) -> None:
super().__init__()
# TODO: make this stateful
shared.store.new_game_ids = set()
shared.store.duplicate_game_ids = set()
self.game_pipelines = set()
self.sources = set()
@property
def n_games_added(self):
def n_games_added(self) -> int:
return sum(
1 if not (pipeline.game.blacklisted or pipeline.game.removed) else 0
for pipeline in self.game_pipelines
)
@property
def pipelines_progress(self):
def pipelines_progress(self) -> float:
progress = sum(pipeline.progress for pipeline in self.game_pipelines)
try:
progress = progress / len(self.game_pipelines)
except ZeroDivisionError:
progress = 0
return progress
return progress # type: ignore
@property
def sources_progress(self):
def sources_progress(self) -> float:
try:
progress = self.n_source_tasks_done / self.n_source_tasks_created
except ZeroDivisionError:
@@ -79,16 +88,16 @@ class Importer(ErrorProducer):
return progress
@property
def finished(self):
def finished(self) -> bool:
return (
self.n_source_tasks_created == self.n_source_tasks_done
and len(self.game_pipelines) == self.n_pipelines_done
)
def add_source(self, source):
def add_source(self, source: Source) -> None:
self.sources.add(source)
def run(self):
def run(self) -> None:
"""Use several Gio.Task to import games from added sources"""
shared.win.get_application().lookup_action("import").set_enabled(False)
@@ -113,7 +122,7 @@ class Importer(ErrorProducer):
self.progress_changed_callback()
def create_dialog(self):
def create_dialog(self) -> None:
"""Create the import dialog"""
self.progressbar = Gtk.ProgressBar(margin_start=12, margin_end=12)
self.import_statuspage = Adw.StatusPage(
@@ -130,7 +139,9 @@ class Importer(ErrorProducer):
)
self.import_dialog.present()
def source_task_thread_func(self, _task, _obj, data, _cancellable):
def source_task_thread_func(
self, _task: Any, _obj: Any, data: tuple, _cancellable: Any
) -> None:
"""Source import task code"""
source: Source
@@ -184,27 +195,27 @@ class Importer(ErrorProducer):
pipeline.connect("advanced", self.pipeline_advanced_callback)
self.game_pipelines.add(pipeline)
def update_progressbar(self):
def update_progressbar(self) -> None:
"""Update the progressbar to show the overall import progress"""
# Reserve 10% for the sources discovery, the rest is the pipelines
self.progressbar.set_fraction(
(0.1 * self.sources_progress) + (0.9 * self.pipelines_progress)
)
def source_callback(self, _obj, _result, data):
def source_callback(self, _obj: Any, _result: Any, data: tuple) -> None:
"""Callback executed when a source is fully scanned"""
source, *_rest = data
logging.debug("Import done for source %s", source.source_id)
self.n_source_tasks_done += 1
self.progress_changed_callback()
def pipeline_advanced_callback(self, pipeline: Pipeline):
def pipeline_advanced_callback(self, pipeline: Pipeline) -> None:
"""Callback called when a pipeline for a game has advanced"""
if pipeline.is_done:
self.n_pipelines_done += 1
self.progress_changed_callback()
def progress_changed_callback(self):
def progress_changed_callback(self) -> None:
"""
Callback called when the import process has progressed
@@ -217,19 +228,47 @@ class Importer(ErrorProducer):
if self.finished:
self.import_callback()
def import_callback(self):
def remove_games(self) -> None:
"""Set removed to True for missing games"""
if not shared.schema.get_boolean("remove-missing"):
return
for game in shared.store:
if game.removed:
continue
if game.source == "imported":
continue
if not shared.schema.get_boolean(game.base_source):
continue
if game.game_id in shared.store.duplicate_game_ids:
continue
if game.game_id in shared.store.new_game_ids:
continue
logging.debug("Removing missing game %s (%s)", game.name, game.game_id)
game.removed = True
game.save()
game.update()
self.removed_game_ids.add(game.game_id)
def import_callback(self) -> None:
"""Callback called when importing has finished"""
logging.info("Import done")
self.remove_games()
self.imported_game_ids = shared.store.new_game_ids
shared.store.new_game_ids = set()
shared.store.duplicate_game_ids = set()
self.import_dialog.close()
self.summary_toast = self.create_summary_toast()
self.create_error_dialog()
shared.win.get_application().lookup_action("import").set_enabled(True)
def create_error_dialog(self):
def create_error_dialog(self) -> None:
"""Dialog containing all errors raised by importers"""
# Collect all errors that happened in the importer and the managers
errors: list[Exception] = []
errors = []
errors.extend(self.collect_errors())
for manager in shared.store.managers.values():
errors.extend(manager.collect_errors())
@@ -277,41 +316,78 @@ class Importer(ErrorProducer):
dialog.present()
def create_summary_toast(self):
"""N games imported toast"""
def undo_import(self, *_args: Any) -> None:
for game_id in self.imported_game_ids:
shared.store[game_id].removed = True
shared.store[game_id].update()
shared.store[game_id].save()
for game_id in self.removed_game_ids:
shared.store[game_id].removed = False
shared.store[game_id].update()
shared.store[game_id].save()
self.imported_game_ids = set()
self.removed_game_ids = set()
self.summary_toast.dismiss()
logging.info("Import undone")
def create_summary_toast(self) -> Adw.Toast:
"""N games imported, removed toast"""
toast = Adw.Toast(timeout=0, priority=Adw.ToastPriority.HIGH)
if self.n_games_added == 0:
toast.set_title(_("No new games found"))
toast.set_button_label(_("Preferences"))
toast.connect(
"button-clicked",
self.dialog_response_callback,
"open_preferences",
"import",
)
if not self.n_games_added:
toast_title = _("No new games found")
if not self.removed_game_ids:
toast.set_button_label(_("Preferences"))
toast.connect(
"button-clicked",
self.dialog_response_callback,
"open_preferences",
"import",
)
elif self.n_games_added == 1:
toast.set_title(_("1 game imported"))
toast_title = _("1 game imported")
elif self.n_games_added > 1:
# The variable is the number of games
toast.set_title(_("{} games imported").format(self.n_games_added))
toast_title = _("{} games imported").format(self.n_games_added)
if (removed_length := len(self.removed_game_ids)) == 1:
# A single game removed
toast_title += ", " + _("1 removed")
elif removed_length > 1:
# The variable is the number of games removed
toast_title += ", " + _("{} removed").format(removed_length)
if self.n_games_added or self.removed_game_ids:
toast.set_button_label(_("Undo"))
toast.connect("button-clicked", self.undo_import)
toast.set_title(toast_title)
shared.win.toast_overlay.add_toast(toast)
return toast
def open_preferences(self, page=None, expander_row=None):
def open_preferences(
self,
page_name: Optional[str] = None,
expander_row: Optional[Adw.ExpanderRow] = None,
) -> Adw.PreferencesWindow:
return shared.win.get_application().on_preferences_action(
page_name=page, expander_row=expander_row
page_name=page_name, expander_row=expander_row
)
def timeout_toast(self, *_args):
def timeout_toast(self, *_args: Any) -> None:
"""Manually timeout the toast after the user has dismissed all warnings"""
GLib.timeout_add_seconds(5, self.summary_toast.dismiss)
def dialog_response_callback(self, _widget, response, *args):
def dialog_response_callback(self, _widget: Any, response: str, *args: Any) -> None:
"""Handle after-import dialogs callback"""
logging.debug("After-import dialog response: %s (%s)", response, str(args))
if response == "open_preferences":

32
src/importer/sources/bottles_source.py
View File

@@ -90,18 +90,22 @@ class BottlesSource(URLExecutableSource):
url_format = 'bottles:run/"{bottle_name}"/"{game_name}"'
available_on = {"linux"}
locations = BottlesLocations(
Location(
schema_key="bottles-location",
candidates=(
shared.flatpak_dir / "com.usebottles.bottles" / "data" / "bottles",
shared.data_dir / "bottles/",
shared.home / ".local" / "share" / "bottles",
),
paths={
"library.yml": LocationSubPath("library.yml"),
"data.yml": LocationSubPath("data.yml"),
},
invalid_subtitle=Location.DATA_INVALID_SUBTITLE,
locations: BottlesLocations
def __init__(self) -> None:
super().__init__()
self.locations = BottlesLocations(
Location(
schema_key="bottles-location",
candidates=(
shared.flatpak_dir / "com.usebottles.bottles" / "data" / "bottles",
shared.data_dir / "bottles/",
shared.home / ".local" / "share" / "bottles",
),
paths={
"library.yml": LocationSubPath("library.yml"),
"data.yml": LocationSubPath("data.yml"),
},
invalid_subtitle=Location.DATA_INVALID_SUBTITLE,
)
)
)

30
src/importer/sources/flatpak_source.py
View File

@@ -125,17 +125,21 @@ class FlatpakSource(ExecutableFormatSource):
executable_format = "flatpak run {flatpak_id}"
available_on = {"linux"}
locations = FlatpakLocations(
Location(
schema_key="flatpak-location",
candidates=(
"/var/lib/flatpak/",
shared.data_dir / "flatpak",
),
paths={
"applications": LocationSubPath("exports/share/applications", True),
"icons": LocationSubPath("exports/share/icons", True),
},
invalid_subtitle=Location.DATA_INVALID_SUBTITLE,
locations: FlatpakLocations
def __init__(self) -> None:
super().__init__()
self.locations = FlatpakLocations(
Location(
schema_key="flatpak-location",
candidates=(
"/var/lib/flatpak/",
shared.data_dir / "flatpak",
),
paths={
"applications": LocationSubPath("exports/share/applications", True),
"icons": LocationSubPath("exports/share/icons", True),
},
invalid_subtitle=Location.DATA_INVALID_SUBTITLE,
)
)
)

50
src/importer/sources/heroic_source.py
View File

@@ -21,12 +21,12 @@
import json
import logging
from abc import abstractmethod
from functools import cached_property
from hashlib import sha256
from json import JSONDecodeError
from pathlib import Path
from time import time
from typing import Iterable, NamedTuple, Optional, TypedDict
from functools import cached_property
from src import shared
from src.game import Game
@@ -108,7 +108,9 @@ class SubSourceIterable(Iterable):
"game_id": self.source.game_id_format.format(
service=self.service, game_id=app_name
),
"executable": self.source.executable_format.format(runner=runner, app_name=app_name),
"executable": self.source.executable_format.format(
runner=runner, app_name=app_name
),
"hidden": self.source_iterable.is_hidden(app_name),
}
game = Game(values)
@@ -239,7 +241,7 @@ class LegendaryIterable(StoreSubSourceIterable):
else:
# Heroic native
logging.debug("Using Heroic native <= 2.8 legendary file")
path = Path.home() / ".config"
path = shared.home / ".config"
path = path / "legendary" / "installed.json"
logging.debug("Using Heroic %s installed.json path %s", self.name, path)
@@ -363,27 +365,31 @@ class HeroicSource(URLExecutableSource):
url_format = "heroic://launch/{runner}/{app_name}"
available_on = {"linux", "win32"}
locations = HeroicLocations(
Location(
schema_key="heroic-location",
candidates=(
shared.config_dir / "heroic",
shared.home / ".config" / "heroic",
shared.flatpak_dir
/ "com.heroicgameslauncher.hgl"
/ "config"
/ "heroic",
shared.appdata_dir / "heroic",
),
paths={
"config.json": LocationSubPath("config.json"),
"store_config.json": LocationSubPath("store/config.json"),
},
invalid_subtitle=Location.CONFIG_INVALID_SUBTITLE,
)
)
locations: HeroicLocations
@property
def game_id_format(self) -> str:
"""The string format used to construct game IDs"""
return self.source_id + "_{service}_{game_id}"
def __init__(self) -> None:
super().__init__()
self.locations = HeroicLocations(
Location(
schema_key="heroic-location",
candidates=(
shared.config_dir / "heroic",
shared.home / ".config" / "heroic",
shared.flatpak_dir
/ "com.heroicgameslauncher.hgl"
/ "config"
/ "heroic",
shared.appdata_dir / "heroic",
),
paths={
"config.json": LocationSubPath("config.json"),
"store_config.json": LocationSubPath("store/config.json"),
},
invalid_subtitle=Location.CONFIG_INVALID_SUBTITLE,
)
)

32
src/importer/sources/itch_source.py
View File

@@ -86,18 +86,22 @@ class ItchSource(URLExecutableSource):
url_format = "itch://caves/{cave_id}/launch"
available_on = {"linux", "win32"}
locations = ItchLocations(
Location(
schema_key="itch-location",
candidates=(
shared.flatpak_dir / "io.itch.itch" / "config" / "itch",
shared.config_dir / "itch",
shared.home / ".config" / "itch",
shared.appdata_dir / "itch",
),
paths={
"butler.db": LocationSubPath("db/butler.db"),
},
invalid_subtitle=Location.CONFIG_INVALID_SUBTITLE,
locations: ItchLocations
def __init__(self) -> None:
super().__init__()
self.locations = ItchLocations(
Location(
schema_key="itch-location",
candidates=(
shared.flatpak_dir / "io.itch.itch" / "config" / "itch",
shared.config_dir / "itch",
shared.home / ".config" / "itch",
shared.appdata_dir / "itch",
),
paths={
"butler.db": LocationSubPath("db/butler.db"),
},
invalid_subtitle=Location.CONFIG_INVALID_SUBTITLE,
)
)
)

30
src/importer/sources/legendary_source.py
View File

@@ -104,17 +104,21 @@ class LegendarySource(ExecutableFormatSource):
available_on = {"linux"}
iterable_class = LegendarySourceIterable
locations = LegendaryLocations(
Location(
schema_key="legendary-location",
candidates=(
shared.config_dir / "legendary",
shared.home / ".config" / "legendary",
),
paths={
"installed.json": LocationSubPath("installed.json"),
"metadata": LocationSubPath("metadata", True),
},
invalid_subtitle=Location.CONFIG_INVALID_SUBTITLE,
locations: LegendaryLocations
def __init__(self) -> None:
super().__init__()
self.locations = LegendaryLocations(
Location(
schema_key="legendary-location",
candidates=(
shared.config_dir / "legendary",
shared.home / ".config" / "legendary",
),
paths={
"installed.json": LocationSubPath("installed.json"),
"metadata": LocationSubPath("metadata", True),
},
invalid_subtitle=Location.CONFIG_INVALID_SUBTITLE,
)
)
)

14
src/importer/sources/location.py
View File

@@ -1,7 +1,7 @@
import logging
from pathlib import Path
from typing import Mapping, Iterable, NamedTuple
from os import PathLike
from pathlib import Path
from typing import Iterable, Mapping, NamedTuple, Optional
from src import shared
@@ -41,7 +41,7 @@ class Location:
paths: Mapping[str, LocationSubPath]
invalid_subtitle: str
root: Path = None
root: Optional[Path] = None
def __init__(
self,
@@ -70,7 +70,7 @@ class Location:
def resolve(self) -> None:
"""Choose a root path from the candidates for the location.
If none fits, raise a UnresolvableLocationError"""
If none fits, raise an UnresolvableLocationError"""
if self.root is not None:
return
@@ -94,7 +94,9 @@ class Location:
shared.schema.set_string(self.schema_key, value)
logging.debug("Resolved value for schema key %s: %s", self.schema_key, value)
def __getitem__(self, key: str):
def __getitem__(self, key: str) -> Optional[Path]:
"""Get the computed path from its key for the location"""
self.resolve()
return self.root / self.paths[key].segment
if self.root:
return self.root / self.paths[key].segment
return None

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

@@ -100,33 +100,37 @@ class LutrisSource(URLExecutableSource):
# FIXME possible bug: config picks ~/.var... and cache picks ~/.local...
locations = LutrisLocations(
Location(
schema_key="lutris-location",
candidates=(
shared.flatpak_dir / "net.lutris.Lutris" / "data" / "lutris",
shared.data_dir / "lutris",
shared.home / ".local" / "share" / "lutris",
),
paths={
"pga.db": LocationSubPath("pga.db"),
},
invalid_subtitle=Location.DATA_INVALID_SUBTITLE,
),
Location(
schema_key="lutris-cache-location",
candidates=(
shared.flatpak_dir / "net.lutris.Lutris" / "cache" / "lutris",
shared.cache_dir / "lutris",
shared.home / ".cache" / "lutris",
),
paths={
"coverart": LocationSubPath("coverart", True),
},
invalid_subtitle=Location.CACHE_INVALID_SUBTITLE,
),
)
locations: LutrisLocations
@property
def game_id_format(self):
return self.source_id + "_{runner}_{game_id}"
def __init__(self) -> None:
super().__init__()
self.locations = LutrisLocations(
Location(
schema_key="lutris-location",
candidates=(
shared.flatpak_dir / "net.lutris.Lutris" / "data" / "lutris",
shared.data_dir / "lutris",
shared.home / ".local" / "share" / "lutris",
),
paths={
"pga.db": LocationSubPath("pga.db"),
},
invalid_subtitle=Location.DATA_INVALID_SUBTITLE,
),
Location(
schema_key="lutris-cache-location",
candidates=(
shared.flatpak_dir / "net.lutris.Lutris" / "cache" / "lutris",
shared.cache_dir / "lutris",
shared.home / ".cache" / "lutris",
),
paths={
"coverart": LocationSubPath("coverart", True),
},
invalid_subtitle=Location.CACHE_INVALID_SUBTITLE,
),
)

23
src/importer/sources/retroarch_source.py
View File

@@ -147,28 +147,7 @@ class RetroarchSource(Source):
available_on = {"linux"}
iterable_class = RetroarchSourceIterable
locations = RetroarchLocations(
Location(
schema_key="retroarch-location",
candidates=[
shared.flatpak_dir / "org.libretro.RetroArch" / "config" / "retroarch",
shared.config_dir / "retroarch",
shared.home / ".config" / "retroarch",
# TODO: Windows support, waiting for executable path setting improvement
# Path("C:\\RetroArch-Win64"),
# Path("C:\\RetroArch-Win32"),
# TODO: UWP support (URL handler - https://github.com/libretro/RetroArch/pull/13563)
# shared.local_appdata_dir
# / "Packages"
# / "1e4cf179-f3c2-404f-b9f3-cb2070a5aad8_8ngdn9a6dx1ma"
# / "LocalState",
],
paths={
"retroarch.cfg": LocationSubPath("retroarch.cfg"),
},
invalid_subtitle=Location.CONFIG_INVALID_SUBTITLE,
)
)
locations: RetroarchLocations
def __init__(self) -> None:
super().__init__()

20
src/importer/sources/source.py
View File

@@ -20,19 +20,19 @@
import sys
from abc import abstractmethod
from collections.abc import Iterable
from typing import Any, Generator, Collection
from typing import Any, Collection, Generator, Optional
from src.game import Game
from src.importer.sources.location import Location
# Type of the data returned by iterating on a Source
SourceIterationResult = None | Game | tuple[Game, tuple[Any]]
SourceIterationResult = Optional[Game | tuple[Game, tuple[Any]]]
class SourceIterable(Iterable):
"""Data producer for a source of games"""
source: "Source" = None
source: "Source"
def __init__(self, source: "Source") -> None:
self.source = source
@@ -53,16 +53,19 @@ class Source(Iterable):
source_id: str
name: str
variant: str = None
variant: Optional[str] = None
available_on: set[str] = set()
iterable_class: type[SourceIterable]
# NOTE: Locations must be set at __init__ time, not in the class definition.
# They must not be shared between source instances.
locations: Collection[Location]
@property
def full_name(self) -> str:
"""The source's full name"""
full_name_ = self.name
if self.variant is not None:
if self.variant:
full_name_ += f" ({self.variant})"
return full_name_
@@ -72,7 +75,7 @@ class Source(Iterable):
return self.source_id + "_{game_id}"
@property
def is_available(self):
def is_available(self) -> bool:
return sys.platform in self.available_on
@abstractmethod
@@ -87,10 +90,7 @@ class Source(Iterable):
Get an iterator for the source
:raises UnresolvableLocationError: Not iterable if any of the locations are unresolvable
"""
for location_name in ("data", "cache", "config"):
location = getattr(self, f"{location_name}_location", None)
if location is None:
continue
for location in self.locations:
location.resolve()
return iter(self.iterable_class(self))

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

@@ -120,19 +120,25 @@ class SteamSource(URLExecutableSource):
iterable_class = SteamSourceIterable
url_format = "steam://rungameid/{game_id}"
locations = SteamLocations(
Location(
schema_key="steam-location",
candidates=(
shared.home / ".steam" / "steam",
shared.data_dir / "Steam",
shared.flatpak_dir / "com.valvesoftware.Steam" / "data" / "Steam",
shared.programfiles32_dir / "Steam",
),
paths={
"libraryfolders.vdf": LocationSubPath("steamapps/libraryfolders.vdf"),
"librarycache": LocationSubPath("appcache/librarycache", True),
},
invalid_subtitle=Location.DATA_INVALID_SUBTITLE,
locations: SteamLocations
def __init__(self) -> None:
super().__init__()
self.locations = SteamLocations(
Location(
schema_key="steam-location",
candidates=(
shared.home / ".steam" / "steam",
shared.data_dir / "Steam",
shared.flatpak_dir / "com.valvesoftware.Steam" / "data" / "Steam",
shared.programfiles32_dir / "Steam",
),
paths={
"libraryfolders.vdf": LocationSubPath(
"steamapps/libraryfolders.vdf"
),
"librarycache": LocationSubPath("appcache/librarycache", True),
},
invalid_subtitle=Location.DATA_INVALID_SUBTITLE,
)
)
)

2
src/logging/color_log_formatter.py
View File

@@ -29,7 +29,7 @@ class ColorLogFormatter(Formatter):
RED = "\033[31m"
YELLOW = "\033[33m"
def format(self, record: LogRecord):
def format(self, record: LogRecord) -> str:
super_format = super().format(record)
match record.levelname:
case "CRITICAL":

10
src/logging/session_file_handler.py
View File

@@ -18,11 +18,12 @@
# SPDX-License-Identifier: GPL-3.0-or-later
import lzma
from io import StringIO
from io import TextIOWrapper
from logging import StreamHandler
from lzma import FORMAT_XZ, PRESET_DEFAULT
from os import PathLike
from pathlib import Path
from typing import Optional
from src import shared
@@ -37,7 +38,7 @@ class SessionFileHandler(StreamHandler):
backup_count: int
filename: Path
log_file: StringIO = None
log_file: Optional[TextIOWrapper] = None
def create_dir(self) -> None:
"""Create the log dir if needed"""
@@ -83,7 +84,7 @@ class SessionFileHandler(StreamHandler):
logfiles.sort(key=self.file_sort_key, reverse=True)
return logfiles
def rotate_file(self, path: Path):
def rotate_file(self, path: Path) -> None:
"""Rotate a file's number suffix and remove it if it's too old"""
# If uncompressed, compress
@@ -128,5 +129,6 @@ class SessionFileHandler(StreamHandler):
super().__init__(self.log_file)
def close(self) -> None:
self.log_file.close()
if self.log_file:
self.log_file.close()
super().close()

4
src/logging/setup.py
View File

@@ -27,7 +27,7 @@ import sys
from src import shared
def setup_logging():
def setup_logging() -> None:
"""Intitate the app's logging"""
is_dev = shared.PROFILE == "development"
@@ -89,7 +89,7 @@ def setup_logging():
logging_dot_config.dictConfig(config)
def log_system_info():
def log_system_info() -> None:
"""Log system debug information"""
logging.debug("Starting %s v%s (%s)", shared.APP_ID, shared.VERSION, shared.PROFILE)

76
src/main.py
View File

@@ -21,6 +21,7 @@ import json
import lzma
import os
import sys
from typing import Any, Optional
import gi
@@ -55,15 +56,15 @@ from src.window import CartridgesWindow
class CartridgesApplication(Adw.Application):
win = None
win: CartridgesWindow
def __init__(self):
def __init__(self) -> None:
shared.store = Store()
super().__init__(
application_id=shared.APP_ID, flags=Gio.ApplicationFlags.FLAGS_NONE
)
def do_activate(self): # pylint: disable=arguments-differ
def do_activate(self) -> None: # pylint: disable=arguments-differ
"""Called on app creation"""
setup_logging()
@@ -141,14 +142,17 @@ class CartridgesApplication(Adw.Application):
self.win.present()
def load_games_from_disk(self):
def load_games_from_disk(self) -> None:
if shared.games_dir.is_dir():
for game_file in shared.games_dir.iterdir():
data = json.load(game_file.open())
try:
data = json.load(game_file.open())
except (OSError, json.decoder.JSONDecodeError):
continue
game = Game(data)
shared.store.add_game(game, {"skip_save": True})
def on_about_action(self, *_args):
def on_about_action(self, *_args: Any) -> None:
# Get the debug info from the log files
debug_str = ""
for i, path in enumerate(shared.log_files):
@@ -192,8 +196,12 @@ class CartridgesApplication(Adw.Application):
about.present()
def on_preferences_action(
self, _action=None, _parameter=None, page_name=None, expander_row=None
):
self,
_action: Any = None,
_parameter: Any = None,
page_name: Optional[str] = None,
expander_row: Optional[str] = None,
) -> CartridgesWindow:
win = PreferencesWindow()
if page_name:
win.set_visible_page_name(page_name)
@@ -203,76 +211,76 @@ class CartridgesApplication(Adw.Application):
return win
def on_launch_game_action(self, *_args):
def on_launch_game_action(self, *_args: Any) -> None:
self.win.active_game.launch()
def on_hide_game_action(self, *_args):
def on_hide_game_action(self, *_args: Any) -> None:
self.win.active_game.toggle_hidden()
def on_edit_game_action(self, *_args):
def on_edit_game_action(self, *_args: Any) -> None:
DetailsWindow(self.win.active_game)
def on_add_game_action(self, *_args):
def on_add_game_action(self, *_args: Any) -> None:
DetailsWindow()
def on_import_action(self, *_args):
importer = Importer()
def on_import_action(self, *_args: Any) -> None:
shared.importer = Importer()
if shared.schema.get_boolean("lutris"):
importer.add_source(LutrisSource())
shared.importer.add_source(LutrisSource())
if shared.schema.get_boolean("steam"):
importer.add_source(SteamSource())
shared.importer.add_source(SteamSource())
if shared.schema.get_boolean("heroic"):
importer.add_source(HeroicSource())
shared.importer.add_source(HeroicSource())
if shared.schema.get_boolean("bottles"):
importer.add_source(BottlesSource())
shared.importer.add_source(BottlesSource())
if shared.schema.get_boolean("flatpak"):
importer.add_source(FlatpakSource())
shared.importer.add_source(FlatpakSource())
if shared.schema.get_boolean("itch"):
importer.add_source(ItchSource())
shared.importer.add_source(ItchSource())
if shared.schema.get_boolean("legendary"):
importer.add_source(LegendarySource())
shared.importer.add_source(LegendarySource())
if shared.schema.get_boolean("retroarch"):
importer.add_source(RetroarchSource())
shared.importer.add_source(RetroarchSource())
importer.run()
shared.importer.run()
def on_remove_game_action(self, *_args):
def on_remove_game_action(self, *_args: Any) -> None:
self.win.active_game.remove_game()
def on_remove_game_details_view_action(self, *_args):
def on_remove_game_details_view_action(self, *_args: Any) -> None:
if self.win.stack.get_visible_child() == self.win.details_view:
self.on_remove_game_action()
def search(self, uri):
def search(self, uri: str) -> None:
Gio.AppInfo.launch_default_for_uri(f"{uri}{self.win.active_game.name}")
def on_igdb_search_action(self, *_args):
def on_igdb_search_action(self, *_args: Any) -> None:
self.search("https://www.igdb.com/search?type=1&q=")
def on_sgdb_search_action(self, *_args):
def on_sgdb_search_action(self, *_args: Any) -> None:
self.search("https://www.steamgriddb.com/search/grids?term=")
def on_protondb_search_action(self, *_args):
def on_protondb_search_action(self, *_args: Any) -> None:
self.search("https://www.protondb.com/search?q=")
def on_lutris_search_action(self, *_args):
def on_lutris_search_action(self, *_args: Any) -> None:
self.search("https://lutris.net/games?q=")
def on_hltb_search_action(self, *_args):
def on_hltb_search_action(self, *_args: Any) -> None:
self.search("https://howlongtobeat.com/?q=")
def on_quit_action(self, *_args):
def on_quit_action(self, *_args: Any) -> None:
self.quit()
def create_actions(self, actions):
def create_actions(self, actions: set) -> None:
for action in actions:
simple_action = Gio.SimpleAction.new(action[0], None)
@@ -288,7 +296,7 @@ class CartridgesApplication(Adw.Application):
scope.add_action(simple_action)
def main(_version):
def main(_version: int) -> Any:
"""App entry point"""
app = CartridgesApplication()
return app.run(sys.argv)

77
src/preferences.py
View File

@@ -21,10 +21,12 @@ import logging
import re
from pathlib import Path
from shutil import rmtree
from typing import Any, Callable, Optional
from gi.repository import Adw, Gio, GLib, Gtk
from src import shared
from src.game import Game
from src.importer.sources.bottles_source import BottlesSource
from src.importer.sources.flatpak_source import FlatpakSource
from src.importer.sources.heroic_source import HeroicSource
@@ -52,6 +54,8 @@ class PreferencesWindow(Adw.PreferencesWindow):
cover_launches_game_switch = Gtk.Template.Child()
high_quality_images_switch = Gtk.Template.Child()
remove_missing_switch = Gtk.Template.Child()
steam_expander_row = Gtk.Template.Child()
steam_data_action_row = Gtk.Template.Child()
steam_data_file_chooser_button = Gtk.Template.Child()
@@ -105,10 +109,10 @@ class PreferencesWindow(Adw.PreferencesWindow):
reset_button = Gtk.Template.Child()
remove_all_games_button = Gtk.Template.Child()
removed_games = set()
warning_menu_buttons = {}
removed_games: set[Game] = set()
warning_menu_buttons: dict = {}
def __init__(self, **kwargs):
def __init__(self, **kwargs: Any) -> None:
super().__init__(**kwargs)
self.win = shared.win
self.file_chooser = Gtk.FileDialog()
@@ -155,7 +159,7 @@ class PreferencesWindow(Adw.PreferencesWindow):
self.init_source_row(source)
# SteamGridDB
def sgdb_key_changed(*_args):
def sgdb_key_changed(*_args: Any) -> None:
shared.schema.set_string("sgdb-key", self.sgdb_key_entry_row.get_text())
self.sgdb_key_entry_row.set_text(shared.schema.get_string("sgdb-key"))
@@ -169,7 +173,7 @@ class PreferencesWindow(Adw.PreferencesWindow):
)
)
def set_sgdb_sensitive(widget):
def set_sgdb_sensitive(widget: Adw.EntryRow) -> None:
if not widget.get_text():
shared.schema.set_boolean("sgdb", False)
@@ -180,10 +184,11 @@ class PreferencesWindow(Adw.PreferencesWindow):
# Switches
self.bind_switches(
(
{
"exit-after-launch",
"cover-launches-game",
"high-quality-images",
"remove-missing",
"lutris-import-steam",
"lutris-import-flatpak",
"heroic-import-epic",
@@ -194,13 +199,13 @@ class PreferencesWindow(Adw.PreferencesWindow):
"sgdb",
"sgdb-prefer",
"sgdb-animated",
)
}
)
def get_switch(self, setting):
def get_switch(self, setting: str) -> Any:
return getattr(self, f'{setting.replace("-", "_")}_switch')
def bind_switches(self, settings):
def bind_switches(self, settings: set[str]) -> None:
for setting in settings:
shared.schema.bind(
setting,
@@ -209,10 +214,12 @@ class PreferencesWindow(Adw.PreferencesWindow):
Gio.SettingsBindFlags.DEFAULT,
)
def choose_folder(self, _widget, callback, callback_data=None):
def choose_folder(
self, _widget: Any, callback: Callable, callback_data: Optional[str] = None
) -> None:
self.file_chooser.select_folder(self.win, None, callback, callback_data)
def undo_remove_all(self, *_args):
def undo_remove_all(self, *_args: Any) -> None:
for game in self.removed_games:
game.removed = False
game.save()
@@ -221,7 +228,7 @@ class PreferencesWindow(Adw.PreferencesWindow):
self.removed_games = set()
self.toast.dismiss()
def remove_all_games(self, *_args):
def remove_all_games(self, *_args: Any) -> None:
for game in shared.store:
if not game.removed:
self.removed_games.add(game)
@@ -234,7 +241,7 @@ class PreferencesWindow(Adw.PreferencesWindow):
self.add_toast(self.toast)
def reset_app(self, *_args):
def reset_app(self, *_args: Any) -> None:
rmtree(shared.data_dir / "cartridges", True)
rmtree(shared.config_dir / "cartridges", True)
rmtree(shared.cache_dir / "cartridges", True)
@@ -252,28 +259,24 @@ class PreferencesWindow(Adw.PreferencesWindow):
shared.win.get_application().quit()
def update_source_action_row_paths(self, source):
def update_source_action_row_paths(self, source: Source) -> None:
"""Set the dir subtitle for a source's action rows"""
for location in ("data", "config", "cache"):
for location_name, location in source.locations._asdict().items():
# Get the action row to subtitle
action_row = getattr(
self, f"{source.source_id}_{location}_action_row", None
self, f"{source.source_id}_{location_name}_action_row", None
)
if not action_row:
continue
infix = "-cache" if location == "cache" else ""
key = f"{source.source_id}{infix}-location"
path = Path(shared.schema.get_string(key)).expanduser()
path = Path(shared.schema.get_string(location.schema_key)).expanduser()
# Remove the path prefix if picked via Flatpak portal
subtitle = re.sub("/run/user/\\d*/doc/.*/", "", str(path))
action_row.set_subtitle(subtitle)
def resolve_locations(self, source: Source):
def resolve_locations(self, source: Source) -> None:
"""Resolve locations and add a warning if location cannot be found"""
def clear_warning_selection(_widget, label):
def clear_warning_selection(_widget: Any, label: Gtk.Label) -> None:
label.select_region(-1, -1)
for location_name, location in source.locations._asdict().items():
@@ -323,40 +326,30 @@ class PreferencesWindow(Adw.PreferencesWindow):
action_row.add_prefix(menu_button)
self.warning_menu_buttons[source.source_id] = menu_button
def init_source_row(self, source: Source):
def init_source_row(self, source: Source) -> None:
"""Initialize a preference row for a source class"""
def set_dir(_widget, result, location_name):
def set_dir(_widget: Any, result: Gio.Task, location_name: str) -> None:
"""Callback called when a dir picker button is clicked"""
try:
path = Path(self.file_chooser.select_folder_finish(result).get_path())
except GLib.GError:
return
# Good picked location
location = getattr(source.locations, location_name)
location = source.locations._asdict()[location_name]
if location.check_candidate(path):
# Set the schema
match location_name:
case "config" | "data":
infix = ""
case _:
infix = f"-{location_name}"
key = f"{source.source_id}{infix}-location"
value = str(path)
shared.schema.set_string(key, value)
# Update the row
shared.schema.set_string(location.schema_key, str(path))
self.update_source_action_row_paths(source)
if self.warning_menu_buttons.get(source.source_id):
action_row = getattr(
self, f"{source.source_id}_{location_name}_action_row", None
)
action_row.remove(self.warning_menu_buttons[source.source_id])
action_row.remove( # type: ignore
self.warning_menu_buttons[source.source_id]
)
self.warning_menu_buttons.pop(source.source_id)
logging.debug("User-set value for schema key %s: %s", key, value)
logging.debug("User-set value for %s is %s", location.schema_key, path)
# Bad picked location, inform user
else:
@@ -369,7 +362,7 @@ class PreferencesWindow(Adw.PreferencesWindow):
_("Set Location"),
)
def on_response(widget, response):
def on_response(widget: Any, response: str) -> None:
if response == "choose_folder":
self.choose_folder(widget, set_dir, location_name)

12
src/store/managers/manager.py
View File

@@ -46,7 +46,7 @@ class Manager(ErrorProducer):
max_tries: int = 3
@property
def name(self):
def name(self) -> str:
return type(self).__name__
@abstractmethod
@@ -59,13 +59,13 @@ class Manager(ErrorProducer):
* May raise other exceptions that will be reported
"""
def run(self, game: Game, additional_data: dict):
def run(self, game: Game, additional_data: dict) -> None:
"""Handle errors (retry, ignore or raise) that occur in the manager logic"""
# Keep track of the number of tries
tries = 1
def handle_error(error: Exception):
def handle_error(error: Exception) -> None:
nonlocal tries
# If FriendlyError, handle its cause instead
@@ -83,11 +83,11 @@ class Manager(ErrorProducer):
retrying_format = "Retrying %s in %s for %s"
unretryable_format = "Unretryable %s in %s for %s"
if error in self.continue_on:
if type(error) in self.continue_on:
# Handle skippable errors (skip silently)
return
if error in self.retryable_on:
if type(error) in self.retryable_on:
if tries > self.max_tries:
# Handle being out of retries
logging.error(out_of_retries_format, *log_args)
@@ -104,7 +104,7 @@ class Manager(ErrorProducer):
logging.error(unretryable_format, *log_args, exc_info=error)
self.report_error(base_error)
def try_manager_logic():
def try_manager_logic() -> None:
try:
self.main(game, additional_data)
except Exception as error: # pylint: disable=broad-exception-caught

4
src/store/pipeline.py
View File

@@ -83,7 +83,7 @@ class Pipeline(GObject.Object):
progress = 1
return progress
def advance(self):
def advance(self) -> None:
"""Spawn tasks for managers that are able to run for a game"""
# Separate blocking / async managers
@@ -106,5 +106,5 @@ class Pipeline(GObject.Object):
self.advance()
@GObject.Signal(name="advanced")
def advanced(self) -> None:
def advanced(self): # type: ignore
"""Signal emitted when the pipeline has advanced"""

29
src/store/store.py
View File

@@ -18,7 +18,7 @@
# SPDX-License-Identifier: GPL-3.0-or-later
import logging
from typing import MutableMapping, Generator, Any
from typing import Any, Generator, MutableMapping, Optional
from src import shared
from src.game import Game
@@ -33,12 +33,16 @@ class Store:
pipeline_managers: set[Manager]
pipelines: dict[str, Pipeline]
source_games: MutableMapping[str, MutableMapping[str, Game]]
new_game_ids: set[str]
duplicate_game_ids: set[str]
def __init__(self) -> None:
self.managers = {}
self.pipeline_managers = set()
self.pipelines = {}
self.source_games = {}
self.new_game_ids = set()
self.duplicate_game_ids = set()
def __contains__(self, obj: object) -> bool:
"""Check if the game is present in the store with the `in` keyword"""
@@ -73,13 +77,15 @@ class Store:
except KeyError:
return default
def add_manager(self, manager: Manager, in_pipeline=True):
def add_manager(self, manager: Manager, in_pipeline: bool = True) -> None:
"""Add a manager to the store"""
manager_type = type(manager)
self.managers[manager_type] = manager
self.toggle_manager_in_pipelines(manager_type, in_pipeline)
def toggle_manager_in_pipelines(self, manager_type: type[Manager], enable: bool):
def toggle_manager_in_pipelines(
self, manager_type: type[Manager], enable: bool
) -> None:
"""Change if a manager should run in new pipelines"""
if enable:
self.pipeline_managers.add(self.managers[manager_type])
@@ -87,7 +93,7 @@ class Store:
self.pipeline_managers.discard(self.managers[manager_type])
def cleanup_game(self, game: Game) -> None:
"""Remove a game's files"""
"""Remove a game's files, dismiss any loose toasts"""
for path in (
shared.games_dir / f"{game.game_id}.json",
shared.covers_dir / f"{game.game_id}.tiff",
@@ -95,9 +101,17 @@ class Store:
):
path.unlink(missing_ok=True)
# TODO: don't run this if the state is startup
for undo in ("remove", "hide"):
try:
shared.win.toasts[(game, undo)].dismiss()
shared.win.toasts.pop((game, undo))
except KeyError:
pass
def add_game(
self, game: Game, additional_data: dict, run_pipeline=True
) -> Pipeline | None:
self, game: Game, additional_data: dict, run_pipeline: bool = True
) -> Optional[Pipeline]:
"""Add a game to the app"""
# Ignore games from a newer spec version
@@ -114,6 +128,7 @@ class Store:
if not stored_game:
# New game, do as normal
logging.debug("New store game %s (%s)", game.name, game.game_id)
self.new_game_ids.add(game.game_id)
elif stored_game.removed:
# Will replace a removed game, cleanup its remains
logging.debug(
@@ -122,9 +137,11 @@ class Store:
game.game_id,
)
self.cleanup_game(stored_game)
self.new_game_ids.add(game.game_id)
else:
# Duplicate game, ignore it
logging.debug("Duplicate store game %s (%s)", game.name, game.game_id)
self.duplicate_game_ids.add(game.game_id)
return None
# Connect signals

12
src/utils/create_dialog.py
View File

@@ -17,10 +17,18 @@
#
# SPDX-License-Identifier: GPL-3.0-or-later
from gi.repository import Adw
from typing import Optional
from gi.repository import Adw, Gtk
def create_dialog(win, heading, body, extra_option=None, extra_label=None):
def create_dialog(
win: Gtk.Window,
heading: str,
body: str,
extra_option: Optional[str] = None,
extra_label: Optional[str] = None,
) -> Adw.MessageDialog:
dialog = Adw.MessageDialog.new(win, heading, body)
dialog.add_response("dismiss", _("Dismiss"))

6
src/utils/migrate_files_v1_to_v2.py
View File

@@ -23,14 +23,14 @@ from pathlib import Path
from src import shared
old_data_dir = Path.home() / ".local" / "share"
old_data_dir = shared.home / ".local" / "share"
old_cartridges_data_dir = old_data_dir / "cartridges"
migrated_file_path = old_cartridges_data_dir / ".migrated"
old_games_dir = old_cartridges_data_dir / "games"
old_covers_dir = old_cartridges_data_dir / "covers"
def migrate_game_covers(game_path: Path):
def migrate_game_covers(game_path: Path) -> None:
"""Migrate a game covers from a source game path to the current dir"""
for suffix in (".tiff", ".gif"):
cover_path = old_covers_dir / game_path.with_suffix(suffix).name
@@ -41,7 +41,7 @@ def migrate_game_covers(game_path: Path):
cover_path.rename(destination_cover_path)
def migrate_files_v1_to_v2():
def migrate_files_v1_to_v2() -> None:
"""
Migrate user data from the v1.X locations to the latest location.

90
src/utils/rate_limiter.py
View File

@@ -17,11 +17,11 @@
#
# SPDX-License-Identifier: GPL-3.0-or-later
from typing import Optional, Sized
from threading import Lock, Thread, BoundedSemaphore
from time import sleep, time
from collections import deque
from contextlib import AbstractContextManager
from threading import BoundedSemaphore, Lock, Thread
from time import sleep, time
from typing import Any, Sized
class PickHistory(Sized):
@@ -30,22 +30,22 @@ class PickHistory(Sized):
period: int
timestamps: list[int] = None
timestamps_lock: Lock = None
timestamps: list[float]
timestamps_lock: Lock
def __init__(self, period: int) -> None:
self.period = period
self.timestamps = []
self.timestamps_lock = Lock()
def remove_old_entries(self):
def remove_old_entries(self) -> None:
"""Remove history entries older than the period"""
now = time()
cutoff = now - self.period
with self.timestamps_lock:
self.timestamps = [entry for entry in self.timestamps if entry > cutoff]
def add(self, *new_timestamps: Optional[int]):
def add(self, *new_timestamps: float) -> None:
"""Add timestamps to the history.
If none given, will add the current timestamp"""
if len(new_timestamps) == 0:
@@ -60,7 +60,7 @@ class PickHistory(Sized):
return len(self.timestamps)
@property
def start(self) -> int:
def start(self) -> float:
"""Get the time at which the history started"""
self.remove_old_entries()
with self.timestamps_lock:
@@ -70,7 +70,7 @@ class PickHistory(Sized):
entry = time()
return entry
def copy_timestamps(self) -> str:
def copy_timestamps(self) -> list[float]:
"""Get a copy of the timestamps history"""
self.remove_old_entries()
with self.timestamps_lock:
@@ -79,51 +79,55 @@ class PickHistory(Sized):
# pylint: disable=too-many-instance-attributes
class RateLimiter(AbstractContextManager):
"""Rate limiter implementing the token bucket algorithm"""
"""
Base rate limiter implementing the token bucket algorithm.
Do not use directly, create a child class to tailor the rate limiting to the
underlying service's limits.
Subclasses must provide values to the following attributes:
* refill_period_seconds - Period in which we have a max amount of tokens
* refill_period_tokens - Number of tokens allowed in this period
* burst_tokens - Max number of tokens that can be consumed instantly
"""
# Period in which we have a max amount of tokens
refill_period_seconds: int
# Number of tokens allowed in this period
refill_period_tokens: int
# Max number of tokens that can be consumed instantly
burst_tokens: int
pick_history: PickHistory = None
bucket: BoundedSemaphore = None
queue: deque[Lock] = None
queue_lock: Lock = None
pick_history: PickHistory
bucket: BoundedSemaphore
queue: deque[Lock]
queue_lock: Lock
# Protect the number of tokens behind a lock
__n_tokens_lock: Lock = None
__n_tokens_lock: Lock
__n_tokens = 0
@property
def n_tokens(self):
def n_tokens(self) -> int:
with self.__n_tokens_lock:
return self.__n_tokens
@n_tokens.setter
def n_tokens(self, value: int):
def n_tokens(self, value: int) -> None:
with self.__n_tokens_lock:
self.__n_tokens = value
def __init__(
self,
refill_period_seconds: Optional[int] = None,
refill_period_tokens: Optional[int] = None,
burst_tokens: Optional[int] = None,
) -> None:
def _init_pick_history(self) -> None:
"""
Initialize the tocken pick history
(only for use in this class and its children)
By default, creates an empty pick history.
Should be overriden or extended by subclasses.
"""
self.pick_history = PickHistory(self.refill_period_seconds)
def __init__(self) -> None:
"""Initialize the limiter"""
# Initialize default values
if refill_period_seconds is not None:
self.refill_period_seconds = refill_period_seconds
if refill_period_tokens is not None:
self.refill_period_tokens = refill_period_tokens
if burst_tokens is not None:
self.burst_tokens = burst_tokens
if self.pick_history is None:
self.pick_history = PickHistory(self.refill_period_seconds)
self._init_pick_history()
# Create synchronization data
self.__n_tokens_lock = Lock()
@@ -147,8 +151,8 @@ class RateLimiter(AbstractContextManager):
"""
# Compute ideal spacing
tokens_left = self.refill_period_tokens - len(self.pick_history)
seconds_left = self.pick_history.start + self.refill_period_seconds - time()
tokens_left = self.refill_period_tokens - len(self.pick_history) # type: ignore
seconds_left = self.pick_history.start + self.refill_period_seconds - time() # type: ignore
try:
spacing_seconds = seconds_left / tokens_left
except ZeroDivisionError:
@@ -159,7 +163,7 @@ class RateLimiter(AbstractContextManager):
natural_spacing = self.refill_period_seconds / self.refill_period_tokens
return max(natural_spacing, spacing_seconds)
def refill(self):
def refill(self) -> None:
"""Add a token back in the bucket"""
sleep(self.refill_spacing)
try:
@@ -170,7 +174,7 @@ class RateLimiter(AbstractContextManager):
else:
self.n_tokens += 1
def refill_thread_func(self):
def refill_thread_func(self) -> None:
"""Entry point for the daemon thread that is refilling the bucket"""
while True:
self.refill()
@@ -200,18 +204,18 @@ class RateLimiter(AbstractContextManager):
self.queue.appendleft(lock)
return lock
def acquire(self):
def acquire(self) -> None:
"""Acquires a token from the bucket when it's your turn in queue"""
lock = self.add_to_queue()
self.update_queue()
# Wait until our turn in queue
lock.acquire() # pylint: disable=consider-using-with
self.pick_history.add()
self.pick_history.add() # type: ignore
# --- Support for use in with statements
def __enter__(self):
def __enter__(self) -> None:
self.acquire()
def __exit__(self, *_args):
def __exit__(self, *_args: Any) -> None:
pass

3
src/utils/relative_date.py
View File

@@ -18,11 +18,12 @@
# SPDX-License-Identifier: GPL-3.0-or-later
from datetime import datetime
from typing import Any
from gi.repository import GLib
def relative_date(timestamp): # pylint: disable=too-many-return-statements
def relative_date(timestamp: int) -> Any: # pylint: disable=too-many-return-statements
days_no = ((today := datetime.today()) - datetime.fromtimestamp(timestamp)).days
if days_no == 0:

9
src/utils/save_cover.py
View File

@@ -20,14 +20,17 @@
from pathlib import Path
from shutil import copyfile
from typing import Optional
from gi.repository import Gdk, Gio, GLib
from gi.repository import Gdk, GdkPixbuf, Gio, GLib
from PIL import Image, ImageSequence, UnidentifiedImageError
from src import shared
def resize_cover(cover_path=None, pixbuf=None):
def resize_cover(
cover_path: Optional[Path] = None, pixbuf: Optional[GdkPixbuf.Pixbuf] = None
) -> Optional[Path]:
if not cover_path and not pixbuf:
return None
@@ -74,7 +77,7 @@ def resize_cover(cover_path=None, pixbuf=None):
return tmp_path
def save_cover(game_id, cover_path):
def save_cover(game_id: str, cover_path: Path) -> None:
shared.covers_dir.mkdir(parents=True, exist_ok=True)
animated_path = shared.covers_dir / f"{game_id}.gif"

27
src/utils/steam.py
View File

@@ -21,13 +21,14 @@
import json
import logging
import re
from pathlib import Path
from typing import TypedDict
import requests
from requests.exceptions import HTTPError
from src import shared
from src.utils.rate_limiter import PickHistory, RateLimiter
from src.utils.rate_limiter import RateLimiter
class SteamError(Exception):
@@ -71,16 +72,18 @@ class SteamRateLimiter(RateLimiter):
refill_period_tokens = 200
burst_tokens = 100
def __init__(self) -> None:
# Load pick history from schema
# (Remember API limits through restarts of Cartridges)
def _init_pick_history(self) -> None:
"""
Load the pick history from schema.
Allows remembering API limits through restarts of Cartridges.
"""
super()._init_pick_history()
timestamps_str = shared.state_schema.get_string("steam-limiter-tokens-history")
self.pick_history = PickHistory(self.refill_period_seconds)
self.pick_history.add(*json.loads(timestamps_str))
self.pick_history.remove_old_entries()
super().__init__()
def acquire(self):
def acquire(self) -> None:
"""Get a token from the bucket and store the pick history in the schema"""
super().acquire()
timestamps_str = json.dumps(self.pick_history.copy_timestamps())
@@ -90,7 +93,7 @@ class SteamRateLimiter(RateLimiter):
class SteamFileHelper:
"""Helper for steam file formats"""
def get_manifest_data(self, manifest_path) -> SteamManifestData:
def get_manifest_data(self, manifest_path: Path) -> SteamManifestData:
"""Get local data for a game from its manifest"""
with open(manifest_path, "r", encoding="utf-8") as file:
@@ -104,7 +107,11 @@ class SteamFileHelper:
raise SteamInvalidManifestError()
data[key] = match.group(1)
return SteamManifestData(**data)
return SteamManifestData(
name=data["name"],
appid=data["appid"],
stateflags=data["stateflags"],
)
class SteamAPIHelper:
@@ -116,7 +123,7 @@ class SteamAPIHelper:
def __init__(self, rate_limiter: RateLimiter) -> None:
self.rate_limiter = rate_limiter
def get_api_data(self, appid) -> SteamAPIData:
def get_api_data(self, appid: str) -> SteamAPIData:
"""
Get online data for a game from its appid.
May block to satisfy the Steam web API limitations.

10
src/utils/steamgriddb.py
View File

@@ -20,12 +20,14 @@
import logging
from pathlib import Path
from typing import Any
import requests
from gi.repository import Gio
from requests.exceptions import HTTPError
from src import shared
from src.game import Game
from src.utils.save_cover import resize_cover, save_cover
@@ -55,12 +57,12 @@ class SGDBHelper:
base_url = "https://www.steamgriddb.com/api/v2/"
@property
def auth_headers(self):
def auth_headers(self) -> dict[str, str]:
key = shared.schema.get_string("sgdb-key")
headers = {"Authorization": f"Bearer {key}"}
return headers
def get_game_id(self, game):
def get_game_id(self, game: Game) -> Any:
"""Get grid results for a game. Can raise an exception."""
uri = f"{self.base_url}search/autocomplete/{game.name}"
res = requests.get(uri, headers=self.auth_headers, timeout=5)
@@ -74,7 +76,7 @@ class SGDBHelper:
case _:
res.raise_for_status()
def get_image_uri(self, game_id, animated=False):
def get_image_uri(self, game_id: str, animated: bool = False) -> Any:
"""Get the image for a SGDB game id"""
uri = f"{self.base_url}grids/game/{game_id}?dimensions=600x900"
if animated:
@@ -93,7 +95,7 @@ class SGDBHelper:
case _:
res.raise_for_status()
def conditionaly_update_cover(self, game):
def conditionaly_update_cover(self, game: Game) -> None:
"""Update the game's cover if appropriate"""
# Obvious skips

29
src/utils/task.py
View File

@@ -18,25 +18,28 @@
# SPDX-License-Identifier: GPL-3.0-or-later
from functools import wraps
from typing import Any, Callable
from gi.repository import Gio
def create_task_thread_func_closure(func, data):
def create_task_thread_func_closure(func: Callable, data: Any) -> Callable:
"""Wrap a Gio.TaskThreadFunc with the given data in a closure"""
def closure(task, source_object, _data, cancellable):
def closure(
task: Gio.Task, source_object: object, _data: Any, cancellable: Gio.Cancellable
) -> Any:
func(task, source_object, data, cancellable)
return closure
def decorate_set_task_data(task):
def decorate_set_task_data(task: Gio.Task) -> Callable:
"""Decorate Gio.Task.set_task_data to replace it"""
def decorator(original_method):
def decorator(original_method: Callable) -> Callable:
@wraps(original_method)
def new_method(task_data):
def new_method(task_data: Any) -> None:
task.task_data = task_data
return new_method
@@ -44,13 +47,13 @@ def decorate_set_task_data(task):
return decorator
def decorate_run_in_thread(task):
def decorate_run_in_thread(task: Gio.Task) -> Callable:
"""Decorate Gio.Task.run_in_thread to pass the task data correctly
Creates a closure around task_thread_func with the task data available."""
def decorator(original_method):
def decorator(original_method: Callable) -> Callable:
@wraps(original_method)
def new_method(task_thread_func):
def new_method(task_thread_func: Callable) -> None:
closure = create_task_thread_func_closure(task_thread_func, task.task_data)
original_method(closure)
@@ -64,11 +67,17 @@ class Task:
"""Wrapper around Gio.Task to patch task data not being passed"""
@classmethod
def new(cls, source_object, cancellable, callback, callback_data):
def new(
cls,
source_object: object,
cancellable: Gio.Cancellable,
callback: Callable,
callback_data: Any,
) -> Gio.Task:
"""Create a new, monkey-patched Gio.Task.
The `set_task_data` and `run_in_thread` methods are decorated.
As of 2023-05-19, pygobject does not work well with Gio.Task, so to pass data
As of 2023-05-19, PyGObject does not work well with Gio.Task, so to pass data
the only viable way it to create a closure with the thread function and its data.
This class is supposed to make Gio.Task comply with its expected behaviour
per the docs:

87
src/window.py
View File

@@ -17,9 +17,13 @@
#
# SPDX-License-Identifier: GPL-3.0-or-later
from gi.repository import Adw, Gtk
from typing import Any, Optional
from gi.repository import Adw, Gio, GLib, Gtk
from src import shared
from src.game import Game
from src.game_cover import GameCover
from src.utils.relative_date import relative_date
@@ -64,13 +68,13 @@ class CartridgesWindow(Adw.ApplicationWindow):
hidden_search_entry = Gtk.Template.Child()
hidden_search_button = Gtk.Template.Child()
game_covers = {}
toasts = {}
active_game = None
details_view_game_cover = None
sort_state = "a-z"
game_covers: dict = {}
toasts: dict = {}
active_game: Game
details_view_game_cover: Optional[GameCover] = None
sort_state: str = "a-z"
def __init__(self, **kwargs):
def __init__(self, **kwargs: Any) -> None:
super().__init__(**kwargs)
self.previous_page = self.library_view
@@ -110,11 +114,11 @@ class CartridgesWindow(Adw.ApplicationWindow):
style_manager.connect("notify::dark", self.set_details_view_opacity)
style_manager.connect("notify::high-contrast", self.set_details_view_opacity)
def search_changed(self, _widget, hidden):
def search_changed(self, _widget: Any, hidden: bool) -> None:
# Refresh search filter on keystroke in search box
(self.hidden_library if hidden else self.library).invalidate_filter()
def set_library_child(self):
def set_library_child(self) -> None:
child, hidden_child = self.notice_empty, self.hidden_notice_empty
for game in shared.store:
@@ -134,7 +138,7 @@ class CartridgesWindow(Adw.ApplicationWindow):
self.library_bin.set_child(child)
self.hidden_library_bin.set_child(hidden_child)
def filter_func(self, child):
def filter_func(self, child: Gtk.Widget) -> bool:
game = child.get_child()
text = (
(
@@ -156,10 +160,10 @@ class CartridgesWindow(Adw.ApplicationWindow):
return not filtered
def set_active_game(self, _widget, _pspec, game):
def set_active_game(self, _widget: Any, _pspec: Any, game: Game) -> None:
self.active_game = game
def show_details_view(self, game):
def show_details_view(self, game: Game) -> None:
self.active_game = game
self.details_view_cover.set_opacity(int(not game.loading))
@@ -207,7 +211,7 @@ class CartridgesWindow(Adw.ApplicationWindow):
self.set_details_view_opacity()
def set_details_view_opacity(self, *_args):
def set_details_view_opacity(self, *_args: Any) -> None:
if self.stack.get_visible_child() != self.details_view:
return
@@ -218,12 +222,12 @@ class CartridgesWindow(Adw.ApplicationWindow):
return
self.details_view_blurred_cover.set_opacity(
1 - self.details_view_game_cover.luminance[0]
1 - self.details_view_game_cover.luminance[0] # type: ignore
if style_manager.get_dark()
else self.details_view_game_cover.luminance[1]
else self.details_view_game_cover.luminance[1] # type: ignore
)
def sort_func(self, child1, child2):
def sort_func(self, child1: Gtk.Widget, child2: Gtk.Widget) -> int:
var, order = "name", True
if self.sort_state in ("newest", "oldest"):
@@ -233,7 +237,7 @@ class CartridgesWindow(Adw.ApplicationWindow):
elif self.sort_state == "a-z":
order = False
def get_value(index):
def get_value(index: int) -> str:
return str(
getattr((child1.get_child(), child2.get_child())[index], var)
).lower()
@@ -243,7 +247,7 @@ class CartridgesWindow(Adw.ApplicationWindow):
return ((get_value(0) > get_value(1)) ^ order) * 2 - 1
def navigate(self, next_page):
def navigate(self, next_page: Gtk.Widget) -> None:
levels = (self.library_view, self.hidden_library_view, self.details_view)
self.stack.set_transition_type(
Gtk.StackTransitionType.UNDER_RIGHT
@@ -260,13 +264,13 @@ class CartridgesWindow(Adw.ApplicationWindow):
self.stack.set_visible_child(next_page)
def on_go_back_action(self, *_args):
def on_go_back_action(self, *_args: Any) -> None:
if self.stack.get_visible_child() == self.hidden_library_view:
self.navigate(self.library_view)
elif self.stack.get_visible_child() == self.details_view:
self.on_go_to_parent_action()
def on_go_to_parent_action(self, *_args):
def on_go_to_parent_action(self, *_args: Any) -> None:
if self.stack.get_visible_child() == self.details_view:
self.navigate(
self.hidden_library_view
@@ -274,20 +278,20 @@ class CartridgesWindow(Adw.ApplicationWindow):
else self.library_view
)
def on_go_home_action(self, *_args):
def on_go_home_action(self, *_args: Any) -> None:
self.navigate(self.library_view)
def on_show_hidden_action(self, *_args):
def on_show_hidden_action(self, *_args: Any) -> None:
self.navigate(self.hidden_library_view)
def on_sort_action(self, action, state):
def on_sort_action(self, action: Gio.SimpleAction, state: GLib.Variant) -> None:
action.set_state(state)
self.sort_state = str(state).strip("'")
self.library.invalidate_sort()
shared.state_schema.set_string("sort-mode", self.sort_state)
def on_toggle_search_action(self, *_args):
def on_toggle_search_action(self, *_args: Any) -> None:
if self.stack.get_visible_child() == self.library_view:
search_bar = self.search_bar
search_entry = self.search_entry
@@ -304,7 +308,7 @@ class CartridgesWindow(Adw.ApplicationWindow):
search_entry.set_text("")
def on_escape_action(self, *_args):
def on_escape_action(self, *_args: Any) -> None:
if (
self.get_focus() == self.search_entry.get_focus_child()
or self.hidden_search_entry.get_focus_child()
@@ -313,7 +317,7 @@ class CartridgesWindow(Adw.ApplicationWindow):
else:
self.on_go_back_action()
def show_details_view_search(self, widget):
def show_details_view_search(self, widget: Gtk.Widget) -> None:
library = (
self.hidden_library if widget == self.hidden_search_entry else self.library
)
@@ -329,30 +333,39 @@ class CartridgesWindow(Adw.ApplicationWindow):
index += 1
def on_undo_action(self, _widget, game=None, undo=None):
def on_undo_action(
self, _widget: Any, game: Optional[Game] = None, undo: Optional[str] = None
) -> None:
if not game: # If the action was activated via Ctrl + Z
if shared.importer and (
shared.importer.imported_game_ids or shared.importer.removed_game_ids
):
shared.importer.undo_import()
return
try:
game = tuple(self.toasts.keys())[-1][0]
undo = tuple(self.toasts.keys())[-1][1]
except IndexError:
return
if undo == "hide":
game.toggle_hidden(False)
if game:
if undo == "hide":
game.toggle_hidden(False)
elif undo == "remove":
game.removed = False
game.save()
game.update()
elif undo == "remove":
game.removed = False
game.save()
game.update()
self.toasts[(game, undo)].dismiss()
self.toasts.pop((game, undo))
self.toasts[(game, undo)].dismiss()
self.toasts.pop((game, undo))
def on_open_menu_action(self, *_args):
def on_open_menu_action(self, *_args: Any) -> None:
if self.stack.get_visible_child() == self.library_view:
self.primary_menu_button.popup()
elif self.stack.get_visible_child() == self.hidden_library_view:
self.hidden_primary_menu_button.popup()
def on_close_action(self, *_args):
def on_close_action(self, *_args: Any) -> None:
self.close()