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 1c7886463d
6 changed files with 92 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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)