h3n3/kiauh
1
0
Fork 0
Code Issues 46 Pull Requests 10 Packages Projects Releases Wiki Activity

refactor(sys_utils): rename systemctl method and add new one

Browse Source 
Signed-off-by: Dominik Willner <th33xitus@gmail.com>
This commit is contained in:
dw-0
2024-05-03 22:20:00 +02:00
parent 5f1e42b88b
commit 799892500a
6 changed files with 43 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
kiauh/components/crowsnest/crowsnest.py
View File

@@ -30,7 +30,7 @@ from utils.input_utils import get_confirm
from utils.logger import Logger
from utils.sys_utils import (
parse_packages_from_file,
control_systemd_service,
cmd_sysctl_service,
)
@@ -98,7 +98,7 @@ def install_crowsnest() -> None:
def update_crowsnest() -> None:
try:
control_systemd_service("crowsnest", "stop")
cmd_sysctl_service("crowsnest", "stop")
if not CROWSNEST_DIR.exists():
git_clone_wrapper(CROWSNEST_REPO, CROWSNEST_DIR, "master")
@@ -111,7 +111,7 @@ def update_crowsnest() -> None:
deps = parse_packages_from_file(script)
check_install_dependencies(deps)
control_systemd_service("crowsnest", "restart")
cmd_sysctl_service("crowsnest", "restart")
Logger.print_ok("Crowsnest updated successfully.", end="\n\n")
except CalledProcessError as e:

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

