install_service Namespace Reference

Data Structures
class	DummyFore
class	DummyStyle

Functions
	get_lang ()
	set_lang (lang)
	colorama_auto_install ()
	setup_logging (verbose=False, quiet=False)
	t (key)
	get_user_input (prompt, valid_inputs, max_attempts=3)
	list_yaml_files ()
	test_yaml_file (file_path)
	detect_yaml_type (file_path)
	execute (command, dry_run=False)
	verify_service (service_path)
	install_service (yaml_file, dry_run=False)
	remove_service (service_name, dry_run=False)
	remove_menu (dry_run=False)
	init_language ()
	main (dry_run=False)

Variables
	BASE_DIR = Path(__file__).resolve().parent
str	BW_DIR = '/opt/boswatch3'
	SERVICE_DIR = Path('/etc/systemd/system')
tuple	CONFIG_DIR = (BASE_DIR / 'config').resolve()
tuple	LOG_FILE = (BASE_DIR / 'log' / 'install' / 'service_install.log').resolve()
	parent
	exist_ok
str	_lang = 'de'
dict	TEXT
	colorama_available
	Fore
	Style = DummyStyle()
	lang_parser
	remaining_argv
	parser
	action
	help
	args = parser.parse_args(remaining_argv)
	verbose
	quiet
	dry_run

Detailed Description

---

