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

refactor(scp): replace old config parser with new one, remove ConfigManager

Browse Source 
Signed-off-by: Dominik Willner <th33xitus@gmail.com>
This commit is contained in:
dw-0
2024-06-16 21:46:21 +02:00
parent 7b9f9b1a67
commit 802eaccf57
7 changed files with 77 additions and 132 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -34,10 +34,12 @@ from components.webui_client.client_config.client_config_setup import (
create_client_config_symlink,
)
from core.backup_manager.backup_manager import BackupManager
from core.config_manager.config_manager import ConfigManager
from core.instance_manager.base_instance import BaseInstance
from core.instance_manager.instance_manager import InstanceManager
from core.instance_manager.name_scheme import NameScheme
from core.submodules.simple_config_parser.src.simple_config_parser.simple_config_parser import (
SimpleConfigParser,
)
from utils import PRINTER_CFG_BACKUP_DIR
from utils.common import get_install_status
from utils.constants import CURRENT_USER
@@ -186,10 +188,11 @@ def klipper_to_multi_conversion(new_name: str) -> None:
# patch the virtual_sdcard sections path
# value to match the new printer_data foldername
cm = ConfigManager(new_instance.cfg_file)
if cm.config.has_section("virtual_sdcard"):
cm.set_value("virtual_sdcard", "path", str(new_instance.gcodes_dir))
cm.write_config()
scp = SimpleConfigParser()
scp.read(new_instance.cfg_file)
if scp.has_section("virtual_sdcard"):
scp.set("virtual_sdcard", "path", str(new_instance.gcodes_dir))
scp.write(new_instance.cfg_file)
# finalize creating the new instance
im.create_instance()
@@ -292,18 +295,19 @@ def create_example_printer_cfg(
Logger.print_error(f"Unable to create example printer.cfg:\n{e}")
return
cm = ConfigManager(target)
cm.set_value("virtual_sdcard", "path", str(instance.gcodes_dir))
scp = SimpleConfigParser()
scp.read(target)
scp.set("virtual_sdcard", "path", str(instance.gcodes_dir))
# include existing client configs in the example config
if clients is not None and len(clients) > 0:
for c in clients:
client_config = c.client_config
section = client_config.config_section
cm.config.add_section(section=section)
scp.add_section(section=section)
create_client_config_symlink(client_config, [instance])
cm.write_config()
scp.write(target)
Logger.print_ok(f"Example printer.cfg created in '{instance.cfg_dir}'")

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

@@ -6,14 +6,17 @@
# #
# This file may be distributed under the terms of the GNU GPLv3 license #
# ======================================================================= #
from __future__ import annotations
import subprocess
from pathlib import Path
from typing import List, Union
from typing import List
from components.moonraker import MODULE_PATH, MOONRAKER_DIR, MOONRAKER_ENV_DIR
from core.config_manager.config_manager import ConfigManager
from core.instance_manager.base_instance import BaseInstance
from core.submodules.simple_config_parser.src.simple_config_parser.simple_config_parser import (
SimpleConfigParser,
)
from utils.constants import SYSTEMD
from utils.logger import Logger
@@ -144,11 +147,12 @@ class Moonraker(BaseInstance):
)
return env_file_content
def _get_port(self) -> Union[int, None]:
def _get_port(self) -> int | None:
if not self.cfg_file.is_file():
return None
cm = ConfigManager(cfg_file=self.cfg_file)
port = cm.get_value("server", "port")
scp = SimpleConfigParser()
scp.read(self.cfg_file)
port = scp.getint("server", "port", fallback=None)
return int(port) if port is not None else port
return port

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

