ED_LRR/ed_lrr_gui/__main__.py

import sys
import multiprocessing as MP
import queue
import ctypes
import os
from datetime import datetime
from math import floor
import click
from tqdm import tqdm
from click_default_group import DefaultGroup
import requests as RQ
from urllib.parse import urljoin
from ed_lrr_gui import Router, Preprocessor, cfg
from _ed_lrr import find_sys

CONTEXT_SETTINGS = dict(help_option_names=["-h", "--help"])

stars_path = os.path.join(cfg["folders.data_dir"], "stars.csv")
for folder in cfg["history.out_path"]:
    file = os.path.join(folder, "stars.csv")
    if os.path.isfile(file):
        stars_path = file
        break

systems_path = os.path.join(cfg["folders.data_dir"], "systemsWithCoordinates.json")
for file in cfg["history.systems_path"]:
    if os.path.isfile(file):
        systems_path = file
        break

bodies_path = os.path.join(cfg["folders.data_dir"], "bodies.json")
for file in cfg["history.bodies_path"][::-1]:
    if os.path.isfile(file):
        bodies_path = file
        break


@click.group(invoke_without_command=True, context_settings=CONTEXT_SETTINGS)
@click.pass_context
@click.version_option()
def main(ctx):
    "Elite: Dangerous long range router, command line interface"
    MP.freeze_support()
    if ctx.invoked_subcommand != "config":
        os.makedirs(cfg["folders.data_dir"], exist_ok=True)
    if ctx.invoked_subcommand is None:
        ctx.invoke(gui)
        return
    return


@main.command()
@click.argument("option", default=None, required=False)
@click.argument("value", default=None, required=False)
def config(option, value):
    """Change configuration
    

    If "key" and "value" are both omitted the current configuration is printed
    """

    def print_config(key):
        default = cfg.section(key).default()
        comment = cfg.section(key).comment
        value = cfg[key]
        is_default = value == default
        if (
            isinstance(value, list)
            and all(isinstance(element, str) for element in value)
            and len(value) != 0
        ):
            value = "[{}]".format(", ".join(map("'{}'".format, value)))
        key = click.style("{}".format(key), fg="cyan")
        value = click.style("{}".format(value), fg="green")
        default = click.style("{}".format(default), fg="blue")
        comment = click.style("{}".format(comment), fg="yellow")
        if is_default:
            print("{}: {} # {}".format(key, default, comment))
        else:
            print("{}: {} (default: {}) # {}".format(key, value, default, comment))

    if option is None and value is None:
        click.secho("Config path: {}".format(cfg.sources[0]), bold=True)
        print()
        for key in cfg:
            print_config(key)
        return
    if value is None:
        if option in cfg:
            print_config(option)
        else:
            print("Invalid option:", option)
        return
    cfg[option] = value
    cfg.sync()
    return


@main.command()
def explore():
    "Open file manager in data folder"
    click.launch(cfg["folders.data_dir"], locate=True)


@main.command()
@click.option("--debug", help="Debug print", is_flag=True)
def gui(debug):
    "Run the ED LRR GUI (default)"
    import ed_lrr_gui.gui as ED_LRR_GUI

    if (not debug) and os.name == "nt":
        ctypes.windll.kernel32.FreeConsole()
        sys.stdin = open("NUL", "rt")
        sys.stdout = open("NUL", "wt")
        sys.stderr = open("NUL", "wt")
    sys.exit(ED_LRR_GUI.main())


@main.command()
@click.option(
    "--url",
    "-u",
    help="Base URL",
    default="https://www.edsm.net/dump/",
    show_default=True,
)
@click.option(
    "--folder",
    "-f",
    help="Target folder for downloads",
    default=cfg["folders.data_dir"],
    type=click.Path(exists=True, dir_okay=True, file_okay=False),
    show_default=True,
)
def download(url, folder):
    "Download EDSM dumps"
    os.makedirs(folder, exist_ok=True)
    for file_name in ["systemsWithCoordinates.json", "bodies.json"]:
        download_url = urljoin(url, file_name)
        download_path = os.path.join(folder, file_name)
        if os.path.isfile(download_path):
            try:
                if not click.confirm(
                    "{} already exissts, overwrite?".format(file_name)
                ):
                    continue
            except click.Abort:
                exit("Canceled!")
        size = RQ.head(download_url, headers={"Accept-Encoding": "None"})
        size.raise_for_status()
        size = int(size.headers.get("Content-Length", 0))
        with tqdm(
            total=size,
            desc="{}".format(file_name),
            unit="b",
            unit_divisor=1024,
            unit_scale=True,
            ascii=True,
        ) as pbar:
            with open(download_path, "wb") as of:
                resp = RQ.get(
                    download_url, stream=True, headers={"Accept-Encoding": "gzip"}
                )
                for chunk in resp.iter_content(1024 * 1024):
                    of.write(chunk)
                    pbar.update(len(chunk))
    click.pause()


@main.command()
@click.option(
    "--systems",
    "-s",
    default=systems_path,
    metavar="<path>",
    help="Path to stars.csv",
    type=click.Path(exists=True, dir_okay=False),
    show_default=True,
)
@click.option(
    "--bodies",
    "-b",
    default=bodies_path,
    metavar="<path>",
    help="Path to bodies.json",
    type=click.Path(exists=True, dir_okay=False),
    show_default=True,
)
@click.option(
    "--output",
    "-o",
    default=stars_path,
    metavar="<path>",
    help="Path to stars.csv",
    type=click.Path(exists=False, dir_okay=False),
    show_default=True,
)
def preprocess(systems, bodies, output):
    "Preprocess EDSM dumps"
    with click.progressbar(
        length=100, label="", show_percent=True, item_show_func=lambda v: v, width=50
    ) as pbar:
        preproc = Preprocessor(systems, bodies, output)
        preproc.start()
        state = {}
        pstate = {}
        while not (preproc.queue.empty() and preproc.is_alive() == False):
            try:
                event = preproc.queue.get(True, 0.1)
                state.update(event)
                if state != pstate:
                    prc = (state["status"]["done"] / state["status"]["total"]) * 100
                    pbar.pos = prc
                    pbar.update(0)
                    pbar.current_item = state["status"]["message"]
                    pstate = state.copy()
            except queue.Empty:
                pass
        pbar.pos = 100
        pbar.update(0)
    print(state.get("result"))
    print("DONE!")
    click.pause()


