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(style): use black code style / formatter

Browse Source 
Signed-off-by: Dominik Willner <th33xitus@gmail.com>
This commit is contained in:
dw-0
2023-10-29 00:31:34 +02:00
parent 84a530be7d
commit 358c666da9
17 changed files with 261 additions and 170 deletions
Download Patch File Download Diff File Expand all files Collapse all files

58
kiauh/modules/klipper/klipper.py
View File

@@ -10,7 +10,6 @@
# ======================================================================= #
import os
import pwd
import shutil
import subprocess
from pathlib import Path
@@ -29,10 +28,12 @@ class Klipper(BaseInstance):
return ["None", "mcu"]
def __init__(self, name: str):
super().__init__(name=name,
prefix="klipper",
user=CURRENT_USER,
data_dir_name=self._get_data_dir_from_name(name))
super().__init__(
name=name,
prefix="klipper",
user=CURRENT_USER,
data_dir_name=self._get_data_dir_from_name(name),
)
self.klipper_dir = KLIPPER_DIR
self.env_dir = KLIPPER_ENV_DIR
self.cfg_file = f"{self.cfg_dir}/printer.cfg"
@@ -43,26 +44,35 @@ class Klipper(BaseInstance):
def create(self) -> None:
Logger.print_info("Creating Klipper Instance")
module_path = os.path.dirname(os.path.abspath(__file__))
service_template_path = os.path.join(module_path, "res",
"klipper.service")
service_template_path = os.path.join(module_path, "res", "klipper.service")
env_template_file_path = os.path.join(module_path, "res", "klipper.env")
service_file_name = self.get_service_file_name(extension=True)
service_file_target = f"{SYSTEMD}/{service_file_name}"
env_file_target = os.path.abspath(f"{self.sysd_dir}/klipper.env")
# create folder structure
dirs = [self.data_dir, self.cfg_dir, self.log_dir,
self.comms_dir, self.sysd_dir]
dirs = [
self.data_dir,
self.cfg_dir,
self.log_dir,
self.comms_dir,
self.sysd_dir,
]
for _dir in dirs:
create_directory(Path(_dir))
try:
# writing the klipper service file (requires sudo!)
service_content = self._prep_service_file(service_template_path,
env_file_target)
service_content = self._prep_service_file(
service_template_path, env_file_target
)
command = ["sudo", "tee", service_file_target]
subprocess.run(command, input=service_content.encode(),
stdout=subprocess.DEVNULL, check=True)
subprocess.run(
command,
input=service_content.encode(),
stdout=subprocess.DEVNULL,
check=True,
)
Logger.print_ok(f"Service file created: {service_file_target}")
# writing the klipper.env file
@@ -73,11 +83,11 @@ class Klipper(BaseInstance):
except subprocess.CalledProcessError as e:
Logger.print_error(
f"Error creating service file {service_file_target}: {e}")
f"Error creating service file {service_file_target}: {e}"
)
raise
except OSError as e:
Logger.print_error(
f"Error creating env file {env_file_target}: {e}")
Logger.print_error(f"Error creating env file {env_file_target}: {e}")
raise
def read(self) -> None:
@@ -118,7 +128,7 @@ class Klipper(BaseInstance):
Logger.print_ok("Directories successfully deleted.")
def get_service_file_name(self, extension=False) -> str:
name = self.prefix if self.name is None else self.prefix + '-' + self.name
name = self.prefix if self.name is None else self.prefix + "-" + self.name
return name if not extension else f"{name}.service"
def _get_service_file_path(self):
@@ -138,11 +148,11 @@ class Klipper(BaseInstance):
template_content = template_file.read()
except FileNotFoundError:
Logger.print_error(
f"Unable to open {service_template_path} - File not found")
f"Unable to open {service_template_path} - File not found"
)
raise
service_content = template_content.replace("%USER%", self.user)
service_content = service_content.replace("%KLIPPER_DIR%",
self.klipper_dir)
service_content = service_content.replace("%KLIPPER_DIR%", self.klipper_dir)
service_content = service_content.replace("%ENV%", self.env_dir)
service_content = service_content.replace("%ENV_FILE%", env_file_path)
return service_content
@@ -153,10 +163,12 @@ class Klipper(BaseInstance):
env_template_file_content = env_file.read()
except FileNotFoundError:
Logger.print_error(
f"Unable to open {env_template_file_path} - File not found")
f"Unable to open {env_template_file_path} - File not found"
)
raise
env_file_content = env_template_file_content.replace("%KLIPPER_DIR%",
self.klipper_dir)
env_file_content = env_template_file_content.replace(
"%KLIPPER_DIR%", self.klipper_dir
)
env_file_content = env_file_content.replace("%CFG%", self.cfg_file)
env_file_content = env_file_content.replace("%SERIAL%", self.serial)
env_file_content = env_file_content.replace("%LOG%", self.log)