@@ -23,8 +23,10 @@ from components.webui_client.base_data import BaseWebClient
from components.webui_client.client_utils import enable_mainsail_remotemode
from components.webui_client.mainsail_data import MainsailData
from core.backup_manager.backup_manager import BackupManager
from core.config_manager.config_manager import ConfigManager
from core.instance_manager.instance_manager import InstanceManager
from core.submodules.simple_config_parser.src.simple_config_parser.simple_config_parser import (
SimpleConfigParser,
)
from utils.common import get_install_status
from utils.logger import Logger
from utils.sys_utils import (
@@ -76,13 +78,21 @@ def create_example_moonraker_conf(
ip.extend(["0", "0/16"])
uds = instance.comms_dir.joinpath("klippy.sock")
cm = ConfigManager(target)
trusted_clients = f"\n{'.'.join(ip)}"
trusted_clients += cm.get_value("authorization", "trusted_clients")
scp = SimpleConfigParser()
scp.read(target)
trusted_clients: List[str] = [
".".join(ip),
*scp.get("authorization", "trusted_clients"),
]
cm.set_value("server", "port", str(port))
cm.set_value("server", "klippy_uds_address", str(uds))
cm.set_value("authorization", "trusted_clients", trusted_clients)
scp.set("server", "port", str(port))
scp.set("server", "klippy_uds_address", str(uds))
scp.set(
"authorization",
"trusted_clients",
"\n".join(trusted_clients),
True,
)
# add existing client and client configs in the update section
if clients is not None and len(clients) > 0:
@@ -95,9 +105,9 @@ def create_example_moonraker_conf(
("repo", c.repo_path),
("path", c.client_dir),
]
cm.config.add_section(section=c_section)
scp.add_section(section=c_section)
for option in c_options:
cm.config.set(c_section, option[0], option[1])
scp.set(c_section, option[0], option[1])
# client config part
c_config = c.client_config
@@ -110,11 +120,11 @@ def create_example_moonraker_conf(
("origin", c_config.repo_url),
("managed_services", "klipper"),
]
cm.config.add_section(section=c_config_section)
scp.add_section(section=c_config_section)
for option in c_config_options:
cm.config.set(c_config_section, option[0], option[1])
scp.set(c_config_section, option[0], option[1])
cm.write_config()
scp.write(target)
Logger.print_ok(f"Example moonraker.conf created in '{instance.cfg_dir}'")
@@ -150,14 +160,15 @@ def moonraker_to_multi_conversion(new_name: str) -> None:
im.current_instance = new_instance
# patch the server sections klippy_uds_address value to match the new printer_data foldername
cm = ConfigManager(new_instance.cfg_file)
if cm.config.has_section("server"):
cm.set_value(
scp = SimpleConfigParser()
scp.read(new_instance.cfg_file)
if scp.has_section("server"):
scp.set(
"server",
"klippy_uds_address",
str(new_instance.comms_dir.joinpath("klippy.sock")),
)
cm.write_config()
scp.write(new_instance.cfg_file)
# create, enable and start the new moonraker instance
im.create_instance()

0
kiauh/core/config_manager/__init__.py
View File

83
kiauh/core/config_manager/config_manager.py
View File

@@ -1,83 +0,0 @@
# ======================================================================= #
# Copyright (C) 2020 - 2024 Dominik Willner <th33xitus@gmail.com> #
# #
# This file is part of KIAUH - Klipper Installation And Update Helper #
# https://github.com/dw-0/kiauh #
# #
# This file may be distributed under the terms of the GNU GPLv3 license #
# ======================================================================= #
import configparser
from pathlib import Path
from typing import Union
from utils.logger import Logger
# noinspection PyMethodMayBeStatic
class ConfigManager:
def __init__(self, cfg_file: Path):
self.config_file = cfg_file
self.config = CustomConfigParser()
if cfg_file.is_file():
self.read_config()
def read_config(self) -> None:
if not self.config_file:
Logger.print_error("Unable to read config file. File not found.")
return
self.config.read_file(open(self.config_file, "r"))
def write_config(self) -> None:
with open(self.config_file, "w") as cfg:
self.config.write(cfg)
def get_value(self, section: str, key: str, silent=True) -> Union[str, bool, None]:
if not self.config.has_section(section):
if not silent:
log = f"Section not defined. Unable to read section: [{section}]."
Logger.print_error(log)
return None
if not self.config.has_option(section, key):
if not silent:
log = f"Option not defined in section [{section}]. Unable to read option: '{key}'."
Logger.print_error(log)
return None
value = self.config.get(section, key)
if value == "True" or value == "true":
return True
elif value == "False" or value == "false":
return False
else:
return value
def set_value(self, section: str, key: str, value: str):
self.config.set(section, key, value)
class CustomConfigParser(configparser.ConfigParser):
"""
A custom ConfigParser class overwriting the write() method of configparser.Configparser.
Key and value will be delimited by a ": ".
Note the whitespace AFTER the colon, which is the whole reason for that overwrite.
"""
def write(self, fp, space_around_delimiters=False):
if self._defaults:
fp.write("[%s]\n" % configparser.DEFAULTSECT)
for key, value in self._defaults.items():
fp.write("%s: %s\n" % (key, str(value).replace("\n", "\n\t")))
fp.write("\n")
for section in self._sections:
fp.write("[%s]\n" % section)
for key, value in self._sections[section].items():
if key == "__name__":
continue
if (value is not None) or (self._optcre == self.OPTCRE):
key = ": ".join((key, str(value).replace("\n", "\n\t")))
fp.write("%s\n" % key)
fp.write("\n")

13
kiauh/extensions/gcode_shell_cmd/gcode_shell_cmd_extension.py
View File

@@ -13,8 +13,10 @@ from typing import List
from components.klipper.klipper import Klipper
from core.backup_manager.backup_manager import BackupManager
from core.config_manager.config_manager import ConfigManager
from core.instance_manager.instance_manager import InstanceManager
from core.submodules.simple_config_parser.src.simple_config_parser.simple_config_parser import (
SimpleConfigParser,
)
from extensions.base_extension import BaseExtension
from extensions.gcode_shell_cmd import (
EXAMPLE_CFG_SRC,
@@ -118,10 +120,11 @@ class GcodeShellCmdExtension(BaseExtension):
cfg_files = [instance.cfg_file for instance in instances]
for cfg_file in cfg_files:
Logger.print_status(f"Include shell_command.cfg in '{cfg_file}' ...")
cm = ConfigManager(cfg_file)
if cm.config.has_section(section):
scp = SimpleConfigParser()
scp.read(cfg_file)
if scp.has_section(section):
Logger.print_info("Section already defined! Skipping ...")
continue
cm.config.add_section(section)
cm.write_config()
scp.add_section(section)
scp.write(cfg_file)
Logger.print_ok("Done!")

32
kiauh/utils/config_utils.py
View File

@@ -12,7 +12,9 @@ from typing import List, Optional, Tuple, TypeVar
from components.klipper.klipper import Klipper
from components.moonraker.moonraker import Moonraker
from core.config_manager.config_manager import ConfigManager
from core.submodules.simple_config_parser.src.simple_config_parser.simple_config_parser import (
SimpleConfigParser,
)
from utils.logger import Logger
B = TypeVar("B", Klipper, Moonraker)
@@ -32,27 +34,30 @@ def add_config_section(
Logger.print_warn(f"'{cfg_file}' not found!")
continue
cm = ConfigManager(cfg_file)
if cm.config.has_section(section):
scp = SimpleConfigParser()
scp.read(cfg_file)
if scp.has_section(section):
Logger.print_info("Section already exist. Skipped ...")
continue
cm.config.add_section(section)
scp.add_section(section)
if options is not None:
for option in options:
cm.config.set(section, option[0], option[1])
scp.set(section, option[0], option[1])
cm.write_config()
scp.write(cfg_file)
def add_config_section_at_top(section: str, instances: List[B]):
# TODO: this could be implemented natively in SimpleConfigParser
for instance in instances:
tmp_cfg = tempfile.NamedTemporaryFile(mode="w", delete=False)
tmp_cfg_path = Path(tmp_cfg.name)
cmt = ConfigManager(tmp_cfg_path)
cmt.config.add_section(section)
cmt.write_config()
scp = SimpleConfigParser()
scp.read(tmp_cfg_path)
scp.add_section(section)
scp.write(tmp_cfg_path)
tmp_cfg.close()
cfg_file = instance.cfg_file
@@ -74,10 +79,11 @@ def remove_config_section(section: str, instances: List[B]) -> None:
Logger.print_warn(f"'{cfg_file}' not found!")
continue
cm = ConfigManager(cfg_file)
if not cm.config.has_section(section):
scp = SimpleConfigParser()
scp.read(cfg_file)
if not scp.has_section(section):
Logger.print_info("Section does not exist. Skipped ...")
continue
cm.config.remove_section(section)
cm.write_config()
scp.remove_section(section)
scp.write(cfg_file)