refactor: greatly enchances import/export data

examples/trade_finder.py

@@ -0,0 +1,64 @@
+from dataclasses import dataclass
+import os
+import asyncio
+from dotenv import load_dotenv
+from pymerc.client import Client
+
+# Load the API_USER and API_TOKEN from the environment
+load_dotenv()
+
+@dataclass
+class TradeRoute:
+    buy_town: str = ""
+    sell_town: str = ""
+    profit: float = float("-inf")
+    volume: int = 0
+
+async def main():
+    client = Client(os.environ["API_USER"], os.environ["API_TOKEN"])
+
+    filter = ["Aderhampton", "Eindburg", "Eindweiller", "Eshagen", "Hogbach", "Utheim"]
+    towns = await client.towns(filter)
+
+    results: dict[str, TradeRoute] = {}
+    for town in towns:
+        for item, details in town.market.items():
+            if item == "labour":
+                continue
+
+            if item not in results:
+                results[item] = TradeRoute()
+
+            sell_price = details.average_price
+            sell_volume = details.volume
+
+            for other_town in towns:
+                if other_town == town:
+                    continue
+
+                other_details = other_town.market.get(item)
+                if other_details is None:
+                    continue
+
+                buy_price = other_details.average_price
+                buy_volume = other_details.volume
+
+                trade_volume = min(sell_volume, buy_volume)
+                potential_profit = (sell_price - buy_price) * trade_volume
+
+                if potential_profit > results[item].profit:
+                    results[item].buy_town = other_town.data.name
+                    results[item].sell_town = town.data.name
+                    results[item].profit = potential_profit
+                    results[item].volume = trade_volume
+
+    results = dict(sorted(results.items(), key=lambda item: item[1].profit, reverse=True))
+
+    print(f"{'Item':<20} {'Buy Town':<15} {'Sell Town':<15} {'Profit':<10} {'Volume':<10}")
+    print("=" * 70)
+    for item, route in results.items():
+        if route.profit > 0:
+            print(f"{item.title():<20} {route.buy_town:<15} {route.sell_town:<15} ${route.profit:<10.2f} {route.volume:<10}")
+
+if __name__ == "__main__":
+    asyncio.run(main())

pymerc/api/models/common.py

@@ -594,6 +594,10 @@ class Inventory(BaseModel):
     previous_flows: Optional[dict[Item, InventoryFlow]] = {}
     reserved: Optional[int] = None
 
+    @property
+    def items(self) -> dict[Item, InventoryAccountAsset]:
+        """The items in the inventory."""
+        return self.account.assets
 
 class InventoryAccount(BaseModel):
     """Represents an inventory account."""
@@ -619,6 +623,31 @@ class InventoryAccountAsset(BaseModel):
     sale_price: Optional[float] = None
     unit_cost: Optional[float] = None
 
+    @property
+    def purchased(self) -> bool:
+        """Whether the asset was purchased."""
+        return self.purchase is not None
+
+    @property
+    def sold(self) -> bool:
+        """Whether the asset was sold."""
+        return self.sale is not None
+
+    @property
+    def total_purchase(self) -> float:
+        """The total purchase cost of the asset."""
+        return self.purchase * self.purchase_price
+
+    @property
+    def total_sale(self) -> float:
+        """The total sale cost of the asset."""
+        return self.sale * self.sale_price
+
+    @property
+    def total_value(self) -> float:
+        """The total value of the asset."""
+        return self.balance * self.unit_cost
+
 
 class InventoryManager(BaseModel):
     """Represents an inventory manager."""
@@ -630,6 +659,25 @@ class InventoryManager(BaseModel):
     sell_price: Optional[float] = None
     sell_volume: Optional[int] = None
 
+    @property
+    def buying(self) -> bool:
+        """Whether the manager is buying."""
+        return self.buy_price is not None and self.buy_volume is not None
+
+    @property
+    def max_buy_price(self) -> float:
+        """The maximum buy price of the manager."""
+        return self.buy_price * self.buy_volume
+
+    @property
+    def max_sell_price(self) -> float:
+        """The maximum sell price of the manager."""
+        return self.sell_price * self.sell_volume
+
+    @property
+    def selling(self) -> bool:
+        """Whether the manager is selling."""
+        return self.sell_price is not None and self.sell_volume is not None
 
 class InventoryFlow(BaseModel):
     """Represents an inventory flow."""
@@ -649,6 +697,16 @@ class Operation(BaseModel):
     production: float
     provision: float
 
