Quick and dirty initial commit to share progress

.gitignore

@@ -0,0 +1,4 @@
+__pycache__/
+*.pyc
+*.egg-info/
+/venv/

README.md

@@ -0,0 +1,8 @@
+Quick & dirty commit of current progress to share with others.
+
+```
+python -m venv venv
+source venv/bin/activate
+pip install --editable .[dev]
+d2dump /path/to/stash.d2i
+```

d2warehouse/__init__.py

@@ -0,0 +1 @@
+__version__ = "0.1.0"

d2warehouse/dump.py

@@ -0,0 +1,34 @@
+import argparse
+from pathlib import Path
+from d2warehouse.parser import parse_stash
+
+from bitarray import bitarray
+
+
+def txtbits(bits: bitarray) -> str:
+    txt = "".join(str(b) for b in bits)
+    grouped = [txt[i : i + 8] for i in range(0, len(txt), 8)]
+    return " ".join(grouped)
+
+
+def read_stash(path):
+    with path.open("rb") as f:
+        stash = parse_stash(f.read())
+
+    print(f"Stash has {len(stash.tabs)} tabs")
+    for i, tab in enumerate(stash.tabs):
+        print(f" - tab {i}")
+        print(f"    - {len(tab.items)} items")
+        print(f"    - {tab.item_data.hex()}")
+        for j, item in enumerate(tab.items):
+            item.print()
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        prog="d2dump",
+        description="dump a text description of the items in d2r shared stash files",
+    )
+    parser.add_argument("stashfile", type=Path)
+    args = parser.parse_args()
+    read_stash(args.stashfile)

d2warehouse/huffman.py

@@ -0,0 +1,57 @@
+from bitarray import bitarray, decodetree
+import itertools
+
+code = {
+    "a": bitarray("11110"),
+    "b": bitarray("0101"),
+    "c": bitarray("01000"),
+    "d": bitarray("110001"),
+    "e": bitarray("110000"),
+    "f": bitarray("010011"),
+    "g": bitarray("11010"),
+    "h": bitarray("00011"),
+    "i": bitarray("1111110"),
+    "j": bitarray("000101110"),
+    "k": bitarray("010010"),
+    "l": bitarray("11101"),
+    "m": bitarray("01101"),
+    "n": bitarray("001101"),
+    "o": bitarray("1111111"),
+    "p": bitarray("11001"),
+    "q": bitarray("11011001"),
+    "r": bitarray("11100"),
+    "s": bitarray("0010"),
+    "t": bitarray("01100"),
+    "u": bitarray("00001"),
+    "v": bitarray("1101110"),
+    "w": bitarray("00000"),
+    "x": bitarray("00111"),
+    "y": bitarray("0001010"),
+    "z": bitarray("11011000"),
+    "0": bitarray("11111011"),
+    "1": bitarray("1111100"),
+    "2": bitarray("001100"),
+    "3": bitarray("1101101"),
+    "4": bitarray("11111010"),
+    "5": bitarray("00010110"),
+    "6": bitarray("1101111"),
+    "7": bitarray("01111"),
+    "8": bitarray("000100"),
+    "9": bitarray("01110"),
+    " ": bitarray("10"),
+}
+
+decode_tree = decodetree(code)
+
+
+def decode(bits: bitarray, n) -> tuple[str, int]:
+    s = "".join(itertools.islice(bits.iterdecode(decode_tree), n))
+    l = len(encode(s))
+    return s, l
+
+
+def encode(s: str) -> bitarray:
+    bits = bitarray(endian="little")
+    bits.encode(code, s)
+    bits.encode(code, " ")
+    return bits

d2warehouse/item.py

