diff --git a/data/gtk/details-window.blp b/data/gtk/details-window.blp index d84b4f8..21d12c1 100644 --- a/data/gtk/details-window.blp +++ b/data/gtk/details-window.blp @@ -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; } }; diff --git a/data/gtk/preferences.blp b/data/gtk/preferences.blp index 9654c29..c779347 100644 --- a/data/gtk/preferences.blp +++ b/data/gtk/preferences.blp @@ -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"); diff --git a/data/hu.kramo.Cartridges.gschema.xml.in b/data/hu.kramo.Cartridges.gschema.xml.in index d31c2ed..9003a93 100644 --- a/data/hu.kramo.Cartridges.gschema.xml.in +++ b/data/hu.kramo.Cartridges.gschema.xml.in @@ -10,6 +10,9 @@ false + + true + true diff --git a/po/POTFILES b/po/POTFILES index d48bf15..c46d11b 100644 --- a/po/POTFILES +++ b/po/POTFILES @@ -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 \ No newline at end of file diff --git a/po/cartridges.pot b/po/cartridges.pot index 55050fe..cc46d3b 100644 --- a/po/cartridges.pot +++ b/po/cartridges.pot @@ -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 \n" "Language-Team: LANGUAGE \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." diff --git a/src/details_window.py b/src/details_window.py index c437d53..bb234da 100644 --- a/src/details_window.py +++ b/src/details_window.py @@ -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("", "").replace("", ""), + ), # 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) diff --git a/src/errors/error_producer.py b/src/errors/error_producer.py index 7f31fea..f3c9b53 100644 --- a/src/errors/error_producer.py +++ b/src/errors/error_producer.py @@ -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) diff --git a/src/errors/friendly_error.py b/src/errors/friendly_error.py index 4c9ca7f..586d784 100644 --- a/src/errors/friendly_error.py +++ b/src/errors/friendly_error.py @@ -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 diff --git a/src/game.py b/src/game.py index 34dbf84..8dc981e 100644 --- a/src/game.py +++ b/src/game.py @@ -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""" diff --git a/src/game_cover.py b/src/game_cover.py index 073f0c9..221075d 100644 --- a/src/game_cover.py +++ b/src/game_cover.py @@ -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, diff --git a/src/importer/importer.py b/src/importer/importer.py index 756ea3b..e987760 100644 --- a/src/importer/importer.py +++ b/src/importer/importer.py @@ -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": diff --git a/src/importer/sources/bottles_source.py b/src/importer/sources/bottles_source.py index 041247b..d85529c 100644 --- a/src/importer/sources/bottles_source.py +++ b/src/importer/sources/bottles_source.py @@ -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, + ) ) - ) diff --git a/src/importer/sources/flatpak_source.py b/src/importer/sources/flatpak_source.py index b4af7c4..fe4d643 100644 --- a/src/importer/sources/flatpak_source.py +++ b/src/importer/sources/flatpak_source.py @@ -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, + ) ) - ) diff --git a/src/importer/sources/heroic_source.py b/src/importer/sources/heroic_source.py index c06a266..f31bbe7 100644 --- a/src/importer/sources/heroic_source.py +++ b/src/importer/sources/heroic_source.py @@ -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, + ) + ) diff --git a/src/importer/sources/itch_source.py b/src/importer/sources/itch_source.py index 36e02e0..e39d090 100644 --- a/src/importer/sources/itch_source.py +++ b/src/importer/sources/itch_source.py @@ -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, + ) ) - ) diff --git a/src/importer/sources/legendary_source.py b/src/importer/sources/legendary_source.py index 6c46ada..bfaaa66 100644 --- a/src/importer/sources/legendary_source.py +++ b/src/importer/sources/legendary_source.py @@ -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, + ) ) - ) diff --git a/src/importer/sources/location.py b/src/importer/sources/location.py index 55684b4..36f592b 100644 --- a/src/importer/sources/location.py +++ b/src/importer/sources/location.py @@ -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 diff --git a/src/importer/sources/lutris_source.py b/src/importer/sources/lutris_source.py index 023d3be..dd5defd 100644 --- a/src/importer/sources/lutris_source.py +++ b/src/importer/sources/lutris_source.py @@ -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, + ), + ) diff --git a/src/importer/sources/retroarch_source.py b/src/importer/sources/retroarch_source.py index 9eb3ee6..796c7ee 100644 --- a/src/importer/sources/retroarch_source.py +++ b/src/importer/sources/retroarch_source.py @@ -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__() diff --git a/src/importer/sources/source.py b/src/importer/sources/source.py index e434d14..e19b4dd 100644 --- a/src/importer/sources/source.py +++ b/src/importer/sources/source.py @@ -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)) diff --git a/src/importer/sources/steam_source.py b/src/importer/sources/steam_source.py index 904fb0b..90460bd 100644 --- a/src/importer/sources/steam_source.py +++ b/src/importer/sources/steam_source.py @@ -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, + ) ) - ) diff --git a/src/logging/color_log_formatter.py b/src/logging/color_log_formatter.py index 53fe261..6f5545e 100644 --- a/src/logging/color_log_formatter.py +++ b/src/logging/color_log_formatter.py @@ -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": diff --git a/src/logging/session_file_handler.py b/src/logging/session_file_handler.py index bc6e06e..a6f46b3 100644 --- a/src/logging/session_file_handler.py +++ b/src/logging/session_file_handler.py @@ -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() diff --git a/src/logging/setup.py b/src/logging/setup.py index e9737cd..f3498cc 100644 --- a/src/logging/setup.py +++ b/src/logging/setup.py @@ -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) diff --git a/src/main.py b/src/main.py index 958aa20..6c59d22 100644 --- a/src/main.py +++ b/src/main.py @@ -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) diff --git a/src/preferences.py b/src/preferences.py index 6b7d105..9adf84b 100644 --- a/src/preferences.py +++ b/src/preferences.py @@ -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) diff --git a/src/store/managers/manager.py b/src/store/managers/manager.py index 4ef50d0..a727a22 100644 --- a/src/store/managers/manager.py +++ b/src/store/managers/manager.py @@ -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 diff --git a/src/store/pipeline.py b/src/store/pipeline.py index f3f2883..f552f04 100644 --- a/src/store/pipeline.py +++ b/src/store/pipeline.py @@ -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""" diff --git a/src/store/store.py b/src/store/store.py index 231bbdc..159da90 100644 --- a/src/store/store.py +++ b/src/store/store.py @@ -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 diff --git a/src/utils/create_dialog.py b/src/utils/create_dialog.py index c77f619..06219e2 100644 --- a/src/utils/create_dialog.py +++ b/src/utils/create_dialog.py @@ -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")) diff --git a/src/utils/migrate_files_v1_to_v2.py b/src/utils/migrate_files_v1_to_v2.py index 7c00c7e..41c166b 100644 --- a/src/utils/migrate_files_v1_to_v2.py +++ b/src/utils/migrate_files_v1_to_v2.py @@ -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. diff --git a/src/utils/rate_limiter.py b/src/utils/rate_limiter.py index 09b17a9..431af60 100644 --- a/src/utils/rate_limiter.py +++ b/src/utils/rate_limiter.py @@ -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 diff --git a/src/utils/relative_date.py b/src/utils/relative_date.py index 6ecde67..fbded0d 100644 --- a/src/utils/relative_date.py +++ b/src/utils/relative_date.py @@ -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: diff --git a/src/utils/save_cover.py b/src/utils/save_cover.py index 0bbcd4b..0bb2381 100644 --- a/src/utils/save_cover.py +++ b/src/utils/save_cover.py @@ -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" diff --git a/src/utils/steam.py b/src/utils/steam.py index e0cb0f2..bdc8d84 100644 --- a/src/utils/steam.py +++ b/src/utils/steam.py @@ -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. diff --git a/src/utils/steamgriddb.py b/src/utils/steamgriddb.py index 57dda3e..c87da1c 100644 --- a/src/utils/steamgriddb.py +++ b/src/utils/steamgriddb.py @@ -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 diff --git a/src/utils/task.py b/src/utils/task.py index 190d7c4..30ef68f 100644 --- a/src/utils/task.py +++ b/src/utils/task.py @@ -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: diff --git a/src/window.py b/src/window.py index 16e3588..b3af2f4 100644 --- a/src/window.py +++ b/src/window.py @@ -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()