412 lines
14 KiB
Python
412 lines
14 KiB
Python
import argparse
|
|
import json
|
|
import sys
|
|
|
|
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 SystemExit("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 SystemExit("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)
|
|
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)
|
|
|
|
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 SystemExit("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 SystemExit("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 _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)
|
|
|
|
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")
|
|
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'")
|
|
if not isinstance(count, int) or count <= 0:
|
|
raise SystemExit("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")
|
|
fit_text = entry.get("fit")
|
|
if not isinstance(fit_text, str):
|
|
raise SystemExit("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)
|
|
|
|
output = _build_output(main_fit)
|
|
json.dump(output, sys.stdout, indent=2, sort_keys=True)
|
|
sys.stdout.write("\n")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|
|
|