Fetch from EVE for cargo before / cargo after

EveProfitManager/cargo.py

@@ -0,0 +1,41 @@
+"""Read what the active ship's cargo hold is worth from ESI, in the app's millions of ISK."""
+from dataclasses import dataclass
+
+from loot import _items, _whole, get_prices
+
+
+@dataclass(frozen=True)
+class Cargo:
+    millions: float
+    unpriced: int  # item types with no market price, left out of `millions`
+    fetched_at: float  # when ESI last answered for the asset list; it caches it for an hour
+
+
+def get_cargo(client, character_id):
+    """
+    The cargo value of the ship EVE says the character is flying now.
+
+    Counts only items directly in that ship with flag Cargo (ammo included), not
+    its drone bay, fitted modules, other ships, hangars or containers. Blueprint
+    copies are left out, as in the asset reader, because their type's price is
+    the original's.
+    """
+    ship = client.get(f"/characters/{character_id}/ship")
+    if not isinstance(ship, dict) or not _whole(ship.get("ship_item_id")):
+        raise ValueError(f"Unexpected ship response: {ship!r}")
+    path = f"/characters/{character_id}/assets"
+    first = f"{path}?page=1"
+    items = _items(client.get(first))
+    for page in range(2, client.pages(first) + 1):
+        items += _items(client.get(f"{path}?page={page}"))
+    quantities = {}
+    for entry in items:
+        if not _whole(entry.get("location_id")) or not isinstance(entry.get("location_flag"), str):
+            raise ValueError(f"Unexpected asset: {entry!r}")
+        if entry["location_id"] == ship["ship_item_id"] and entry["location_flag"] == "Cargo" \
+                and not entry.get("is_blueprint_copy"):
+            quantities[entry["type_id"]] = quantities.get(entry["type_id"], 0) + entry["quantity"]
+    prices = get_prices(client)
+    isk = sum(count * prices[type_id] for type_id, count in quantities.items() if type_id in prices)
+    unpriced = sum(1 for type_id in quantities if type_id not in prices)
+    return Cargo(millions=isk / 1e6, unpriced=unpriced, fetched_at=client.fetched_at(first))

EveProfitManager/gui.py

@@ -31,6 +31,7 @@
 from clock import read_clock
 from journal import get_incursion_payouts
 from location import AbyssWatcher, get_location
+from cargo import get_cargo
 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
@@ -929,6 +930,42 @@ def build_log_form(self, frame):
         self.cargo_before = self.field(frame, 0, "Cargo before (M)", 150, "e.g. 120")
         self.cargo_after = self.field(frame, 1, "Cargo after (M)", 150, "e.g. 164.5")
         self.time_left = self.field(frame, 2, "Time left (min)", 120, "blank = timer")
+        self.fetch_buttons = {}
+        for column, which in enumerate(("before", "after")):
+            button = self.paint(
+                ctk.CTkButton(frame, text="Fetch from EVE", width=150, height=28, font=font(12),
+                              border_width=1, command=lambda w=which: self.fetch_cargo(w)),
+                fg_color=lambda _t: "transparent", border_color="line", hover_color="line", text_color="soft",
+            )
+            button.grid(row=2, column=column, sticky="w", padx=(0, 12), pady=(6, 0))
+            self.fetch_buttons[which] = button
+            note = self.paint(ctk.CTkLabel(frame, text="", font=font(11), wraplength=150, justify="left"),
+                              text_color="soft")
+            note.grid(row=3, column=column, sticky="nw", padx=(0, 12))
+            setattr(self, f"cargo_{which}_note", note)
+
+    def say_cargo_note(self, which, text, problem=False):
+        note = getattr(self, f"cargo_{which}_note")
+        note.configure(text=text, text_color=CRITICAL_TEXT if problem else self.theme.soft)
+
+    def fetch_cargo(self, which):
+        """Fill Cargo before or after with the active ship's cargo value from ESI, off the UI thread."""
+        client, why = self.signed_in_client()
+        if client is None:
+            self.say_cargo_note(which, why, problem=True)
+            return
+        self.say_cargo_note(which, "reading cargo...")
+
+        def done(cargo, error):
+            if error is not None:
+                self.say_cargo_note(which, str(error), problem=True)
+                return
+            set_entry(getattr(self, f"cargo_{which}"), f"{cargo.millions:.2f}")
+            stamp = f"From EVE at {datetime.fromtimestamp(cargo.fetched_at):%H:%M}"
+            unpriced = f" · {cargo.unpriced} unpriced" if cargo.unpriced else ""
+            self.say_cargo_note(which, f"{stamp}{unpriced}. ESI can lag up to an hour; edit if it's off.")
+
+        self.app.run_async(lambda: get_cargo(client, client.character.id), done)
 
     def reset_watcher(self):
         """Detection switched on: the next reading is a fresh baseline."""