@main.command()
@click.option(
    "--path",
    "-i",
    required=True,
    metavar="<path>",
    help="Path to stars.csv",
    default=stars_path,
    type=click.Path(exists=True, dir_okay=False),
    show_default=True,
)
@click.option(
    "--precomp_file",
    "-pf",
    metavar="<path>",
    help="Precomputed routing graph to use",
    type=click.Path(exists=True, dir_okay=False),
)
@click.option(
    "--range",
    "-r",
    default=cfg["route.range"],
    metavar="<float>",
    help="Jump range (Ly)",
    type=click.FloatRange(min=0),
    show_default=True,
)
@click.option(
    "--prune",
    "-d",
    default=(cfg["route.prune.steps"], cfg["route.prune.min_improvement"]),
    metavar="<n> <m>",
    help="Prune search branches",
    nargs=2,
    type=click.Tuple([click.IntRange(min=0), click.FloatRange(min=0)]),
    show_default=True,
)
@click.option(
    "--permute",
    "-p",
    type=click.Choice(["all", "keep_first", "keep_last", "keep_both"]),
    default=None,
    help="Permute hops to find shortest route",
    show_default=True,
)
@click.option(
    "--primary/--no-primary",
    "+ps/-ps",
    is_flag=True,
    default=cfg["route.primary"],
    help="Only route through primary stars",
    show_default=True,
)
@click.option(
    "--factor",
    "-g",
    metavar="<float>",
    default=cfg["route.greediness"],
    help="Greedyness factor for A-Star",
    show_default=True,
)
@click.option(
    "--mode",
    "-m",
    default=cfg["route.mode"],
    help="Search mode",
    type=click.Choice(["bfs", "a-star", "greedy"]),
    show_default=True,
)
@click.argument("systems", nargs=-1)
def route(**kwargs):
    "Compute a route"
    if len(kwargs["systems"]) < 2:
        exit("Need at least two systems to plot a route")
    if kwargs["prune"] == (0, 0):
        kwargs["prune"] = None

    def to_string(state):
        if state:
            return "{prc_done:.2f}% [N:{depth} Q:{queue_size} D:{d_rem:.2f} Ly] {system}".format(
                **state
            )

    keep_first, keep_last = {
        "all": (False, False),
        "keep_first": (True, False),
        "keep_last": (False, True),
        "keep_both": (True, True),
        None: (False, False),
    }[kwargs["permute"]]
    print("Resolving systems...")
    t = datetime.today()
    matches = find_sys(kwargs["systems"], kwargs["path"])
    kwargs["systems"] = [str(matches[key][1]["id"]) for key in kwargs["systems"]]
    print("Done in", datetime.today() - t)
    args = [
        kwargs["systems"],
        kwargs["range"],
        kwargs["prune"],
        kwargs["mode"],
        kwargs["primary"],
        kwargs["permute"] != None,
        keep_first,
        keep_last,
        kwargs["factor"],
        None,
        kwargs["path"],
    ]
    with click.progressbar(
        length=100,
        label="Computing route",
        show_percent=False,
        item_show_func=to_string,
        width=50,
    ) as pbar:
        router = Router(*args)
        t = datetime.today()
        router.start()
        state = {}
        pstate = {}
        while not (router.queue.empty() and router.is_alive() == False):
            try:
                event = router.queue.get(True, 0.1)
                state.update(event)
                if state != pstate:
                    pbar.current_item = state.get("status")
                    if pbar.current_item:
                        pbar.pos = pbar.current_item["prc_done"]
                        pbar.update(0)
                    pstate = state.copy()
            except queue.Empty:
                pass
        pbar.pos = 100
        pbar.update(0)
    for n, jump in enumerate(state.get("return", []), 1):
        jump["n"] = n
        if jump["body"].index(jump["system"]) == -1:
            jump["where"] = "[{body}] in [{system}]".format(**jump)
        else:
            jump["where"] = "[{body}]".format(**jump)
        if jump["distance"] > 0:
            print("({n}) {where}: {star_type} ({distance} Ls)".format(**jump))
        else:
            print("({n}) {where}: {star_type}".format(**jump))
    print("Done in", datetime.today() - t)


@main.command()
@click.option(
    "--path",
    "-i",
    required=True,
    help="Path to stars.csv",
    default=stars_path,
    type=click.Path(exists=True, dir_okay=False),
    show_default=True,
)
@click.option(
    "--range", "-r", required=True, help="Jump range (Ly)", type=click.FloatRange(min=0)
)
@click.option("--primary", "-ps", help="Only route through primary stars")
@click.option(
    "--output",
    "-o",
    required=True,
    help="Output path",
    default="./stars.idx",
    type=click.Path(exists=False, dir_okay=False),
    show_default=True,
)
@click.argument("systems", nargs=-1)
def precompute(*args, **kwargs):
    "Precompute routing graph"
    print("PreComp:", args, kwargs)


if __name__ == "__main__":
    main()