@@ -0,0 +1,67 @@
+from bitarray import bitarray
+from dataclasses import dataclass
+from enum import Enum
+
+
+class Quality(Enum):
+    LOW = 1
+    NORMAL = 2
+    HIGH = 3
+    MAGIC = 4
+    SET = 5
+    RARE = 6
+    UNIQUE = 7
+    CRAFTED = 8
+
+    def __str__(self) -> str:
+        return self.name.capitalize()
+
+
+def txtbits(bits: bitarray) -> str:
+    txt = "".join(str(b) for b in bits)
+    grouped = [txt[i : i + 8] for i in range(0, len(txt), 8)]
+    return " ".join(grouped)
+
+
+@dataclass
+class Item:
+    raw_data: bytes
+    is_identified: bool
+    is_socketed: bool
+    is_beginner: bool
+    is_simple: bool
+    is_ethereal: bool
+    is_personalized: bool
+    is_runeword: bool
+    pos_x: int
+    pos_y: int
+    kind: str
+    uid: int | None
+    lvl: int | None
+    quality: Quality | None
+    graphic: int | None
+    inherent: int | None
+
+    def print(self, indent=5):
+        properties = []
+        print(" " * indent, self.kind)
+        if self.lvl:
+            print(" " * indent, f"ilvl {self.lvl}")
+        if self.is_simple:
+            properties.append("Simple")
+        else:
+            properties.append("Extended")
+        if self.is_ethereal:
+            properties.append("Ethereal")
+        if not self.is_identified:
+            properties.append("Unidentified")
+        if self.is_socketed:
+            properties.append("Socketed")
+        if properties:
+            print(" " * indent, ", ".join(properties))
+        print(" " * indent, f"at {self.pos_x}, {self.pos_y}")
+        bits = bitarray(endian="little")
+        bits.frombytes(self.raw_data)
+        print(" " * indent, txtbits(bits))
+        print("")
+        print("")

d2warehouse/parser.py