@@ -995,6 +1032,8 @@ def log_run(self):
 
     def after_log(self):
         self.prefill_cargo_before()
+        for which in ("before", "after"):
+            self.say_cargo_note(which, "")
         self.cargo_after.delete(0, "end")
         self.time_left.delete(0, "end")
         self.read_wallet()
@@ -2047,7 +2086,7 @@ def set_icon(self):
             self.iconbitmap(asset_path("icon.ico"))
         else:
             # Keep a reference: Tk drops the image once Python garbage-collects it
-            self.icon = tk.PhotoImage(file=asset_path("icon.png"))
+            self.icon = tk.PhotoImage(master=self, file=asset_path("icon.png"))
             self.iconphoto(True, self.icon)
 
     def bring_to_front(self):

EveProfitManager/test_cargo.py

@@ -0,0 +1,82 @@
+import unittest
+
+from cargo import Cargo, get_cargo
+from test_loot import FakeClient, item
+
+SHIP = "/characters/123/ship"
+ASSETS = "/characters/123/assets"
+SHIP_ID = 1_050_000_000_001
+
+
+def in_hold(type_id, quantity=1, **extra):
+    return item(type_id, quantity, location_id=SHIP_ID, location_flag="Cargo", location_type="item", **extra)
+
+
+def client(first_page, second_page=None, prices=None, ship=None, fetched_at=1_700_000_000.0):
+    bodies = {
+        SHIP: {"ship_item_id": SHIP_ID, "ship_name": "x", "ship_type_id": 17_738} if ship is None else ship,
+        f"{ASSETS}?page=1": first_page,
+        "/markets/prices": [{"type_id": 34, "average_price": 5.0}, {"type_id": 47_762, "adjusted_price": 1_000_000.0}]
+        if prices is None else prices,
+    }
+    pages = {}
+    if second_page is not None:
+        bodies[f"{ASSETS}?page=2"] = second_page
+        pages[f"{ASSETS}?page=1"] = 2
+    return FakeClient(bodies, pages=pages, fetched_at=fetched_at)
+
+
+class GetCargoTests(unittest.TestCase):
+    def test_values_the_cargo_in_millions_with_the_time_esi_last_answered(self):
+        c = client([in_hold(34, 2_000_000), in_hold(47_762, 3)], fetched_at=1_700_000_123.0)
+        self.assertEqual(get_cargo(c, 123), Cargo(millions=13.0, unpriced=0, fetched_at=1_700_000_123.0))
+
+    def test_ammo_in_the_cargo_hold_counts(self):
+        c = client([in_hold(47_762, 100)])
+        self.assertEqual(get_cargo(c, 123).millions, 100.0)
+
+    def test_only_the_active_ships_cargo_flag_counts(self):
+        c = client([
+            in_hold(34, 1_000_000),
+            item(34, 1_000_000, location_id=SHIP_ID, location_flag="DroneBay", location_type="item"),
+            item(34, 1_000_000, location_id=SHIP_ID, location_flag="HiSlot0", location_type="item"),
+            item(34, 1_000_000, location_id=SHIP_ID, location_flag="LoSlot0", location_type="item"),
+            item(34, 1_000_000, location_id=SHIP_ID + 1, location_flag="Cargo", location_type="item"),  # another ship
+            item(34, 1_000_000),  # a station hangar
+            item(34, 1_000_000, location_id=9_000_000_000_001, location_flag="Unlocked", location_type="item"),  # a container
+        ])
+        self.assertEqual(get_cargo(c, 123).millions, 5.0)
+
+    def test_follows_the_asset_pages(self):
+        c = client([in_hold(34, 1_000_000)], [in_hold(34, 1_000_000)])
+        self.assertEqual(get_cargo(c, 123).millions, 10.0)
+
+    def test_leaves_out_blueprint_copies(self):
+        c = client([in_hold(34, 1_000_000), in_hold(47_762, is_blueprint_copy=True)])
+        self.assertEqual(get_cargo(c, 123).millions, 5.0)
+
+    def test_counts_unpriced_item_types_and_leaves_them_out_of_the_value(self):
+        c = client([in_hold(34, 1_000_000), in_hold(52_227), in_hold(52_227), in_hold(52_228, 4)])
+        cargo = get_cargo(c, 123)
+        self.assertEqual((cargo.millions, cargo.unpriced), (5.0, 2))
+
+    def test_an_empty_hold_is_worth_nothing(self):
+        self.assertEqual(get_cargo(client([item(34)]), 123).millions, 0.0)
+
+    def test_rejects_malformed_bodies(self):
+        good = [in_hold(34)]
+        cases = {
+            "ship not a dict": client(good, ship=[1]),
+            "ship without an item id": client(good, ship={"ship_type_id": 1}),
+            "ship id not whole": client(good, ship={"ship_item_id": "7"}),
+            "assets not a list": client({"error": "x"}),
+            "asset without a location": client([{"type_id": 34, "quantity": 1}]),
+            "prices not a list": client(good, prices={"error": "x"}),
+        }
+        for name, c in cases.items():
+            with self.subTest(name), self.assertRaises(ValueError):
+                get_cargo(c, 123)
+
+
+if __name__ == "__main__":
+    unittest.main()

