461 lines
18 KiB
Python
461 lines
18 KiB
Python
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
|
|
|
|
# POST /simulate request and response shape; included in 400 response body
|
|
SIMULATE_SCHEMA = {
|
|
"request": {
|
|
"fit": "string (required). EFT or multi-line text export of a single fit, e.g. [ShipName, FitName] followed by modules/drones/etc.",
|
|
"projected_fits": "array (optional). Each element: { \"fit\": string, \"count\": positive integer }.",
|
|
"command_fits": "array (optional). Each element: { \"fit\": string }.",
|
|
},
|
|
"response_200": {
|
|
"fit": {"id": "int", "name": "string", "ship_type": "string"},
|
|
"resources": {"hardpoints": {}, "drones": {}, "fighters": {}, "calibration": {}, "powergrid": {}, "cpu": {}, "cargo": {}},
|
|
"defense": {"hp": {}, "ehp": {}, "resonance": {}, "tank": {}, "effective_tank": {}, "sustainable_tank": {}, "effective_sustainable_tank": {}},
|
|
"capacitor": {"capacity": {}, "recharge": {}, "use": {}, "delta": {}, "stable": {}, "state": {}, "neutralizer_resistance": {}},
|
|
"firepower": {"weapon_dps": {}, "drone_dps": {}, "total_dps": {}, "weapon_volley": {}, "drone_volley": {}, "total_volley": {}, "weapons": []},
|
|
"remote_reps_outgoing": {"current": {}, "pre_spool": {}, "full_spool": {}},
|
|
"targeting_misc": {"targets_max": {}, "target_range": {}, "scan_resolution": {}, "scan_strength": {}, "scan_type": {}, "jam_chance": {}, "drone_control_range": {}, "speed": {}, "align_time": {}, "signature_radius": {}, "warp_speed": {}, "max_warp_distance": {}, "probe_size": {}, "cargo_capacity": {}, "cargo_used": {}},
|
|
"price": {"ship": {}, "fittings": {}, "drones_and_fighters": {}, "cargo": {}, "character": {}, "total": {}},
|
|
},
|
|
}
|
|
|
|
|
|
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:
|
|
def send_error_json(code: int, msg: str, traceback_str: str | None = None):
|
|
body = {"error": msg, "schema": SIMULATE_SCHEMA}
|
|
if traceback_str:
|
|
body["traceback"] = traceback_str
|
|
return json.dumps(body, indent=2) + "\n"
|
|
|
|
class SimulateHandler(BaseHTTPRequestHandler):
|
|
def end_headers(self):
|
|
self.send_header("Access-Control-Allow-Origin", "*")
|
|
super().end_headers()
|
|
|
|
def send_error(self, code: int, message: str | None = None, explain: str | None = None):
|
|
msg = message or explain or "Error"
|
|
self.send_response(code)
|
|
self.send_header("Content-Type", "application/json; charset=utf-8")
|
|
self.end_headers()
|
|
self.wfile.write(send_error_json(code, msg).encode("utf-8"))
|
|
|
|
def _reply_error(self, code: int, msg: str, traceback_str: str | None = None):
|
|
self.send_response(code)
|
|
self.send_header("Content-Type", "application/json; charset=utf-8")
|
|
self.end_headers()
|
|
self.wfile.write(send_error_json(code, msg, traceback_str).encode("utf-8"))
|
|
|
|
def do_OPTIONS(self):
|
|
self.send_response(204)
|
|
self.send_header("Access-Control-Allow-Methods", "POST, OPTIONS")
|
|
self.send_header("Access-Control-Allow-Headers", "Content-Type")
|
|
self.send_header("Access-Control-Max-Age", "86400")
|
|
self.end_headers()
|
|
|
|
def do_GET(self):
|
|
if self.path == "/simulate":
|
|
self._reply_error(405, "Method not allowed. Use POST.")
|
|
else:
|
|
self._reply_error(404, "Not found. POST /simulate with JSON body.")
|
|
|
|
def do_POST(self):
|
|
if self.path != "/simulate":
|
|
self._reply_error(404, "Not found. POST /simulate with JSON body.")
|
|
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._reply_error(400, "Invalid JSON: %s" % exc)
|
|
return
|
|
if not isinstance(payload, dict):
|
|
self._reply_error(400, "Top-level JSON must be an object")
|
|
return
|
|
try:
|
|
output = compute_stats(payload, savepath)
|
|
except ValueError as e:
|
|
self._reply_error(400, str(e))
|
|
return
|
|
except Exception as e:
|
|
import traceback
|
|
tb = traceback.format_exc()
|
|
sys.stderr.write(tb)
|
|
sys.stderr.flush()
|
|
self._reply_error(500, str(e), tb)
|
|
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)
|