From eaef4590c51b6601a90ad3aa47d7e683a0700939 Mon Sep 17 00:00:00 2001 From: Fazorath Date: Sun, 25 Oct 2026 14:11:34 -0400 Subject: [PATCH] Show live incursion status on the Incursion page --- EveProfitManager/gui.py | 52 +++++++++++++++++++++ EveProfitManager/incursions.py | 57 +++++++++++++++++++++++ EveProfitManager/test_incursions.py | 70 +++++++++++++++++++++++++++++ 3 files changed, 179 insertions(+) create mode 100644 EveProfitManager/incursions.py create mode 100644 EveProfitManager/test_incursions.py diff --git a/EveProfitManager/gui.py b/EveProfitManager/gui.py index 11817fd..3b4cded 100644 --- a/EveProfitManager/gui.py +++ b/EveProfitManager/gui.py @@ -27,6 +27,8 @@ tier_label, ) from clock import read_clock +from esi import EsiClient +from incursions import get_incursions from wallet import get_wallet from account import SIGN_IN_ERRORS, Account from theme import NEUTRAL_THEME, asset_path, load_fonts, theme_for, ui_family @@ -894,6 +896,56 @@ class IncursionPage(SessionPage): ) unit = "site" pocket_label = "Payout (M)" + INCURSIONS_POLL_MS = 300000 # ESI caches the incursions for five minutes + STATE_COLOR = {"established": CRITICAL_TEXT, "mobilizing": WARNING, "withdrawing": GOOD_TEXT} + + def __init__(self, master, app): + super().__init__(master, app) + self.public_client = EsiClient(None, None) # incursions are public: no sign-in + self._incursions_job = None + self.build_incursions_panel(self.start_form.master) + + def build_incursions_panel(self, parent): + """Live incursions, read from ESI's public endpoint, below the start form.""" + head = ctk.CTkFrame(parent, fg_color="transparent") + head.pack(anchor="w", fill="x", pady=(28, 6)) + ctk.CTkLabel(head, text="Live incursions", font=font(15, "bold"), text_color=INK).pack(side="left") + self.paint( + ctk.CTkButton(head, text="Refresh", width=80, height=26, fg_color="transparent", border_width=1, + text_color=INK, command=self.read_incursions), + border_color="line", hover_color="panel", + ).pack(side="left", padx=12) + self.incursions_box = ctk.CTkFrame(parent, fg_color="transparent") + self.incursions_box.pack(anchor="w") + self.incursions_note = self.paint(ctk.CTkLabel(parent, text="Loading…", font=font(12)), text_color="soft") + self.incursions_note.pack(anchor="w", pady=(4, 0)) + self.read_incursions() + + def read_incursions(self): + """Read the incursions off the UI thread, then check again in five minutes.""" + if self._incursions_job: + self.after_cancel(self._incursions_job) + self._incursions_job = self.after(self.INCURSIONS_POLL_MS, self.read_incursions) + self.app.run_async(lambda: get_incursions(self.public_client), self.show_incursions) + + def show_incursions(self, incursions, error): + if error is not None: + self.incursions_note.configure(text=f"Couldn't read incursions: {error}") + return + for row in self.incursions_box.winfo_children(): + row.destroy() + for i, inc in enumerate(incursions): + boss = " · boss" if inc.has_boss else "" + ctk.CTkLabel(self.incursions_box, text=inc.state.capitalize(), font=font(13, "bold"), + text_color=self.STATE_COLOR[inc.state]).grid(row=i, column=0, sticky="w", padx=(0, 16)) + ctk.CTkLabel(self.incursions_box, text=f"{inc.constellation} · staging {inc.staging}", + font=font(13), text_color=INK).grid(row=i, column=1, sticky="w", padx=(0, 16)) + ctk.CTkLabel(self.incursions_box, font=font(13), text_color=INK_2, + text=f"{inc.systems} systems · {inc.influence:.0%} influence{boss}", + ).grid(row=i, column=2, sticky="w") + self.incursions_note.configure( + text="" if incursions else "No incursions are active right now." + ) def build_start_form(self, frame): self.per_site = self.field(frame, 0, "Payout per site (million ISK)", 240, "e.g. 31") diff --git a/EveProfitManager/incursions.py b/EveProfitManager/incursions.py new file mode 100644 index 0000000..1994d3c --- /dev/null +++ b/EveProfitManager/incursions.py @@ -0,0 +1,57 @@ +"""Read where Sansha incursions are active, and their state, from ESI's public incursions endpoint.""" +from dataclasses import dataclass + +STATES = ("established", "mobilizing", "withdrawing") # in the order an incursion is worth running + + +@dataclass(frozen=True) +class Incursion: + constellation: str + staging: str # the system the fleet stages in + state: str + systems: int # infested solar systems + influence: float # 0 to 1; how much of the constellation is still Sansha-held + has_boss: bool + + +def get_incursions(client): + """ + The active incursions, established ones first. Needs no sign-in. + + A constellation or system whose name can't be read shows as its ID instead. + """ + body = client.get("/incursions", auth=False) + if not isinstance(body, list): + raise ValueError(f"Unexpected incursions response: {body!r}") + names = {} # path -> name, so a repeated constellation or system is read once + found = [] + for entry in body: + state = entry.get("state") if isinstance(entry, dict) else None + constellation, staging = _id(entry, "constellation_id"), _id(entry, "staging_solar_system_id") + systems = entry.get("infested_solar_systems") if isinstance(entry, dict) else None + if state not in STATES or constellation is None or staging is None or not isinstance(systems, list): + raise ValueError(f"Unexpected incursion: {entry!r}") + found.append(Incursion( + constellation=_name(client, names, f"/universe/constellations/{constellation}", + f"Constellation {constellation}"), + staging=_name(client, names, f"/universe/systems/{staging}", f"System {staging}"), + state=state, + systems=len(systems), + influence=float(entry.get("influence") or 0), + has_boss=bool(entry.get("has_boss")), + )) + return sorted(found, key=lambda i: STATES.index(i.state)) + + +def _id(entry, key): + value = entry.get(key) if isinstance(entry, dict) else None + return value if isinstance(value, int) and not isinstance(value, bool) else None + + +def _name(client, names, path, fallback): + if path not in names: + try: + names[path] = client.get(path, auth=False)["name"] + except Exception: # a missing name shouldn't hide the incursion + names[path] = fallback + return names[path] diff --git a/EveProfitManager/test_incursions.py b/EveProfitManager/test_incursions.py new file mode 100644 index 0000000..b43fe5a --- /dev/null +++ b/EveProfitManager/test_incursions.py @@ -0,0 +1,70 @@ +import unittest + +from incursions import Incursion, get_incursions + + +class FakeClient: + """Stands in for EsiClient: serves canned bodies by path and records the auth flag.""" + + def __init__(self, bodies): + self.bodies = bodies + self.auth_used = [] + + def get(self, path, auth=True): + self.auth_used.append(auth) + if path not in self.bodies: + raise RuntimeError("not found") + return self.bodies[path] + + +def raw(**over): + body = { + "constellation_id": 20000100, "faction_id": 500019, "has_boss": True, + "infested_solar_systems": [30000001, 30000002, 30000003], "influence": 0.5, + "staging_solar_system_id": 30000001, "state": "established", "type": "Incursion", + } + body.update(over) + return body + + +NAMES = { + "/universe/constellations/20000100": {"name": "Tanoo"}, + "/universe/systems/30000001": {"name": "Jita"}, +} + + +class GetIncursionsTests(unittest.TestCase): + def test_returns_named_incursions_without_signing_in(self): + client = FakeClient({"/incursions": [raw()], **NAMES}) + self.assertEqual( + get_incursions(client), + [Incursion(constellation="Tanoo", staging="Jita", state="established", systems=3, + influence=0.5, has_boss=True)], + ) + self.assertEqual(set(client.auth_used), {False}) + + def test_falls_back_to_ids_when_a_name_cannot_be_read(self): + client = FakeClient({"/incursions": [raw()]}) + incursion = get_incursions(client)[0] + self.assertEqual(incursion.constellation, "Constellation 20000100") + self.assertEqual(incursion.staging, "System 30000001") + + def test_orders_established_before_mobilizing_before_withdrawing(self): + client = FakeClient({"/incursions": [ + raw(state="withdrawing"), raw(state="established"), raw(state="mobilizing"), + ], **NAMES}) + self.assertEqual([i.state for i in get_incursions(client)], + ["established", "mobilizing", "withdrawing"]) + + def test_no_active_incursions_is_an_empty_list(self): + self.assertEqual(get_incursions(FakeClient({"/incursions": []})), []) + + def test_rejects_an_unexpected_response(self): + with self.assertRaises(ValueError): + get_incursions(FakeClient({"/incursions": {"error": "x"}})) + with self.assertRaises(ValueError): + get_incursions(FakeClient({"/incursions": [raw(state="weird")]})) + + +if __name__ == "__main__": + unittest.main()