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

Typing

Browse Source
This commit is contained in:
kramo
2023-08-16 19:18:03 +02:00
parent a3aa7f9ccf
commit eeb18eb017
17 changed files with 102 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/details_window.py
View File

@@ -19,7 +19,7 @@
import os
from time import time
from typing import Any
from typing import Any, Optional
from gi.repository import Adw, Gio, GLib, Gtk
from PIL import Image
@@ -55,7 +55,7 @@ class DetailsWindow(Adw.Window):
cover_changed: bool = False
def __init__(self, game: Game | None = None, **kwargs: Any):
def __init__(self, game: Optional[Game] = None, **kwargs: Any):
super().__init__(**kwargs)
self.game: Game = game

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

8
src/game.py
View File

@@ -23,7 +23,7 @@ import shlex
import subprocess
from pathlib import Path
from time import time
from typing import Any
from typing import Any, Optional
from gi.repository import Adw, GLib, GObject, Gtk
@@ -57,7 +57,7 @@ class Game(Gtk.Box):
hidden: bool = False
last_played: int = 0
name: str
developer: str | None = None
developer: Optional[str] = None
removed: bool = False
blacklisted: bool = False
game_cover: GameCover = None
@@ -97,7 +97,7 @@ class Game(Gtk.Box):
def save(self) -> None:
self.emit("save-ready", {})
def create_toast(self, title: str, action: str | None = None) -> 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)
@@ -180,7 +180,7 @@ class Game(Gtk.Box):
self.cover.set_opacity(int(not loading))
self.spinner.set_spinning(loading)
def get_cover_path(self) -> Path | None:
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 # type: ignore

20
src/game_cover.py
View File

@@ -18,7 +18,7 @@
# SPDX-License-Identifier: GPL-3.0-or-later
from pathlib import Path
from typing import Any, Callable
from typing import Any, Callable, Optional
from gi.repository import Gdk, GdkPixbuf, Gio, GLib, Gtk
from PIL import Image, ImageFilter, ImageStat
@@ -27,12 +27,12 @@ from src import shared
class GameCover:
texture: Gdk.Texture | None
blurred: Gdk.Texture | None
luminance: tuple[float, float] | None
path: Path | None
animation: GdkPixbuf.PixbufAnimation | None
anim_iter: GdkPixbuf.PixbufAnimationIter | 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"
@@ -41,12 +41,12 @@ class GameCover:
shared.PREFIX + "/library_placeholder_small.svg"
)
def __init__(self, pictures: set[Gtk.Picture], path: Path | None = None) -> 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: Path | None) -> Callable:
def create_func(self, path: Optional[Path]) -> Callable:
self.animation = GdkPixbuf.PixbufAnimation.new_from_file(str(path))
self.anim_iter = self.animation.get_iter()
@@ -55,7 +55,7 @@ class GameCover:
return wrapper
def new_cover(self, path: Path | None = None) -> None:
def new_cover(self, path: Optional[Path] = None) -> None:
self.animation = None
self.texture = None
self.blurred = None

67
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,22 +38,22 @@ 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]
removed_game_ids: set[str] = set()
imported_game_ids: set[str] = set()
def __init__(self):
def __init__(self) -> None:
super().__init__()
# TODO: make this stateful
@@ -63,23 +64,23 @@ class Importer(ErrorProducer):
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:
@@ -87,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)
@@ -121,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(
@@ -138,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
@@ -192,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
@@ -225,7 +228,7 @@ class Importer(ErrorProducer):
if self.finished:
self.import_callback()
def remove_games(self):
def remove_games(self) -> None:
"""Set removed to True for missing games"""
if not shared.schema.get_boolean("remove-missing"):
return
@@ -249,7 +252,7 @@ class Importer(ErrorProducer):
game.update()
self.removed_game_ids.add(game.game_id)
def import_callback(self):
def import_callback(self) -> None:
"""Callback called when importing has finished"""
logging.info("Import done")
self.remove_games()
@@ -261,11 +264,11 @@ class Importer(ErrorProducer):
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())
@@ -313,7 +316,7 @@ class Importer(ErrorProducer):
dialog.present()
def undo_import(self, *_args):
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()
@@ -330,7 +333,7 @@ class Importer(ErrorProducer):
logging.info("Import undone")
def create_summary_toast(self):
def create_summary_toast(self) -> Adw.Toast:
"""N games imported, removed toast"""
toast = Adw.Toast(timeout=0, priority=Adw.ToastPriority.HIGH)
@@ -371,16 +374,20 @@ class Importer(ErrorProducer):
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":

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

@@ -1,7 +1,7 @@
import logging
from os import PathLike
from pathlib import Path
from typing import Iterable, Mapping, NamedTuple
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,
@@ -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

12
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, Collection, Generator
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,7 +53,7 @@ class Source(Iterable):
source_id: str
name: str
variant: str = None
variant: Optional[str] = None
available_on: set[str] = set()
iterable_class: type[SourceIterable]
@@ -65,7 +65,7 @@ class Source(Iterable):
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_
@@ -75,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
@property

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)

6
src/main.py
View File

@@ -21,7 +21,7 @@ import json
import lzma
import os
import sys
from typing import Any
from typing import Any, Optional
import gi
@@ -199,8 +199,8 @@ class CartridgesApplication(Adw.Application):
self,
_action: Any = None,
_parameter: Any = None,
page_name: (str | None) = None,
expander_row: (str | None) = None,
page_name: Optional[str] = None,
expander_row: Optional[str] = None,
) -> CartridgesWindow:
win = PreferencesWindow()
if page_name:

4
src/preferences.py
View File

@@ -21,7 +21,7 @@ import logging
import re
from pathlib import Path
from shutil import rmtree
from typing import Any, Callable
from typing import Any, Callable, Optional
from gi.repository import Adw, Gio, GLib, Gtk
@@ -215,7 +215,7 @@ class PreferencesWindow(Adw.PreferencesWindow):
)
def choose_folder(
self, _widget: Any, callback: Callable, callback_data: str | None = None
self, _widget: Any, callback: Callable, callback_data: Optional[str] = None
) -> None:
self.file_chooser.select_folder(self.win, None, callback, callback_data)

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

2
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

12
src/store/store.py
View File

@@ -18,7 +18,7 @@
# SPDX-License-Identifier: GPL-3.0-or-later
import logging
from typing import Any, Generator, MutableMapping
from typing import Any, Generator, MutableMapping, Optional
from src import shared
from src.game import Game
@@ -77,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])
@@ -108,8 +110,8 @@ class Store:
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

6
src/window.py
View File

@@ -17,7 +17,7 @@
#
# SPDX-License-Identifier: GPL-3.0-or-later
from typing import Any
from typing import Any, Optional
from gi.repository import Adw, Gio, GLib, Gtk
@@ -71,7 +71,7 @@ class CartridgesWindow(Adw.ApplicationWindow):
game_covers: dict = {}
toasts: dict = {}
active_game: Game
details_view_game_cover: GameCover | None = None
details_view_game_cover: Optional[GameCover] = None
sort_state: str = "a-z"
def __init__(self, **kwargs: Any) -> None:
@@ -334,7 +334,7 @@ class CartridgesWindow(Adw.ApplicationWindow):
index += 1
def on_undo_action(
self, _widget: Any, game: Game | None = None, undo: str | None = None
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 (