@@ -0,0 +1,200 @@
+import struct
+from bitarray import bitarray
+from bitarray.util import ba2int
+from d2warehouse.stash import Stash, StashTab
+from d2warehouse.item import Item, Quality
+import d2warehouse.huffman as huffman
+
+STASH_TAB_MAGIC = b"\x55\xAA\x55\xAA"
+ITEM_DATA_MAGIC = b"JM"
+
+
+class ParseError(RuntimeError):
+    pass
+
+
+class UnsupportedItemError(ParseError):
+    pass
+
+
+def ensure_length(data, n):
+    if len(data) < n:
+        nearby = data[:10].hex()
+        raise ParseError(
+            f"Expected {n} bytes but only {len(data)} bytes are available near {nearby}"
+        )
+
+
+def parse_fixed(data: bytes, prefix: bytes) -> bytes:
+    if data.startswith(prefix):
+        return data[len(prefix) :]
+    raise ParseError(f"Expected {prefix} near {data[:10].hex()}")
+
+
+def parse_bytes(data: bytes, n: int) -> tuple[bytes, bytes]:
+    ensure_length(data, n)
+    return data[n:], data[:n]
+
+
+def parse_u32(data: bytes) -> tuple[bytes, int]:
+    ensure_length(data, 4)
+    return data[4:], struct.unpack("<I", data[:4])[0]
+
+
+def parse_u16(data: bytes) -> tuple[bytes, int]:
+    ensure_length(data, 2)
+    return data[2:], struct.unpack("<H", data[:2])[0]
+
+
+def parse_stash(data) -> Stash:
+    stash = Stash()
+    while len(data) > 0:
+        data, tab = parse_stash_tab(data)
+        stash.tabs.append(tab)
+    return stash
+
+
+def parse_stash_tab(data: bytes) -> tuple[bytes, StashTab]:
+    data = parse_fixed(data, STASH_TAB_MAGIC)
+    data, unknown = parse_u32(data)
+    data, version = parse_u32(data)
+
+    if unknown != 1:
+        ParseError("Unknown stash tab field is not 1")
+    if version != 99:
+        ParseError(f"Unsupported stash tab version ({version} instead of 99)")
+
+    tab = StashTab()
+    data, gold = parse_u32(data)
+    tab.gold = gold
+
+    # Length is the total length of the tab data in bytes. This includes all
+    # fields that are already parsed above.
+    data, length = parse_u32(data)
+
+    # Skip what is probably zero padding (?) but check just in case it contains data
+    data, pad = parse_bytes(data, 44)
+    if pad != b"\x00" * 44:
+        raise ParseError(f"Unexpecteed data in zero padding: {pad}")
+
+    # Separate out the item data so we can easilly verify items parsed the correct
+    # number of bytes
+    data, item_data = parse_bytes(data, length - 64)
+
+    tab.item_data = item_data
+    tab.items = parse_items(item_data)
+
+    return data, tab
+
+
+def parse_item(data: bytes) -> tuple[bytes, Item]:
+    bits = bitarray(endian="little")
+    bits.frombytes(data)
+
+    is_identified = bool(bits[4])
+    is_socketed = bool(bits[11])
+    is_ear = bool(bits[16])
+    is_beginner = bool(bits[17])
+    is_simple = bool(bits[21])
+    is_ethereal = bool(bits[22])
+    is_personalized = bool(bits[24])
+    is_runeword = bool(bits[26])
+    pos_x = ba2int(bits[42:46])
+    pos_y = ba2int(bits[46:50])
+    kind, kind_end = huffman.decode(bits[53:], 3)
+    kind_end += 53
+    # TODO: verify that this socket thing is really 1 bit for simple items...?
+    sockets_end = kind_end + 1 if is_simple else kind_end + 3
+    sockets_count = ba2int(bits[kind_end:sockets_end])
+    print("sockets", sockets_count)
+    if is_ear:
+        raise UnsupportedItemError("Ear items are not supported")
+
+    uid, lvl, quality, graphic, inherent = None, None, None, None, None
+    if not is_simple:
+        uid, lvl, quality = parse_extended_item(bits[sockets_end:])
+
+        extended_end = sockets_end + 32 + 7 + 4
+
+        graphic, graphic_end = parse_item_graphic(bits[extended_end:])
+        graphic_end += extended_end
+
+        inherent, inherent_end = parse_inherent_mod(bits[graphic_end:])
+        inherent_end += graphic_end
+
+    item = Item(
+        data[:],  # TODO: only take parsed bytes
+        is_identified,
+        is_socketed,
+        is_beginner,
+        is_simple,
+        is_ethereal,
+        is_personalized,
+        is_runeword,
+        pos_x,
+        pos_y,
+        kind,
+        uid,
+        lvl,
+        quality,
+        graphic,
+        inherent,
+    )
+    return b"", item  # TODO: properly return remaining data
+
+
+def parse_item_graphic(bits: bitarray) -> tuple[int | None, int]:
+    if bits[0]:
+        return None, 1
+    else:
+        return ba2int(bits[1:4]), 4
+
+
+def parse_inherent_mod(bits: bitarray) -> tuple[int | None, int]:
+    if bits[0]:
+        return None, 1
+    else:
+        return ba2int(bits[1:4]), 4
+
+
+def parse_extended_item(bits: bitarray) -> tuple[int, int, Quality]:
+    uid = ba2int(bits[:32])
+    lvl = ba2int(bits[32:39])
+    quality = ba2int(bits[39:43])
+    return uid, lvl, Quality(quality)
+
+
+# TODO: figure out what we want to return here
+# TODO: Introduce enums for quality
+def parse_quality_data(bits: bitarray, quality: int):
+    if quality == 1:
+        return parse_low_quality_data(bits)
+    elif quality == 2:
+        return None, 0
+    elif quality == 3:
+        return parse_high_quality_data(bits)
+    elif quality == 4:
+        return parse_magic_data(bits)
+    elif quality == 5:
+        return parse_set_data(bits)
+    elif quality == 6:
+        return parse_rare_data(bits)
+    elif quality == 7:
+        return parse_unique_data(bits)
+    elif quality == 8:
+        return parse_crafted_data(bits)
+        # magic
+
+
+def parse_items(data: bytes) -> list[Item]:
+    data = parse_fixed(data, ITEM_DATA_MAGIC)
+    data, num = parse_u16(data)
+
+    items: list[Item] = []
+    while data:
+        data, item = parse_item(data)
+        items.append(item)
+
+    # TODO: check if num == len(items)
+
+    return items