+    @property
+    def surplus(self) -> float:
+        """The surplus of the operation."""
+        return self.production - self.target
+
+    @property
+    def shortfall(self) -> float:
+        """The shortfall of the operation."""
+        return self.target - self.production
+
 class Path(BaseModel):
     """Represents part of a path."""
     x: int

pymerc/game/exports.py

@@ -0,0 +1,129 @@
+from __future__ import annotations
+
+from collections import UserDict, UserList
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+from pymerc.api.models import common
+from pymerc.game.town import Town
+
+if TYPE_CHECKING:
+    from pymerc.game.transport import Transport
+
+@dataclass
+class Export:
+    """A representation of an export in the game."""
+
+    asset: common.InventoryAccountAsset
+    flow: common.InventoryFlow
+    manager: common.InventoryManager
+    town: Town
+    transport: Transport
+
+    @property
+    def flowed(self) -> int:
+        """How much of the import flowed in the last turn."""
+        return self.flow.export or 0
+
+    @property
+    def value(self) -> float:
+        """The value of the export if it was sold at max price."""
+        return self.manager.max_sell_price
+
+    @property
+    def value_flowed(self) -> float:
+        """The value of the export that flowed in the last turn."""
+        if not self.flowed:
+            return 0.0
+
+        return self.asset.sale * self.asset.sale_price
+
+    @property
+    def volume(self) -> int:
+        """The volume of the export if it was sold at max volume."""
+        return self.manager.buy_volume
+
+class Exports(UserDict[common.Item, Export]):
+    """A collection of exports for a transport in the game."""
+
+    @property
+    def flowed(self) -> Exports:
+        """The exports that flowed in the last turn."""
+        return Exports({item: exp for item, exp in self.data.items() if exp.flowed})
+
+    @property
+    def value(self) -> float:
+        """The total value of all exports if they were sold at max price."""
+        return sum([exp.value for exp in self.data.values()])
+
+    @property
+    def value_flowed(self) -> float:
+        """The total value of all exports that flowed in the last turn."""
+        return sum([exp.value_flowed for exp in self.data.values()])
+
+    @property
+    def volume(self) -> int:
+        """The total volume of all exports if they were sold at max volume."""
+        return sum([exp.volume for exp in self.data.values()])
+
+class ExportsList(UserList[Export]):
+    """A collection of exports for a transport in the game."""
+
+    @property
+    def flowed(self) -> ExportsList:
+        """The exports that flowed in the last turn."""
+        return ExportsList([exp for exp in self.data if exp.flowed])
+
+    @property
+    def value(self) -> float:
+        """The total value of all exports if they were sold at max price."""
+        return sum([exp.value for exp in self.data])
+
+    @property
+    def value_flowed(self) -> float:
+        """The total value of all exports that flowed in the last turn."""
+        return sum([exp.value_flowed for exp in self.data])
+
+    @property
+    def volume(self) -> int:
+        """The total volume of all exports if they were sold at max volume."""
+        return sum([exp.volume for exp in self.data])
+
+    def by_town_id(self, id: int) -> ExportsList:
+        """Returns the exports for a town by id."""
+        return ExportsList([exp for exp in self.data if exp.town.data.id == id])
+
+    def by_town_name(self, name: str) -> ExportsList:
+        """Returns the exports for a town by name."""
+        return ExportsList([exp for exp in self.data if exp.town.data.name == name])
+
+class ExportsSummed(UserDict[common.Item, ExportsList]):
+    """A collection of exports for a player in the game."""
+
+    @property
+    def flowed(self) -> ExportsSummed:
+        """The exports that flowed in the last turn."""
+        return ExportsSummed({item: exps for item, exps in self.data.items() if any([exp.flowed for exp in exps])})
+
+    @property
+    def value(self) -> float:
+        """The total value of all exports if they were sold at max price."""
+        return sum([sum([exp.value for exp in exps]) for exps in self.data.values()])
+
+    @property
+    def value_flowed(self) -> float:
+        """The total value of all exports that flowed in the last turn."""
+        return sum([sum([exp.value_flowed for exp in exps]) for exps in self.data.values()])
+
+    @property
+    def volume(self) -> int:
+        """The total volume of all exports if they were sold at max volume."""
+        return sum([sum([exp.volume for exp in exps]) for exps in self.data.values()])
+
+    def by_town_id(self, town_id: int) -> ExportsSummed:
+        """Returns the exports for a town by id."""
+        return ExportsSummed({item: exps for item, exps in self.data.items() if exps[0].town.data.id == town_id})
+
+    def by_town_name(self, town_name: str) -> ExportsSummed:
+        """Returns the exports for a town by name."""
+        return ExportsSummed({item: exps for item, exps in self.data.items() if exps[0].town.data.name == town_name})