✨ New shiny drop data management all in one place.

coderanger · coderanger · commit dbba35432e79 · 2022-02-13T03:15:08.000-08:00
Along with a new script to estimate base drop rates and a revamped explore vs lemonade drop comparison tool.
diff --git a/py/base_droprates.py b/py/base_droprates.py
@@ -0,0 +1,19 @@
+from fractions import Fraction
+
+import typer
+
+import droprates
+
+
+def base_droprates(cider: bool = False, limit: int = 20) -> None:
+    for loc, loc_explores in sorted(
+        droprates.compile_drops(explore=True, cider=cider).locations.items()
+    ):
+        drops_per_explore = loc_explores.drops / loc_explores.explores
+        rounded = Fraction(drops_per_explore).limit_denominator(limit)
+        num, denom = rounded.as_integer_ratio()
+        print(f"{loc}: {num}/{denom} ({loc_explores.explores})")
+
+
+if __name__ == "__main__":
+    typer.run(base_droprates)
diff --git a/py/droprates.py b/py/droprates.py
@@ -1,8 +1,141 @@
 import sys
+import time
+from collections import defaultdict
+from typing import Any, Union
 
+import attrs
+
+import fixtures
 import parse_logs
 
 
+def when_dropped(item: fixtures.Item) -> range:
+    first_dropped = item.first_dropped or item.first_seen or 0
+    last_dropped = item.last_dropped or round((time.time() + 10000000) * 1000)
+    return range(first_dropped, last_dropped + 1)
+
+
+@attrs.define
+class ItemDrops:
+    explores: int = 0
+    lemonades: int = 0
+    ciders: int = 0
+    fishes: int = 0
+    nets: int = 0
+    drops: int = 0
+
+
+@attrs.define
+class LocationDrops:
+    explores: int = 0
+    lemonades: int = 0
+    ciders: int = 0
+    fishes: int = 0
+    nets: int = 0
+    drops: int = 0
+    items: dict[str, ItemDrops] = attrs.Factory(lambda: defaultdict(ItemDrops))
+
+
+@attrs.define
+class Drops:
+    explores: int = 0
+    lemonades: int = 0
+    ciders: int = 0
+    fishes: int = 0
+    nets: int = 0
+    drops: int = 0
+    locations: dict[str, LocationDrops] = attrs.Factory(
+        lambda: defaultdict(LocationDrops)
+    )
+
+
+def count_sources(
+    drops: Union[Drops, LocationDrops, ItemDrops], row: dict[str, Any]
+) -> None:
+    if row["type"] == "explore":
+        drops.explores += row["results"]["stamina"]
+    elif row["type"] == "lemonade":
+        drops.lemonades += 1
+    elif row["type"] == "cider":
+        drops.ciders += 1
+        # This is kind of wrong for global and location stats since not all explores count
+        # for all items but it's more correct than not.
+        drops.explores += row["results"].get("explores", row["results"]["stamina"])
+    elif row["type"] == "fish":
+        drops.fishes += 1
+    elif row["type"] == "net":
+        drops.nets += 1
+
+
+def compile_drops(
+    explore: bool = False,
+    lemonade: bool = False,
+    cider: bool = False,
+    fish: bool = False,
+    net: bool = False,
+) -> Drops:
+    when_items_dropped = {
+        item.name: when_dropped(item) for item in fixtures.load_items()
+    }
+    # loc_items = {loc.name: loc.items for loc in fixtures.load_locations()}
+    types: set[str] = set()
+    if explore:
+        types.add("explore")
+    if lemonade:
+        types.add("lemonade")
+    if cider:
+        types.add("cider")
+    if fish:
+        types.add("fish")
+    if net:
+        types.add("net")
+    explores = Drops()
+    logs = [row for row in parse_logs.parse_logs() if row["type"] in types]
+    # First run through to count drops and work out which items are in each locations.
+    for row in logs:
+        location_name = row["results"].get("location")
+        if not location_name:
+            continue
+        overflow_items: set[str] = {
+            item["item"] for item in row["results"]["items"] if item["overflow"]
+        }
+        for item in row["results"]["items"]:
+            if row["ts"] not in when_items_dropped[item["item"]]:
+                # Ignore out-of-bounds drops. This allows accounting for stuff like drop
+                # rates changing substantially by manually resetting firstDropped.
+                continue
+            if row["type"] == "cider" and item["item"] in overflow_items:
+                # Cider overflow always reports 0 drops so any item that overflows during
+                # a cider has to be ignored.
+                continue
+            explores.locations[location_name].items[item["item"]].drops += item.get(
+                "quantity", 1
+            )
+            explores.locations[location_name].drops += item.get("quantity", 1)
+            explores.drops += item.get("quantity", 1)
+    # Second pass to get the explore counts.
+    for row in logs:
+        location_name = row["results"].get("location")
+        if not location_name:
+            continue
+        overflow_items: set[str] = {
+            item["item"] for item in row["results"]["items"] if item["overflow"]
+        }
+        for item, item_explores in explores.locations[location_name].items.items():
+            if row["ts"] not in when_items_dropped[item]:
+                # Item couldn't drop, this doesn't count.
+                continue
+            if row["type"] == "cider" and item in overflow_items:
+                # Cider overflow always reports 0 drops so any item that overflows during
+                # a cider has to be ignored.
+                continue
+            count_sources(item_explores, row)
+        count_sources(explores.locations[location_name], row)
+        count_sources(explores, row)
+
+    return explores
+
+
 def total_drops() -> dict[str, dict[str, int]]:
     totals = {}
     for row in parse_logs.parse_logs("explore"):