EveProfitManager/test_cargo_fetch.py

@@ -0,0 +1,123 @@
+import os
+import tempfile
+import tkinter
+import unittest
+from datetime import datetime
+from types import SimpleNamespace
+from unittest import mock
+
+from cargo import Cargo
+from core import ABYSS, Tracker
+from theme import load_fonts
+
+READ_AT = datetime(2026, 10, 25, 18, 40).timestamp()
+
+
+class FetchCargoFromEveTest(unittest.TestCase):
+    """Drives the two Fetch from EVE buttons through gui.App with the cargo reader mocked."""
+
+    def setUp(self):
+        load_fonts()
+        try:
+            tkinter.Tk().destroy()
+        except tkinter.TclError as err:
+            self.skipTest(f"Tk cannot open a window: {err}")
+        import customtkinter as ctk
+        import gui
+
+        self.gui = gui
+        ctk.set_appearance_mode("dark")
+        self.client = SimpleNamespace(character=SimpleNamespace(id=123))
+        account = mock.patch.object(gui.Account, "client", side_effect=lambda: self.client)
+        account.start()
+        self.addCleanup(account.stop)
+        self.cargo = mock.Mock(return_value=Cargo(millions=164.5678, unpriced=0, fetched_at=READ_AT))
+        reader = mock.patch.object(gui, "get_cargo", self.cargo)
+        reader.start()
+        self.addCleanup(reader.stop)
+        wallet = mock.patch.object(gui, "get_wallet", side_effect=RuntimeError("no wallet here"))
+        wallet.start()
+        self.addCleanup(wallet.stop)
+        folder = tempfile.TemporaryDirectory()
+        self.addCleanup(folder.cleanup)
+        self.tracker = Tracker(os.path.join(folder.name, "tracker.db"))
+        self.addCleanup(self.tracker.close)
+        self.tracker.start_session(ABYSS, weather="Gamma", tier=2)
+        sync = mock.patch.object(gui.App, "run_async", lambda _app, work, done: self.finish(work, done))
+        sync.start()
+        self.addCleanup(sync.stop)
+        self.app = gui.App(self.tracker)
+        self.addCleanup(self.app.destroy)
+        self.app.update()
+        self.page = self.app.pages["abyss"]
+
+    @staticmethod
+    def finish(work, done):
+        try:
+            result = work()
+        except Exception as error:
+            done(None, error)
+        else:
+            done(result, None)
+
+    def note(self, which):
+        return getattr(self.page, f"{which}_note").cget("text")
+
+    def test_each_button_fills_its_own_field_to_two_decimals_and_stamps_when_eve_answered(self):
+        self.page.fetch_cargo("before")
+        self.assertEqual(self.page.cargo_before.get(), "164.57")
+        self.assertEqual(self.page.cargo_after.get(), "")
+        self.assertIn("From EVE at 18:40", self.note("cargo_before"))
+        self.assertEqual(self.note("cargo_after"), "")
+        self.page.fetch_cargo("after")
+        self.assertEqual(self.page.cargo_after.get(), "164.57")
+        self.assertIn("From EVE at 18:40", self.note("cargo_after"))
+        self.assertIn("up to an hour", self.note("cargo_after"))
+        self.assertIn("edit", self.note("cargo_after"))
+
+    def test_the_buttons_are_on_the_page_and_call_fetch(self):
+        self.page.fetch_buttons["after"].invoke()
+        self.assertEqual(self.page.cargo_after.get(), "164.57")
+        self.page.fetch_buttons["before"].invoke()
+        self.assertEqual(self.page.cargo_before.get(), "164.57")
+
+    def test_names_the_unpriced_item_types(self):
+        self.cargo.return_value = Cargo(millions=10.0, unpriced=2, fetched_at=READ_AT)
+        self.page.fetch_cargo("after")
+        self.assertIn("From EVE at 18:40 · 2 unpriced", self.note("cargo_after"))
+
+    def test_an_error_shows_in_the_note_in_the_critical_colour_and_leaves_the_field_alone(self):
+        self.gui.set_entry(self.page.cargo_after, "88")
+        self.cargo.side_effect = ValueError("Unexpected ship response")
+        self.page.fetch_cargo("after")
+        self.assertEqual(self.page.cargo_after.get(), "88")
+        self.assertIn("Unexpected ship response", self.note("cargo_after"))
+        self.assertEqual(self.page.cargo_after_note.cget("text_color"), self.gui.CRITICAL_TEXT)
+
+    def test_a_fetch_after_an_error_clears_the_error_colour(self):
+        self.cargo.side_effect = ValueError("x")
+        self.page.fetch_cargo("after")
+        self.cargo.side_effect = None
+        self.page.fetch_cargo("after")
+        self.assertNotEqual(self.page.cargo_after_note.cget("text_color"), self.gui.CRITICAL_TEXT)
+
+    def test_signed_out_fetch_says_to_sign_in_and_the_field_still_works_by_hand(self):
+        self.client = None
+        self.gui.set_entry(self.page.cargo_before, "10")
+        self.page.fetch_cargo("before")
+        self.cargo.assert_not_called()
+        self.assertIn("Sign in", self.note("cargo_before"))
+        self.assertEqual(self.page.cargo_before.get(), "10")
+        self.assertEqual(self.page.cargo_before_note.cget("text_color"), self.gui.CRITICAL_TEXT)
+
+    def test_logging_a_run_clears_the_notes(self):
+        self.page.fetch_cargo("before")
+        self.page.fetch_cargo("after")
+        self.gui.set_entry(self.page.time_left, "5")
+        self.page.log()
+        self.assertEqual(self.note("cargo_before"), "")
+        self.assertEqual(self.note("cargo_after"), "")
+
+
+if __name__ == "__main__":
+    unittest.main()