From 1c7886463dd445e43bcf2585641cb313d6ddb794 Mon Sep 17 00:00:00 2001 From: PhatPhuckDave Date: Sat, 28 Feb 2026 19:33:38 +0100 Subject: [PATCH] Rework the "cli" to a server instead --- build.sh | 2 +- pyfa.py | 10 --- pyfa.spec | 27 +++++--- pyfa_headless.py | 5 ++ scripts/pyfa_cli_stats.py | 127 ++++++++++++++++++----------------- tests/test_pyfa_cli_stats.py | 19 +----- 6 files changed, 92 insertions(+), 98 deletions(-) create mode 100644 pyfa_headless.py diff --git a/build.sh b/build.sh index a8c58be99..2e196b79b 100644 --- a/build.sh +++ b/build.sh @@ -29,4 +29,4 @@ if [ -f dist/pyfa_headless/pyfa-headless.exe ]; then fi echo "" -echo "Build complete! dist/pyfa/pyfa.exe (GUI), dist/pyfa/pyfa-headless.exe (CLI, use with --headless)" +echo "Build complete! dist/pyfa/pyfa.exe (GUI), dist/pyfa/pyfa-headless.exe (HTTP server POST /simulate :9123)" diff --git a/pyfa.py b/pyfa.py index a70b76c64..31028e6aa 100755 --- a/pyfa.py +++ b/pyfa.py @@ -74,19 +74,9 @@ parser.add_option("-s", "--savepath", action="store", dest="savepath", help="Set parser.add_option("-l", "--logginglevel", action="store", dest="logginglevel", help="Set desired logging level [Critical|Error|Warning|Info|Debug]", default="Error") parser.add_option("-p", "--profile", action="store", dest="profile_path", help="Set location to save profileing.", default=None) parser.add_option("-i", "--language", action="store", dest="language", help="Sets the language for pyfa. Overrides user's saved settings. Format: xx_YY (eg: en_US). If translation doesn't exist, defaults to en_US", default=None) -parser.add_option("--headless", action="store_true", dest="headless", help="Run CLI stats only: read JSON fit from first positional arg (or stdin), print stats JSON to stdout, then exit. No GUI.", default=False) (options, args) = parser.parse_args() -if getattr(options, "headless", False): - from scripts.pyfa_cli_stats import main as headless_main - headless_argv = [] - if options.savepath: - headless_argv.extend(["--savepath", options.savepath]) - if args: - headless_argv.append(args[0]) - sys.exit(headless_main(headless_argv)) - if __name__ == "__main__": try: diff --git a/pyfa.spec b/pyfa.spec index a198f62f8..93350e4f9 100644 --- a/pyfa.spec +++ b/pyfa.spec @@ -79,8 +79,20 @@ a = Analysis(['pyfa.py'], win_private_assemblies=False, cipher=block_cipher) -pyz = PYZ(a.pure, a.zipped_data, cipher=block_cipher) +a_headless = Analysis(['pyfa_headless.py'], + pathex=pathex, + binaries=[], + datas=added_files, + hiddenimports=import_these, + hookspath=['dist_assets/pyinstaller_hooks'], + runtime_hooks=[], + excludes=['Tkinter'], + win_no_prefer_redirects=False, + win_private_assemblies=False, + cipher=block_cipher) +pyz = PYZ(a.pure, a.zipped_data, cipher=block_cipher) +pyz_headless = PYZ(a_headless.pure, a_headless.zipped_data, cipher=block_cipher) exe = EXE( pyz, @@ -96,10 +108,10 @@ exe = EXE( contents_directory='app', ) -# Headless CLI: console=True so stdout/stdin exist; use: pyfa-headless.exe --headless +# Headless: server only. POST /simulate on port 9123. exe_headless = EXE( - pyz, - a.scripts, + pyz_headless, + a_headless.scripts, exclude_binaries=True, name='pyfa-headless', debug=debug, @@ -119,12 +131,11 @@ coll = COLLECT( name='pyfa', ) -# Headless exe; build puts it in dist/pyfa_headless/; copy pyfa-headless.exe into dist/pyfa/ after build coll_headless = COLLECT( exe_headless, - a.binaries, - a.zipfiles, - a.datas, + a_headless.binaries, + a_headless.zipfiles, + a_headless.datas, strip=False, upx=upx, name='pyfa_headless', diff --git a/pyfa_headless.py b/pyfa_headless.py new file mode 100644 index 000000000..04cc7fbfe --- /dev/null +++ b/pyfa_headless.py @@ -0,0 +1,5 @@ +#!/usr/bin/env python3 +# Headless sim daemon only. POST /simulate with JSON body. +from scripts.pyfa_cli_stats import _run_http_server + +_run_http_server(9123, None) diff --git a/scripts/pyfa_cli_stats.py b/scripts/pyfa_cli_stats.py index f37b12b45..3770f320e 100644 --- a/scripts/pyfa_cli_stats.py +++ b/scripts/pyfa_cli_stats.py @@ -1,6 +1,6 @@ -import argparse import json import sys +from http.server import BaseHTTPRequestHandler, HTTPServer import config import eos.config @@ -50,7 +50,7 @@ def _import_single_fit(raw_text: str) -> tuple[object, set[str]]: import_type, import_data = Port.importFitFromBuffer(cleaned_text) if not import_data or len(import_data) != 1: - raise SystemExit("Expected exactly one fit in input; got %d" % (len(import_data) if import_data else 0)) + raise ValueError("Expected exactly one fit in input; got %d" % (len(import_data) if import_data else 0)) fit = import_data[0] @@ -69,19 +69,19 @@ def _add_projected_fit(s_fit, target_fit, projected_fit, amount: int) -> None: fit = s_fit.getFit(target_fit.ID) projected = s_fit.getFit(projected_fit.ID, projected=True) if projected is None: - raise SystemExit("Projected fit %s is not available" % projected_fit.ID) + raise ValueError("Projected fit %s is not available" % projected_fit.ID) if projected in fit.projectedFits and projected.ID in fit.projectedFitDict: projection_info = projected.getProjectionInfo(fit.ID) if projection_info is None: - raise SystemExit("Projection info missing for projected fit %s" % projected_fit.ID) + raise ValueError("Projection info missing for projected fit %s" % projected_fit.ID) else: fit.projectedFitDict[projected.ID] = projected eos.db.saveddata_session.flush() eos.db.saveddata_session.refresh(projected) projection_info = projected.getProjectionInfo(fit.ID) if projection_info is None: - raise SystemExit("Projection info missing after linking projected fit %s" % projected_fit.ID) + raise ValueError("Projection info missing after linking projected fit %s" % projected_fit.ID) projection_info.amount = amount projection_info.active = True @@ -91,7 +91,7 @@ def _add_command_fit(s_fit, target_fit, command_fit) -> None: fit = s_fit.getFit(target_fit.ID) command = s_fit.getFit(command_fit.ID) if command is None: - raise SystemExit("Command fit %s is not available" % command_fit.ID) + raise ValueError("Command fit %s is not available" % command_fit.ID) if command in fit.commandFits or command.ID in fit.commandFitDict: return @@ -101,7 +101,7 @@ def _add_command_fit(s_fit, target_fit, command_fit) -> None: eos.db.saveddata_session.refresh(command) info = command.getCommandInfo(fit.ID) if info is None: - raise SystemExit("Command info missing for command fit %s" % command_fit.ID) + raise ValueError("Command info missing for command fit %s" % command_fit.ID) info.active = True @@ -320,92 +320,95 @@ def _build_output(main_fit) -> dict: } -def _parse_args(argv: list[str]) -> argparse.Namespace: - parser = argparse.ArgumentParser(description="Headless pyfa CLI fit statistics exporter") - parser.add_argument( - "data", - nargs="?", - help="JSON payload (inline). If omitted, read from stdin.", - ) - parser.add_argument( - "--savepath", - help="Override pyfa save path (saveddata, DB, logs). Defaults to standard user location.", - ) - return parser.parse_args(argv) - - -def _load_payload(data: str) -> dict: - try: - payload = json.loads(data) - except json.JSONDecodeError as exc: - raise SystemExit("Invalid JSON: %s" % exc) from exc - if not isinstance(payload, dict): - raise SystemExit("Top-level JSON must be an object") - return payload - - -def main(argv: list[str] | None = None) -> int: - if argv is None: - argv = sys.argv[1:] - - args = _parse_args(argv) - _init_pyfa(args.savepath) - +def compute_stats(payload: dict, savepath: str | None = None) -> dict: + if "fit" not in payload or not isinstance(payload["fit"], str): + raise ValueError("Payload must contain a 'fit' field with EFT/text export") + _init_pyfa(savepath) from service.fit import Fit as FitService - s_fit = FitService.getInstance() if s_fit.character is None: from eos.saveddata.character import Character as saveddata_Character s_fit.character = saveddata_Character.getAll5() - - raw = args.data if args.data is not None else sys.stdin.read() - payload = _load_payload(raw) - - if "fit" not in payload or not isinstance(payload["fit"], str): - raise SystemExit("Payload must contain a 'fit' field with EFT/text export") - main_fit, _ = _import_single_fit(payload["fit"]) - projected_defs = payload.get("projected_fits", []) command_defs = payload.get("command_fits", []) - projected_fits: list[tuple[object, int]] = [] for entry in projected_defs: if not isinstance(entry, dict): - raise SystemExit("Each projected_fits entry must be an object") + raise ValueError("Each projected_fits entry must be an object") fit_text = entry.get("fit") count = entry.get("count") if not isinstance(fit_text, str): - raise SystemExit("Each projected_fits entry must contain a string 'fit'") + raise ValueError("Each projected_fits entry must contain a string 'fit'") if not isinstance(count, int) or count <= 0: - raise SystemExit("Each projected_fits entry must contain a positive integer 'count'") + raise ValueError("Each projected_fits entry must contain a positive integer 'count'") pf, _ = _import_single_fit(fit_text) projected_fits.append((pf, count)) - command_fits: list[object] = [] for entry in command_defs: if not isinstance(entry, dict): - raise SystemExit("Each command_fits entry must be an object") + raise ValueError("Each command_fits entry must be an object") fit_text = entry.get("fit") if not isinstance(fit_text, str): - raise SystemExit("Each command_fits entry must contain a string 'fit'") + raise ValueError("Each command_fits entry must contain a string 'fit'") cf, _ = _import_single_fit(fit_text) command_fits.append(cf) - for pf, count in projected_fits: _add_projected_fit(s_fit, main_fit, pf, count) for cf in command_fits: _add_command_fit(s_fit, main_fit, cf) - s_fit.recalc(main_fit) s_fit.fill(main_fit) + return _build_output(main_fit) - output = _build_output(main_fit) - json.dump(output, sys.stdout, indent=2, sort_keys=True) - sys.stdout.write("\n") - return 0 + +def _run_http_server(port: int, savepath: str | None) -> None: + class SimulateHandler(BaseHTTPRequestHandler): + def do_POST(self): + if self.path != "/simulate": + self.send_response(404) + self.end_headers() + return + content_length = int(self.headers.get("Content-Length", 0)) + body = self.rfile.read(content_length).decode("utf-8") + try: + payload = json.loads(body) + except json.JSONDecodeError as exc: + self.send_response(400) + self.send_header("Content-Type", "text/plain; charset=utf-8") + self.end_headers() + self.wfile.write(("Invalid JSON: %s" % exc).encode("utf-8")) + return + if not isinstance(payload, dict): + self.send_response(400) + self.send_header("Content-Type", "text/plain; charset=utf-8") + self.end_headers() + self.wfile.write(b"Top-level JSON must be an object") + return + try: + output = compute_stats(payload, savepath) + except ValueError as e: + self.send_response(400) + self.send_header("Content-Type", "text/plain; charset=utf-8") + self.end_headers() + self.wfile.write(str(e).encode("utf-8")) + return + except Exception as e: + self.send_response(500) + self.send_header("Content-Type", "text/plain; charset=utf-8") + self.end_headers() + self.wfile.write(str(e).encode("utf-8")) + return + self.send_response(200) + self.send_header("Content-Type", "application/json; charset=utf-8") + self.end_headers() + self.wfile.write(json.dumps(output, indent=2, sort_keys=True).encode("utf-8")) + self.wfile.write(b"\n") + + with HTTPServer(("", port), SimulateHandler) as httpd: + print("POST /simulate on http://127.0.0.1:%s" % port, file=sys.stderr, flush=True) + httpd.serve_forever() if __name__ == "__main__": - raise SystemExit(main()) - + _run_http_server(9123, None) diff --git a/tests/test_pyfa_cli_stats.py b/tests/test_pyfa_cli_stats.py index 39ce52847..03df61e30 100644 --- a/tests/test_pyfa_cli_stats.py +++ b/tests/test_pyfa_cli_stats.py @@ -2,8 +2,6 @@ import json import os import sys import tempfile -from io import StringIO -from contextlib import redirect_stdout script_dir = os.path.dirname(os.path.abspath(__file__)) @@ -41,22 +39,9 @@ Berserker II x5 def test_ishtar_spider_remote_armor_reps(): - payload = { - "fit": ISHTAR_SPIDER_FIT, - } + payload = {"fit": ISHTAR_SPIDER_FIT} with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmp: - argv = ["--savepath", tmp, json.dumps(payload)] - - buf = StringIO() - with redirect_stdout(buf): - rc = pyfa_cli_stats.main(argv) - - assert rc == 0 - - out = buf.getvalue() - data = json.loads(out) - + data = pyfa_cli_stats.compute_stats(payload, tmp) armor_rps = data["remote_reps_outgoing"]["current"]["armor"] - assert armor_rps is not None assert int(round(armor_rps)) == 171