/ __ )/ __ \/ ___/ | / /___ _/ /______/ /_ |__ / / __ / / / /__ | | /| / / __ `/ __/ ___/ __ \ /_ < / /_/ / /_/ /___/ /| |/ |/ / /_/ / /_/ /__/ / / / ___/ / /_____/____//____/ |__/|__/__,_/__/___/_/ /_/ /____/ German BOS Information Script by Bastian Schroll

Function Documentation

get_lang()

install_service.get_lang

(

)

def get_lang():
    return _lang
 
 

set_lang()

install_service.set_lang

(

lang

)

def set_lang(lang):
    global _lang
    _lang = lang
 
 
#  === text-dictionary ===

colorama_auto_install()

install_service.colorama_auto_install

(

)

    Auto-installs colorama if missing.
    Note: Language detection happens before colorama is available.

def colorama_auto_install():
    r"""
    Auto-installs colorama if missing.
    Note: Language detection happens before colorama is available.
    """
    # recognize language early (before colorama installation)
    import argparse
    early_parser = argparse.ArgumentParser(add_help=False)
    early_parser.add_argument('--lang', '-l', choices=['de', 'en'], default='de')
    early_args, _ = early_parser.parse_known_args()
    lang = early_args.lang
 
    # use text from global TEXT dictionary
    txt = TEXT[lang]
 
    try:
        from colorama import init as colorama_init, Fore, Style
        colorama_init(autoreset=True)
        return True, Fore, Style
    except ImportError:
        print(txt["colorama_missing"])
 
        # install Colorama
        print(txt["colorama_install"])
        subprocess.run(["sudo", "apt", "install", "-y", "python3-colorama"], check=False)
 
        # retry importing Colorama
        try:
            from colorama import init as colorama_init, Fore, Style
            colorama_init(autoreset=True)
            print(txt["colorama_install_ok"])
            return True, Fore, Style
        except ImportError:
            print(txt["colorama_install_fail"])
            return False, None, None
 
 
# === import / install colorama ===

setup_logging()

install_service.setup_logging	(		verbose = False,
			quiet = False
	)

Setup logging to file and console with colorized output.

def setup_logging(verbose=False, quiet=False):
    r"""
    Setup logging to file and console with colorized output.
    """
    log_level = logging.INFO
    if quiet:
        log_level = logging.WARNING
    elif verbose:
        log_level = logging.DEBUG
 
    logger = logging.getLogger()
    logger.setLevel(log_level)
 
    formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
 
    # File Handler (plain)
    fh = logging.FileHandler(LOG_FILE)
    fh.setFormatter(formatter)
    logger.addHandler(fh)
 
    # Console Handler (colorized)
    class ColorFormatter(logging.Formatter):
        COLORS = {
            logging.DEBUG: Fore.CYAN,
            logging.INFO: Fore.GREEN,
            logging.WARNING: Fore.YELLOW,
            logging.ERROR: Fore.RED,
            logging.CRITICAL: Fore.RED + Style.BRIGHT,
        }
 
        def format(self, record):
            color = self.COLORS.get(record.levelno, Fore.RESET)
            message = super().format(record)
            return f"{color}{message}{Style.RESET_ALL}"
 
    ch = logging.StreamHandler(sys.stdout)
    ch.setFormatter(ColorFormatter('%(levelname)s: %(message)s'))
    logger.addHandler(ch)
 
    return logger
 
 

t()

install_service.t

(

key

)

Translation helper: returns the localized string for the given key.

def t(key):
    r"""
    Translation helper: returns the localized string for the given key.
    """
    lang = get_lang()
    return TEXT.get(lang, TEXT['de']).get(key, key)
 
 

get_user_input()

install_service.get_user_input	(		prompt,
			valid_inputs,
			max_attempts = 3
	)

Prompt user for input until a valid input from valid_inputs is entered or max_attempts exceeded.
Raises RuntimeError on failure.

def get_user_input(prompt, valid_inputs, max_attempts=3):
    r"""
    Prompt user for input until a valid input from valid_inputs is entered or max_attempts exceeded.
    Raises RuntimeError on failure.
    """
    attempts = 0
    while attempts < max_attempts:
        value = input(prompt).strip().lower()
        if value in valid_inputs:
            return value
        logging.warning(t("invalid_input").format(", ".join(valid_inputs)))
        attempts += 1
    raise RuntimeError(t("max_attempts_exceeded"))
 
 

list_yaml_files()

install_service.list_yaml_files

(

)

Returns a list of .yaml or .yml files in the config directory.

def list_yaml_files():
    r"""
    Returns a list of .yaml or .yml files in the config directory.
    """
    return sorted([f.name for f in CONFIG_DIR.glob("*.y*ml")])
 
 

test_yaml_file()

install_service.test_yaml_file

(

file_path

)

Tests if YAML file can be loaded without error.

def test_yaml_file(file_path):
    r"""
    Tests if YAML file can be loaded without error.
    """
    try:
        content = file_path.read_text(encoding='utf-8')
        yaml.safe_load(content)
        return True
    except Exception as e:
        logging.error(t("yaml_error").format(file_path, e))
        return False
 
 

detect_yaml_type()

install_service.detect_yaml_type

(

file_path

)

Detects if YAML config is 'client' or 'server' type.

def detect_yaml_type(file_path):
    r"""
    Detects if YAML config is 'client' or 'server' type.
    """
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            data = yaml.safe_load(f)
        if 'client' in data:
            return 'client'
        elif 'server' in data:
            return 'server'
        else:
            logging.error(t("unknown_yaml_type").format(os.path.basename(file_path)))
            return None
    except Exception as e:
        logging.error(t("yaml_read_error").format(file_path, e))
        return None
 
 

execute()

install_service.execute	(		command,
			dry_run = False
	)

Executes shell command unless dry_run is True.

def execute(command, dry_run=False):
    r"""
    Executes shell command unless dry_run is True.
    """
    logging.debug(f"→ {command}")
    if not dry_run:
        subprocess.run(command, shell=True, check=False)
 
 

verify_service()

install_service.verify_service

(

service_path

)

Runs 'systemd-analyze verify' on the service file and logs warnings/errors.

def verify_service(service_path):
    r"""
    Runs 'systemd-analyze verify' on the service file and logs warnings/errors.
    """
    try:
        result = subprocess.run(
            ['systemd-analyze', 'verify', service_path],
            capture_output=True,
            text=True,
            timeout=10
        )
        if result.returncode != 0 or result.stderr:
            logging.warning(t("verify_warn").format(result.stderr.strip()))
        else:
            logging.debug(t("verify_ok").format(os.path.basename(service_path)))
    except subprocess.TimeoutExpired:
        logging.warning(t("verify_timeout").format(os.path.basename(service_path)))
    except Exception as e:
        logging.error(t("yaml_error").format(service_path, e))
 
 

install_service()

install_service.install_service	(		yaml_file,
			dry_run = False
	)

Creates and installs systemd service based on YAML config.

def install_service(yaml_file, dry_run=False):
    r"""
    Creates and installs systemd service based on YAML config.
    """
    yaml_path = CONFIG_DIR / yaml_file
    yaml_type = detect_yaml_type(yaml_path)
    if yaml_type == 'server':
        is_server = True
    elif yaml_type == 'client':
        is_server = False
    else:
        logging.error(t("unknown_yaml_type").format(yaml_file))
        return
 
    service_name = f"bw3_{Path(yaml_file).stem}.service"
    service_path = SERVICE_DIR / service_name
 
    if is_server:
        exec_line = f"/usr/bin/python3 {BW_DIR}/bw_server.py -c {yaml_file}"
        description = "BOSWatch Server"
        after = "network-online.target"
        wants = "Wants=network-online.target"
    else:
        exec_line = f"/usr/bin/python3 {BW_DIR}/bw_client.py -c {yaml_file}"
        description = "BOSWatch Client"
        after = "network.target"
        wants = ""
 
    service_content = f"""[Unit]
Description={description}
After={after}
{wants}
 
[Service]
Type=simple
WorkingDirectory={BW_DIR}
ExecStart={exec_line}
Restart=on-abort
 
[Install]
WantedBy=multi-user.target
"""
 
    logging.info(t("creating_service_file").format(yaml_file, service_name))
 
    if not dry_run:
        try:
            with open(service_path, 'w', encoding='utf-8') as f:
                f.write(service_content)
        except IOError as e:
            logging.error(t("file_write_error").format(service_path, e))
            return
        verify_service(service_path)
 
    execute("systemctl daemon-reload", dry_run=dry_run)
    execute(f"systemctl enable {service_name}", dry_run=dry_run)
    execute(f"systemctl start {service_name}", dry_run=dry_run)
 
    if not dry_run:
        try:
            subprocess.run(
                ["systemctl", "is-active", "--quiet", service_name],
                check=True,
                timeout=5
            )
            logging.info(t("service_active").format(service_name))
        except subprocess.CalledProcessError:
            logging.warning(t("service_inactive").format(service_name))
        except subprocess.TimeoutExpired:
            logging.warning(t("status_timeout").format(service_name))
    else:
        logging.info(t("dryrun_status_check").format(service_name))
 
 

remove_service()

install_service.remove_service	(		service_name,
			dry_run = False
	)

Stops, disables and removes the given systemd service.

def remove_service(service_name, dry_run=False):
    r"""
    Stops, disables and removes the given systemd service.
    """
    logging.warning(t("removing_service").format(service_name))
    execute(f"systemctl stop {service_name}", dry_run=dry_run)
    execute(f"systemctl disable {service_name}", dry_run=dry_run)
 
    service_path = SERVICE_DIR / service_name
    if not dry_run and service_path.exists():
        try:
            service_path.unlink()
            logging.info(t("service_deleted").format(service_name))
        except Exception as e:
            logging.error(t("file_write_error").format(service_path, e))
    else:
        logging.warning(t("service_not_found").format(service_name))
 
    execute("systemctl daemon-reload", dry_run=dry_run)
 
 

remove_menu()

install_service.remove_menu

(

dry_run = False

)

Interactive menu to remove services.

def remove_menu(dry_run=False):
    r"""
    Interactive menu to remove services.
    """
    while True:
        services = sorted([
            f for f in os.listdir(SERVICE_DIR)
            if f.startswith('bw3_') and f.endswith('.service')
        ])
 
        if not services:
            print(Fore.YELLOW + t("no_services") + Style.RESET_ALL)
            return
 
        print(Fore.CYAN + "\n" + t("available_services") + Style.RESET_ALL)
        for i, s in enumerate(services):
            print(f" [{i}] {s}")
        print(" " + t("all"))
        print(" " + t("exit"))
 
        try:
            auswahl = get_user_input(
                t("remove_prompt"),
                ['e', 'a'] + [str(i) for i in range(len(services))]
            )
        except RuntimeError:
            logging.error(t("max_attempts_exceeded"))
            break
 
        if auswahl == 'e':
            break
        elif auswahl == 'a':
            for s in services:
                remove_service(s, dry_run=dry_run)
            # directly continue to the next loop (updated list!)
            continue
        else:
            remove_service(services[int(auswahl)], dry_run=dry_run)
            # also directly continue to the next loop (updated list!)
            continue
 
 

init_language()

install_service.init_language

(

)

Parses --lang/-l argument early to set language before other parsing.

def init_language():
    r"""
    Parses --lang/-l argument early to set language before other parsing.
    """
    lang_parser = argparse.ArgumentParser(add_help=False)
    lang_parser.add_argument(
        '--lang', '-l',
        choices=['de', 'en'],
        default='de',
        metavar='LANG',
        help=TEXT["en"]["help_lang"]
    )
    lang_args, remaining_argv = lang_parser.parse_known_args()
    set_lang(lang_args.lang)
    return lang_parser, remaining_argv
 
 

main()

install_service.main

(

dry_run = False

)

main program: install or remove service.

def main(dry_run=False):
    r"""
    main program: install or remove service.
    """
    print(Fore.GREEN + Style.BRIGHT + t("script_title") + Style.RESET_ALL)
    print(t('mode_dry') if dry_run else t('mode_live'))
    print()
 
    yaml_files = list_yaml_files()
    if not yaml_files:
        print(Fore.RED + t("no_yaml") + Style.RESET_ALL)
        sys.exit(1)
 
    print(Fore.GREEN + t("found_yaml").format(len(yaml_files)) + Style.RESET_ALL)
    for f in yaml_files:
        file_path = CONFIG_DIR / f
        valid = test_yaml_file(file_path)
        status = Fore.GREEN + "✅" if valid else Fore.RED + "❌"
        print(f" - {f} {status}{Style.RESET_ALL}")
 
    try:
        action = get_user_input(t("action_prompt"), ['i', 'r', 'e'])
    except RuntimeError:
        logging.error(t("max_retries_exit"))
        sys.exit(1)
 
    if action == 'e':
        sys.exit(0)
    elif action == 'r':
        remove_menu(dry_run=dry_run)
        return
 
    try:
        edited = get_user_input(t("edited_prompt"), ['y', 'n'])
    except RuntimeError:
        logging.error(t("max_retries_exit"))
        sys.exit(1)
 
    if edited == 'n':
        print(Fore.YELLOW + t("edit_abort") + Style.RESET_ALL)
        sys.exit(0)
 
    installed = 0
    skipped = 0
 
    for yaml_file in yaml_files:
        file_path = CONFIG_DIR / yaml_file
        if not test_yaml_file(file_path):
            print(Fore.RED + t("skip_invalid_yaml").format(yaml_file) + Style.RESET_ALL)
            skipped += 1
            continue
 
        try:
            install = get_user_input(t("install_confirm").format(yaml_file), ['y', 'n'])
        except RuntimeError:
            logging.error(t("max_retries_skip"))
            skipped += 1
            continue
 
        if install == 'y':
            install_service(yaml_file, dry_run=dry_run)
            installed += 1
        else:
            logging.info(t("install_skipped").format(yaml_file))
            skipped += 1
 
    print()
    logging.info(t("install_done").format(installed, skipped))
 
 

Variable Documentation

BASE_DIR

install_service.BASE_DIR = Path(__file__).resolve().parent

BW_DIR

str install_service.BW_DIR = '/opt/boswatch3'

SERVICE_DIR

install_service.SERVICE_DIR = Path('/etc/systemd/system')

CONFIG_DIR

tuple install_service.CONFIG_DIR = (BASE_DIR / 'config').resolve()

LOG_FILE

tuple install_service.LOG_FILE = (BASE_DIR / 'log' / 'install' / 'service_install.log').resolve()

parent

install_service.parent

exist_ok

install_service.exist_ok

_lang

str install_service._lang = 'de'

protected

TEXT

dict install_service.TEXT

colorama_available

install_service.colorama_available

Fore

install_service.Fore

Style

install_service.Style = DummyStyle()

lang_parser

install_service.lang_parser

remaining_argv

install_service.remaining_argv

parser

install_service.parser

Initial value:

=  argparse.ArgumentParser(
        description=t("script_title"),
        parents=[lang_parser]
    )

action

install_service.action

help

install_service.help

args

install_service.args = parser.parse_args(remaining_argv)

verbose

install_service.verbose

quiet

install_service.quiet

dry_run

install_service.dry_run