dave/pyfa
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 21 Wiki Activity

Rework the "cli" to a server instead

Browse Source
This commit is contained in:
PhatPhuckDave
2026-02-28 19:33:38 +01:00
parent bc23f380db
commit 53390e85ac
6 changed files with 92 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
build.sh
View File

@@ -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)"

10
pyfa.py
View File

@@ -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:

27
pyfa.spec
View File

@@ -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',

5
pyfa_headless.py Normal file
View File

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

127
scripts/pyfa_cli_stats.py
View File

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

19
tests/test_pyfa_cli_stats.py
View File

@@ -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