Skip to content
Snippets Groups Projects
config.py 19.6 KiB
Newer Older
  • Learn to ignore specific revisions
  • David Huss's avatar
    David Huss committed
    #!/usr/bin/env python 
    #-*- coding: utf-8 -*-
    
    
    import os, getpass, sys
    import ast
    
    import logging
    
    David Huss's avatar
    David Huss committed
    import toml
    from pathlib import Path
    
    from logging.config import dictConfig
    from typing import List
    import collections.abc
    
    
    # Affects config directories etc
    APPLICATION_NAME = "stechuhr-client"
    
    # Do not change here, just use an override instead
    DEFAULT_CONFIG = """
    
    [application]
    # Valid Log Levels are Debug, Info, Warning, Error, Critical
    loglevel = "Debug"
    dryrun = true
    
    
    [server]
    address = "127.0.0.1"
    port = 80
    
    David Huss's avatar
    David Huss committed
    https = true
    
    timeout = 5
    verify_cert = true
    
    
    # Frequency of requesting pattern updates from the server in seconds
    update_frequency = 600
    
    
    [client]
    location = "lerchenfeld/mensa"
    entrance = "haupteingang"
    
    # Valid directions are "in", "out", "both"
    
    direction = "in"
    
    
    # Duration for which to ignore repeated register attempt for a card (in seconds)
    ignore_duration = 10
    
    
    # A list of possible python regex patterns for the id (logical OR!)
    id_patterns = [
        "^806[A-Z0-9]{9}04$",
        "^FB6A1E60$",
        "^FB6D6950$",
        "^FB6A9DE0$",
        "^FB67D500$",
    ]
    
    
    [reader]
    # Connect to the reader using this
    # Find this using udevadm info --query=all /dev/input/eventNUMBER
    vendor_id = "413d"
    model_id = "2107"
    
    
    [led]
    # Colors are RGB (Red/Green/Blue)
    standby_color = "[(100, 100, 100)]"
    
    # LED color when the scan was unsucessful
    # Note: color, off_time and on_time can be lists to allow changing colors over
    #       time. If the list is shorter than the repetitions it is looped over again
    failure_color = "[(255, 0, 0)]"
    failure_repetitions = 3
    failure_off_time = "[0.2, 0.2, 0.3]"
    failure_on_time = "[0.2, 0.2, 0.3]"
    
    # LED blinking when the scan was successful
    # Note: color, off_time and on_time can be lists to allow changing colors over
    #       time. If the list is shorter than the repetitions it is looped over again
    success_color  = "[(0, 255, 0), (0, 255, 255), (0, 0, 255), (255, 0, 255), (255, 255, 0)]"
    success_repetitions = 5
    success_off_time = "[0.05]"
    success_on_time = "[0.05]"
    
    
    [buzzer]
    active = true
    
    # Midi note numbers and lengths in seconds - played on startup
    startup_notes = [80, 92, 80, 81]
    startup_note_lengths = [0.1, 0.05, 0.1, 0.6]
    startup_note_stop_lengths = [0.1, 0.05, 0.3, 0.1]
    
    # Midi note numbers and lengths in seconds - played on successful scan
    success_notes = [60, 72]
    success_note_lengths = [0.05, 0.05]
    success_note_stop_lengths = [0.05, 0.05]
    
    # Midi note numbers and lengths in seconds - played on successful scan
    failure_notes = [72, 58, 56, 53]
    failure_note_lengths = [0.05, 0.05, 0.05, 0.2]
    failure_note_stop_lengths = [0.05, 0.05, 0.05, 0.1]
    """
    
    # Config for the logger, there should be no need to make
    # manual changes here
    dictConfig({
        'version': 1,
        'formatters': {'default': {
            'format': '[%(asctime)s] %(levelname)s in %(module)s: %(message)s',
        }},
        'handlers': {'wsgi': {
            'class': 'logging.StreamHandler',
            'stream': 'ext://flask.logging.wsgi_errors_stream',
            'formatter': 'default'
        }},
        'root': {
            'level': 'INFO',
            'handlers': ['wsgi']
        }
    })
    
    David Huss's avatar
    David Huss committed
    
    
    def this_or_else(this: str, other: str) -> str:
        """
    
        Return this appended with the application name
        if this is a non-empty string otherwise return other
    
    David Huss's avatar
    David Huss committed
        """
        if this is None or this.strip() == "":
            return other
        else:
    
            return "{}/{}".format(this, APPLICATION_NAME)
    
    David Huss's avatar
    David Huss committed
    
    
    def get_home() -> str:
        """
        Get the home directory from the environment variable
        """
        return os.environ.get("HOME")
    
    
    
    def get_config_directories() -> List[dict]:
    
    David Huss's avatar
    David Huss committed
        """
    
        Returns a list of potential directories where the config could be stored.
        Existence of the directories is _not_ checked
    
    David Huss's avatar
    David Huss committed
        """
    
    
        # Generate a etc directory (usually /etc/stechuhr-client)
        etc_directory = "/etc/{}".format(APPLICATION_NAME)
    
    David Huss's avatar
    David Huss committed
    
    
        # Get the default config dir (usually /home/user/.config/stechuhr-client)
        default_config_home = "{}/.config/{}".format(get_home(), APPLICATION_NAME)
    
    David Huss's avatar
    David Huss committed
    
    
        # Unless the XDG_CONFIG_HOME environment variable has been set, then use this instead
        default_config_home = this_or_else(os.environ.get("XDG_CONFIG_HOME"), default_config_home)
    
    David Huss's avatar
    David Huss committed
    
    
        # Create the list of directories
        config_directories = [
            { "path": Path(etc_directory), "kind": "default", "source": None },
            { "path": Path(default_config_home), "kind": "default", "source": None }
        ]
    
    David Huss's avatar
    David Huss committed
    
    
        # If the STECHUHR_SERVER_CONFIG_PATH environment variable exists append to list
        key = "{}_CONFIG_DIR".format(APPLICATION_NAME.upper()).replace("-", "_").replace(" ", "_")
        if os.getenv(key) is not None:
            env_dir = { "path": Path(os.getenv(key)), "kind": "env", "source": key }
            config_directories.append(env_dir)
    
    David Huss's avatar
    David Huss committed
    
    
        return config_directories
    
    def get_potential_config_file_paths() -> List[dict]:
        """
        Returns a list of config file paths in reverse order of importance
        (last overrides first, non-existing paths may be contained)
        """
    
        config_paths = []
    
        for directory in get_config_directories():
            for path in sorted(directory["path"].glob('*.toml')):
                path = { "path": path, "kind": "default", "source": None }
                config_paths.append(path)
    
        # If the STECHUHR_SERVER_CONFIG_PATH environment variable exists append to list
        key = "{}_CONFIG_PATH".format(APPLICATION_NAME.upper()).replace("-", "_").replace(" ", "_")
        if os.getenv(key) is not None:
            env_path = { "path": Path(os.getenv(key)), "kind": "env", "source": key }
            config_paths.append(env_path)
    
        # Add information about a paths existence
        for p in config_paths:
            p["exists"] = p["path"].is_file()
    
        return config_paths
    
    def get_existing_config_file_paths() -> List[Path]:
        """
        Returns a list of existing config file paths in reverse order of importance
        (last overrides first)
        """
    
        return [p["path"] for p in get_potential_config_file_paths() if p["path"].is_file()]
    
    def has_no_existing_config() -> bool:
        """
        Returns true if there is no existing config
        """
        return len(get_existing_config_file_paths()) == 0
    
    
    def merge(this: dict, that: dict) -> dict:
        """
        Merge dict this in to dict that
        """
        for key, value in that.items():
            if isinstance(value, collections.abc.Mapping):
                this[key] = merge(this.get(key, {}), value)
            else:
                this[key] = value
        return this
    
    
    def initialize_config(logger=None) -> dict:
    
    David Huss's avatar
    David Huss committed
        """
        Initialize a configuration. If none exists, create a default one
        """
    
        config = toml.loads(DEFAULT_CONFIG)
    
        # Return if there is no other config
        if has_no_existing_config():
            if logger is not None:
                logger.warning("Using default configuration, create an override by running config create")
            else:
                print("Using default configuration, create an override by running config create")
            return config
    
        if logger is not None:
            logger.info("Reading Configs in this order:")
            logger.info("Config [1]: DEFAULT_CONFIG (hardcoded)")
    
    David Huss's avatar
    David Huss committed
        else:
    
            print("Reading Configs in this order:")
            print("Config [1]: DEFAULT_CONFIG (hardcoded)")
    
        # Read all existing configs in order and merge/override the default one
        for i, p in enumerate(get_existing_config_file_paths()):
            next_config = read_config(p)
            config = merge(config, next_config)
            if logger is not None:
                logger.info("Config [{}]: {} (overrides previous configs)".format(i+2, p))
            else:
                print("Config [{}]: {} (overrides previous configs)".format(i+2, p))
    
    David Huss's avatar
    David Huss committed
    
    
        if logger is not None:
            logger = set_loglevel(config, logger)
    
        return config
    
    
    
    def set_loglevel(config, logger):
        """
        Set the loglevel based on the config settings
        """
    
        if config["application"]["loglevel"].lower().strip() == "debug":
    
            logger.setLevel(logging.DEBUG)
    
        elif config["application"]["loglevel"].lower().strip() == "info":
    
            logger.setLevel(logging.INFO)
    
        elif config["application"]["loglevel"].lower().strip() == "warning":
    
            logger.setLevel(logging.WARNING)
    
        elif config["application"]["loglevel"].lower().strip() == "error":
    
            logger.setLevel(logging.ERROR)
    
        elif config["application"]["loglevel"].lower().strip() == "critical":
    
            logger.setLevel(logging.CRITICAL)
        else:
    
            logger.critical("The loglevel \"{}\" set in config.toml is invalid use one of the following: \"Debug\", \"Info\", \"Warning\", \"Error\" or \"Critical\"".format(config["application"]["loglevel"]))
    
            exit(1)
        return logger
    
    David Huss's avatar
    David Huss committed
    
    
    David Huss's avatar
    David Huss committed
    
    
    David Huss's avatar
    David Huss committed
    def process_led_list_value(config, key):
    
    David Huss's avatar
    David Huss committed
        """
        Convert the value at a given list to a list
        """
        value = ast.literal_eval(config["led"][key])
        if type(value) == list:
            config["led"][key] = value
        elif type(value) == tuple:
            config["led"][key] = list(value)
    
    
    
    David Huss's avatar
    David Huss committed
    def read_config(config_path: str) -> dict:
        """
        Read a config.toml from the given path,
        return a dict containing the config
        """
        with open(config_path, "r", encoding="utf-8") as f:
            config = toml.load(f)
    
    David Huss's avatar
    David Huss committed
    
        # Evaluate lists for LED colors
    
    David Huss's avatar
    David Huss committed
        process_led_list_value(config, "standby_color")
        process_led_list_value(config, "failure_color")
        process_led_list_value(config, "success_color")
        process_led_list_value(config, "failure_off_time")
        process_led_list_value(config, "failure_on_time")
        process_led_list_value(config, "success_off_time")
        process_led_list_value(config, "success_on_time")
    
        return config
    
    
    David Huss's avatar
    David Huss committed
    def main():
        """
        Gets run only if config.py is called directly or via `poetry run config`
        Entry point for the CLI application
        """
    
        # List of available commands and their respective functions
        commands = {
            "default" : print_default,
            "paths": print_paths,
            "directories": print_directories,
            "create": create_config,
            "test": test
        }
    
        # List of available options
        availaible_options = [
            ["-h", "--help"]
        ]
    
        # If no argument has been passed display the help and exit
        if len(sys.argv) == 1:
            print_help()
            exit()
    
        # Extract the command arguments
        command_args = [c for c in sys.argv[1:] if not c.strip().startswith("-")]
    
        # Extract the short_options
        short_options = [c.lstrip("-") for c in sys.argv[1:] if c.strip().startswith("-") and not c.strip().startswith("--")]
    
        # Flatten the short_options to e.g. so -1234 will result in ["1", "2", "3", "4"]
        short_options = [item for sublist in short_options for item in sublist]
    
        # Extract the long options
        long_options = [c.lstrip("--") for c in sys.argv[1:] if c.strip().startswith("--")]
    
        errored = False
        # Short Options
        for o in short_options:
            if o not in [a[0].lstrip("-") for a in availaible_options]:
                print("Error: the option \"-{}\" does not exist.".format(o), file=sys.stderr)
                errored = True
    
        # Long Options
        for o in long_options:
            if o not in [a[0].lstrip("--") for a in availaible_options]:
                print("Error: the option \"--{}\" does not exist.".format(o), file=sys.stderr)
                errored = True
    
        # If any of the above errored, exit. This allows to display all errors at once
        if errored:
            print("\nCheck the available commands and options below:")
            print()
            print_help()
            exit()
    
        # Currently we only handle a single command
        if len(command_args) == 1:
            command = sys.argv[1]
            # Short commands are allowed if they are not ambigous. E.g "te" will trigger "test"
            if not any([c.startswith(command.strip().lower()) for c in commands.keys()]):
                # No fitting command has been found, print helpt and exit
                print_help()
                exit()
            elif len([c for c in commands.keys() if c.startswith(command.strip().lower())]) > 1:
                # More than one fitting command has been found, display this message
                print("Ambiguous Input: There are {} commands starting with \"{}\": {}".format(
                    len([c for c in commands.keys() if c.startswith(command.strip().lower())]),
                    command,
                    ", ".join([c for c in commands.keys() if c.startswith(command.strip().lower())])
                ))
            else:
                # A command has been found:
                choice = [c for c in commands.keys() if c.startswith(command.strip().lower())][0]
    
                # If there is a -h or --help option, display the function's docstring
                # otherwise execute the function
                if "h" in short_options or "help" in long_options:
                    print("Help: config {}".format(commands[choice].__name__))
                    print(commands[choice].__doc__)
                else:
                    commands[choice]()
        else:
            # If more than one command is given, display the help
            print_help()
    
    
    
    
    def test():
        """
        Reads all configs like it would in production, prints out the order in which the config files are read and spits out the final resulting toml config
        """
        import pprint
        config = initialize_config()
        print("\nvvvvvvvvvvvvv Below is the resulting config vvvvvvvvvvvvv\n")
        print(toml.dumps(config))
    
    def print_default():
        """
        Prints the default config.toml
        """
        print(DEFAULT_CONFIG)
    
    
    def print_paths():
        """
        Prints the potential paths where a config could or should be. If environment variables are used to specify said path, this will be mentioned. If a file doesn't exist, it will be mentioned as well.
        """
        paths = get_potential_config_file_paths()
        if paths is not None:
            for p in paths:
                if p["kind"] == "env":
                    if p["exists"]:
                        print("{} (set by environment variable {})".format(p["path"], p["source"]))
                    else:
                        print("{} (set by environment variable {}, but doesn't exist)".format(p["path"], p["source"]))
                else:
                    if p["exists"]:
                        print("{}".format(p["path"]))
                    else:
                        print("{} (doesn't exist yet)".format(p["path"]))
        else:
            print("There are no paths..")
    
    
    def print_directories():
        """
        Prints a list of directories where configs are searched for.
        Lower directories override higher directories.
        """
        directories = get_config_directories()
        if directories is not None:
            for d in directories:
                if d["kind"] == "env":
                    print("{} (set by environment variable {})".format(d["path"], d["source"]))
                else:
                    print("{}".format(d["path"]))
        else:
            print("There are no directories..")
    
    
    def create_config():
        """
        Interactivally create a config directory at a choice of different places with a default config in it.
        """
        helptext = """Configs are read from the following directories (later overrides earlier):
    1. DEFAULT_CONFIG (use config default to inspect)
    2. /etc/stechuhr-client/*.toml (in alphabetical order)
    3. $XDG_CONFIG_HOME/stechuhr-client/*.toml (in alphabetical order)
    4. $STECHUHR_SERVER_CONFIG_DIR/*.toml (in alphabetical order)
    5. $STECHUHR_SERVER_CONFIG_PATH (final override)
    """
        print(helptext)
        print()
        print("Select one of the following options to create a new config:")
        config_directories = get_config_directories()
        for i, p in enumerate(config_directories):
            # Create a source string describing the origin of the directory
            if p["kind"] == "env":
                source = "set via environment variable {}, ".format(p["source"])
            else:
                source = ""
            # Display some options
            if p["path"].is_dir():
                config_path = Path("{}/{}".format(p, "00-config.toml"))
                if not config_path.is_file():
                    print("  [{}] {} ({}create 00-config.toml there)".format(i, p["path"], source))
                else:
                    print("  [{}] {} ({}override existing 00-config.toml!)".format(i, p["path"], source))
            elif p["path"].is_file():
                pass
            else:
                print("  [{}] {} ({}dir doesn't exist: create, then write 00-config.toml there)".format(i, p["path"], source))
        print("  [x] Do nothing")
        print()
    
        # Collect the selection input
        selection = None
        while selection is None or not selection in [str(i) for i, p in enumerate(config_directories)]:
            selection = input("Select one of the above: ")
            if selection.lower() in ["x"]:
                break
    
        # If nothing has been selected, exit
        if selection.lower() == "x":
            exit()
    
        # Store the selected directory here
        selection = config_directories[int(selection)]
    
        # Create the directory if it doesn't exist yet
        try:
            selection["path"].mkdir(mode=0o755, parents=True, exist_ok=True)
        except PermissionError:
            print()
            print("Error: Didn't have the permissions to create the config directory at {}".format(selection["path"]), file=sys.stderr)
            print("Hint:  Change the owner of the directory temporarily to {} or run {} config create with more permissions".format(getpass.getuser(), APPLICATION_NAME))
            exit()
    
        config_path = Path("{}/{}".format(selection["path"], "00-config.toml"))
    
    
        # If the 00-config.toml already exists, ask whether it shall be moved to 00-config.toml.old
        if config_path.is_file():
            # Create an alternate config path, incrementing up if .old already exists
            alt_config_path = None
            i = 1
            while alt_config_path is None or alt_config_path.exists():
                if i == 1:
                    alt_config_path = Path("{}.old".format(config_path))
                else:
                    alt_config_path = Path("{}.old{}".format(config_path, i))
                i += 1
    
            # Ask for confirmation
            confirmation = None
            while confirmation is None or not confirmation.lower().strip() in ["y", "n"]:
                confirmation = input("Move existing file at \"{}\" to \"{}\"? [Y/n]:\t".format(config_path, alt_config_path))
            if confirmation.lower().strip() != "y":
                exit()
            else:
                config_path.rename(alt_config_path)
                print("Moved existing file \"{}\" to \"{}\"".format(config_path, alt_config_path))
    
        # Create the config.toml or display a hint if the permissions don't suffice
        try:
            config_path.write_text(DEFAULT_CONFIG)
        except PermissionError as e:
            print()
            print("Error: Didn't have the permissions to write the file to {}".format(config_path), file=sys.stderr)
            print("       Directory \"{}\" belongs to user {} (was running as user {})".format(selection["path"], selection["path"].owner(), getpass.getuser()), file=sys.stderr)
            print()
            print("Hint:  Change the owner of the directory temporarily to {} or run {} config create with more permissions".format(getpass.getuser(), APPLICATION_NAME))
            exit()
        print("Default Config has been written to {}".format(config_path))
    
    
    def print_help():
        """
        Print the help
        """
        helptext = """========= {} CONFIG =========
    
    Helper tool for managing and installing a stechuhr-client config.
    
    Configs are read from the following directories (later overrides earlier):
    1. DEFAULT_CONFIG (see below)
    2. /etc/stechuhr-client/*.toml (in alphabetical order)
    3. $XDG_CONFIG_HOME/stechuhr-client/*.toml (in alphabetical order)
    4. $STECHUHR_SERVER_CONFIG_DIR/*.toml (in alphabetical order)
    5. $STECHUHR_SERVER_CONFIG_PATH (final override)
    
    Commands:
        create  . . . . . Interactivly create a default config file
        default . . . . . Prints default config.toml to stdout
        directories . . . Prints which config directories are read
        paths . . . . . . Prints which config files are read
        test  . . . . . . Read in the configs and print the resulting combined toml
    
    Options:
        -h, --help  . . . Display the help message of a command
        """.format(APPLICATION_NAME.upper())
    
        print(helptext)
    
    
    if __name__ == "__main__":
        main()