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

Rename cli to server and add a docker distro

Browse Source
This commit is contained in:
PhatPhuckDave
2026-02-28 19:54:16 +01:00
parent 1c7886463d
commit 564ba1d85d
10 changed files with 88 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

414
scripts/pyfa_sim.py Normal file
View File

@@ -0,0 +1,414 @@
import json
import sys
from http.server import BaseHTTPRequestHandler, HTTPServer
import config
import eos.config
from eos.const import FittingHardpoint
from eos.utils.spoolSupport import SpoolOptions, SpoolType
def _init_pyfa(savepath: str | None) -> None:
config.debug = False
config.loggingLevel = config.LOGLEVEL_MAP["error"]
config.defPaths(savepath)
config.defLogging()
import eos.db # noqa: F401
eos.db.saveddata_meta.create_all() # type: ignore[name-defined]
import eos.events # noqa: F401
from service import prefetch # noqa: F401
def _parse_drone_markers(text: str) -> tuple[str, set[str]]:
active_names: set[str] = set()
cleaned_lines: list[str] = []
for raw_line in text.splitlines():
line = raw_line.rstrip()
if " xx" in line:
idx = line.find(" xx")
marker_segment = line[idx + 1 :].strip()
if marker_segment == "xx":
payload = line[:idx].rstrip()
parts = payload.split(" x", 1)
if len(parts) == 2:
type_name = parts[0].strip()
if type_name:
active_names.add(type_name)
line = payload
cleaned_lines.append(line)
return "\n".join(cleaned_lines), active_names
def _import_single_fit(raw_text: str) -> tuple[object, set[str]]:
from service.port.port import Port
cleaned_text, active_names = _parse_drone_markers(raw_text)
import_type, import_data = Port.importFitFromBuffer(cleaned_text)
if not import_data or len(import_data) != 1:
raise ValueError("Expected exactly one fit in input; got %d" % (len(import_data) if import_data else 0))
fit = import_data[0]
if active_names:
for drone in fit.drones:
if getattr(drone.item, "typeName", None) in active_names:
drone.amountActive = drone.amount
for fighter in fit.fighters:
if getattr(fighter.item, "typeName", None) in active_names:
fighter.amount = fighter.amount
return fit, active_names
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 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 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 ValueError("Projection info missing after linking projected fit %s" % projected_fit.ID)
projection_info.amount = amount
projection_info.active = True
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 ValueError("Command fit %s is not available" % command_fit.ID)
if command in fit.commandFits or command.ID in fit.commandFitDict:
return
fit.commandFitDict[command.ID] = command
eos.db.saveddata_session.flush()
eos.db.saveddata_session.refresh(command)
info = command.getCommandInfo(fit.ID)
if info is None:
raise ValueError("Command info missing for command fit %s" % command_fit.ID)
info.active = True
def _collect_resources(fit) -> dict:
ship = fit.ship
resources: dict = {}
resources["hardpoints"] = {
"turret": {
"used": fit.getHardpointsUsed(FittingHardpoint.TURRET),
"total": ship.getModifiedItemAttr("turretSlotsLeft"),
},
"launcher": {
"used": fit.getHardpointsUsed(FittingHardpoint.MISSILE),
"total": ship.getModifiedItemAttr("launcherSlotsLeft"),
},
}
resources["drones"] = {
"active": fit.activeDrones,
"max_active": fit.extraAttributes.get("maxActiveDrones"),
"bay_used": fit.droneBayUsed,
"bay_capacity": ship.getModifiedItemAttr("droneCapacity"),
"bandwidth_used": fit.droneBandwidthUsed,
"bandwidth_capacity": ship.getModifiedItemAttr("droneBandwidth"),
}
resources["fighters"] = {
"tubes_used": fit.fighterTubesUsed,
"tubes_total": fit.fighterTubesTotal,
"bay_used": fit.fighterBayUsed,
"bay_capacity": ship.getModifiedItemAttr("fighterCapacity"),
}
resources["calibration"] = {
"used": fit.calibrationUsed,
"total": ship.getModifiedItemAttr("upgradeCapacity"),
}
resources["powergrid"] = {
"used": fit.pgUsed,
"output": ship.getModifiedItemAttr("powerOutput"),
}
resources["cpu"] = {
"used": fit.cpuUsed,
"output": ship.getModifiedItemAttr("cpuOutput"),
}
resources["cargo"] = {
"used": fit.cargoBayUsed,
"capacity": ship.getModifiedItemAttr("capacity"),
}
return resources
def _collect_defense(fit) -> dict:
ship = fit.ship
def _res(attr_name):
return ship.getModifiedItemAttr(attr_name)
resonance = {
"armor": {"em": _res("armorEmDamageResonance"), "exp": _res("armorExplosiveDamageResonance"), "kin": _res("armorKineticDamageResonance"), "therm": _res("armorThermalDamageResonance")},
"shield": {"em": _res("shieldEmDamageResonance"), "exp": _res("shieldExplosiveDamageResonance"), "kin": _res("shieldKineticDamageResonance"), "therm": _res("shieldThermalDamageResonance")},
"hull": {"em": _res("emDamageResonance"), "exp": _res("explosiveDamageResonance"), "kin": _res("kineticDamageResonance"), "therm": _res("thermalDamageResonance")},
}
defense: dict = {}
defense["hp"] = fit.hp
defense["ehp"] = fit.ehp
defense["resonance"] = resonance
defense["tank"] = fit.tank
defense["effective_tank"] = fit.effectiveTank
defense["sustainable_tank"] = fit.sustainableTank
defense["effective_sustainable_tank"] = fit.effectiveSustainableTank
return defense
def _collect_capacitor(fit) -> dict:
ship = fit.ship
cap: dict = {}
cap["capacity"] = ship.getModifiedItemAttr("capacitorCapacity")
cap["recharge"] = fit.capRecharge
cap["use"] = fit.capUsed
cap["delta"] = fit.capDelta
cap["stable"] = fit.capStable
cap["state"] = fit.capState
cap["neutralizer_resistance"] = ship.getModifiedItemAttr("energyWarfareResistance", 1)
return cap
def _collect_firepower(fit) -> dict:
default_spool = eos.config.settings["globalDefaultSpoolupPercentage"]
spool = SpoolOptions(SpoolType.SPOOL_SCALE, default_spool, False)
wdps = fit.getWeaponDps(spoolOptions=spool)
ddps = fit.getDroneDps()
wvol = fit.getWeaponVolley(spoolOptions=spool)
dvol = fit.getDroneVolley()
return {
"weapon_dps": wdps.total,
"drone_dps": ddps.total,
"total_dps": fit.getTotalDps(spoolOptions=spool).total,
"weapon_volley": wvol.total,
"drone_volley": dvol.total,
"total_volley": fit.getTotalVolley(spoolOptions=spool).total,
"weapons": [],
}
def _collect_remote_reps(fit) -> dict:
default_spool = eos.config.settings["globalDefaultSpoolupPercentage"]
pre = fit.getRemoteReps(spoolOptions=SpoolOptions(SpoolType.SPOOL_SCALE, 0, True))
full = fit.getRemoteReps(spoolOptions=SpoolOptions(SpoolType.SPOOL_SCALE, 1, True))
current = fit.getRemoteReps(spoolOptions=SpoolOptions(SpoolType.SPOOL_SCALE, default_spool, False))
return {
"current": {
"capacitor": current.capacitor,
"shield": current.shield,
"armor": current.armor,
"hull": current.hull,
},
"pre_spool": {
"capacitor": pre.capacitor,
"shield": pre.shield,
"armor": pre.armor,
"hull": pre.hull,
},
"full_spool": {
"capacitor": full.capacitor,
"shield": full.shield,
"armor": full.armor,
"hull": full.hull,
},
}
def _collect_targeting_misc(fit) -> dict:
ship = fit.ship
misc: dict = {}
misc["targets_max"] = fit.maxTargets
misc["target_range"] = fit.maxTargetRange
misc["scan_resolution"] = ship.getModifiedItemAttr("scanResolution")
misc["scan_strength"] = fit.scanStrength
misc["scan_type"] = fit.scanType
misc["jam_chance"] = fit.jamChance
misc["drone_control_range"] = fit.extraAttributes.get("droneControlRange")
misc["speed"] = fit.maxSpeed
misc["align_time"] = fit.alignTime
misc["signature_radius"] = ship.getModifiedItemAttr("signatureRadius")
misc["warp_speed"] = fit.warpSpeed
misc["max_warp_distance"] = fit.maxWarpDistance
misc["probe_size"] = fit.probeSize
misc["cargo_capacity"] = ship.getModifiedItemAttr("capacity")
misc["cargo_used"] = fit.cargoBayUsed
return misc
def _collect_price(fit) -> dict:
ship_price = 0.0
module_price = 0.0
drone_price = 0.0
fighter_price = 0.0
cargo_price = 0.0
booster_price = 0.0
implant_price = 0.0
if fit:
ship_price = getattr(fit.ship.item.price, "price", 0.0)
for module in fit.modules:
if not module.isEmpty:
module_price += getattr(module.item.price, "price", 0.0)
for drone in fit.drones:
drone_price += getattr(drone.item.price, "price", 0.0) * drone.amount
for fighter in fit.fighters:
fighter_price += getattr(fighter.item.price, "price", 0.0) * fighter.amount
for cargo in fit.cargo:
cargo_price += getattr(cargo.item.price, "price", 0.0) * cargo.amount
for booster in fit.boosters:
booster_price += getattr(booster.item.price, "price", 0.0)
for implant in fit.appliedImplants:
implant_price += getattr(implant.item.price, "price", 0.0)
total_price = ship_price + module_price + drone_price + fighter_price + cargo_price + booster_price + implant_price
return {
"ship": ship_price,
"fittings": module_price,
"drones_and_fighters": drone_price + fighter_price,
"cargo": cargo_price,
"character": booster_price + implant_price,
"total": total_price,
}
def _build_output(main_fit) -> dict:
return {
"fit": {
"id": main_fit.ID,
"name": main_fit.name,
"ship_type": main_fit.ship.item.typeName,
},
"resources": _collect_resources(main_fit),
"defense": _collect_defense(main_fit),
"capacitor": _collect_capacitor(main_fit),
"firepower": _collect_firepower(main_fit),
"remote_reps_outgoing": _collect_remote_reps(main_fit),
"targeting_misc": _collect_targeting_misc(main_fit),
"price": _collect_price(main_fit),
}
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()
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 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 ValueError("Each projected_fits entry must contain a string 'fit'")
if not isinstance(count, int) or count <= 0:
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 ValueError("Each command_fits entry must be an object")
fit_text = entry.get("fit")
if not isinstance(fit_text, str):
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)
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__":
_run_http_server(9123, None)