diff --git a/py/exp_vs_lem.py b/py/exp_vs_lem.py
@@ -1,21 +1,73 @@
+import statistics
+from typing import Optional
+
+import typer
+
 import droprates
-import lemonade
-
-lem = lemonade.drop_rates()
-# exp = droprates.rates_per_stam()
-exp = droprates.drop_rates()
-
-locations = set(lem.keys()) | set(exp.keys())
-
-for loc in sorted(locations):
-    print(f"{loc}:")
-    lem_data = lem.get(loc, {})
-    exp_data = exp.get(loc, {})
-    items = set(lem_data.keys()) | set(exp_data.keys())
-    for item in sorted(items):
-        lem_rate = lem_data.get(item, 0)
-        exp_rate = exp_data.get(item, 0)
-        if lem_rate == 0 or exp_rate == 0:
-            print(f"\t{item}: NO MATCH")
-        else:
-            print(f"\t{item}: {lem_rate / exp_rate}")
+
+BASE_DROP_RATES = {
+    "Black Rock Canyon": 1 / 3,
+    "Cane Pole Ridge": 2 / 7,
+    "Ember Lagoon": 1 / 3,
+    "Forest": 1 / 3,
+    "Highland Hills": 1 / 4,
+    "Misty Forest": 1 / 3,
+    "Mount Banon": 1 / 3,
+    "Small Cave": 2 / 5,
+    "Small Spring": 1 / 3,
+    "Whispering Creek": 4 / 15,
+}
+
+
+def exp_vs_lem(drop_threshold: Optional[int] = None) -> None:
+    explores = droprates.compile_drops(explore=True)
+    lemonade = droprates.compile_drops(lemonade=True)
+    for location in sorted(explores.locations.keys() | lemonade.locations.keys()):
+        explore_loc = explores.locations[location]
+        lemonade_loc = lemonade.locations[location]
+        base_drop_rate = BASE_DROP_RATES[location]
+
+        if explore_loc.explores == 0:
+            print(f"{location} NO EXPLORES")
+            continue
+        if lemonade_loc.lemonades == 0:
+            print(f"{location} NO LEMONADES")
+            continue
+
+        exp_vs_lem_rates = []
+        adj_exp_vs_lem_rates = []
+
+        for item in sorted(explore_loc.items.keys() | lemonade_loc.items.keys()):
+            if explore_loc.items[item].drops == 0:
+                print(f"{location} {item} NO EXPLORE DROPS")
+                continue
+            if lemonade_loc.items[item].drops == 0:
+                print(f"{location} {item} NO LEMONADE DROPS")
+                continue
+            if drop_threshold is not None and (
+                explore_loc.items[item].drops < drop_threshold
+                or lemonade_loc.items[item].drops < drop_threshold
+            ):
+                print(f"{location} {item} NOT ENOUGH DROPS")
+                continue
+
+            explore_drop_rate = explore_loc.items[item].drops / explore_loc.drops
+            lemonade_drop_rate = lemonade_loc.items[item].drops / lemonade_loc.drops
+            drops_per_explore = (
+                explore_loc.items[item].drops / explore_loc.items[item].explores
+            )
+            adjusted_drops_per_explore = drops_per_explore / base_drop_rate
+            exp_vs_lem_rate = explore_drop_rate / lemonade_drop_rate
+            adj_exp_vs_lem_rate = adjusted_drops_per_explore / lemonade_drop_rate
+            print(f"{location} {item} {exp_vs_lem_rate:.3f} {adj_exp_vs_lem_rate:.3f}")
+            exp_vs_lem_rates.append(exp_vs_lem_rate)
+            adj_exp_vs_lem_rates.append(adj_exp_vs_lem_rate)
+
+        if exp_vs_lem_rates and adj_exp_vs_lem_rates:
+            print(
+                f"{location} AVERAGE {statistics.mean(exp_vs_lem_rates):.3f} {statistics.mean(adj_exp_vs_lem_rates):.3f}"
+            )
+
+
+if __name__ == "__main__":
+    typer.run(exp_vs_lem)
diff --git a/py/fixtures.py b/py/fixtures.py
@@ -4,6 +4,7 @@
 from typing import Iterable, Optional
 
 import attr
