version 1.0

README.md

@@ -0,0 +1,15 @@
+# A tool to quickly assess missing passives points in PoE
+
+## Usage
+
+1. In PoE, type /passives
+2. Run tool
+
+## Example output
+```
+[~/dev/poe/passives]% ./passives.py
+Missing passives (3):
+Kitava's Torments                 A5: Reliquary (3 torments) - Lani
+Reflection of Terror              A8: High Gardens via Bath House (YUGUUUUUUUUUL) - Sin
+An End to Hunger                  Kitava: Oriath - Lani
+```

passives.py

@@ -0,0 +1,248 @@
+#!/usr/bin/python3
+
+import logging
+from argparse import ArgumentParser, ArgumentError
+from datetime import datetime
+from os import fstat
+from re import compile
+
+
+TIME_FORMAT = "%Y/%m/%d %H:%M:%S"
+
+
+def extract_delta(logline: str) -> int:
+    """
+    Extracts a delta from a Client.txt log line.
+
+    The 3rd column is a millisecond delta from some specific but unknowable time in the past.
+    This is useful to group together /passives messages which, historically, have all had the
+    same delta.
+    """
+    return int(logline.split(" ")[2])
+
+
+def extract_time(line: str):
+    """
+    Extracts a time from a Client.txt log line.
+    """
+    # ls = line.split(' ')[0:2]
+    # strptime(f"{ls[0]} {ls[1]} {ls[2][-3:]}000", "%Y/%m/%d %H:%M:%S %f")
+    return datetime.strptime(" ".join(line.split(" ")[0:2]), TIME_FORMAT)
+
+
+class PassivesList:
+    """
+    An object encapsulating all of the data output by one /passives command
+    """
+
+    MATCHLINE = compile(r": (\d+) Passive Skill Points from quests:$")
+    QUESTLINE = compile(r": \((\d) from ([^\)]+)")
+
+    def __init__(self, logline):
+        m = self.MATCHLINE.search(logline)
+        if not m:
+            raise ValueError("Invalid init logline")
+
+        self.start_time = extract_time(logline)
+        self.delta = extract_delta(logline)
+        self.uncompleted_quests = {
+            "The Marooned Mariner": "A1: Ship Graveyard (Allflame) - Fairgraves)",
+            "The Dweller of the Deep": "A1: Flood depths 🦀 - Tarkleigh",
+            "The Way Forward": "A2: Western Forest (Thaumetic Emblem, Blackguards) - Bestel, A1",
+            "Victario's Secrets": "A3: Sewer (busts) - Hargan",
+            "Piety's Pets": "A3 - Grigor",
+            "An Indomitable Spirit": " A4: Mines level 2 (Deshret spirit) - Tasuni",
+            "In Service to Science": "A5: Control Blocks (Miasmeter) - Vilenta",
+            "Kitava's Torments": "A5: Reliquary (3 torments) - Lani",
+            "The Father of War": "A6: Karui Fortress (Tukohama) - Tarkleigh",
+            "The Cloven One": "A6: Prisoner's Gate (Abberath) - Bestel",
+            "The Puppet Mistress": "A6: The Wetlands via Riverways (Ryslatha) - Tarkleigh",
+            "The Master of a Million Faces": "A7 - Eramir",
+            "Queen of Despair": "A7: Dread Thicket via Northern Forest (Gruthkul) - Eramir",
+            "Kishara's Star": "A7: Causeway (Kishara's Star) - Weylam",
+            "Love is Dead": "A8: Quay (Ankh) - Clarissa",
+            "Reflection of Terror": "A8: High Gardens via Bath House (YUGUUUUUUUUUL) - Sin",
+            "The Gemling Legion": "A8: Grain Gate (Legionnaires) - Maramoa",
+            "Queen of the Sands": "A9: Oasis (Shakari) - Irasha",
+            "The Ruler of Highgate": "A9: Quarry (Garukhan) - Irasha or Tasuni",
+            "Vilenta's Vengeance": "A10: Control Blocks via Ravaged Square (Vilenta) - Lani",
+            "An End to Hunger": "Kitava: Oriath - Lani",
+        }
+        self.completed_quests = {}
+        self.completed_count = int(m.group(1))
+        self.parsed_count = 0
+        self._parsed_line_count = 0
+
+    @property
+    def name(self) -> str:
+        return self.start_time.strftime(TIME_FORMAT)
+
+    @property
+    def full(self) -> bool:
+        return self.parsed_count == self.completed_count
+
+    def try_append(self, logline) -> bool:
+        m = self.QUESTLINE.search(logline)
+        if m:
+            passive_count = int(m.group(1))
+            quest_name = m.group(2)
+            logging.debug(f"Found quest line: {passive_count}, {quest_name}")
+            if extract_delta(logline) == self.delta:
+                logging.debug(f"Delta matches >{self.delta}<")
+                if "Bandit" in quest_name:
+                    self.parsed_count += passive_count
+                    logging.debug(
+                        f"{self.name} | {self.parsed_count}/{self.completed_count} | {self.full}"
+                    )
+                    return True
+                else:
+                    try:
+                        # Move description over because why not
+                        self.completed_quests[quest_name] = self.uncompleted_quests[
+                            quest_name
+                        ]
+                        del self.uncompleted_quests[quest_name]
+                        self.parsed_count += passive_count
+                        logging.debug(
+                            f"{self.name} | {self.parsed_count}/{self.completed_count} | {self.full}"
+                        )
+                        return True
+                    except KeyError as e:
+                        raise KeyError(
+                            f"Quest '{quest_name}' was not defined or already parsed."
+                        ) from e
+            else:
+                # TODO: Make this a parsing exception?
+                # Delta mismatch
+                pass
+        return False
+
+
+class PassivesListContainer:
+    FASTMATCH = "quests"
+
+    def __init__(self):
+        self.container = []
+        self.parsed_count = 0
+
+    @property
+    def _is_open(self):
+        return self.container and not self._last.full
+
+    @property
+    def _last(self):
+        if self.container:
+            return self.container[-1]
+
+    def parse(self, logline):
+        if self._is_open:
+            # logging.debug(f"OPEN: {logline}")
+            self._last.try_append(logline)
+        if self.FASTMATCH in logline:
+            # logging.debug(f"Passed FASTMATCH: {logline}")
+            try:
+                n = PassivesList(logline)
+                self.container.append(n)
+            except ValueError as e:
+                # Line impersonated /passives starting flow
+                pass
+        if self.parsed_count <= 10:
+            logging.info(f"Starting log: {logline}")
+
+        self.parsed_count += 1
+
+    def print(self, index=-1):
+        if not self.container:
+            raise Exception("No data loaded")
+        c = self.container[index]
+        if not c:
+            raise KeyError("Invalid passives index (check skip value?)")
+        c = self.container[index]
+        if self.validate(index):
+            if not c.uncompleted_quests or not len(c.uncompleted_quests):
+                print("PASSIVES GET!")
+            else:
+                uq = c.uncompleted_quests
+                if "End to Hunger" in uq.keys():
+                    missing = 1 + len(uq.keys())
+                else:
+                    missing = len(uq.keys())
+                print(f"Missing passives ({missing}):")
+                for key, value in c.uncompleted_quests.items():
+                    print("{: <34}{}".format(key, value))
+        else:
+            logging.info(f"UnFound: {self.container[index].uncompleted_quests.items()}")
+            logging.info(f"Found: {self.container[index].completed_quests.items()}")
+            raise ValueError(
+                f"Parser didn't find all of the quests for /passive call {self.container[index].name}!"
+            )
+
+    def printall(self):
+        for i in range(0, len(self.container)):
+            self.print(i)
+            print("-----")
+
+    def validate(self, index):
+        return self.container[index].full
+
+
+def main():
+    parser = ArgumentParser(
+        description="reveal which passives are missing",
+    )
+    parser.add_argument(
+        "--file",
+        "-f",
+        default="/mnt/c/Program Files (x86)/Grinding Gear Games/Path of Exile/logs/Client.txt",
+        dest="client_file",
+        help="path to Client.txt",
+    )
+    parser.add_argument(
+        "--bytes",
+        type=int,
+        default=2000000,
+        help="number of bytes from the end of Client.txt to check",
+    )
+    parser.add_argument(
+        "--skip",
+        type=int,
+        default=None,
+        help="for debugging only: view data for a /passives call other than the most recent",
+    )
+    parser.add_argument(
+        "--debug",
+        "-d",
+        action="count",
+        default=0,
+        help="can be specified multiple times to increase log level",
+    )
+    args = parser.parse_args()
+    if args.bytes < 1:
+        raise ArgumentError("--line count must be greater than 1")
+
+    if args.debug:
+        logging.getLogger().setLevel(logging.DEBUG)
+
+    # Create datasets
+    passives = PassivesListContainer()
+    with open(args.client_file, "r") as f:
+        # Client.txt stores the whole world, only parse the last ~1MB
+        size = fstat(f.fileno()).st_size
+        if size > args.bytes:
+            f.seek(size - args.bytes)
+        f.readline()  # discard the partial line
+
+        for line in f:
+            passives.parse(line)
+
+    # Print dataset
+    if not args.skip:
+        passives.print()
+    elif args.skip == -1:
+        passives.printall()
+    else:
+        passives.print(-1 - args.skip)
+
+
+if __name__ == "__main__":
+    main()