linux-tools/modules/swap/__init__.py

import subprocess
from typing import List

from PyInquirer import prompt

from print_helpers import print_bold, print_warn, print_gr


def run_shell_command(cmd: List[str]) -> str:
    return subprocess.check_output(cmd).decode("UTF-8").strip()


def create_resize_persistent_swap():
    create_resize_swap()
    make_swap_persistent()


def get_swap_location():
    output_swaps = run_shell_command(["cat", "/proc/swaps"])
    try:
        swap_location = output_swaps.split()[5]
        print_bold(f"Swap is located here: {swap_location}")
        return swap_location
    except IndexError:
        print_warn("Swap file doesn´t exist!")
        return None


def get_swap_size():
    swap_size = run_shell_command(['swapon', '--show'])
    try:
        swap_size = swap_size.split()[7]
        print_gr(f"Swap is {swap_size}b big!")
    except IndexError:
        print_warn("Swap file doesn´t exist!")


def create_resize_swap():
    output_swapon = run_shell_command(['swapon', '--show'])
    if output_swapon is not "":
        swap_size = output_swapon.split()[7]
        print("")
        print_bold("Swap already installed! You can resize it!")
        print_bold(f"Curr. swap size: {swap_size}b")
        resize = input("How big should your swap become (numbers in Gigabyte)? ")
        resize_swapfile = "swapoff /swapfile && " + \
                          f"fallocate -l {resize}G /swapfile && " + \
                          "mkswap /swapfile && " + \
                          "swapon /swapfile"
        subprocess.call(resize_swapfile, shell=True)
        output_free = run_shell_command(["free", "-h"])
        print_gr(output_free)

    else:
        size = input("How big should the swap be (numbers in Gigabyte)? ")
        create_swapfile = f"fallocate -l {size}G /swapfile && " + \
                          "chmod 600 /swapfile && " + \
                          "mkswap /swapfile && " + \
                          "swapon /swapfile"
        subprocess.call(create_swapfile, shell=True)

        output_free = run_shell_command(["free", "-h"])
        print_gr(output_free)


def make_swap_persistent():
    swap_location = get_swap_location()
    if swap_location is None:
        print_warn("Swap file doesn't exist!")
        return
    backup_fstab = "cp /etc/fstab /etc/fstab.bak"
    enable_persistence = f"echo '{swap_location} none swap sw 0 0' | tee -a /etc/fstab"
    if fstab_entry_exists():
        print_warn("Swap is already persistent!")
    else:
        subprocess.call(backup_fstab, shell=True)
        subprocess.call(enable_persistence, shell=True)
        print_gr("Swap is now persistent!")


def show_swapiness():
    print_gr(run_shell_command(["cat", "/proc/sys/vm/swappiness"]))


def adjust_swapiness_temp():
    options = {
        "Light": 25,
        "Default": 60,
        "Aggressive": 100
    }
    menu = [
        {
            'type': 'list',
            'message': 'Select action',
            'name': 'action',
            'choices': list(map(lambda x: {"name": x}, options.keys()))
        }
    ]
    selected_swapiness = prompt(menu)['action']

    adjust = "sysctl vm.swappiness=" + str(options[selected_swapiness])
    subprocess.call(adjust, shell=True)
    print_gr("Temporary swapiness is " + str(options[selected_swapiness]))


def delete_swap():
    swap_location = get_swap_location()
    if swap_location is None:
        return
    disable_swapfile = f"swapoff {swap_location} && " + \
                       f"rm {swap_location}"
    if fstab_entry_exists():
        with open("/etc/fstab", "r") as fstab_out:
            content = fstab_out.readlines()
        with open("/etc/fstab", "w") as fstab_in:
            content = content[:-1]
            for line in content:
                fstab_in.write(line)
    else:
        print_warn("No entry in /etc/fstab!")
    subprocess.call(disable_swapfile, shell=True)
    output_swapon = run_shell_command(['swapon', '--show'])
    output_free = run_shell_command(["free", "-h"])
    if not output_swapon:
        print_gr("Swap deleted!")
        print_gr(output_free)


def fstab_entry_exists():
    swap_location = get_swap_location()
    fstab_entry = f"{swap_location} none swap sw 0 0\n"
    with open("/etc/fstab", "r") as fstab_file:
        line = fstab_file.readlines()[-1]
        if line != fstab_entry:
            print_warn("No entry in /etc/fstab")
            return False
        else:
            print_gr(f"fstab entry: {line.strip()}")
            return True


def run():
    actions = {
        "Create/Resize temp. swap": create_resize_swap,
        "Create/Resize persistent swap": create_resize_persistent_swap,
        "Delete swap": delete_swap,
        "Make temp. swap persistent": make_swap_persistent,
        "Get swap location": get_swap_location,
        "Get swap size": get_swap_size,
        "Show swapiness": show_swapiness,
        "Adjust temp. swapiness": adjust_swapiness_temp,
        "Check fstab for entry": fstab_entry_exists,
    }
    menu = [
        {
            'type': 'list',
            'message': 'Select action',
            'name': 'action',
            'choices': list(map(lambda x: {"name": x}, actions.keys()))
        }
    ]
    selected_action = prompt(menu)['action']
    actions[selected_action]()


functions = run