@@ -45,7 +45,7 @@ from utils.constants import CURRENT_USER
from utils.git_utils import get_repo_name, get_remote_commit, get_local_commit
from utils.input_utils import get_confirm, get_string_input, get_number_input
from utils.logger import Logger
from utils.sys_utils import control_systemd_service
from utils.sys_utils import cmd_sysctl_service
def get_klipper_status() -> (
@@ -258,7 +258,7 @@ def handle_disruptive_system_packages() -> None:
for service in services if services else []:
try:
control_systemd_service(service, "mask")
cmd_sysctl_service(service, "mask")
except subprocess.CalledProcessError:
warn_msg = textwrap.dedent(
f"""

12
kiauh/components/klipperscreen/klipperscreen.py
View File

@@ -37,7 +37,7 @@ from utils.input_utils import get_confirm
from utils.logger import Logger, DialogType
from utils.sys_utils import (
check_python_version,
control_systemd_service,
cmd_sysctl_service,
install_python_requirements,
)
@@ -104,7 +104,7 @@ def patch_klipperscreen_update_manager(instances: List[Moonraker]) -> None:
def update_klipperscreen() -> None:
try:
control_systemd_service("KlipperScreen", "stop")
cmd_sysctl_service("KlipperScreen", "stop")
if not KLIPPERSCREEN_DIR.exists():
Logger.print_info(
@@ -114,7 +114,7 @@ def update_klipperscreen() -> None:
Logger.print_status("Updating KlipperScreen ...")
control_systemd_service("KlipperScreen", "stop")
cmd_sysctl_service("KlipperScreen", "stop")
settings = KiauhSettings()
if settings.get("kiauh", "backup_before_update"):
@@ -127,7 +127,7 @@ def update_klipperscreen() -> None:
)
install_python_requirements(KLIPPERSCREEN_ENV, requirements)
control_systemd_service("KlipperScreen", "start")
cmd_sysctl_service("KlipperScreen", "start")
Logger.print_ok("KlipperScreen updated successfully.", end="\n\n")
except CalledProcessError as e:
@@ -176,8 +176,8 @@ def remove_klipperscreen() -> None:
service = SYSTEMD.joinpath("KlipperScreen.service")
if service.exists():
Logger.print_status("Removing KlipperScreen service ...")
control_systemd_service(service, "stop")
control_systemd_service(service, "disable")
cmd_sysctl_service(service, "stop")
cmd_sysctl_service(service, "disable")
remove_with_sudo(service)
Logger.print_ok("KlipperScreen service successfully removed!")

4
kiauh/components/webui_client/client_setup.py
View File

@@ -55,7 +55,7 @@ from utils.sys_utils import (
download_file,
set_nginx_permissions,
get_ipv4_addr,
control_systemd_service,
cmd_sysctl_service,
)
@@ -145,7 +145,7 @@ def install_client(client: BaseWebClient) -> None:
create_client_nginx_cfg(client, port)
if kl_instances:
symlink_webui_nginx_log(kl_instances)
control_systemd_service("nginx", "restart")
cmd_sysctl_service("nginx", "restart")
except Exception as e:
Logger.print_error(f"{client.display_name} installation failed!\n{e}")

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

@@ -15,7 +15,7 @@ from typing import List, Optional, Union, TypeVar
from core.instance_manager.base_instance import BaseInstance
from utils.constants import SYSTEMD
from utils.logger import Logger
from utils.sys_utils import control_systemd_service
from utils.sys_utils import cmd_sysctl_service
T = TypeVar(name="T", bound=BaseInstance, covariant=True)
@@ -109,7 +109,7 @@ class InstanceManager:
def enable_instance(self) -> None:
Logger.print_status(f"Enabling {self.instance_service_full} ...")
try:
control_systemd_service(self.instance_service_full, "enable")
cmd_sysctl_service(self.instance_service_full, "enable")
except subprocess.CalledProcessError as e:
Logger.print_error(f"Error enabling service {self.instance_service_full}:")
Logger.print_error(f"{e}")
@@ -117,7 +117,7 @@ class InstanceManager:
def disable_instance(self) -> None:
Logger.print_status(f"Disabling {self.instance_service_full} ...")
try:
control_systemd_service(self.instance_service_full, "disable")
cmd_sysctl_service(self.instance_service_full, "disable")
except subprocess.CalledProcessError as e:
Logger.print_error(f"Error disabling {self.instance_service_full}:")
Logger.print_error(f"{e}")
@@ -125,7 +125,7 @@ class InstanceManager:
def start_instance(self) -> None:
Logger.print_status(f"Starting {self.instance_service_full} ...")
try:
control_systemd_service(self.instance_service_full, "start")
cmd_sysctl_service(self.instance_service_full, "start")
except subprocess.CalledProcessError as e:
Logger.print_error(f"Error starting {self.instance_service_full}:")
Logger.print_error(f"{e}")
@@ -133,7 +133,7 @@ class InstanceManager:
def restart_instance(self) -> None:
Logger.print_status(f"Restarting {self.instance_service_full} ...")
try:
control_systemd_service(self.instance_service_full, "restart")
cmd_sysctl_service(self.instance_service_full, "restart")
except subprocess.CalledProcessError as e:
Logger.print_error(f"Error restarting {self.instance_service_full}:")
Logger.print_error(f"{e}")
@@ -151,7 +151,7 @@ class InstanceManager:
def stop_instance(self) -> None:
Logger.print_status(f"Stopping {self.instance_service_full} ...")
try:
control_systemd_service(self.instance_service_full, "stop")
cmd_sysctl_service(self.instance_service_full, "stop")
except subprocess.CalledProcessError as e:
Logger.print_error(f"Error stopping {self.instance_service_full}:")
Logger.print_error(f"{e}")

29
kiauh/utils/sys_utils.py
View File

@@ -26,6 +26,19 @@ from utils.input_utils import get_confirm
from utils.logger import Logger
SysCtlServiceAction = Literal[
"start",
"stop",
"restart",
"reload",
"enable",
"disable",
"mask",
"unmask",
]
SysCtlManageAction = Literal["deamon-reload", "reset-failed"]
def kill(opt_err_msg: str = "") -> None:
"""
Kills the application |
@@ -328,9 +341,7 @@ def set_nginx_permissions() -> None:
Logger.print_ok("Permissions granted.")
def control_systemd_service(
name: str, action: Literal["start", "stop", "restart", "enable", "disable", "mask"]
) -> None:
def cmd_sysctl_service(name: str, action: SysCtlServiceAction) -> None:
"""
Helper method to execute several actions for a specific systemd service. |
:param name: the service name
@@ -339,8 +350,7 @@ def control_systemd_service(
"""
try:
Logger.print_status(f"{action.capitalize()} {name} ...")
command = ["sudo", "systemctl", action, name]
run(command, stderr=PIPE, check=True)
run(["sudo", "systemctl", action, name], stderr=PIPE, check=True)
Logger.print_ok("OK!")
except CalledProcessError as e:
log = f"Failed to {action} {name}: {e.stderr.decode()}"
@@ -348,6 +358,15 @@ def control_systemd_service(
raise
def cmd_sysctl_manage(action: SysCtlManageAction) -> None:
try:
run(["sudo", "systemctl", action], stderr=PIPE, check=True)
except CalledProcessError as e:
log = f"Failed to run {action}: {e.stderr.decode()}"
Logger.print_error(log)
raise
def log_process(process: Popen) -> None:
"""
Helper method to print stdout of a process in near realtime to the console.