ED_LRR/ed_lrr_gui/__main__.py

403 lines
12 KiB
Python

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()