86
kiauh/modules/klipper/klipper_setup.py
View File

@@ -19,15 +19,27 @@ from typing import Optional, List, Union
from kiauh.instance_manager.instance_manager import InstanceManager
from kiauh.modules.klipper.klipper import Klipper
from kiauh.modules.klipper.klipper_utils import print_instance_overview, print_missing_usergroup_dialog
from kiauh.modules.klipper.klipper_utils import (
print_instance_overview,
print_missing_usergroup_dialog,
)
from kiauh.utils.constants import CURRENT_USER, KLIPPER_DIR, KLIPPER_ENV_DIR
from kiauh.utils.input_utils import get_user_confirm, get_user_number_input, \
get_user_string_input, get_user_selection_input
from kiauh.utils.input_utils import (
get_user_confirm,
get_user_number_input,
get_user_string_input,
get_user_selection_input,
)
from kiauh.utils.logger import Logger
from kiauh.utils.system_utils import parse_packages_from_file, \
clone_repo, create_python_venv, \
install_python_requirements, update_system_package_lists, \
install_system_packages, mask_system_service
from kiauh.utils.system_utils import (
parse_packages_from_file,
clone_repo,
create_python_venv,
install_python_requirements,
update_system_package_lists,
install_system_packages,
mask_system_service,
)
def run_klipper_setup(install: bool) -> None:
@@ -81,8 +93,8 @@ def install_klipper(instance_manager: InstanceManager) -> None:
instance_list = instance_manager.get_instances()
if_adding = " additional" if len(instance_list) > 0 else ""
install_count = get_user_number_input(
f"Number of{if_adding} Klipper instances to set up",
1, default=1)
f"Number of{if_adding} Klipper instances to set up", 1, default=1
)
instance_names = set_instance_names(instance_list, install_count)
@@ -132,8 +144,7 @@ def install_klipper_packages(klipper_dir: Path) -> None:
install_system_packages(packages)
def set_instance_names(instance_list, install_count: int) -> List[
Union[str, None]]:
def set_instance_names(instance_list, install_count: int) -> List[Union[str, None]]:
instance_count = len(instance_list)
# default single instance install
@@ -141,9 +152,11 @@ def set_instance_names(instance_list, install_count: int) -> List[
return [None]
# new multi instance install
elif ((instance_count == 0 and install_count > 1)
# or convert single instance install to multi instance install
or (instance_count == 1 and install_count >= 1)):
elif (
(instance_count == 0 and install_count > 1)
# or convert single instance install to multi instance install
or (instance_count == 1 and install_count >= 1)
):
if get_user_confirm("Assign custom names?", False):
return assign_custom_names(instance_count, install_count, None)
else:
@@ -153,8 +166,7 @@ def set_instance_names(instance_list, install_count: int) -> List[
# existing multi instance install
elif instance_count > 1:
if has_custom_names(instance_list):
return assign_custom_names(instance_count, install_count,
instance_list)
return assign_custom_names(instance_count, install_count, instance_list)
else:
start = get_highest_index(instance_list) + 1
_range = range(start, start + install_count)
@@ -170,8 +182,9 @@ def has_custom_names(instance_list: List[Klipper]) -> bool:
return False
def assign_custom_names(instance_count: int, install_count: int,
instance_list: Optional[List[Klipper]]) -> List[str]:
def assign_custom_names(
instance_count: int, install_count: int, instance_list: Optional[List[Klipper]]
) -> List[str]:
instance_names = []
exclude = Klipper.blacklist()
@@ -190,7 +203,7 @@ def assign_custom_names(instance_count: int, install_count: int,
def get_highest_index(instance_list: List[Klipper]) -> int:
indices = [int(instance.name.split('-')[-1]) for instance in instance_list]
indices = [int(instance.name.split("-")[-1]) for instance in instance_list]
return max(indices)
@@ -209,14 +222,12 @@ def remove_single_instance(instance_manager: InstanceManager) -> None:
def remove_multi_instance(instance_manager: InstanceManager) -> None:
instance_list = instance_manager.get_instances()
print_instance_overview(instance_list, show_index=True,
show_select_all=True)
print_instance_overview(instance_list, show_index=True, show_select_all=True)
options = [str(i) for i in range(len(instance_list))]
options.extend(["a", "A", "b", "B"])
selection = get_user_selection_input(
"Select Klipper instance to remove", options)
selection = get_user_selection_input("Select Klipper instance to remove", options)
print(selection)
if selection == "b".lower():
@@ -231,7 +242,8 @@ def remove_multi_instance(instance_manager: InstanceManager) -> None:
else:
instance = instance_list[int(selection)]
Logger.print_info(
f"Removing Klipper instance: {instance.get_service_file_name()}")
f"Removing Klipper instance: {instance.get_service_file_name()}"
)
instance_manager.set_current_instance(instance)
instance_manager.stop_instance()
instance_manager.disable_instance()
@@ -254,7 +266,9 @@ def check_user_groups():
print_missing_usergroup_dialog(missing_groups)
if not get_user_confirm(f"Add user '{CURRENT_USER}' to group(s) now?"):
Logger.warn("Skipped adding user to required groups. You might encounter issues.")
Logger.warn(
"Skipped adding user to required groups. You might encounter issues."
)
return
try:
@@ -267,13 +281,19 @@ def check_user_groups():
Logger.print_error(f"Unable to add user to usergroups: {e}")
raise
Logger.print_warn("Remember to relog/restart this machine for the group(s) to be applied!")
Logger.print_warn(
"Remember to relog/restart this machine for the group(s) to be applied!"
)
def handle_disruptive_system_packages() -> None:
services = []
brltty_status = subprocess.run(["systemctl", "is-enabled", "brltty"], capture_output=True, text=True)
modem_manager_status = subprocess.run(["systemctl", "is-enabled", "ModemManager"], capture_output=True, text=True)
brltty_status = subprocess.run(
["systemctl", "is-enabled", "brltty"], capture_output=True, text=True
)
modem_manager_status = subprocess.run(
["systemctl", "is-enabled", "ModemManager"], capture_output=True, text=True
)
if "enabled" in brltty_status.stdout:
services.append("brltty")
@@ -282,13 +302,17 @@ def handle_disruptive_system_packages() -> None:
for service in services if services else []:
try:
Logger.print_info(f"{service} service detected! Masking {service} service ...")
Logger.print_info(
f"{service} service detected! Masking {service} service ..."
)
mask_system_service(service)
Logger.print_ok(f"{service} service masked!")
except subprocess.CalledProcessError:
warn_msg = textwrap.dedent(f"""
warn_msg = textwrap.dedent(
f"""
KIAUH was unable to mask the {service} system service.
Please fix the problem manually. Otherwise, this may have
undesirable effects on the operation of Klipper.
""")[1:]
"""
)[1:]
Logger.print_warn(warn_msg)

29
kiauh/modules/klipper/klipper_utils.py
View File

@@ -13,12 +13,12 @@ from typing import List
from kiauh.instance_manager.base_instance import BaseInstance
from kiauh.menus.base_menu import print_back_footer
from kiauh.utils.constants import COLOR_GREEN, COLOR_CYAN, COLOR_YELLOW, \
RESET_FORMAT
from kiauh.utils.constants import COLOR_GREEN, COLOR_CYAN, COLOR_YELLOW, RESET_FORMAT
def print_instance_overview(instances: List[BaseInstance], show_index=False,
show_select_all=False):
def print_instance_overview(
instances: List[BaseInstance], show_index=False, show_select_all=False
):
headline = f"{COLOR_GREEN}The following Klipper instances were found:{RESET_FORMAT}"
print("/=======================================================\\")
@@ -38,13 +38,20 @@ def print_instance_overview(instances: List[BaseInstance], show_index=False,
print_back_footer()
def print_missing_usergroup_dialog(missing_groups) -> None:
print("/=======================================================\\")
print(f"| {COLOR_YELLOW}WARNING: Your current user is not in group:{RESET_FORMAT} |")
print(
f"| {COLOR_YELLOW}WARNING: Your current user is not in group:{RESET_FORMAT} |"
)
if "tty" in missing_groups:
print(f"| {COLOR_CYAN}● tty{RESET_FORMAT} |")
print(
f"| {COLOR_CYAN}● tty{RESET_FORMAT} |"
)
if "dialout" in missing_groups:
print(f"| {COLOR_CYAN}● dialout{RESET_FORMAT} |")
print(
f"| {COLOR_CYAN}● dialout{RESET_FORMAT} |"
)
print("| |")
print("| It is possible that you won't be able to successfully |")
print("| connect and/or flash the controller board without |")
@@ -52,6 +59,10 @@ def print_missing_usergroup_dialog(missing_groups) -> None:
print("| If you want to add the current user to the group(s) |")
print("| listed above, answer with 'Y'. Else skip with 'n'. |")
print("| |")
print(f"| {COLOR_YELLOW}INFO:{RESET_FORMAT} |")
print(f"| {COLOR_YELLOW}Relog required for group assignments to take effect!{RESET_FORMAT} |")
print(
f"| {COLOR_YELLOW}INFO:{RESET_FORMAT} |"
)
print(
f"| {COLOR_YELLOW}Relog required for group assignments to take effect!{RESET_FORMAT} |"
)
print("\\=======================================================/")