fa5294b1d5
The factories are also usable in Rare's code to create compatible functions for the callbacks. If they there is no callback they just log what is happening. It also removes the need for `typing-extentions` module.
135 lines
3.6 KiB
Python
135 lines
3.6 KiB
Python
import logging
|
|
from dataclasses import dataclass
|
|
from typing import Callable, List, Optional, Dict
|
|
|
|
from rare.lgndr.models.downloading import UIUpdate
|
|
|
|
logger = logging.getLogger("LgndrMonkeys")
|
|
|
|
|
|
def get_boolean_choice_factory(value: bool = True) -> Callable[[str, bool], bool]:
|
|
def get_boolean_choice(prompt: str, default: bool) -> bool:
|
|
logger.debug("get_boolean_choice: %s, default: %s, choice: %s", prompt, default, value)
|
|
return value
|
|
return get_boolean_choice
|
|
|
|
|
|
def sdl_prompt_factory(install_tag: Optional[List[str]] = None) -> Callable[[Dict, str], List[str]]:
|
|
def sdl_prompt(sdl_data: Dict, title: str) -> List[str]:
|
|
logger.debug("sdl_prompt: %s", title)
|
|
for key in sdl_data.keys():
|
|
logger.debug("%s: %s %s", key, sdl_data[key]["tags"], sdl_data[key]["name"])
|
|
tags = install_tag if install_tag is not None else [""]
|
|
logger.debug("choice: %s, tags: %s", install_tag, tags)
|
|
return tags
|
|
return sdl_prompt
|
|
|
|
|
|
def verify_stdout_factory(
|
|
callback: Callable[[int, int, float, float], None] = None
|
|
) -> Callable[[int, int, float, float], None]:
|
|
def verify_stdout(a0: int, a1: int, a2: float, a3: float) -> None:
|
|
if callback is not None and callable(callback):
|
|
callback(a0, a1, a2, a3)
|
|
else:
|
|
logger.info("Verification progress: %d/%d (%.01f%%) [%.1f MiB/s]", a0, a1, a2, a3)
|
|
return verify_stdout
|
|
|
|
|
|
def ui_update_factory(callback: Callable[[UIUpdate], None] = None) -> Callable[[UIUpdate], None]:
|
|
def ui_update(status: UIUpdate) -> None:
|
|
if callback is not None and callable(callback):
|
|
callback(status)
|
|
else:
|
|
logger.info("Installation progress: %s", status)
|
|
return ui_update
|
|
|
|
|
|
class DLManagerSignals:
|
|
_kill = False
|
|
_update = False
|
|
|
|
@property
|
|
def kill(self) -> bool:
|
|
self._update = False
|
|
return self._kill
|
|
|
|
@kill.setter
|
|
def kill(self, value: bool) -> None:
|
|
self._update = True
|
|
self._kill = value
|
|
|
|
@property
|
|
def update(self) -> bool:
|
|
_update = self._update
|
|
self._update = False
|
|
return _update
|
|
|
|
|
|
@dataclass
|
|
class LgndrIndirectStatus:
|
|
success: bool = False
|
|
message: str = ""
|
|
|
|
def __len__(self):
|
|
if self.message:
|
|
return 2
|
|
else:
|
|
return 0
|
|
|
|
def __bool__(self):
|
|
return self.success
|
|
|
|
def __getitem__(self, item):
|
|
if item == 0:
|
|
return self.success
|
|
elif item == 1:
|
|
return self.message
|
|
else:
|
|
raise IndexError
|
|
|
|
def __iter__(self):
|
|
return iter((self.success, self.message))
|
|
|
|
def __str__(self):
|
|
return self.message
|
|
|
|
|
|
class LgndrIndirectLogger:
|
|
def __init__(
|
|
self, status: LgndrIndirectStatus, logger: logging.Logger = None, level: int = logging.ERROR
|
|
):
|
|
self.logger = logger
|
|
self.level = level
|
|
self.status = status
|
|
|
|
def set_logger(self, logger: logging.Logger):
|
|
self.logger = logger
|
|
|
|
def set_level(self, level: int):
|
|
self.level = level
|
|
|
|
def _log(self, level: int, msg: str):
|
|
self.status.success = True if level < self.level else False
|
|
self.status.message = msg
|
|
if self.logger:
|
|
self.logger.log(level, msg)
|
|
|
|
def debug(self, msg: str):
|
|
self._log(logging.DEBUG, msg)
|
|
|
|
def info(self, msg: str):
|
|
self._log(logging.INFO, msg)
|
|
|
|
def warning(self, msg: str):
|
|
self._log(logging.WARNING, msg)
|
|
|
|
def error(self, msg: str):
|
|
self._log(logging.ERROR, msg)
|
|
|
|
def critical(self, msg: str):
|
|
self._log(logging.CRITICAL, msg)
|
|
|
|
def fatal(self, msg: str):
|
|
self.critical(msg)
|