Core: Utils.py typing (#3064)
* Core: Utils.py typing `get_fuzzy_results` typing There are places that this is called with a `word_list` that is not a `Sequence`, and it is valid (e.g., `set` or `dict`). To decide the right type, we look at how `word_list` is used: - the parameter to `len` - requires `__len__` - the 2nd parameter to `map` - requires `__iter__` Then we look at https://docs.python.org/3/library/collections.abc.html#collections-abstract-base-classes and ask what is the simplest type that includes both `__len__` and `__iter__`: `Collection` (Python 3.8 requires using the alias in `typing`, instead of `collections.abc`) * a bit more typing and cleaning * fine, take away my fun for something that no one is ever going to see anyway... Co-authored-by: black-sliver <59490463+black-sliver@users.noreply.github.com> --------- Co-authored-by: black-sliver <59490463+black-sliver@users.noreply.github.com>
This commit is contained in:
parent
860ab10b0b
commit
d09b214309
56
Utils.py
56
Utils.py
|
@ -101,8 +101,7 @@ def cache_self1(function: typing.Callable[[S, T], RetType]) -> typing.Callable[[
|
||||||
|
|
||||||
@functools.wraps(function)
|
@functools.wraps(function)
|
||||||
def wrap(self: S, arg: T) -> RetType:
|
def wrap(self: S, arg: T) -> RetType:
|
||||||
cache: Optional[Dict[T, RetType]] = typing.cast(Optional[Dict[T, RetType]],
|
cache: Optional[Dict[T, RetType]] = getattr(self, cache_name, None)
|
||||||
getattr(self, cache_name, None))
|
|
||||||
if cache is None:
|
if cache is None:
|
||||||
res = function(self, arg)
|
res = function(self, arg)
|
||||||
setattr(self, cache_name, {arg: res})
|
setattr(self, cache_name, {arg: res})
|
||||||
|
@ -209,10 +208,11 @@ def output_path(*path: str) -> str:
|
||||||
|
|
||||||
def open_file(filename: typing.Union[str, "pathlib.Path"]) -> None:
|
def open_file(filename: typing.Union[str, "pathlib.Path"]) -> None:
|
||||||
if is_windows:
|
if is_windows:
|
||||||
os.startfile(filename)
|
os.startfile(filename) # type: ignore
|
||||||
else:
|
else:
|
||||||
from shutil import which
|
from shutil import which
|
||||||
open_command = which("open") if is_macos else (which("xdg-open") or which("gnome-open") or which("kde-open"))
|
open_command = which("open") if is_macos else (which("xdg-open") or which("gnome-open") or which("kde-open"))
|
||||||
|
assert open_command, "Didn't find program for open_file! Please report this together with system details."
|
||||||
subprocess.call([open_command, filename])
|
subprocess.call([open_command, filename])
|
||||||
|
|
||||||
|
|
||||||
|
@ -300,21 +300,21 @@ def get_options() -> Settings:
|
||||||
return get_settings()
|
return get_settings()
|
||||||
|
|
||||||
|
|
||||||
def persistent_store(category: str, key: typing.Any, value: typing.Any):
|
def persistent_store(category: str, key: str, value: typing.Any):
|
||||||
path = user_path("_persistent_storage.yaml")
|
path = user_path("_persistent_storage.yaml")
|
||||||
storage: dict = persistent_load()
|
storage = persistent_load()
|
||||||
category = storage.setdefault(category, {})
|
category_dict = storage.setdefault(category, {})
|
||||||
category[key] = value
|
category_dict[key] = value
|
||||||
with open(path, "wt") as f:
|
with open(path, "wt") as f:
|
||||||
f.write(dump(storage, Dumper=Dumper))
|
f.write(dump(storage, Dumper=Dumper))
|
||||||
|
|
||||||
|
|
||||||
def persistent_load() -> typing.Dict[str, dict]:
|
def persistent_load() -> Dict[str, Dict[str, Any]]:
|
||||||
storage = getattr(persistent_load, "storage", None)
|
storage: Union[Dict[str, Dict[str, Any]], None] = getattr(persistent_load, "storage", None)
|
||||||
if storage:
|
if storage:
|
||||||
return storage
|
return storage
|
||||||
path = user_path("_persistent_storage.yaml")
|
path = user_path("_persistent_storage.yaml")
|
||||||
storage: dict = {}
|
storage = {}
|
||||||
if os.path.exists(path):
|
if os.path.exists(path):
|
||||||
try:
|
try:
|
||||||
with open(path, "r") as f:
|
with open(path, "r") as f:
|
||||||
|
@ -323,7 +323,7 @@ def persistent_load() -> typing.Dict[str, dict]:
|
||||||
logging.debug(f"Could not read store: {e}")
|
logging.debug(f"Could not read store: {e}")
|
||||||
if storage is None:
|
if storage is None:
|
||||||
storage = {}
|
storage = {}
|
||||||
persistent_load.storage = storage
|
setattr(persistent_load, "storage", storage)
|
||||||
return storage
|
return storage
|
||||||
|
|
||||||
|
|
||||||
|
@ -365,6 +365,7 @@ def store_data_package_for_checksum(game: str, data: typing.Dict[str, Any]) -> N
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logging.debug(f"Could not store data package: {e}")
|
logging.debug(f"Could not store data package: {e}")
|
||||||
|
|
||||||
|
|
||||||
def get_default_adjuster_settings(game_name: str) -> Namespace:
|
def get_default_adjuster_settings(game_name: str) -> Namespace:
|
||||||
import LttPAdjuster
|
import LttPAdjuster
|
||||||
adjuster_settings = Namespace()
|
adjuster_settings = Namespace()
|
||||||
|
@ -383,7 +384,9 @@ def get_adjuster_settings(game_name: str) -> Namespace:
|
||||||
default_settings = get_default_adjuster_settings(game_name)
|
default_settings = get_default_adjuster_settings(game_name)
|
||||||
|
|
||||||
# Fill in any arguments from the argparser that we haven't seen before
|
# Fill in any arguments from the argparser that we haven't seen before
|
||||||
return Namespace(**vars(adjuster_settings), **{k:v for k,v in vars(default_settings).items() if k not in vars(adjuster_settings)})
|
return Namespace(**vars(adjuster_settings), **{
|
||||||
|
k: v for k, v in vars(default_settings).items() if k not in vars(adjuster_settings)
|
||||||
|
})
|
||||||
|
|
||||||
|
|
||||||
@cache_argsless
|
@cache_argsless
|
||||||
|
@ -407,13 +410,13 @@ safe_builtins = frozenset((
|
||||||
class RestrictedUnpickler(pickle.Unpickler):
|
class RestrictedUnpickler(pickle.Unpickler):
|
||||||
generic_properties_module: Optional[object]
|
generic_properties_module: Optional[object]
|
||||||
|
|
||||||
def __init__(self, *args, **kwargs):
|
def __init__(self, *args: Any, **kwargs: Any) -> None:
|
||||||
super(RestrictedUnpickler, self).__init__(*args, **kwargs)
|
super(RestrictedUnpickler, self).__init__(*args, **kwargs)
|
||||||
self.options_module = importlib.import_module("Options")
|
self.options_module = importlib.import_module("Options")
|
||||||
self.net_utils_module = importlib.import_module("NetUtils")
|
self.net_utils_module = importlib.import_module("NetUtils")
|
||||||
self.generic_properties_module = None
|
self.generic_properties_module = None
|
||||||
|
|
||||||
def find_class(self, module, name):
|
def find_class(self, module: str, name: str) -> type:
|
||||||
if module == "builtins" and name in safe_builtins:
|
if module == "builtins" and name in safe_builtins:
|
||||||
return getattr(builtins, name)
|
return getattr(builtins, name)
|
||||||
# used by MultiServer -> savegame/multidata
|
# used by MultiServer -> savegame/multidata
|
||||||
|
@ -437,7 +440,7 @@ class RestrictedUnpickler(pickle.Unpickler):
|
||||||
raise pickle.UnpicklingError(f"global '{module}.{name}' is forbidden")
|
raise pickle.UnpicklingError(f"global '{module}.{name}' is forbidden")
|
||||||
|
|
||||||
|
|
||||||
def restricted_loads(s):
|
def restricted_loads(s: bytes) -> Any:
|
||||||
"""Helper function analogous to pickle.loads()."""
|
"""Helper function analogous to pickle.loads()."""
|
||||||
return RestrictedUnpickler(io.BytesIO(s)).load()
|
return RestrictedUnpickler(io.BytesIO(s)).load()
|
||||||
|
|
||||||
|
@ -493,7 +496,7 @@ def init_logging(name: str, loglevel: typing.Union[str, int] = logging.INFO, wri
|
||||||
file_handler.setFormatter(logging.Formatter(log_format))
|
file_handler.setFormatter(logging.Formatter(log_format))
|
||||||
|
|
||||||
class Filter(logging.Filter):
|
class Filter(logging.Filter):
|
||||||
def __init__(self, filter_name, condition):
|
def __init__(self, filter_name: str, condition: typing.Callable[[logging.LogRecord], bool]) -> None:
|
||||||
super().__init__(filter_name)
|
super().__init__(filter_name)
|
||||||
self.condition = condition
|
self.condition = condition
|
||||||
|
|
||||||
|
@ -544,7 +547,7 @@ def init_logging(name: str, loglevel: typing.Union[str, int] = logging.INFO, wri
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
def stream_input(stream, queue):
|
def stream_input(stream: typing.TextIO, queue: "asyncio.Queue[str]"):
|
||||||
def queuer():
|
def queuer():
|
||||||
while 1:
|
while 1:
|
||||||
try:
|
try:
|
||||||
|
@ -572,7 +575,7 @@ class VersionException(Exception):
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
||||||
def chaining_prefix(index: int, labels: typing.Tuple[str]) -> str:
|
def chaining_prefix(index: int, labels: typing.Sequence[str]) -> str:
|
||||||
text = ""
|
text = ""
|
||||||
max_label = len(labels) - 1
|
max_label = len(labels) - 1
|
||||||
while index > max_label:
|
while index > max_label:
|
||||||
|
@ -595,7 +598,7 @@ def format_SI_prefix(value, power=1000, power_labels=("", "k", "M", "G", "T", "P
|
||||||
return f"{value.quantize(decimal.Decimal('1.00'))} {chaining_prefix(n, power_labels)}"
|
return f"{value.quantize(decimal.Decimal('1.00'))} {chaining_prefix(n, power_labels)}"
|
||||||
|
|
||||||
|
|
||||||
def get_fuzzy_results(input_word: str, wordlist: typing.Sequence[str], limit: typing.Optional[int] = None) \
|
def get_fuzzy_results(input_word: str, word_list: typing.Collection[str], limit: typing.Optional[int] = None) \
|
||||||
-> typing.List[typing.Tuple[str, int]]:
|
-> typing.List[typing.Tuple[str, int]]:
|
||||||
import jellyfish
|
import jellyfish
|
||||||
|
|
||||||
|
@ -603,21 +606,20 @@ def get_fuzzy_results(input_word: str, wordlist: typing.Sequence[str], limit: ty
|
||||||
return (1 - jellyfish.damerau_levenshtein_distance(word1.lower(), word2.lower())
|
return (1 - jellyfish.damerau_levenshtein_distance(word1.lower(), word2.lower())
|
||||||
/ max(len(word1), len(word2)))
|
/ max(len(word1), len(word2)))
|
||||||
|
|
||||||
limit: int = limit if limit else len(wordlist)
|
limit = limit if limit else len(word_list)
|
||||||
return list(
|
return list(
|
||||||
map(
|
map(
|
||||||
lambda container: (container[0], int(container[1]*100)), # convert up to limit to int %
|
lambda container: (container[0], int(container[1]*100)), # convert up to limit to int %
|
||||||
sorted(
|
sorted(
|
||||||
map(lambda candidate:
|
map(lambda candidate: (candidate, get_fuzzy_ratio(input_word, candidate)), word_list),
|
||||||
(candidate, get_fuzzy_ratio(input_word, candidate)),
|
|
||||||
wordlist),
|
|
||||||
key=lambda element: element[1],
|
key=lambda element: element[1],
|
||||||
reverse=True)[0:limit]
|
reverse=True
|
||||||
|
)[0:limit]
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
def open_filename(title: str, filetypes: typing.Sequence[typing.Tuple[str, typing.Sequence[str]]], suggest: str = "") \
|
def open_filename(title: str, filetypes: typing.Iterable[typing.Tuple[str, typing.Iterable[str]]], suggest: str = "") \
|
||||||
-> typing.Optional[str]:
|
-> typing.Optional[str]:
|
||||||
logging.info(f"Opening file input dialog for {title}.")
|
logging.info(f"Opening file input dialog for {title}.")
|
||||||
|
|
||||||
|
@ -734,7 +736,7 @@ def messagebox(title: str, text: str, error: bool = False) -> None:
|
||||||
root.update()
|
root.update()
|
||||||
|
|
||||||
|
|
||||||
def title_sorted(data: typing.Sequence, key=None, ignore: typing.Set = frozenset(("a", "the"))):
|
def title_sorted(data: typing.Iterable, key=None, ignore: typing.AbstractSet[str] = frozenset(("a", "the"))):
|
||||||
"""Sorts a sequence of text ignoring typical articles like "a" or "the" in the beginning."""
|
"""Sorts a sequence of text ignoring typical articles like "a" or "the" in the beginning."""
|
||||||
def sorter(element: Union[str, Dict[str, Any]]) -> str:
|
def sorter(element: Union[str, Dict[str, Any]]) -> str:
|
||||||
if (not isinstance(element, str)):
|
if (not isinstance(element, str)):
|
||||||
|
@ -788,7 +790,7 @@ class DeprecateDict(dict):
|
||||||
log_message: str
|
log_message: str
|
||||||
should_error: bool
|
should_error: bool
|
||||||
|
|
||||||
def __init__(self, message, error: bool = False) -> None:
|
def __init__(self, message: str, error: bool = False) -> None:
|
||||||
self.log_message = message
|
self.log_message = message
|
||||||
self.should_error = error
|
self.should_error = error
|
||||||
super().__init__()
|
super().__init__()
|
||||||
|
|
Loading…
Reference in New Issue