d2warehouse/stash.py

@@ -0,0 +1,13 @@
+from bitarray import bitarray
+
+
+class Stash:
+    def __init__(self) -> None:
+        self.tabs: list[StashTab] = []
+
+
+class StashTab:
+    def __init__(self) -> None:
+        self.gold: int = 0
+        self.item_data: bytes = b""
+        self.items: list[Item] = []

d2warehouse/test.py

@@ -0,0 +1,56 @@
+from bitarray import bitarray
+from d2warehouse.parser import parse_item
+import d2warehouse.huffman as huffman
+
+test_items = [
+    (
+        "Leather armor (16 armor, 24/24 durability)",
+        "102080000500f40e2f087bf2b426041ac0c0f01f",
+    ),
+    (
+        "Ethereal Leather Armor (24 armor, 13/13 durability)",
+        "1000c0000500f40e2f9c8627a70404226868f01f",
+    ),
+    (
+        "Unidentified Ethereal Magic Leather armor (27 armor, 13/13 dura)",
+        "0000c0000500f40e2f08a37fd61388470040091a1a4088f01f",
+    ),
+    (
+        "^ The same item but identified (+17% ED, Sturdy prefix",
+        "1000c0000500f40e2f08a37fd61388470040091a1a4088f01f",
+    ),
+    ("minor healing potion", "1000a008050014cf4f00"),
+    ("beginner minor healing potion", "1020a200050014cf4f00"),
+    ("minor mana potion", "1000a0000500d4ce4f00"),
+    ("minor rejuvenation potion", "1000a0000500f4ec28"),
+    ("stamina potion", "1000a0000500743729"),
+    ("antidote potion", "1000a0000500143529"),
+    ("thawing potion", "1000a000050014580a"),
+    ("beginner scroll of identify", "1000a2000500f44722"),
+    ("scroll of identify", "1000a0000500f44722"),
+]
+
+
+def txtbits(bits: bitarray) -> str:
+    txt = "".join(str(b) for b in bits)
+    grouped = [txt[i : i + 8] for i in range(0, len(txt), 8)]
+    return " ".join(grouped)
+
+
+def main():
+    for descr, hex in test_items:
+        print(descr)
+        print(hex)
+        raw = bytes.fromhex(hex)
+        bits = bitarray(endian="little")
+        bits.frombytes(raw)
+        print(txtbits(bits))
+        _, item = parse_item(raw)
+        print(f"kind: {item.kind}")
+        print(f"simple: {item.is_simple}")
+        print(f"identified: {item.is_identified}")
+        print(f"beginner: {item.is_beginner}")
+        print("--")
+    print(huffman.encode("hp1"))
+    print(huffman.encode("mp1"))
+    print(huffman.encode("rvs"))

pyproject.toml

@@ -0,0 +1,40 @@
+[build-system]
+requires = ["setuptools"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "d2warehouse"
+authors = [
+    {name = "omicron"},
+]
+description = "personal notes keeping"
+classifiers = [
+    "Development Status :: 3 - Alpha",
+    "Environment :: Console",
+
+]
+requires-python = ">=3.9"
+license = {text = "GPLv3 License"}
+dependencies = [
+    "PyYAML",
+    "bitarray",
+]
+dynamic = ["version"]
+
+[project.optional-dependencies]
+dev = [
+    "coverage",
+    "pytest",
+    "pytest-cov",
+    "mypy",
+    "types-PyYAML",
+    "flake8",
+    "black",
+]
+
+[project.scripts]
+d2dump = "d2warehouse.dump:main"
+d2test = "d2warehouse.test:main"
+
+[tool.setuptools.dynamic]
+version = {attr = "d2warehouse.__version__"}