+import attrs
 from frozendict import frozendict
 
 comment_re = re.compile(r"^//.*$")
@@ -17,6 +18,7 @@
     "fleaMarket": "flea_market",
     "growthTime": "growth_time",
     "firstSeen": "first_seen",
+    "firstDropped": "first_dropped",
 }
 
 
@@ -51,13 +53,30 @@ class Item:
     event: bool = False
     growth_time: Optional[int] = None
     first_seen: Optional[int] = None
+    first_dropped: Optional[int] = None
+    last_dropped: Optional[int] = None
 
 
 def load_items() -> Iterable[Item]:
     for item in load_fixture("items"):
         yield Item(**{item_case_mappings.get(k, k): v for k, v in item.items()})
 
 
+@attrs.define(frozen=True)
+class Location:
+    id: str
+    type: str
+    name: str
+    items: tuple[str]
+
+
+def load_locations(type: Optional[str] = None) -> Iterable[Location]:
+    for location in load_fixture("locations"):
+        if type is not None and type != location["type"]:
+            continue
+        yield Location(**location)
+
+
 if __name__ == "__main__":
     items = [
         {item_case_mappings.get(k, k): v for k, v in item.items()}
diff --git a/py/parse_logs.py b/py/parse_logs.py
@@ -24,6 +24,12 @@ def parse_logs(log_type=None, since=int(os.environ.get("SINCE", 0))):
             if items and "item" not in items[0]:
                 continue
 
+            if row["type"] == "cider" and "explores" not in row["results"]:
+                # Backfill this.
+                row["results"]["explores"] = (
+                    1250 if row["results"]["stamina"] >= 750 else 1000
+                )
+
             if row["ts"] not in all_logs:
                 # Fix up the old zone/loc thing.
                 results = row.get("results")
diff --git a/py/test_droprates.py b/py/test_droprates.py
