jeka/kiauh
1
0
Fork 0
mirror of https://github.com/dw-0/kiauh.git synced 2025-12-25 00:33:37 +05:00
Code Issues Packages Projects Releases Wiki Activity

feat: add mypy to the project

Browse Source 
Signed-off-by: Dominik Willner <th33xitus@gmail.com>
This commit is contained in:
dw-0
2024-08-04 16:15:59 +02:00
parent 0d7a471a03
commit 96daf966ee
55 changed files with 513 additions and 430 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
kiauh/components/klipper/klipper.py
View File

@@ -6,6 +6,8 @@
# #
# This file may be distributed under the terms of the GNU GPLv3 license #
# ======================================================================= #
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from subprocess import CalledProcessError, run
@@ -30,13 +32,13 @@ from utils.logger import Logger
class Klipper(BaseInstance):
klipper_dir: Path = KLIPPER_DIR
env_dir: Path = KLIPPER_ENV_DIR
cfg_file: Path = None
log: Path = None
serial: Path = None
uds: Path = None
cfg_file: Path | None = None
log: Path | None = None
serial: Path | None = None
uds: Path | None = None
def __init__(self, suffix: str = "") -> None:
super().__init__(instance_type=self, suffix=suffix)
super().__init__(suffix=suffix)
def __post_init__(self) -> None:
super().__post_init__()
@@ -132,15 +134,15 @@ class Klipper(BaseInstance):
)
env_file_content = env_file_content.replace(
"%SERIAL%",
self.serial.as_posix(),
self.serial.as_posix() if self.serial else "",
)
env_file_content = env_file_content.replace(
"%LOG%",
self.log.as_posix(),
self.log.as_posix() if self.log else "",
)
env_file_content = env_file_content.replace(
"%UDS%",
self.uds.as_posix(),
self.uds.as_posix() if self.uds else "",
)
return env_file_content

6
kiauh/components/klipper/klipper_dialogs.py
View File

@@ -34,7 +34,7 @@ def print_instance_overview(
show_index=False,
start_index=0,
show_select_all=False,
):
) -> None:
dialog = "╔═══════════════════════════════════════════════════════╗\n"
if show_headline:
d_type = (
@@ -64,7 +64,7 @@ def print_instance_overview(
print_back_footer()
def print_select_instance_count_dialog():
def print_select_instance_count_dialog() -> None:
line1 = f"{COLOR_YELLOW}WARNING:{RESET_FORMAT}"
line2 = f"{COLOR_YELLOW}Setting up too many instances may crash your system.{RESET_FORMAT}"
dialog = textwrap.dedent(
@@ -84,7 +84,7 @@ def print_select_instance_count_dialog():
print_back_footer()
def print_select_custom_name_dialog():
def print_select_custom_name_dialog() -> None:
line1 = f"{COLOR_YELLOW}INFO:{RESET_FORMAT}"
line2 = f"{COLOR_YELLOW}Only alphanumeric characters are allowed!{RESET_FORMAT}"
dialog = textwrap.dedent(

9
kiauh/components/klipper/klipper_remove.py
View File

@@ -6,8 +6,9 @@
# #
# This file may be distributed under the terms of the GNU GPLv3 license #
# ======================================================================= #
from __future__ import annotations
from typing import List, Union
from typing import List
from components.klipper import KLIPPER_DIR, KLIPPER_ENV_DIR
from components.klipper.klipper import Klipper
@@ -47,9 +48,7 @@ def run_klipper_removal(
run_remove_routines(KLIPPER_ENV_DIR)
def select_instances_to_remove(
instances: List[Klipper],
) -> Union[List[Klipper], None]:
def select_instances_to_remove(instances: List[Klipper]) -> List[Klipper] | None:
start_index = 1
options = [str(i + start_index) for i in range(len(instances))]
options.extend(["a", "b"])
@@ -76,7 +75,7 @@ def select_instances_to_remove(
def remove_instances(
instance_manager: InstanceManager,
instance_list: List[Klipper],
instance_list: List[Klipper] | None,
) -> None:
if not instance_list:
return

8
kiauh/components/klipper/klipper_setup.py
View File

@@ -129,10 +129,10 @@ def handle_instance_names(
def get_install_count_and_name_dict(
klipper_list: List[Klipper], moonraker_list: List[Moonraker]
) -> Tuple[int, Dict[int, str]]:
install_count: int | None
if len(moonraker_list) > len(klipper_list):
install_count = len(moonraker_list)
name_dict = {i: moonraker.suffix for i, moonraker in enumerate(moonraker_list)}
else:
install_count = get_install_count()
name_dict = {i: klipper.suffix for i, klipper in enumerate(klipper_list)}
@@ -217,11 +217,12 @@ def create_klipper_instance(name: str, create_example_cfg: bool) -> None:
def use_custom_names_or_go_back() -> bool | None:
print_select_custom_name_dialog()
return get_confirm(
_input: bool | None = get_confirm(
"Assign custom names?",
False,
allow_go_back=True,
)
return _input
def display_moonraker_info(moonraker_list: List[Moonraker]) -> bool:
@@ -238,4 +239,5 @@ def display_moonraker_info(moonraker_list: List[Moonraker]) -> bool:
padding_top=0,
padding_bottom=0,
)
return get_confirm("Proceed with installation?")
_input: bool = get_confirm("Proceed with installation?")
return _input

12
kiauh/components/klipper/klipper_utils.py
View File

@@ -46,10 +46,11 @@ def get_klipper_status() -> ComponentStatus:
return get_install_status(KLIPPER_DIR, KLIPPER_ENV_DIR, Klipper)
def add_to_existing() -> bool:
kl_instances = InstanceManager(Klipper).instances
def add_to_existing() -> bool | None:
kl_instances: List[Klipper] = InstanceManager(Klipper).instances
print_instance_overview(kl_instances)
return get_confirm("Add new instances?", allow_go_back=True)
_input: bool | None = get_confirm("Add new instances?", allow_go_back=True)
return _input
def get_install_count() -> int | None:
@@ -66,7 +67,8 @@ def get_install_count() -> int | None:
f"{' additional' if len(kl_instances) > 0 else ''} "
f"Klipper instances to set up"
)
return get_number_input(question, 1, default=1, allow_go_back=True)
_input: int | None = get_number_input(question, 1, default=1, allow_go_back=True)
return _input
def assign_custom_name(key: int, name_dict: Dict[int, str]) -> None:
@@ -79,7 +81,7 @@ def assign_custom_name(key: int, name_dict: Dict[int, str]) -> None:
name_dict[key] = get_string_input(question, exclude=existing_names, regex=pattern)
def check_user_groups():
def check_user_groups() -> None:
user_groups = [grp.getgrgid(gid).gr_name for gid in os.getgroups()]
missing_groups = [g for g in ["tty", "dialout"] if g not in user_groups]

13
kiauh/components/klipper/menus/klipper_remove_menu.py
View File

@@ -6,9 +6,10 @@
# #
# This file may be distributed under the terms of the GNU GPLv3 license #
# ======================================================================= #
from __future__ import annotations
import textwrap
from typing import Optional, Type
from typing import Type
from components.klipper import klipper_remove
from core.menus import FooterType, Option
@@ -18,21 +19,19 @@ from utils.constants import COLOR_CYAN, COLOR_RED, RESET_FORMAT
# noinspection PyUnusedLocal
class KlipperRemoveMenu(BaseMenu):
def __init__(self, previous_menu: Optional[Type[BaseMenu]] = None):
def __init__(self, previous_menu: Type[BaseMenu] | None = None):
super().__init__()
self.previous_menu = previous_menu
self.previous_menu: Type[BaseMenu] | None = previous_menu
self.footer_type = FooterType.BACK
self.remove_klipper_service = False
self.remove_klipper_dir = False
self.remove_klipper_env = False
self.selection_state = False
def set_previous_menu(self, previous_menu: Optional[Type[BaseMenu]]) -> None:
def set_previous_menu(self, previous_menu: Type[BaseMenu] | None) -> None:
from core.menus.remove_menu import RemoveMenu
self.previous_menu: Type[BaseMenu] = (
previous_menu if previous_menu is not None else RemoveMenu
)
self.previous_menu = previous_menu if previous_menu is not None else RemoveMenu
def set_options(self) -> None:
self.options = {

9
kiauh/components/klipper_firmware/firmware_utils.py
View File

@@ -24,12 +24,12 @@ from utils.sys_utils import log_process
def find_firmware_file() -> bool:
target = KLIPPER_DIR.joinpath("out")
target_exists = target.exists()
target_exists: bool = target.exists()
f1 = "klipper.elf.hex"
f2 = "klipper.elf"
f3 = "klipper.bin"
fw_file_exists = (
fw_file_exists: bool = (
target.joinpath(f1).exists() and target.joinpath(f2).exists()
) or target.joinpath(f3).exists()
@@ -75,10 +75,11 @@ def get_sd_flash_board_list() -> List[str]:
try:
cmd = f"{SD_FLASH_SCRIPT} -l"
blist = check_output(cmd, shell=True, text=True)
return blist.splitlines()[1:]
blist: List[str] = check_output(cmd, shell=True, text=True).splitlines()[1:]
return blist
except CalledProcessError as e:
Logger.print_error(f"An unexpected error occured:\n{e}")
return []
def start_flash_process(flash_options: FlashOptions) -> None:

2
kiauh/components/klipper_firmware/flash_options.py
View File

@@ -45,7 +45,7 @@ class FlashOptions:
return cls._instance
@classmethod
def destroy(cls):
def destroy(cls) -> None:
cls._instance = None
@property

14
kiauh/components/klipper_firmware/menus/klipper_build_menu.py
View File

@@ -6,9 +6,10 @@
# #
# This file may be distributed under the terms of the GNU GPLv3 license #
# ======================================================================= #
from __future__ import annotations
import textwrap
from typing import Optional, Type
from typing import Type
from components.klipper import KLIPPER_DIR
from components.klipper_firmware.firmware_utils import (
@@ -30,16 +31,16 @@ from utils.sys_utils import (
# noinspection PyUnusedLocal
# noinspection PyMethodMayBeStatic
class KlipperBuildFirmwareMenu(BaseMenu):
def __init__(self, previous_menu: Optional[Type[BaseMenu]] = None):
def __init__(self, previous_menu: Type[BaseMenu] | None = None):
super().__init__()
self.previous_menu = previous_menu
self.previous_menu: Type[BaseMenu] | None = previous_menu
self.deps = ["build-essential", "dpkg-dev", "make"]
self.missing_deps = check_package_install(self.deps)
def set_previous_menu(self, previous_menu: Optional[Type[BaseMenu]]) -> None:
def set_previous_menu(self, previous_menu: Type[BaseMenu] | None) -> None:
from core.menus.advanced_menu import AdvancedMenu
self.previous_menu: Type[BaseMenu] = (
self.previous_menu = (
previous_menu if previous_menu is not None else AdvancedMenu
)
@@ -109,4 +110,5 @@ class KlipperBuildFirmwareMenu(BaseMenu):
Logger.print_error("Building Klipper Firmware failed!")
finally:
self.previous_menu().run()
if self.previous_menu is not None:
self.previous_menu().run()

16
kiauh/components/klipper_firmware/menus/klipper_flash_error_menu.py
View File

@@ -6,8 +6,10 @@
# #
# This file may be distributed under the terms of the GNU GPLv3 license #
# ======================================================================= #
from __future__ import annotations
import textwrap
from typing import Optional, Type
from typing import Type
from components.klipper_firmware.flash_options import FlashMethod, FlashOptions
from core.menus import FooterType, Option
@@ -18,15 +20,15 @@ from utils.constants import COLOR_RED, RESET_FORMAT
# noinspection PyUnusedLocal
# noinspection PyMethodMayBeStatic
class KlipperNoFirmwareErrorMenu(BaseMenu):
def __init__(self, previous_menu: Optional[Type[BaseMenu]] = None):
def __init__(self, previous_menu: Type[BaseMenu] | None = None):
super().__init__()
self.previous_menu = previous_menu
self.previous_menu: Type[BaseMenu] | None = previous_menu
self.flash_options = FlashOptions()
self.footer_type = FooterType.BLANK
self.input_label_txt = "Press ENTER to go back to [Advanced Menu]"
def set_previous_menu(self, previous_menu: Optional[Type[BaseMenu]]) -> None:
def set_previous_menu(self, previous_menu: Type[BaseMenu] | None) -> None:
self.previous_menu = previous_menu
def set_options(self) -> None:
@@ -67,13 +69,13 @@ class KlipperNoFirmwareErrorMenu(BaseMenu):
# noinspection PyUnusedLocal
# noinspection PyMethodMayBeStatic
class KlipperNoBoardTypesErrorMenu(BaseMenu):
def __init__(self, previous_menu: Optional[Type[BaseMenu]] = None):
def __init__(self, previous_menu: Type[BaseMenu] | None = None):
super().__init__()
self.previous_menu = previous_menu
self.previous_menu: Type[BaseMenu] | None = previous_menu
self.footer_type = FooterType.BLANK
self.input_label_txt = "Press ENTER to go back to [Main Menu]"
def set_previous_menu(self, previous_menu: Optional[Type[BaseMenu]]) -> None:
def set_previous_menu(self, previous_menu: Type[BaseMenu] | None) -> None:
self.previous_menu = previous_menu
def set_options(self) -> None:

28
kiauh/components/klipper_firmware/menus/klipper_flash_help_menu.py
View File

@@ -6,8 +6,10 @@
# #
# This file may be distributed under the terms of the GNU GPLv3 license #
# ======================================================================= #
from __future__ import annotations
import textwrap
from typing import Optional, Type
from typing import Type
from core.menus.base_menu import BaseMenu
from utils.constants import COLOR_CYAN, COLOR_YELLOW, RESET_FORMAT
@@ -15,16 +17,16 @@ from utils.constants import COLOR_CYAN, COLOR_YELLOW, RESET_FORMAT
# noinspection DuplicatedCode
class KlipperFlashMethodHelpMenu(BaseMenu):
def __init__(self, previous_menu: Optional[Type[BaseMenu]] = None):
def __init__(self, previous_menu: Type[BaseMenu] | None = None):
super().__init__()
self.previous_menu = previous_menu
self.previous_menu: Type[BaseMenu] | None = previous_menu
def set_previous_menu(self, previous_menu: Optional[Type[BaseMenu]]) -> None:
def set_previous_menu(self, previous_menu: Type[BaseMenu] | None) -> None:
from components.klipper_firmware.menus.klipper_flash_menu import (
KlipperFlashMethodMenu,
)
self.previous_menu: Type[BaseMenu] = (
self.previous_menu = (
previous_menu if previous_menu is not None else KlipperFlashMethodMenu
)
@@ -73,16 +75,16 @@ class KlipperFlashMethodHelpMenu(BaseMenu):
# noinspection DuplicatedCode
class KlipperFlashCommandHelpMenu(BaseMenu):
def __init__(self, previous_menu: Optional[Type[BaseMenu]] = None):
def __init__(self, previous_menu: Type[BaseMenu] | None = None):
super().__init__()
self.previous_menu = previous_menu
self.previous_menu: Type[BaseMenu] | None = previous_menu
def set_previous_menu(self, previous_menu: Optional[Type[BaseMenu]]) -> None:
def set_previous_menu(self, previous_menu: Type[BaseMenu] | None) -> None:
from components.klipper_firmware.menus.klipper_flash_menu import (
KlipperFlashCommandMenu,
)
self.previous_menu: Type[BaseMenu] = (
self.previous_menu = (
previous_menu if previous_menu is not None else KlipperFlashCommandMenu
)
@@ -117,16 +119,16 @@ class KlipperFlashCommandHelpMenu(BaseMenu):
# noinspection DuplicatedCode
class KlipperMcuConnectionHelpMenu(BaseMenu):
def __init__(self, previous_menu: Optional[Type[BaseMenu]] = None):
def __init__(self, previous_menu: Type[BaseMenu] | None = None):
super().__init__()
self.previous_menu = previous_menu
self.previous_menu: Type[BaseMenu] | None = previous_menu
def set_previous_menu(self, previous_menu: Optional[Type[BaseMenu]]) -> None:
def set_previous_menu(self, previous_menu: Type[BaseMenu] | None) -> None:
from components.klipper_firmware.menus.klipper_flash_menu import (
KlipperSelectMcuConnectionMenu,
)
self.previous_menu: Type[BaseMenu] = (
self.previous_menu = (
previous_menu
if previous_menu is not None
else KlipperSelectMcuConnectionMenu

77
kiauh/components/klipper_firmware/menus/klipper_flash_menu.py
View File

@@ -6,10 +6,11 @@
# #
# This file may be distributed under the terms of the GNU GPLv3 license #
# ======================================================================= #
from __future__ import annotations
import textwrap
import time
from typing import Optional, Type
from typing import Type
from components.klipper_firmware.firmware_utils import (
find_firmware_file,
@@ -44,17 +45,17 @@ from utils.logger import DialogType, Logger
# noinspection PyUnusedLocal
# noinspection PyMethodMayBeStatic
class KlipperFlashMethodMenu(BaseMenu):
def __init__(self, previous_menu: Optional[Type[BaseMenu]] = None):
def __init__(self, previous_menu: Type[BaseMenu] | None = None):
super().__init__()
self.help_menu = KlipperFlashMethodHelpMenu
self.input_label_txt = "Select flash method"
self.footer_type = FooterType.BACK_HELP
self.flash_options = FlashOptions()
def set_previous_menu(self, previous_menu: Optional[Type[BaseMenu]]) -> None:
def set_previous_menu(self, previous_menu: Type[BaseMenu] | None) -> None:
from core.menus.advanced_menu import AdvancedMenu
self.previous_menu: Type[BaseMenu] = (
self.previous_menu = (
previous_menu if previous_menu is not None else AdvancedMenu
)
@@ -108,15 +109,15 @@ class KlipperFlashMethodMenu(BaseMenu):
# noinspection PyUnusedLocal
# noinspection PyMethodMayBeStatic
class KlipperFlashCommandMenu(BaseMenu):
def __init__(self, previous_menu: Optional[Type[BaseMenu]] = None):
def __init__(self, previous_menu: Type[BaseMenu] | None = None):
super().__init__()
self.help_menu = KlipperFlashCommandHelpMenu
self.input_label_txt = "Select flash command"
self.footer_type = FooterType.BACK_HELP
self.flash_options = FlashOptions()
def set_previous_menu(self, previous_menu: Optional[Type[BaseMenu]]) -> None:
self.previous_menu: Type[BaseMenu] = (
def set_previous_menu(self, previous_menu: Type[BaseMenu] | None) -> None:
self.previous_menu = (
previous_menu if previous_menu is not None else KlipperFlashMethodMenu
)
@@ -156,18 +157,18 @@ class KlipperFlashCommandMenu(BaseMenu):
# noinspection PyMethodMayBeStatic
class KlipperSelectMcuConnectionMenu(BaseMenu):
def __init__(
self, previous_menu: Optional[Type[BaseMenu]] = None, standalone: bool = False
self, previous_menu: Type[BaseMenu] | None = None, standalone: bool = False
):
super().__init__()
self.previous_menu = previous_menu
self.previous_menu: Type[BaseMenu] | None = previous_menu
self.__standalone = standalone
self.help_menu = KlipperMcuConnectionHelpMenu
self.input_label_txt = "Select connection type"
self.footer_type = FooterType.BACK_HELP
self.flash_options = FlashOptions()
def set_previous_menu(self, previous_menu: Optional[Type[BaseMenu]]) -> None:
self.previous_menu: Type[BaseMenu] = (
def set_previous_menu(self, previous_menu: Type[BaseMenu] | None) -> None:
self.previous_menu = (
previous_menu if previous_menu is not None else KlipperFlashCommandMenu
)
@@ -243,15 +244,15 @@ class KlipperSelectMcuConnectionMenu(BaseMenu):
# noinspection PyUnusedLocal
# noinspection PyMethodMayBeStatic
class KlipperSelectMcuIdMenu(BaseMenu):
def __init__(self, previous_menu: Optional[Type[BaseMenu]] = None):
def __init__(self, previous_menu: Type[BaseMenu] | None = None):
super().__init__()
self.flash_options = FlashOptions()
self.mcu_list = self.flash_options.mcu_list
self.input_label_txt = "Select MCU to flash"
self.footer_type = FooterType.BACK_HELP
def set_previous_menu(self, previous_menu: Optional[Type[BaseMenu]]) -> None:
self.previous_menu: Type[BaseMenu] = (
def set_previous_menu(self, previous_menu: Type[BaseMenu] | None) -> None:
self.previous_menu = (
previous_menu
if previous_menu is not None
else KlipperSelectMcuConnectionMenu
@@ -288,27 +289,35 @@ class KlipperSelectMcuIdMenu(BaseMenu):
print(menu, end="\n")
def flash_mcu(self, **kwargs):
index = int(kwargs.get("opt_index"))
selected_mcu = self.mcu_list[index]
self.flash_options.selected_mcu = selected_mcu
try:
index: int | None = kwargs.get("opt_index", None)
if index is None:
raise Exception("opt_index is None")
if self.flash_options.flash_method == FlashMethod.SD_CARD:
KlipperSelectSDFlashBoardMenu(previous_menu=self.__class__).run()
elif self.flash_options.flash_method == FlashMethod.REGULAR:
KlipperFlashOverviewMenu(previous_menu=self.__class__).run()
index = int(index)
selected_mcu = self.mcu_list[index]
self.flash_options.selected_mcu = selected_mcu
if self.flash_options.flash_method == FlashMethod.SD_CARD:
KlipperSelectSDFlashBoardMenu(previous_menu=self.__class__).run()
elif self.flash_options.flash_method == FlashMethod.REGULAR:
KlipperFlashOverviewMenu(previous_menu=self.__class__).run()
except Exception as e:
Logger.print_error(e)
Logger.print_error("Flashing failed!")
# noinspection PyUnusedLocal
# noinspection PyMethodMayBeStatic
class KlipperSelectSDFlashBoardMenu(BaseMenu):
def __init__(self, previous_menu: Optional[Type[BaseMenu]] = None):
def __init__(self, previous_menu: Type[BaseMenu] | None = None):
super().__init__()
self.flash_options = FlashOptions()
self.available_boards = get_sd_flash_board_list()
self.input_label_txt = "Select board type"
def set_previous_menu(self, previous_menu: Optional[Type[BaseMenu]]) -> None:
self.previous_menu: Type[BaseMenu] = (
def set_previous_menu(self, previous_menu: Type[BaseMenu] | None) -> None:
self.previous_menu = (
previous_menu if previous_menu is not None else KlipperSelectMcuIdMenu
)
@@ -340,9 +349,17 @@ class KlipperSelectSDFlashBoardMenu(BaseMenu):
print(menu, end="")
def board_select(self, **kwargs):
board = int(kwargs.get("opt_index"))
self.flash_options.selected_board = self.available_boards[board]
self.baudrate_select()
try:
index: int | None = kwargs.get("opt_index", None)
if index is None:
raise Exception("opt_index is None")
index = int(index)
self.flash_options.selected_board = self.available_boards[index]
self.baudrate_select()
except Exception as e:
Logger.print_error(e)
Logger.print_error("Board selection failed!")
def baudrate_select(self, **kwargs):
Logger.print_dialog(
@@ -366,13 +383,13 @@ class KlipperSelectSDFlashBoardMenu(BaseMenu):
# noinspection PyUnusedLocal
# noinspection PyMethodMayBeStatic
class KlipperFlashOverviewMenu(BaseMenu):
def __init__(self, previous_menu: Optional[Type[BaseMenu]] = None):
def __init__(self, previous_menu: Type[BaseMenu] | None = None):
super().__init__()
self.flash_options = FlashOptions()
self.input_label_txt = "Perform action (default=Y)"
def set_previous_menu(self, previous_menu: Optional[Type[BaseMenu]]) -> None:
self.previous_menu: Type[BaseMenu] = previous_menu
def set_previous_menu(self, previous_menu: Type[BaseMenu] | None) -> None:
self.previous_menu: Type[BaseMenu] | None = previous_menu
def set_options(self) -> None:
self.options = {

28
kiauh/components/log_uploads/menus/log_upload_menu.py
View File

@@ -6,29 +6,29 @@
# #
# This file may be distributed under the terms of the GNU GPLv3 license #
# ======================================================================= #
from __future__ import annotations
import textwrap
from typing import Optional, Type
from typing import Type
from components.log_uploads.log_upload_utils import get_logfile_list, upload_logfile
from core.menus import Option
from core.menus.base_menu import BaseMenu
from utils.constants import COLOR_YELLOW, RESET_FORMAT
from utils.logger import Logger
# noinspection PyMethodMayBeStatic
class LogUploadMenu(BaseMenu):
def __init__(self, previous_menu: Optional[Type[BaseMenu]] = None):
def __init__(self, previous_menu: Type[BaseMenu] | None = None):
super().__init__()
self.previous_menu = previous_menu
self.previous_menu: Type[BaseMenu] | None = previous_menu
self.logfile_list = get_logfile_list()
def set_previous_menu(self, previous_menu: Optional[Type[BaseMenu]]) -> None:
def set_previous_menu(self, previous_menu: Type[BaseMenu] | None) -> None:
from core.menus.main_menu import MainMenu
self.previous_menu: Type[BaseMenu] = (
previous_menu if previous_menu is not None else MainMenu
)
self.previous_menu = previous_menu if previous_menu is not None else MainMenu
def set_options(self) -> None:
self.options = {
@@ -36,7 +36,7 @@ class LogUploadMenu(BaseMenu):
for index in range(len(self.logfile_list))
}
def print_menu(self):
def print_menu(self) -> None:
header = " [ Log Upload ] "
color = COLOR_YELLOW
count = 62 - len(color) - len(RESET_FORMAT)
@@ -58,5 +58,13 @@ class LogUploadMenu(BaseMenu):
print(menu, end="")
def upload(self, **kwargs):
index = int(kwargs.get("opt_index"))
upload_logfile(self.logfile_list[index])
try:
index: int | None = kwargs.get("opt_index", None)
if index is None:
raise Exception("opt_index is None")
index = int(index)
upload_logfile(self.logfile_list[index])
except Exception as e:
Logger.print_error(e)
Logger.print_error("Log upload failed!")

13
kiauh/components/moonraker/menus/moonraker_remove_menu.py
View File

@@ -6,9 +6,10 @@
# #
# This file may be distributed under the terms of the GNU GPLv3 license #
# ======================================================================= #
from __future__ import annotations
import textwrap
from typing import Optional, Type
from typing import Type
from components.moonraker import moonraker_remove
from core.menus import Option
@@ -18,21 +19,19 @@ from utils.constants import COLOR_CYAN, COLOR_RED, RESET_FORMAT
# noinspection PyUnusedLocal
class MoonrakerRemoveMenu(BaseMenu):
def __init__(self, previous_menu: Optional[Type[BaseMenu]] = None):
def __init__(self, previous_menu: Type[BaseMenu] | None = None):
super().__init__()
self.previous_menu = previous_menu
self.previous_menu: Type[BaseMenu] | None = previous_menu
self.remove_moonraker_service = False
self.remove_moonraker_dir = False
self.remove_moonraker_env = False
self.remove_moonraker_polkit = False
self.selection_state = False
def set_previous_menu(self, previous_menu: Optional[Type[BaseMenu]]) -> None:
def set_previous_menu(self, previous_menu: Type[BaseMenu] | None) -> None:
from core.menus.remove_menu import RemoveMenu
self.previous_menu: Type[BaseMenu] = (
previous_menu if previous_menu is not None else RemoveMenu
)
self.previous_menu = previous_menu if previous_menu is not None else RemoveMenu
def set_options(self) -> None:
self.options = {

18
kiauh/components/moonraker/moonraker.py
View File

@@ -33,15 +33,15 @@ from utils.logger import Logger
class Moonraker(BaseInstance):
moonraker_dir: Path = MOONRAKER_DIR
env_dir: Path = MOONRAKER_ENV_DIR
cfg_file: Path = None
port: int = None
backup_dir: Path = None
certs_dir: Path = None
db_dir: Path = None
log: Path = None
cfg_file: Path | None = None
port: int | None = None
backup_dir: Path | None = None
certs_dir: Path | None = None
db_dir: Path | None = None
log: Path | None = None
def __init__(self, suffix: str = ""):
super().__init__(instance_type=self, suffix=suffix)
super().__init__(suffix=suffix)
def __post_init__(self) -> None:
super().__post_init__()
@@ -140,11 +140,11 @@ class Moonraker(BaseInstance):
return env_file_content
def _get_port(self) -> int | None:
if not self.cfg_file.is_file():
if not self.cfg_file or not self.cfg_file.is_file():
return None
scp = SimpleConfigParser()
scp.read(self.cfg_file)
port = scp.getint("server", "port", fallback=None)
port: int | None = scp.getint("server", "port", fallback=None)
return port

10
kiauh/components/moonraker/moonraker_remove.py
View File

@@ -6,9 +6,10 @@
# #
# This file may be distributed under the terms of the GNU GPLv3 license #
# ======================================================================= #
from __future__ import annotations
from subprocess import DEVNULL, PIPE, CalledProcessError, run
from typing import List, Union
from typing import List
from components.klipper.klipper_dialogs import print_instance_overview
from components.moonraker import MOONRAKER_DIR, MOONRAKER_ENV_DIR
@@ -57,7 +58,7 @@ def run_moonraker_removal(
def select_instances_to_remove(
instances: List[Moonraker],
) -> Union[List[Moonraker], None]:
) -> List[Moonraker] | None:
start_index = 1
options = [str(i + start_index) for i in range(len(instances))]
options.extend(["a", "b"])
@@ -84,8 +85,11 @@ def select_instances_to_remove(
def remove_instances(
instance_manager: InstanceManager,
instance_list: List[Moonraker],
instance_list: List[Moonraker] | None,
) -> None:
if not instance_list:
Logger.print_info("No Moonraker instances found. Skipped ...")
return
for instance in instance_list:
Logger.print_status(f"Removing instance {instance.get_service_file_name()} ...")
instance_manager.current_instance = instance

5
kiauh/components/moonraker/moonraker_setup.py
View File

@@ -89,7 +89,10 @@ def install_moonraker() -> None:
if selected_option == "a":
instance_names.extend([k.suffix for k in klipper_list])
else:
instance_names.append(options.get(selected_option).suffix)
klipper_instance: Klipper | None = options.get(selected_option)
if klipper_instance is None:
raise Exception("Error selecting instance!")
instance_names.append(klipper_instance.suffix)
create_example_cfg = get_confirm("Create example moonraker.conf?")

2
kiauh/components/moonraker/moonraker_utils.py
View File

@@ -126,7 +126,7 @@ def create_example_moonraker_conf(
Logger.print_ok(f"Example moonraker.conf created in '{instance.cfg_dir}'")
def backup_moonraker_dir():
def backup_moonraker_dir() -> None:
bm = BackupManager()
bm.backup_directory("moonraker", source=MOONRAKER_DIR, target=MOONRAKER_BACKUP_DIR)
bm.backup_directory(

3
kiauh/components/octoeverywhere/octoeverywhere.py
View File

@@ -26,6 +26,7 @@ from core.instance_manager.base_instance import BaseInstance
from utils.logger import Logger
# todo: make this to a dataclass
class Octoeverywhere(BaseInstance):
@classmethod
def blacklist(cls) -> List[str]:
@@ -64,7 +65,7 @@ class Octoeverywhere(BaseInstance):
raise
@staticmethod
def update():
def update() -> None:
try:
run(OE_UPDATE_SCRIPT.as_posix(), check=True, shell=True, cwd=OE_DIR)

3
kiauh/components/webui_client/client_config/client_config_setup.py
View File

@@ -6,6 +6,7 @@
# #
# This file may be distributed under the terms of the GNU GPLv3 license #
# ======================================================================= #
from __future__ import annotations
import shutil
import subprocess
@@ -112,7 +113,7 @@ def update_client_config(client: BaseWebClient) -> None:
def create_client_config_symlink(
client_config: BaseWebClientConfig, klipper_instances: List[Klipper] = None
client_config: BaseWebClientConfig, klipper_instances: List[Klipper] | None = None
) -> None:
if klipper_instances is None:
kl_im = InstanceManager(Klipper)

8
kiauh/components/webui_client/client_dialogs.py
View File

@@ -13,7 +13,7 @@ from components.webui_client.base_data import BaseWebClient
from utils.logger import DialogType, Logger
def print_moonraker_not_found_dialog():
def print_moonraker_not_found_dialog() -> None:
Logger.print_dialog(
DialogType.WARNING,
[
@@ -29,7 +29,7 @@ def print_moonraker_not_found_dialog():
)
def print_client_already_installed_dialog(name: str):
def print_client_already_installed_dialog(name: str) -> None:
Logger.print_dialog(
DialogType.WARNING,
[
@@ -41,7 +41,9 @@ def print_client_already_installed_dialog(name: str):
)
def print_client_port_select_dialog(name: str, port: int, ports_in_use: List[int]):
def print_client_port_select_dialog(
name: str, port: int, ports_in_use: List[int]
) -> None:
Logger.print_dialog(
DialogType.CUSTOM,
[

6
kiauh/components/webui_client/client_utils.py
View File

@@ -73,7 +73,7 @@ def get_current_client_config(clients: List[BaseWebClient]) -> str:
return f"{COLOR_CYAN}-{RESET_FORMAT}"
def backup_mainsail_config_json(is_temp=False) -> None:
def backup_mainsail_config_json(is_temp: bool = False) -> None:
c_json = MainsailData().client_dir.joinpath("config.json")
bm = BackupManager()
if is_temp:
@@ -137,7 +137,7 @@ def get_local_client_version(client: BaseWebClient) -> str | None:
if relinfo_file.is_file():
with open(relinfo_file, "r") as f:
return json.load(f)["version"]
return str(json.load(f)["version"])
else:
with open(version_file, "r") as f:
return f.readlines()[0]
@@ -146,7 +146,7 @@ def get_local_client_version(client: BaseWebClient) -> str | None:
def get_remote_client_version(client: BaseWebClient) -> str | None:
try:
if (tag := get_latest_tag(client.repo_path)) != "":
return tag
return str(tag)
return None
except Exception:
return None

3
kiauh/components/webui_client/fluidd_data.py
View File

@@ -48,7 +48,8 @@ class FluiddData(BaseWebClient):
def download_url(self) -> str:
from components.webui_client.client_utils import get_download_url
return get_download_url(self.BASE_DL_URL, self)
url: str = get_download_url(self.BASE_DL_URL, self)
return url
@property
def client_config(self) -> BaseWebClientConfig:

3
kiauh/components/webui_client/mainsail_data.py
View File

@@ -48,7 +48,8 @@ class MainsailData(BaseWebClient):
def download_url(self) -> str:
from components.webui_client.client_utils import get_download_url
return get_download_url(self.BASE_DL_URL, self)
url: str = get_download_url(self.BASE_DL_URL, self)
return url
@property
def client_config(self) -> BaseWebClientConfig:

23
kiauh/components/webui_client/menus/client_remove_menu.py
View File

@@ -6,9 +6,10 @@
# #
# This file may be distributed under the terms of the GNU GPLv3 license #
# ======================================================================= #
from __future__ import annotations
import textwrap
from typing import Optional, Type
from typing import Type
from components.webui_client import client_remove
from components.webui_client.base_data import BaseWebClient, WebClientType
@@ -20,22 +21,20 @@ from utils.constants import COLOR_CYAN, COLOR_RED, RESET_FORMAT
# noinspection PyUnusedLocal
class ClientRemoveMenu(BaseMenu):
def __init__(
self, client: BaseWebClient, previous_menu: Optional[Type[BaseMenu]] = None
self, client: BaseWebClient, previous_menu: Type[BaseMenu] | None = None
):
super().__init__()
self.previous_menu = previous_menu
self.client = client
self.remove_client = False
self.remove_client_cfg = False
self.backup_mainsail_config_json = False
self.selection_state = False
self.previous_menu: Type[BaseMenu] | None = previous_menu
self.client: BaseWebClient = client
self.remove_client: bool = False
self.remove_client_cfg: bool = False
self.backup_mainsail_config_json: bool = False
self.selection_state: bool = False
def set_previous_menu(self, previous_menu: Optional[Type[BaseMenu]]) -> None:
def set_previous_menu(self, previous_menu: Type[BaseMenu] | None) -> None:
from core.menus.remove_menu import RemoveMenu
self.previous_menu: Type[BaseMenu] = (
previous_menu if previous_menu is not None else RemoveMenu
)
self.previous_menu = previous_menu if previous_menu is not None else RemoveMenu
def set_options(self) -> None:
self.options = {

15
kiauh/core/backup_manager/backup_manager.py
View File

@@ -6,6 +6,7 @@
# #
# This file may be distributed under the terms of the GNU GPLv3 license #
# ======================================================================= #
from __future__ import annotations
import shutil
from pathlib import Path
@@ -20,8 +21,8 @@ from utils.logger import Logger
# noinspection PyMethodMayBeStatic
class BackupManager:
def __init__(self, backup_root_dir: Path = BACKUP_ROOT_DIR):
self._backup_root_dir = backup_root_dir
self._ignore_folders = None
self._backup_root_dir: Path = backup_root_dir
self._ignore_folders: List[str] = []
@property
def backup_root_dir(self) -> Path:
@@ -39,7 +40,7 @@ class BackupManager:
def ignore_folders(self, value: List[str]):
self._ignore_folders = value
def backup_file(self, file: Path, target: Path = None, custom_filename=None):
def backup_file(self, file: Path, target: Path | None = None, custom_filename=None):
Logger.print_status(f"Creating backup of {file} ...")
if not file.exists():
@@ -62,7 +63,9 @@ class BackupManager:
else:
Logger.print_info(f"File '{file}' not found ...")
def backup_directory(self, name: str, source: Path, target: Path = None) -> None:
def backup_directory(
self, name: str, source: Path, target: Path | None = None
) -> None:
Logger.print_status(f"Creating backup of {name} in {target} ...")
if source is None or not Path(source).exists():
@@ -83,9 +86,9 @@ class BackupManager:
Logger.print_error(f"Unable to backup directory '{source}':\n{e}")
return
def ignore_folders_func(self, dirpath, filenames):
def ignore_folders_func(self, dirpath, filenames) -> List[str]:
return (
[f for f in filenames if f in self._ignore_folders]
if self._ignore_folders is not None
if self._ignore_folders
else []
)

39
kiauh/core/instance_manager/base_instance.py
View File

@@ -13,7 +13,7 @@ import re
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from pathlib import Path
from typing import List, Optional
from typing import List
from utils.constants import CURRENT_USER, SYSTEMD
from utils.logger import Logger
@@ -21,26 +21,26 @@ from utils.logger import Logger
@dataclass
class BaseInstance(ABC):
instance_type: BaseInstance
suffix: str
user: str = field(default=CURRENT_USER, init=False)
data_dir: Path = None
data_dir: Path | None = None
data_dir_name: str = ""
is_legacy_instance: bool = False
cfg_dir: Path = None
log_dir: Path = None
comms_dir: Path = None
sysd_dir: Path = None
gcodes_dir: Path = None
cfg_dir: Path | None = None
log_dir: Path | None = None
comms_dir: Path | None = None
sysd_dir: Path | None = None
gcodes_dir: Path | None = None
def __post_init__(self) -> None:
self._set_data_dir()
self._set_is_legacy_instance()
self.cfg_dir = self.data_dir.joinpath("config")
self.log_dir = self.data_dir.joinpath("logs")
self.comms_dir = self.data_dir.joinpath("comms")
self.sysd_dir = self.data_dir.joinpath("systemd")
self.gcodes_dir = self.data_dir.joinpath("gcodes")
if self.data_dir is not None:
self.cfg_dir = self.data_dir.joinpath("config")
self.log_dir = self.data_dir.joinpath("logs")
self.comms_dir = self.data_dir.joinpath("comms")
self.sysd_dir = self.data_dir.joinpath("systemd")
self.gcodes_dir = self.data_dir.joinpath("gcodes")
@classmethod
def blacklist(cls) -> List[str]:
@@ -54,8 +54,8 @@ class BaseInstance(ABC):
def delete(self) -> None:
raise NotImplementedError("Subclasses must implement the delete method")
def create_folders(self, add_dirs: Optional[List[Path]] = None) -> None:
dirs = [
def create_folders(self, add_dirs: List[Path] | None = None) -> None:
dirs: List[Path | None] = [
self.data_dir,
self.cfg_dir,
self.log_dir,
@@ -68,13 +68,15 @@ class BaseInstance(ABC):
dirs.extend(add_dirs)
for _dir in dirs:
if _dir is None:
continue
_dir.mkdir(exist_ok=True)
# todo: refactor into a set method and access the value by accessing the property
def get_service_file_name(self, extension: bool = False) -> str:
from utils.common import convert_camelcase_to_kebabcase
name = convert_camelcase_to_kebabcase(self.__class__.__name__)
name: str = convert_camelcase_to_kebabcase(self.__class__.__name__)
if self.suffix != "":
name += f"-{self.suffix}"
@@ -82,12 +84,13 @@ class BaseInstance(ABC):
# todo: refactor into a set method and access the value by accessing the property
def get_service_file_path(self) -> Path:
return SYSTEMD.joinpath(self.get_service_file_name(extension=True))
path: Path = SYSTEMD.joinpath(self.get_service_file_name(extension=True))
return path
def delete_logfiles(self, log_name: str) -> None:
from utils.fs_utils import run_remove_routines
if not self.log_dir.exists():
if not self.log_dir or not self.log_dir.exists():
return
files = self.log_dir.iterdir()

54
kiauh/core/instance_manager/instance_manager.py
View File

@@ -6,64 +6,66 @@
# #
# This file may be distributed under the terms of the GNU GPLv3 license #
# ======================================================================= #
from __future__ import annotations
import re
import subprocess
from pathlib import Path
from typing import List, Optional, TypeVar, Union
from typing import List, Type, TypeVar
from core.instance_manager.base_instance import BaseInstance
from utils.constants import SYSTEMD
from utils.logger import Logger
from utils.sys_utils import cmd_sysctl_service
T = TypeVar(name="T", bound=BaseInstance, covariant=True)
T = TypeVar("T", bound=BaseInstance, covariant=True)
# noinspection PyMethodMayBeStatic
class InstanceManager:
def __init__(self, instance_type: T) -> None:
def __init__(self, instance_type: Type[T]) -> None:
self._instance_type = instance_type
self._current_instance: Optional[T] = None
self._instance_suffix: Optional[str] = None
self._instance_service: Optional[str] = None
self._instance_service_full: Optional[str] = None
self._instance_service_path: Optional[str] = None
self._current_instance: Type[T] | None = None
self._instance_suffix: str | None = None
self._instance_service: str | None = None
self._instance_service_full: str | None = None
self._instance_service_path: str | None = None
self._instances: List[T] = []
@property
def instance_type(self) -> T:
def instance_type(self) -> Type[T]:
return self._instance_type
@instance_type.setter
def instance_type(self, value: T):
def instance_type(self, value: Type[T]):
self._instance_type = value
@property
def current_instance(self) -> T:
def current_instance(self) -> Type[T] | None:
return self._current_instance
@current_instance.setter
def current_instance(self, value: T) -> None:
def current_instance(self, value: Type[T] | None) -> None:
self._current_instance = value
self.instance_suffix = value.suffix
self.instance_service = value.get_service_file_name()
self.instance_service_path = value.get_service_file_path()
if value is not None:
self.instance_suffix = value.suffix
self.instance_service = value.get_service_file_name()
self.instance_service_path = value.get_service_file_path()
@property
def instance_suffix(self) -> str:
def instance_suffix(self) -> str | None:
return self._instance_suffix
@instance_suffix.setter
def instance_suffix(self, value: str):
def instance_suffix(self, value: str | None):
self._instance_suffix = value
@property
def instance_service(self) -> str:
def instance_service(self) -> str | None:
return self._instance_service
@instance_service.setter
def instance_service(self, value: str):
def instance_service(self, value: str | None) -> None:
self._instance_service = value
@property
@@ -71,19 +73,19 @@ class InstanceManager:
return f"{self._instance_service}.service"
@property
def instance_service_path(self) -> str:
def instance_service_path(self) -> str | None:
return self._instance_service_path
@instance_service_path.setter
def instance_service_path(self, value: str):
def instance_service_path(self, value: str | None) -> None:
self._instance_service_path = value
@property
def instances(self) -> List[T]:
def instances(self) -> List[Type[T]]:
return self.find_instances()
@instances.setter
def instances(self, value: List[T]):
def instances(self, value: List[T]) -> None:
self._instances = value
def create_instance(self) -> None:
@@ -157,7 +159,7 @@ class InstanceManager:
self.current_instance = instance
self.stop_instance()
def find_instances(self) -> List[T]:
def find_instances(self) -> List[Type[T]]:
from utils.common import convert_camelcase_to_kebabcase
name = convert_camelcase_to_kebabcase(self.instance_type.__name__)
@@ -185,10 +187,10 @@ class InstanceManager:
suffix = file_path.stem[len(name) :]
return suffix[1:] if suffix else ""
def _sort_instance_list(self, suffix: Union[int, str, None]):
def _sort_instance_list(self, suffix: int | str | None):
if suffix is None:
return
elif suffix.isdigit():
elif isinstance(suffix, str) and suffix.isdigit():
return f"{int(suffix):04}"
else:
return suffix

5
kiauh/core/menus/__init__.py
View File

@@ -6,10 +6,11 @@
# #
# This file may be distributed under the terms of the GNU GPLv3 license #
# ======================================================================= #
from __future__ import annotations
from dataclasses import dataclass
from enum import Enum
from typing import Any, Callable, Union
from typing import Any, Callable
@dataclass
@@ -22,7 +23,7 @@ class Option:
:param opt_data: Can be used to pass any additional data to the menu option
"""
method: Union[Callable, None] = None
method: Callable | None = None
menu: bool = False
opt_index: str = ""
opt_data: Any = None

29
kiauh/core/menus/advanced_menu.py
View File

@@ -6,9 +6,10 @@
# #
# This file may be distributed under the terms of the GNU GPLv3 license #
# ======================================================================= #
from __future__ import annotations
import textwrap
from typing import Optional, Type
from typing import Type
from components.klipper import KLIPPER_DIR
from components.klipper.klipper import Klipper
@@ -30,18 +31,16 @@ from utils.git_utils import rollback_repository
# noinspection PyUnusedLocal
# noinspection PyMethodMayBeStatic
class AdvancedMenu(BaseMenu):
def __init__(self, previous_menu: Optional[Type[BaseMenu]] = None):
def __init__(self, previous_menu: Type[BaseMenu] | None = None) -> None:
super().__init__()
self.previous_menu = previous_menu
self.previous_menu: Type[BaseMenu] | None = previous_menu
def set_previous_menu(self, previous_menu: Optional[Type[BaseMenu]]) -> None:
def set_previous_menu(self, previous_menu: Type[BaseMenu] | None) -> None:
from core.menus.main_menu import MainMenu
self.previous_menu: Type[BaseMenu] = (
previous_menu if previous_menu is not None else MainMenu
)
self.previous_menu = previous_menu if previous_menu is not None else MainMenu
def set_options(self):
def set_options(self) -> None:
self.options = {
"1": Option(method=self.build, menu=True),
"2": Option(method=self.flash, menu=False),
@@ -51,7 +50,7 @@ class AdvancedMenu(BaseMenu):
"6": Option(method=self.moonraker_rollback, menu=True),
}
def print_menu(self):
def print_menu(self) -> None:
header = " [ Advanced Menu ] "
color = COLOR_YELLOW
count = 62 - len(color) - len(RESET_FORMAT)
@@ -70,23 +69,23 @@ class AdvancedMenu(BaseMenu):
)[1:]
print(menu, end="")
def klipper_rollback(self, **kwargs):
def klipper_rollback(self, **kwargs) -> None:
rollback_repository(KLIPPER_DIR, Klipper)
def moonraker_rollback(self, **kwargs):
def moonraker_rollback(self, **kwargs) -> None:
rollback_repository(MOONRAKER_DIR, Moonraker)
def build(self, **kwargs):
def build(self, **kwargs) -> None:
KlipperBuildFirmwareMenu(previous_menu=self.__class__).run()
def flash(self, **kwargs):
def flash(self, **kwargs) -> None:
KlipperFlashMethodMenu(previous_menu=self.__class__).run()
def build_flash(self, **kwargs):
def build_flash(self, **kwargs) -> None:
KlipperBuildFirmwareMenu(previous_menu=KlipperFlashMethodMenu).run()
KlipperFlashMethodMenu(previous_menu=self.__class__).run()
def get_id(self, **kwargs):
def get_id(self, **kwargs) -> None:
KlipperSelectMcuConnectionMenu(
previous_menu=self.__class__,
standalone=True,

33
kiauh/core/menus/backup_menu.py
View File

@@ -6,9 +6,10 @@
# #
# This file may be distributed under the terms of the GNU GPLv3 license #
# ======================================================================= #
from __future__ import annotations
import textwrap
from typing import Optional, Type
from typing import Type
from components.klipper.klipper_utils import backup_klipper_dir
from components.klipperscreen.klipperscreen import backup_klipperscreen_dir
@@ -31,16 +32,14 @@ from utils.constants import COLOR_CYAN, COLOR_YELLOW, RESET_FORMAT
# noinspection PyUnusedLocal
# noinspection PyMethodMayBeStatic
class BackupMenu(BaseMenu):
def __init__(self, previous_menu: Optional[Type[BaseMenu]] = None):
def __init__(self, previous_menu: Type[BaseMenu] | None = None) -> None:
super().__init__()
self.previous_menu = previous_menu
self.previous_menu: Type[BaseMenu] | None = previous_menu
def set_previous_menu(self, previous_menu: Optional[Type[BaseMenu]]) -> None:
def set_previous_menu(self, previous_menu: Type[BaseMenu] | None) -> None:
from core.menus.main_menu import MainMenu
self.previous_menu: Type[BaseMenu] = (
previous_menu if previous_menu is not None else MainMenu
)
self.previous_menu = previous_menu if previous_menu is not None else MainMenu
def set_options(self) -> None:
self.options = {
@@ -55,7 +54,7 @@ class BackupMenu(BaseMenu):
"9": Option(method=self.backup_klipperscreen, menu=False),
}
def print_menu(self):
def print_menu(self) -> None:
header = " [ Backup Menu ] "
line1 = f"{COLOR_YELLOW}INFO: Backups are located in '~/kiauh-backups'{RESET_FORMAT}"
color = COLOR_CYAN
@@ -81,29 +80,29 @@ class BackupMenu(BaseMenu):
)[1:]
print(menu, end="")
def backup_klipper(self, **kwargs):
def backup_klipper(self, **kwargs) -> None:
backup_klipper_dir()
def backup_moonraker(self, **kwargs):
def backup_moonraker(self, **kwargs) -> None:
backup_moonraker_dir()
def backup_printer_config(self, **kwargs):
def backup_printer_config(self, **kwargs) -> None:
backup_printer_config_dir()
def backup_moonraker_db(self, **kwargs):
def backup_moonraker_db(self, **kwargs) -> None:
backup_moonraker_db_dir()
def backup_mainsail(self, **kwargs):
def backup_mainsail(self, **kwargs) -> None:
backup_client_data(MainsailData())
def backup_fluidd(self, **kwargs):
def backup_fluidd(self, **kwargs) -> None:
backup_client_data(FluiddData())
def backup_mainsail_config(self, **kwargs):
def backup_mainsail_config(self, **kwargs) -> None:
backup_client_config_data(MainsailData())
def backup_fluidd_config(self, **kwargs):
def backup_fluidd_config(self, **kwargs) -> None:
backup_client_config_data(FluiddData())
def backup_klipperscreen(self, **kwargs):
def backup_klipperscreen(self, **kwargs) -> None:
backup_klipperscreen_dir()

34
kiauh/core/menus/base_menu.py
View File

@@ -14,7 +14,7 @@ import sys
import textwrap
import traceback
from abc import abstractmethod
from typing import Dict, Optional, Type
from typing import Dict, Type
from core.menus import FooterType, Option
from utils.constants import (
@@ -27,11 +27,11 @@ from utils.constants import (
from utils.logger import Logger
def clear():
def clear() -> None:
subprocess.call("clear", shell=True)
def print_header():
def print_header() -> None:
line1 = " [ KIAUH ] "
line2 = "Klipper Installation And Update Helper"
line3 = ""
@@ -49,7 +49,7 @@ def print_header():
print(header, end="")
def print_quit_footer():
def print_quit_footer() -> None:
text = "Q) Quit"
color = COLOR_RED
count = 62 - len(color) - len(RESET_FORMAT)
@@ -62,7 +62,7 @@ def print_quit_footer():
print(footer, end="")
def print_back_footer():
def print_back_footer() -> None:
text = "B) « Back"
color = COLOR_GREEN
count = 62 - len(color) - len(RESET_FORMAT)
@@ -75,7 +75,7 @@ def print_back_footer():
print(footer, end="")
def print_back_help_footer():
def print_back_help_footer() -> None:
text1 = "B) « Back"
text2 = "H) Help [?]"
color1 = COLOR_GREEN
@@ -90,7 +90,7 @@ def print_back_help_footer():
print(footer, end="")
def print_blank_footer():
def print_blank_footer() -> None:
print("╚═══════════════════════════════════════════════════════╝")
@@ -109,15 +109,15 @@ class BaseMenu(metaclass=PostInitCaller):
default_option: Option = None
input_label_txt: str = "Perform action"
header: bool = False
previous_menu: Type[BaseMenu] = None
help_menu: Type[BaseMenu] = None
previous_menu: Type[BaseMenu] | None = None
help_menu: Type[BaseMenu] | None = None
footer_type: FooterType = FooterType.BACK
def __init__(self, **kwargs):
def __init__(self, **kwargs) -> None:
if type(self) is BaseMenu:
raise NotImplementedError("BaseMenu cannot be instantiated directly.")
def __post_init__(self):
def __post_init__(self) -> None:
self.set_previous_menu(self.previous_menu)
self.set_options()
@@ -133,18 +133,22 @@ class BaseMenu(metaclass=PostInitCaller):
if self.default_option is not None:
self.options[""] = self.default_option
def __go_back(self, **kwargs):
def __go_back(self, **kwargs) -> None:
if self.previous_menu is None:
return
self.previous_menu().run()
def __go_to_help(self, **kwargs):
def __go_to_help(self, **kwargs) -> None:
if self.help_menu is None:
return
self.help_menu(previous_menu=self).run()
def __exit(self, **kwargs):
def __exit(self, **kwargs) -> None:
Logger.print_ok("###### Happy printing!", False)
sys.exit(0)
@abstractmethod
def set_previous_menu(self, previous_menu: Optional[Type[BaseMenu]]) -> None:
def set_previous_menu(self, previous_menu: Type[BaseMenu] | None) -> None:
raise NotImplementedError
@abstractmethod

35
kiauh/core/menus/install_menu.py
View File

@@ -6,9 +6,10 @@
# #
# This file may be distributed under the terms of the GNU GPLv3 license #
# ======================================================================= #
from __future__ import annotations
import textwrap
from typing import Optional, Type
from typing import Type
from components.crowsnest.crowsnest import install_crowsnest
from components.klipper import klipper_setup
@@ -28,16 +29,14 @@ from utils.constants import COLOR_GREEN, RESET_FORMAT
# noinspection PyUnusedLocal
# noinspection PyMethodMayBeStatic
class InstallMenu(BaseMenu):
def __init__(self, previous_menu: Optional[Type[BaseMenu]] = None):
def __init__(self, previous_menu: Type[BaseMenu] | None = None) -> None:
super().__init__()
self.previous_menu = previous_menu
self.previous_menu: Type[BaseMenu] | None = previous_menu
def set_previous_menu(self, previous_menu: Optional[Type[BaseMenu]]) -> None:
def set_previous_menu(self, previous_menu: Type[BaseMenu] | None) -> None:
from core.menus.main_menu import MainMenu
self.previous_menu: Type[BaseMenu] = (
previous_menu if previous_menu is not None else MainMenu
)
self.previous_menu = previous_menu if previous_menu is not None else MainMenu
def set_options(self) -> None:
self.options = {
@@ -53,7 +52,7 @@ class InstallMenu(BaseMenu):
"10": Option(method=self.install_octoeverywhere, menu=False),
}
def print_menu(self):
def print_menu(self) -> None:
header = " [ Installation Menu ] "
color = COLOR_GREEN
count = 62 - len(color) - len(RESET_FORMAT)
@@ -79,32 +78,32 @@ class InstallMenu(BaseMenu):
)[1:]
print(menu, end="")
def install_klipper(self, **kwargs):
def install_klipper(self, **kwargs) -> None:
klipper_setup.install_klipper()
def install_moonraker(self, **kwargs):
def install_moonraker(self, **kwargs) -> None:
moonraker_setup.install_moonraker()
def install_mainsail(self, **kwargs):
def install_mainsail(self, **kwargs) -> None:
client_setup.install_client(MainsailData())
def install_mainsail_config(self, **kwargs):
def install_mainsail_config(self, **kwargs) -> None:
client_config_setup.install_client_config(MainsailData())
def install_fluidd(self, **kwargs):
def install_fluidd(self, **kwargs) -> None:
client_setup.install_client(FluiddData())
def install_fluidd_config(self, **kwargs):
def install_fluidd_config(self, **kwargs) -> None:
client_config_setup.install_client_config(FluiddData())
def install_klipperscreen(self, **kwargs):
def install_klipperscreen(self, **kwargs) -> None:
install_klipperscreen()
def install_mobileraker(self, **kwargs):
def install_mobileraker(self, **kwargs) -> None:
install_mobileraker()
def install_crowsnest(self, **kwargs):
def install_crowsnest(self, **kwargs) -> None:
install_crowsnest()
def install_octoeverywhere(self, **kwargs):
def install_octoeverywhere(self, **kwargs) -> None:
install_octoeverywhere()

34
kiauh/core/menus/main_menu.py
View File

@@ -6,9 +6,11 @@
# #
# This file may be distributed under the terms of the GNU GPLv3 license #
# ======================================================================= #
from __future__ import annotations
import sys
import textwrap
from typing import Optional, Type
from typing import Callable, Type
from components.crowsnest.crowsnest import get_crowsnest_status
from components.klipper.klipper_utils import get_klipper_status
@@ -47,18 +49,18 @@ from utils.types import ComponentStatus, StatusMap, StatusText
# noinspection PyUnusedLocal
# noinspection PyMethodMayBeStatic
class MainMenu(BaseMenu):
def __init__(self):
def __init__(self) -> None:
super().__init__()
self.header = True
self.footer_type = FooterType.QUIT
self.header: bool = True
self.footer_type: FooterType = FooterType.QUIT
self.kl_status = self.kl_repo = self.mr_status = self.mr_repo = ""
self.ms_status = self.fl_status = self.ks_status = self.mb_status = ""
self.cn_status = self.cc_status = self.oe_status = ""
self._init_status()
def set_previous_menu(self, previous_menu: Optional[Type[BaseMenu]]) -> None:
def set_previous_menu(self, previous_menu: Type[BaseMenu] | None) -> None:
"""MainMenu does not have a previous menu"""
pass
@@ -94,7 +96,7 @@ class MainMenu(BaseMenu):
self._get_component_status("cn", get_crowsnest_status)
self._get_component_status("oe", get_octoeverywhere_status)
def _get_component_status(self, name: str, status_fn: callable, *args) -> None:
def _get_component_status(self, name: str, status_fn: Callable, *args) -> None:
status_data: ComponentStatus = status_fn(*args)
code: int = status_data.status
status: StatusText = StatusMap[code]
@@ -119,7 +121,7 @@ class MainMenu(BaseMenu):
return f"{color}{status}{count}{RESET_FORMAT}"
def print_menu(self):
def print_menu(self) -> None:
self._fetch_status()
header = " [ Main Menu ] "
@@ -155,30 +157,30 @@ class MainMenu(BaseMenu):
)[1:]
print(menu, end="")
def exit(self, **kwargs):
def exit(self, **kwargs) -> None:
Logger.print_ok("###### Happy printing!", False)
sys.exit(0)
def log_upload_menu(self, **kwargs):
def log_upload_menu(self, **kwargs) -> None:
LogUploadMenu().run()
def install_menu(self, **kwargs):
def install_menu(self, **kwargs) -> None:
InstallMenu(previous_menu=self.__class__).run()
def update_menu(self, **kwargs):
def update_menu(self, **kwargs) -> None:
UpdateMenu(previous_menu=self.__class__).run()
def remove_menu(self, **kwargs):
def remove_menu(self, **kwargs) -> None:
RemoveMenu(previous_menu=self.__class__).run()
def advanced_menu(self, **kwargs):
def advanced_menu(self, **kwargs) -> None:
AdvancedMenu(previous_menu=self.__class__).run()
def backup_menu(self, **kwargs):
def backup_menu(self, **kwargs) -> None:
BackupMenu(previous_menu=self.__class__).run()
def settings_menu(self, **kwargs):
def settings_menu(self, **kwargs) -> None:
SettingsMenu(previous_menu=self.__class__).run()
def extension_menu(self, **kwargs):
def extension_menu(self, **kwargs) -> None:
ExtensionsMenu(previous_menu=self.__class__).run()

33
kiauh/core/menus/remove_menu.py
View File

@@ -6,9 +6,10 @@
# #
# This file may be distributed under the terms of the GNU GPLv3 license #
# ======================================================================= #
from __future__ import annotations
import textwrap
from typing import Optional, Type
from typing import Type
from components.crowsnest.crowsnest import remove_crowsnest
from components.klipper.menus.klipper_remove_menu import KlipperRemoveMenu
@@ -29,18 +30,16 @@ from utils.constants import COLOR_RED, RESET_FORMAT
# noinspection PyUnusedLocal
# noinspection PyMethodMayBeStatic
class RemoveMenu(BaseMenu):
def __init__(self, previous_menu: Optional[Type[BaseMenu]] = None):
def __init__(self, previous_menu: Type[BaseMenu] | None = None) -> None:
super().__init__()
self.previous_menu = previous_menu
self.previous_menu: Type[BaseMenu] | None = previous_menu
def set_previous_menu(self, previous_menu: Optional[Type[BaseMenu]]) -> None:
def set_previous_menu(self, previous_menu: Type[BaseMenu] | None) -> None:
from core.menus.main_menu import MainMenu
self.previous_menu: Type[BaseMenu] = (
previous_menu if previous_menu is not None else MainMenu
)
self.previous_menu = previous_menu if previous_menu is not None else MainMenu
def set_options(self):
def set_options(self) -> None:
self.options = {
"1": Option(method=self.remove_klipper, menu=True),
"2": Option(method=self.remove_moonraker, menu=True),
@@ -52,7 +51,7 @@ class RemoveMenu(BaseMenu):
"8": Option(method=self.remove_octoeverywhere, menu=True),
}
def print_menu(self):
def print_menu(self) -> None:
header = " [ Remove Menu ] "
color = COLOR_RED
count = 62 - len(color) - len(RESET_FORMAT)
@@ -78,26 +77,26 @@ class RemoveMenu(BaseMenu):
)[1:]
print(menu, end="")
def remove_klipper(self, **kwargs):
def remove_klipper(self, **kwargs) -> None:
KlipperRemoveMenu(previous_menu=self.__class__).run()
def remove_moonraker(self, **kwargs):
def remove_moonraker(self, **kwargs) -> None:
MoonrakerRemoveMenu(previous_menu=self.__class__).run()
def remove_mainsail(self, **kwargs):
def remove_mainsail(self, **kwargs) -> None:
ClientRemoveMenu(previous_menu=self.__class__, client=MainsailData()).run()
def remove_fluidd(self, **kwargs):
def remove_fluidd(self, **kwargs) -> None:
ClientRemoveMenu(previous_menu=self.__class__, client=FluiddData()).run()
def remove_klipperscreen(self, **kwargs):
def remove_klipperscreen(self, **kwargs) -> None:
remove_klipperscreen()
def remove_mobileraker(self, **kwargs):
def remove_mobileraker(self, **kwargs) -> None:
remove_mobileraker()
def remove_crowsnest(self, **kwargs):
def remove_crowsnest(self, **kwargs) -> None:
remove_crowsnest()
def remove_octoeverywhere(self, **kwargs):
def remove_octoeverywhere(self, **kwargs) -> None:
remove_octoeverywhere()

40
kiauh/core/menus/settings_menu.py
View File

@@ -6,10 +6,12 @@
# #
# This file may be distributed under the terms of the GNU GPLv3 license #
# ======================================================================= #
from __future__ import annotations
import shutil
import textwrap
from pathlib import Path
from typing import Optional, Tuple, Type
from typing import Tuple, Type
from components.klipper import KLIPPER_DIR
from components.klipper.klipper import Klipper
@@ -28,22 +30,20 @@ from utils.logger import DialogType, Logger
# noinspection PyUnusedLocal
# noinspection PyMethodMayBeStatic
class SettingsMenu(BaseMenu):
def __init__(self, previous_menu: Optional[Type[BaseMenu]] = None):
def __init__(self, previous_menu: Type[BaseMenu] | None = None) -> None:
super().__init__()
self.previous_menu = previous_menu
self.klipper_repo = None
self.moonraker_repo = None
self.mainsail_unstable = None
self.fluidd_unstable = None
self.auto_backups_enabled = None
self.previous_menu: Type[BaseMenu] | None = previous_menu
self.klipper_repo: str | None = None
self.moonraker_repo: str | None = None
self.mainsail_unstable: bool | None = None
self.fluidd_unstable: bool | None = None
self.auto_backups_enabled: bool | None = None
self._load_settings()
def set_previous_menu(self, previous_menu: Optional[Type[BaseMenu]]) -> None:
def set_previous_menu(self, previous_menu: Type[BaseMenu] | None) -> None:
from core.menus.main_menu import MainMenu
self.previous_menu: Type[BaseMenu] = (
previous_menu if previous_menu is not None else MainMenu
)
self.previous_menu = previous_menu if previous_menu is not None else MainMenu
def set_options(self) -> None:
self.options = {
@@ -54,7 +54,7 @@ class SettingsMenu(BaseMenu):
"5": Option(method=self.toggle_backup_before_update, menu=False),
}
def print_menu(self):
def print_menu(self) -> None:
header = " [ KIAUH Settings ] "
color = COLOR_CYAN
count = 62 - len(color) - len(RESET_FORMAT)
@@ -94,7 +94,7 @@ class SettingsMenu(BaseMenu):
)[1:]
print(menu, end="")
def _load_settings(self):
def _load_settings(self) -> None:
self.settings = KiauhSettings()
self._format_repo_str("klipper")
@@ -133,7 +133,7 @@ class SettingsMenu(BaseMenu):
return repo, branch
def _set_repo(self, repo_name: str):
def _set_repo(self, repo_name: str) -> None:
repo_url, branch = self._gather_input()
display_name = repo_name.capitalize()
Logger.print_dialog(
@@ -186,23 +186,23 @@ class SettingsMenu(BaseMenu):
im.start_all_instance()
def set_klipper_repo(self, **kwargs):
def set_klipper_repo(self, **kwargs) -> None:
self._set_repo("klipper")
def set_moonraker_repo(self, **kwargs):
def set_moonraker_repo(self, **kwargs) -> None:
self._set_repo("moonraker")
def toggle_mainsail_release(self, **kwargs):
def toggle_mainsail_release(self, **kwargs) -> None:
self.mainsail_unstable = not self.mainsail_unstable
self.settings.mainsail.unstable_releases = self.mainsail_unstable
self.settings.save()
def toggle_fluidd_release(self, **kwargs):
def toggle_fluidd_release(self, **kwargs) -> None:
self.fluidd_unstable = not self.fluidd_unstable
self.settings.fluidd.unstable_releases = self.fluidd_unstable
self.settings.save()
def toggle_backup_before_update(self, **kwargs):
def toggle_backup_before_update(self, **kwargs) -> None:
self.auto_backups_enabled = not self.auto_backups_enabled
self.settings.kiauh.backup_before_update = self.auto_backups_enabled
self.settings.save()

44
kiauh/core/menus/update_menu.py
View File

@@ -6,8 +6,10 @@
# #
# This file may be distributed under the terms of the GNU GPLv3 license #
# ======================================================================= #
from __future__ import annotations
import textwrap
from typing import Optional, Type
from typing import Callable, Type
from components.crowsnest.crowsnest import get_crowsnest_status, update_crowsnest
from components.klipper.klipper_setup import update_klipper
@@ -54,9 +56,9 @@ from utils.types import ComponentStatus
# noinspection PyUnusedLocal
# noinspection PyMethodMayBeStatic
class UpdateMenu(BaseMenu):
def __init__(self, previous_menu: Optional[Type[BaseMenu]] = None):
def __init__(self, previous_menu: Type[BaseMenu] | None = None) -> None:
super().__init__()
self.previous_menu = previous_menu
self.previous_menu: Type[BaseMenu] | None = previous_menu
self.klipper_local = self.klipper_remote = ""
self.moonraker_local = self.moonraker_remote = ""
@@ -84,12 +86,10 @@ class UpdateMenu(BaseMenu):
"octoeverywhere": {"installed": False, "local": None, "remote": None},
}
def set_previous_menu(self, previous_menu: Optional[Type[BaseMenu]]) -> None:
def set_previous_menu(self, previous_menu: Type[BaseMenu] | None) -> None:
from core.menus.main_menu import MainMenu
self.previous_menu: Type[BaseMenu] = (
previous_menu if previous_menu is not None else MainMenu
)
self.previous_menu = previous_menu if previous_menu is not None else MainMenu
def set_options(self) -> None:
self.options = {
@@ -107,7 +107,7 @@ class UpdateMenu(BaseMenu):
"11": Option(self.upgrade_system_packages, menu=False),
}
def print_menu(self):
def print_menu(self) -> None:
spinner = Spinner()
spinner.start()
@@ -149,52 +149,52 @@ class UpdateMenu(BaseMenu):
)[1:]
print(menu, end="")
def update_all(self, **kwargs):
def update_all(self, **kwargs) -> None:
print("update_all")
def update_klipper(self, **kwargs):
def update_klipper(self, **kwargs) -> None:
if self._check_is_installed("klipper"):
update_klipper()
def update_moonraker(self, **kwargs):
def update_moonraker(self, **kwargs) -> None:
if self._check_is_installed("moonraker"):
update_moonraker()
def update_mainsail(self, **kwargs):
def update_mainsail(self, **kwargs) -> None:
if self._check_is_installed("mainsail"):
update_client(self.mainsail_data)
def update_mainsail_config(self, **kwargs):
def update_mainsail_config(self, **kwargs) -> None:
if self._check_is_installed("mainsail_config"):
update_client_config(self.mainsail_data)
def update_fluidd(self, **kwargs):
def update_fluidd(self, **kwargs) -> None:
if self._check_is_installed("fluidd"):
update_client(self.fluidd_data)
def update_fluidd_config(self, **kwargs):
def update_fluidd_config(self, **kwargs) -> None:
if self._check_is_installed("fluidd_config"):
update_client_config(self.fluidd_data)
def update_klipperscreen(self, **kwargs):
def update_klipperscreen(self, **kwargs) -> None:
if self._check_is_installed("klipperscreen"):
update_klipperscreen()
def update_mobileraker(self, **kwargs):
def update_mobileraker(self, **kwargs) -> None:
if self._check_is_installed("mobileraker"):
update_mobileraker()
def update_crowsnest(self, **kwargs):
def update_crowsnest(self, **kwargs) -> None:
if self._check_is_installed("crowsnest"):
update_crowsnest()
def update_octoeverywhere(self, **kwargs):
def update_octoeverywhere(self, **kwargs) -> None:
if self._check_is_installed("octoeverywhere"):
update_octoeverywhere()
def upgrade_system_packages(self, **kwargs): ...
def upgrade_system_packages(self, **kwargs) -> None: ...
def _fetch_update_status(self):
def _fetch_update_status(self) -> None:
self._set_status_data("klipper", get_klipper_status)
self._set_status_data("moonraker", get_moonraker_status)
self._set_status_data("mainsail", get_client_status, self.mainsail_data, True)
@@ -221,7 +221,7 @@ class UpdateMenu(BaseMenu):
return f"{color}{local_version or '-'}{RESET_FORMAT}"
def _set_status_data(self, name: str, status_fn: callable, *args) -> None:
def _set_status_data(self, name: str, status_fn: Callable, *args) -> None:
comp_status: ComponentStatus = status_fn(*args)
self.status_data[name]["installed"] = True if comp_status.status == 2 else False

17
kiauh/core/settings/kiauh_settings.py
View File

@@ -8,8 +8,6 @@
# ======================================================================= #
from __future__ import annotations
from typing import Union
from core.submodules.simple_config_parser.src.simple_config_parser.simple_config_parser import (
NoOptionError,
NoSectionError,
@@ -61,10 +59,11 @@ class KiauhSettings:
def __new__(cls, *args, **kwargs) -> "KiauhSettings":
if cls._instance is None:
cls._instance = super(KiauhSettings, cls).__new__(cls, *args, **kwargs)
cls._instance.__initialized = False
return cls._instance
def __init__(self) -> None:
if not hasattr(self, "__initialized"):
self.__initialized = False
if self.__initialized:
return
self.__initialized = True
@@ -87,7 +86,7 @@ class KiauhSettings:
self._load_config()
def get(self, section: str, option: str) -> Union[str, int, bool]:
def get(self, section: str, option: str) -> str | int | bool:
"""
Get a value from the settings state by providing the section and option name as strings.
Prefer direct access to the properties, as it is usually safer!
@@ -99,11 +98,11 @@ class KiauhSettings:
try:
section = getattr(self, section)
value = getattr(section, option)
return value
return value # type: ignore
except AttributeError:
raise
def set(self, section: str, option: str, value: Union[str, int, bool]) -> None:
def set(self, section: str, option: str, value: str | int | bool) -> None:
"""
Set a value in the settings state by providing the section and option name as strings.
Prefer direct access to the properties, as it is usually safer!
@@ -113,7 +112,7 @@ class KiauhSettings:
"""
try:
section = getattr(self, section)
section.option = value
section.option = value # type: ignore
except AttributeError:
raise
@@ -172,7 +171,7 @@ class KiauhSettings:
if v.isdigit() or v.lower() == "true" or v.lower() == "false":
raise ValueError
def _read_settings(self):
def _read_settings(self) -> None:
self.kiauh.backup_before_update = self.config.getboolean(
"kiauh", "backup_before_update"
)
@@ -189,7 +188,7 @@ class KiauhSettings:
"fluidd", "unstable_releases"
)
def _set_config_options(self):
def _set_config_options(self) -> None:
self.config.set(
"kiauh",
"backup_before_update",

23
kiauh/extensions/extensions_menu.py
View File

@@ -6,13 +6,14 @@
# #
# This file may be distributed under the terms of the GNU GPLv3 license #
# ======================================================================= #
from __future__ import annotations
import importlib
import inspect
import json
import textwrap
from pathlib import Path
from typing import Dict, List, Optional, Type
from typing import Dict, List, Type
from core.menus import Option
from core.menus.base_menu import BaseMenu
@@ -25,17 +26,15 @@ from utils.logger import Logger
# noinspection PyUnusedLocal
# noinspection PyMethodMayBeStatic
class ExtensionsMenu(BaseMenu):
def __init__(self, previous_menu: Optional[Type[BaseMenu]] = None):
def __init__(self, previous_menu: Type[BaseMenu] | None = None):
super().__init__()
self.previous_menu = previous_menu
self.previous_menu: Type[BaseMenu] | None = previous_menu
self.extensions: Dict[str, BaseExtension] = self.discover_extensions()
def set_previous_menu(self, previous_menu: Optional[Type[BaseMenu]]) -> None:
def set_previous_menu(self, previous_menu: Type[BaseMenu] | None) -> None:
from core.menus.main_menu import MainMenu
self.previous_menu: Type[BaseMenu] = (
previous_menu if previous_menu is not None else MainMenu
)
self.previous_menu = previous_menu if previous_menu is not None else MainMenu
def set_options(self) -> None:
self.options = {
@@ -80,7 +79,7 @@ class ExtensionsMenu(BaseMenu):
def extension_submenu(self, **kwargs):
ExtensionSubmenu(kwargs.get("opt_data"), self.__class__).run()
def print_menu(self):
def print_menu(self) -> None:
header = " [ Extensions Menu ] "
color = COLOR_CYAN
line1 = f"{COLOR_YELLOW}Available Extensions:{RESET_FORMAT}"
@@ -108,14 +107,14 @@ class ExtensionsMenu(BaseMenu):
# noinspection PyMethodMayBeStatic
class ExtensionSubmenu(BaseMenu):
def __init__(
self, extension: BaseExtension, previous_menu: Optional[Type[BaseMenu]] = None
self, extension: BaseExtension, previous_menu: Type[BaseMenu] | None = None
):
super().__init__()
self.extension = extension
self.previous_menu = previous_menu
self.previous_menu: Type[BaseMenu] | None = previous_menu
def set_previous_menu(self, previous_menu: Optional[Type[BaseMenu]]) -> None:
self.previous_menu: Type[BaseMenu] = (
def set_previous_menu(self, previous_menu: Type[BaseMenu] | None) -> None:
self.previous_menu = (
previous_menu if previous_menu is not None else ExtensionsMenu
)

9
kiauh/extensions/klipper_backup/klipper_backup_extension.py
View File

@@ -45,8 +45,7 @@ class KlipperbackupExtension(BaseExtension):
subprocess.run([str(KLIPPERBACKUP_DIR / "install.sh"), "check_updates"])
def remove_extension(self, **kwargs) -> None:
def uninstall_service(service_name):
def uninstall_service(service_name) -> bool:
try:
subprocess.run(["sudo", "systemctl", "stop", service_name], check=True)
subprocess.run(
@@ -59,7 +58,7 @@ class KlipperbackupExtension(BaseExtension):
except subprocess.CalledProcessError:
return False
def check_crontab_entry(entry):
def check_crontab_entry(entry) -> bool:
try:
crontab_content = subprocess.check_output(
["crontab", "-l"], stderr=subprocess.DEVNULL, text=True
@@ -123,8 +122,8 @@ class KlipperbackupExtension(BaseExtension):
)
else:
Logger.print_error(
f"Error uninstalling the {service_name} service."
)
f"Error uninstalling the {service_name} service."
)
else:
Logger.print_info(f"Service {service_name} NOT detected.")
except:

7
kiauh/extensions/mainsail_theme_installer/mainsail_theme_installer_extension.py
View File

@@ -6,12 +6,13 @@
# #
# This file may be distributed under the terms of the GNU GPLv3 license #
# ======================================================================= #
from __future__ import annotations
import csv
import shutil
import textwrap
import urllib.request
from typing import List, Optional, Type, TypedDict, Union
from typing import List, Type, TypedDict, Union
from components.klipper.klipper import Klipper
from components.klipper.klipper_dialogs import (
@@ -81,10 +82,10 @@ class MainsailThemeInstallMenu(BaseMenu):
self.themes: List[ThemeData] = self.load_themes()
self.instances = instances
def set_previous_menu(self, previous_menu: Optional[Type[BaseMenu]]) -> None:
def set_previous_menu(self, previous_menu: Type[BaseMenu] | None) -> None:
from extensions.extensions_menu import ExtensionsMenu
self.previous_menu: Type[BaseMenu] = (
self.previous_menu = (
previous_menu if previous_menu is not None else ExtensionsMenu
)

3
kiauh/extensions/obico/moonraker_obico.py
View File

@@ -33,6 +33,7 @@ OBICO_MACROS_CFG = "moonraker_obico_macros.cfg"
# noinspection PyMethodMayBeStatic
# todo: make this to a dataclass
class MoonrakerObico(BaseInstance):
@classmethod
def blacklist(cls) -> List[str]:
@@ -106,7 +107,7 @@ class MoonrakerObico(BaseInstance):
service_content = self._prep_service_file(
service_template_path, env_file_target
)
command = ["sudo", "tee", service_file_target]
command = ["sudo", "tee", service_file_target.as_posix()]
run(
command,
input=service_content.encode(),

4
kiauh/extensions/obico/moonraker_obico_extension.py
View File

@@ -297,11 +297,11 @@ class ObicoExtension(BaseExtension):
def _patch_moonraker_conf(self, instances: List[Moonraker]) -> None:
add_config_section(section=f"include {OBICO_UPDATE_CFG}", instances=instances)
def _link_obico_instances(self, unlinked_instances):
def _link_obico_instances(self, unlinked_instances) -> None:
for obico in unlinked_instances:
obico.link()
def _check_and_opt_link_instances(self):
def _check_and_opt_link_instances(self) -> None:
Logger.print_status("Checking link status of Obico instances ...")
ob_im = InstanceManager(MoonrakerObico)
ob_instances: List[MoonrakerObico] = ob_im.instances

3
kiauh/extensions/telegram_bot/moonraker_telegram_bot.py
View File

@@ -23,6 +23,7 @@ TELEGRAM_BOT_REPO = "https://github.com/nlef/moonraker-telegram-bot.git"
# noinspection PyMethodMayBeStatic
# todo: make this to a dataclass
class MoonrakerTelegramBot(BaseInstance):
@classmethod
def blacklist(cls) -> List[str]:
@@ -95,7 +96,7 @@ class MoonrakerTelegramBot(BaseInstance):
service_content = self._prep_service_file(
service_template_path, env_file_target
)
command = ["sudo", "tee", service_file_target]
command = ["sudo", "tee", service_file_target.as_posix()]
run(
command,
input=service_content.encode(),

2
kiauh/main.py
View File

@@ -12,7 +12,7 @@ from core.settings.kiauh_settings import KiauhSettings
from utils.logger import Logger
def main():
def main() -> None:
try:
KiauhSettings()
MainMenu().run()

9
kiauh/utils/common.py
View File

@@ -96,12 +96,13 @@ def get_install_status(
for f in files:
checks.append(f.exists())
status: StatusCode
if all(checks):
status: StatusCode = 2 # installed
status = 2 # installed
elif not any(checks):
status: StatusCode = 0 # not installed
status = 0 # not installed
else:
status: StatusCode = 1 # incomplete
status = 1 # incomplete
return ComponentStatus(
status=status,
@@ -112,7 +113,7 @@ def get_install_status(
)
def backup_printer_config_dir():
def backup_printer_config_dir() -> None:
# local import to prevent circular import
from core.backup_manager.backup_manager import BackupManager

2
kiauh/utils/config_utils.py
View File

@@ -49,7 +49,7 @@ def add_config_section(
scp.write(cfg_file)
def add_config_section_at_top(section: str, instances: List[B]):
def add_config_section_at_top(section: str, instances: List[B]) -> None:
# TODO: this could be implemented natively in SimpleConfigParser
for instance in instances:
tmp_cfg = tempfile.NamedTemporaryFile(mode="w", delete=False)

6
kiauh/utils/decorators.py
View File

@@ -6,12 +6,14 @@
# #
# This file may be distributed under the terms of the GNU GPLv3 license #
# ======================================================================= #
from __future__ import annotations
import warnings
from typing import Callable
def deprecated(info: str = "", replaced_by: Callable = None) -> Callable:
def decorator(func):
def deprecated(info: str = "", replaced_by: Callable | None = None) -> Callable:
def decorator(func) -> Callable:
def wrapper(*args, **kwargs):
msg = f"{info}{replaced_by.__name__ if replaced_by else ''}"
warnings.warn(msg, category=DeprecationWarning, stacklevel=2)

6
kiauh/utils/fs_utils.py
View File

@@ -36,7 +36,7 @@ def check_file_exist(file_path: Path, sudo=False) -> bool:
"""
if sudo:
try:
command = ["sudo", "find", file_path]
command = ["sudo", "find", file_path.as_posix()]
check_output(command, stderr=DEVNULL)
return True
except CalledProcessError:
@@ -50,7 +50,7 @@ def check_file_exist(file_path: Path, sudo=False) -> bool:
def create_symlink(source: Path, target: Path, sudo=False) -> None:
try:
cmd = ["ln", "-sf", source, target]
cmd = ["ln", "-sf", source.as_posix(), target.as_posix()]
if sudo:
cmd.insert(0, "sudo")
run(cmd, stderr=PIPE, check=True)
@@ -61,7 +61,7 @@ def create_symlink(source: Path, target: Path, sudo=False) -> None:
def remove_with_sudo(file: Path) -> None:
try:
cmd = ["sudo", "rm", "-rf", file]
cmd = ["sudo", "rm", "-rf", file.as_posix()]
run(cmd, stderr=PIPE, check=True)
except CalledProcessError as e:
Logger.print_error(f"Failed to remove file: {e}")

10
kiauh/utils/git_utils.py
View File

@@ -7,7 +7,7 @@ from http.client import HTTPResponse
from json import JSONDecodeError
from pathlib import Path
from subprocess import DEVNULL, PIPE, CalledProcessError, check_output, run
from typing import List, Optional, Type
from typing import List, Type
from core.instance_manager.base_instance import BaseInstance
from core.instance_manager.instance_manager import InstanceManager
@@ -16,7 +16,7 @@ from utils.logger import Logger
def git_clone_wrapper(
repo: str, target_dir: Path, branch: Optional[str] = None, force: bool = False
repo: str, target_dir: Path, branch: str | None = None, force: bool = False
) -> None:
"""
Clones a repository from the given URL and checks out the specified branch if given.
@@ -75,7 +75,7 @@ def get_repo_name(repo: Path) -> str | None:
return "-"
try:
cmd = ["git", "-C", repo, "config", "--get", "remote.origin.url"]
cmd = ["git", "-C", repo.as_posix(), "config", "--get", "remote.origin.url"]
result = check_output(cmd, stderr=DEVNULL)
return "/".join(result.decode().strip().split("/")[-2:])
except CalledProcessError:
@@ -159,7 +159,7 @@ def get_remote_commit(repo: Path) -> str | None:
def git_cmd_clone(repo: str, target_dir: Path) -> None:
try:
command = ["git", "clone", repo, target_dir]
command = ["git", "clone", repo, target_dir.as_posix()]
run(command, check=True)
Logger.print_ok("Clone successful!")
@@ -169,7 +169,7 @@ def git_cmd_clone(repo: str, target_dir: Path) -> None:
raise
def git_cmd_checkout(branch: str, target_dir: Path) -> None:
def git_cmd_checkout(branch: str | None, target_dir: Path) -> None:
if branch is None:
return

20
kiauh/utils/input_utils.py
View File

@@ -9,7 +9,7 @@
from __future__ import annotations
import re
from typing import Dict, List, Union
from typing import Dict, List
from utils import INVALID_CHOICE
from utils.constants import COLOR_CYAN, RESET_FORMAT
@@ -53,9 +53,9 @@ def get_confirm(question: str, default_choice=True, allow_go_back=False) -> bool
def get_number_input(
question: str,
min_count: int,
max_count=None,
default=None,
allow_go_back=False,
max_count: int | None = None,
default: int | None = None,
allow_go_back: bool = False,
) -> int | None:
"""
Helper method to get a number input from the user
@@ -73,7 +73,7 @@ def get_number_input(
if allow_go_back and _input in options_go_back:
return None
if _input == "":
if _input == "" and default is not None:
return default
try:
@@ -84,10 +84,10 @@ def get_number_input(
def get_string_input(
question: str,
regex: Union[str, None] = None,
exclude: Union[List, None] = None,
allow_special_chars=False,
default=None,
regex: str | None = None,
exclude: List[str] | None = None,
allow_special_chars: bool = False,
default: str | None = None,
) -> str:
"""
Helper method to get a string input from the user
@@ -157,7 +157,7 @@ def format_question(question: str, default=None) -> str:
return f"{COLOR_CYAN}###### {formatted_q}: {RESET_FORMAT}"
def validate_number_input(value: str, min_count: int, max_count: int) -> int:
def validate_number_input(value: str, min_count: int, max_count: int | None) -> int:
"""
Helper method for a simple number input validation. |
:param value: The value to validate

25
kiauh/utils/logger.py
View File

@@ -6,6 +6,8 @@
# #
# This file may be distributed under the terms of the GNU GPLv3 license #
# ======================================================================= #
from __future__ import annotations
import textwrap
from enum import Enum
from typing import List
@@ -44,17 +46,17 @@ LINE_WIDTH = 53
class Logger:
@staticmethod
def info(msg):
def info(msg) -> None:
# log to kiauh.log
pass
@staticmethod
def warn(msg):
def warn(msg) -> None:
# log to kiauh.log
pass
@staticmethod
def error(msg):
def error(msg) -> None:
# log to kiauh.log
pass
@@ -88,8 +90,8 @@ class Logger:
title: DialogType,
content: List[str],
center_content: bool = False,
custom_title: str = None,
custom_color: DialogCustomColor = None,
custom_title: str | None = None,
custom_color: DialogCustomColor | None = None,
padding_top: int = 1,
padding_bottom: int = 1,
) -> None:
@@ -122,18 +124,23 @@ class Logger:
print("\n" * padding_bottom)
@staticmethod
def _get_dialog_title(title: DialogType, custom_title: str = None) -> str:
def _get_dialog_title(
title: DialogType, custom_title: str | None = None
) -> str | None:
if title == DialogType.CUSTOM and custom_title:
return f"[ {custom_title} ]"
return f"[ {title.value[0]} ]" if title.value[0] else None
@staticmethod
def _get_dialog_color(
title: DialogType, custom_color: DialogCustomColor = None
title: DialogType, custom_color: DialogCustomColor | None = None
) -> str:
if title == DialogType.CUSTOM and custom_color:
return str(custom_color.value)
return title.value[1] if title.value[1] else DialogCustomColor.WHITE.value
color: str = title.value[1] if title.value[1] else DialogCustomColor.WHITE.value
return color
@staticmethod
def _format_top_border(color: str) -> str:
@@ -146,7 +153,7 @@ class Logger:
)
@staticmethod
def _format_dialog_title(title: str) -> str:
def _format_dialog_title(title: str | None) -> str:
if title is not None:
return textwrap.dedent(f"""
┃ {title:^{LINE_WIDTH}} ┃

11
kiauh/utils/spinner.py
View File

@@ -1,17 +1,18 @@
import sys
import threading
import time
from typing import List
class Spinner:
def __init__(self, message="Loading", delay=0.2):
def __init__(self, message: str = "Loading", delay: float = 0.2) -> None:
self.message = f"{message} ..."
self.delay = delay
self._stop_event = threading.Event()
self._thread = threading.Thread(target=self._animate)
def _animate(self):
animation = ["◜", "◝", "◞", "◟"]
def _animate(self) -> None:
animation: List[str] = ["◜", "◝", "◞", "◟"]
while not self._stop_event.is_set():
for char in animation:
sys.stdout.write(f"\r{char} {self.message}")
@@ -22,12 +23,12 @@ class Spinner:
sys.stdout.write("\r" + " " * (len(self.message) + 1) + "\r")
sys.stdout.flush()
def start(self):
def start(self) -> None:
self._stop_event.clear()
if not self._thread.is_alive():
self._thread = threading.Thread(target=self._animate)
self._thread.start()
def stop(self):
def stop(self) -> None:
self._stop_event.set()
self._thread.join()

41
kiauh/utils/sys_utils.py
View File

@@ -6,6 +6,7 @@
# #
# This file may be distributed under the terms of the GNU GPLv3 license #
# ======================================================================= #
from __future__ import annotations
import os
import re
@@ -122,12 +123,13 @@ def update_python_pip(target: Path) -> None:
"""
Logger.print_status("Updating pip ...")
try:
pip_location = target.joinpath("bin/pip")
pip_exists = check_file_exist(pip_location)
pip_location: Path = target.joinpath("bin/pip")
pip_exists: bool = check_file_exist(pip_location)
if not pip_exists:
raise FileNotFoundError("Error updating pip! Not found.")
command = [pip_location, "install", "-U", "pip"]
command = [pip_location.as_posix(), "install", "-U", "pip"]
result = run(command, stderr=PIPE, text=True)
if result.returncode != 0 or result.stderr:
Logger.print_error(f"{result.stderr}", False)
@@ -156,7 +158,7 @@ def install_python_requirements(target: Path, requirements: Path) -> None:
Logger.print_status("Installing Python requirements ...")
command = [
target.joinpath("bin/pip"),
target.joinpath("bin/pip").as_posix(),
"install",
"-r",
f"{requirements}",
@@ -182,8 +184,8 @@ def update_system_package_lists(silent: bool, rls_info_change=False) -> None:
:param rls_info_change: Flag for "--allow-releaseinfo-change"
:return: None
"""
cache_mtime = 0
cache_files = [
cache_mtime: float = 0
cache_files: List[Path] = [
Path("/var/lib/apt/periodic/update-success-stamp"),
Path("/var/lib/apt/lists"),
]
@@ -269,7 +271,7 @@ def get_ipv4_addr() -> str:
try:
# doesn't even have to be reachable
s.connect(("192.255.255.255", 1))
return s.getsockname()[0]
return str(s.getsockname()[0])
except Exception:
return "127.0.0.1"
finally:
@@ -329,9 +331,9 @@ def set_nginx_permissions() -> None:
"""
cmd = f"ls -ld {Path.home()} | cut -d' ' -f1"
homedir_perm = run(cmd, shell=True, stdout=PIPE, text=True)
homedir_perm = homedir_perm.stdout
permissions = homedir_perm.stdout
if homedir_perm.count("x") < 3:
if permissions.count("x") < 3:
Logger.print_status("Granting NGINX the required permissions ...")
run(["chmod", "og+x", Path.home()])
Logger.print_ok("Permissions granted.")
@@ -363,7 +365,7 @@ def cmd_sysctl_manage(action: SysCtlManageAction) -> None:
raise
def service_instance_exists(name: str, exclude: List[str] = None) -> bool:
def service_instance_exists(name: str, exclude: List[str] | None = None) -> bool:
"""
Checks if a systemd service instance exists.
:param name: the service name
@@ -387,15 +389,16 @@ def log_process(process: Popen) -> None:
:return: None
"""
while True:
reads = [process.stdout.fileno()]
ret = select.select(reads, [], [])
for fd in ret[0]:
if fd == process.stdout.fileno():
line = process.stdout.readline()
if line:
print(line.strip(), flush=True)
else:
break
if process.stdout is not None:
reads = [process.stdout.fileno()]
ret = select.select(reads, [], [])
for fd in ret[0]:
if fd == process.stdout.fileno():
line = process.stdout.readline()
if line:
print(line.strip(), flush=True)
else:
break
if process.poll() is not None:
break

13
pyproject.toml
View File

@@ -2,7 +2,7 @@
requires-python = ">=3.8"
[project.optional-dependencies]
dev=["ruff"]
dev=["ruff", "mypy"]
[tool.ruff]
required-version = ">=0.3.4"
@@ -19,3 +19,14 @@ quote-style = "double"
[tool.ruff.lint]
extend-select = ["I"]
[tool.mypy]
python_version = "3.8"
platform = "linux"
# strict = true # TODO: enable this once everything is else is handled
check_untyped_defs = true
ignore_missing_imports = true
warn_redundant_casts = true
warn_unused_ignores = true
warn_return_any = true
warn_unreachable = true