From 3c8834e98f727be0215bacee7c684dc116ae9b01 Mon Sep 17 00:00:00 2001 From: goober Date: Fri, 21 Aug 2026 11:38:27 +0300 Subject: [PATCH] feat: capture and export gear, cartridges and loadouts Adds a third export alongside pull history and achievements: the cartridges, modules, character levels and equipped console boards the game sends at login. Shaped as added files plus a handful of one-line calls, so this repo can keep merging upstream: - decoder/gear.py: self-contained bit-level decoder over a list of datagrams. - export/gear_export.py: the nte-gear-export envelope. - live_capture/gear_collector.py: UDP flow accumulation, the mid-capture decode, the fallback scan, the console line and the export file. Gear is reported during the capture rather than only at the stop. The burst arrives in a rush and then stops, so "this flow went quiet" is the only mid-stream marker available; a flow is retried only once it has grown, since decoding per packet would be quadratic in flow size. Flows are tried largest first and the first one that decodes wins - picking the single biggest was wrong, history traffic can outweigh a short login burst. Verified against a live capture: 817 items (310 cartridges, 507 modules), 20 characters, 13 loadouts, every bucket matching the in-game counts exactly. 26 synthetic tests, no real capture used as a fixture; 149 total. Co-Authored-By: Claude Opus 5 --- pyproject.toml | 2 +- src/nte_history_exporter/__init__.py | 2 +- src/nte_history_exporter/console.py | 4 + src/nte_history_exporter/constants.py | 2 +- src/nte_history_exporter/decoder/gear.py | 499 ++++++++++++ .../export/gear_export.py | 99 +++ .../live_capture/gear_collector.py | 191 +++++ .../live_capture/runner.py | 25 +- tests/test_export_contract.py | 2 +- tests/test_gear_decoding.py | 747 ++++++++++++++++++ 10 files changed, 1568 insertions(+), 5 deletions(-) create mode 100644 src/nte_history_exporter/decoder/gear.py create mode 100644 src/nte_history_exporter/export/gear_export.py create mode 100644 src/nte_history_exporter/live_capture/gear_collector.py create mode 100644 tests/test_gear_decoding.py diff --git a/pyproject.toml b/pyproject.toml index 47b3bb8..81e3d68 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "nte-history-exporter" -version = "0.3.0" +version = "0.4.0" description = "Cross-platform Neverness to Everness pull-history and achievement exporter." requires-python = ">=3.10" dependencies = [] diff --git a/src/nte_history_exporter/__init__.py b/src/nte_history_exporter/__init__.py index 493f741..6a9beea 100644 --- a/src/nte_history_exporter/__init__.py +++ b/src/nte_history_exporter/__init__.py @@ -1 +1 @@ -__version__ = "0.3.0" +__version__ = "0.4.0" diff --git a/src/nte_history_exporter/console.py b/src/nte_history_exporter/console.py index 0fc8950..15a3dc4 100644 --- a/src/nte_history_exporter/console.py +++ b/src/nte_history_exporter/console.py @@ -97,6 +97,10 @@ def print_live_instructions(local_ip: str, backend: str = "windows_raw", detail: print(" Your tracked achievements are captured automatically when") print(" you log in after following the step above.") print() + print(style(" Gear", BOLD)) + print(" Your cartridges, modules and equipped loadouts are captured") + print(" automatically at login, in the same step as achievements.") + print() print(style(" Ready - use the game, then press any key here when finished.", BOLD, GREEN)) print(rule()) diff --git a/src/nte_history_exporter/constants.py b/src/nte_history_exporter/constants.py index 7f7d059..6fe8b4e 100644 --- a/src/nte_history_exporter/constants.py +++ b/src/nte_history_exporter/constants.py @@ -13,7 +13,7 @@ ARC_BANNER_NAME = "Arc Miracle Box" MYSTERY_BOX_BANNER_ID = "Gashapon_MysteryBox" MYSTERY_BOX_BANNER_NAME = "Mystery Box" EXPORTER_NAME = "nte-history-exporter" -EXPORTER_VERSION = "0.3.0" +EXPORTER_VERSION = "0.4.0" HISTORY_REQUEST_BANNER = 4220 HISTORY_REQUEST_LENGTH = 45 diff --git a/src/nte_history_exporter/decoder/gear.py b/src/nte_history_exporter/decoder/gear.py new file mode 100644 index 0000000..87a1ef6 --- /dev/null +++ b/src/nte_history_exporter/decoder/gear.py @@ -0,0 +1,499 @@ +"""Decode cartridges, modules and character loadouts from the gameplay UDP flow. + +The game speaks Unreal's bit-packed net protocol. Each datagram is a bit stream: +a 15-byte packet and bunch header, the payload, then two 1 bits - the packet +handler's terminator and the bit writer's - zero-padded to the next byte +boundary. Records routinely straddle datagram boundaries, so the payload bit +ranges have to be spliced; concatenating whole datagrams splices the framing into +the middle of a record and corrupts it. + +Inside the spliced stream, an item record reads: + + length == len(id) + 1 (self-validating frame check) + \0 + <5 bytes> <8-byte instance id> ... + ... <12-byte owner reference at +312 bits, all zero when unequipped> ... + then N stat entries, each: + \0 <4 zero bytes> + +Consecutive entries drift one bit apart, so entries are chained forward from the +previous one rather than found by global search - that way a neighbouring +record's entries can never be stolen. + +A character block is marked by its `GA__Melee|Skill|UltraSkill|QTE` ability +strings, with the character's level and breakthrough count at fixed bit offsets +before it, and its equipped modules listed afterwards one entry per occupied +grid cell. + +Every failure path returns empty rather than raising. +""" + +from __future__ import annotations + +import re +import struct +from dataclasses import dataclass, field + +# --- framing --------------------------------------------------------------- + +HEADER_BITS = 120 +# Datagrams with no room for a payload are pure acks and carry nothing. +MIN_PAYLOAD_BITS = 32 + +# --- record layout --------------------------------------------------------- + +LEVEL_OFFSET_BYTES = 38 +MAX_ITEM_LEVEL = 20 +OWNER_OFFSET_BITS = 312 +OWNER_BYTES = 12 +# An item packs its entries tightly; anything further away belongs to something +# else, so a wider scan would wander into unrelated data. +CHAIN_SCAN_BITS = 1400 +# Observed entry spacing is (len+12) bytes plus a 1-bit separator, tightest at +# ~145 bits. Advancing by a floor below that and rescanning avoids overshooting +# entries that are shorter than their computed width - main stats run two bytes +# shorter than substats. +MIN_ENTRY_SPACING_BITS = 100 +MODULE_ENTRIES = 6 +CARTRIDGE_ENTRIES = 5 + +# --- character layout ------------------------------------------------------ + +# Bit offsets relative to the start of the character's first ability frame. +CHARACTER_MARKER_OFFSET_BITS = -466 +CHARACTER_LEVEL_OFFSET_BITS = -434 +CHARACTER_BREAKTHROUGH_OFFSET_BITS = -402 +CHARACTER_MARKER = 67806 +MAX_CHARACTER_LEVEL = 90 +MAX_BREAKTHROUGHS = 10 +# A loadout cell entry stores doubled grid coordinates after the item id. +CELL_ROW_OFFSET_BITS = 96 +CELL_COL_OFFSET_BITS = 128 + +# --- vocabulary ------------------------------------------------------------ + +# Ids are whitelisted: the length prefix validates a frame but cannot catch a +# single flipped bit inside the name - `Ingantation` and `GetEfviciency` both +# passed the length check before the whitelist went in. +ITEM_ID = re.compile( + rb"(cell[234]_style[1-6]_\d_(?:Orange|Purple|Blue)" + # Psychically must precede Psyche: both ids exist and name different sets. + rb"|(?:Attack|Chaos|Cosmos|GetEfficiency|Heal|Incantation|Lakshana|Mag|Nature" + rb"|Psychically|Psyche|Shield)_(?:orange|purple|blue))\x00" +) +ABILITY_ID = re.compile(rb"GA_([A-Za-z0-9]+)_(?:Melee|Skill|UltraSkill|QTE)\x00") + +STAT_NAMES = frozenset({ + "AtkAdd", "AtkUp", "DefAdd", "DefUp", "HPMaxAdd", "HPMaxUp", "CritBase", + "CritDamageBase", "MagBase", "UnbalIntensityBase", "HealUp", +}) + +# Substat values are fixed per stat and item size, with no roll variance, so any +# value off this table means the parse went wrong. Main stats read 0.0. +CANONICAL_VALUES = frozenset({ + 0.0, 200.0, 300.0, 400.0, 1000.0, 16.0, 24.0, 32.0, 80.0, 48.0, 64.0, 600.0, 800.0, + 12.0, 18.0, 60.0, 36.0, + 0.025, 0.0375, 0.05, 0.125, 0.035, 0.0525, 0.07, 0.075, 0.175, 0.02, 0.03, 0.04, + 0.1, 0.06, 0.08, 0.2, 0.105, 0.14, +}) + +# A main stat's displayed value is never transmitted: it is base * (1 + level/5), +# so +0 shows the base and +20 shows five times it. +CARTRIDGE_MAIN_BASE = { + "HPMaxUp": 0.075, "AtkUp": 0.075, "DefUp": 0.105, "CritBase": 0.06, + "CritDamageBase": 0.12, "MagBase": 36.0, "UnbalIntensityBase": 36.0, + "DamageUpGeneralBase": 0.075, + # Healing Bonus reads 6.90% at +0 in game - the one main that is not round. + "HealUp": 0.069, +} +# Module mains scale per cell: 56 HP and 4.2 ATK each at +0. Measured in game as +# Type III 168/12.6 -> 840/63 and Type IV 224/16.8 -> 1120/84. The client floors +# flat values, which is why ATK displays as 12 and 16 at +0. +MODULE_HP_PER_CELL = 56.0 +MODULE_ATK_PER_CELL = 4.2 + +# Confirmed against everness's cartridge boxes, which name all twelve sets. The +# three the capture has never carried are Psyche, Shield and Heal - NOT Blood, +# Night and Kingdom, which were a guess and would have decoded as nothing. +SET_NAMES = { + "Attack": "Shadow Creed", "Chaos": "Diabolos", "Cosmos": "Lost Radiance", + "GetEfficiency": "Speedy Hedgehog", "Incantation": "Crimson: Twin Butterflies", + "Lakshana": "Street Boxer", "Nature": "Fireflies and the Forest", + "Psychically": "Quiet Manor", "Mag": "Tiny Big Adventure", + "Psyche": "Devil's Blood: Curse", "Heal": "Thea's Night Tavern", + "Shield": "Kingdom's Guard", +} + +MODULE_TYPE_NAMES = {2: "II", 3: "III", 4: "IV"} + + +@dataclass(frozen=True) +class GearItem: + instance: str + kind: str + item_id: str + level: int + rarity: str + main_stats: tuple[tuple[str, float | None], ...] + substats: tuple[tuple[str, float], ...] + owner_group: str | None + shape: str | None = None + module_type: str | None = None + set_name: str | None = None + + +@dataclass(frozen=True) +class GearCharacter: + key: str + level: int | None + breakthroughs: int | None + board: tuple[tuple[int, int], ...] = () + + +@dataclass(frozen=True) +class GearSnapshot: + items: tuple[GearItem, ...] = () + characters: tuple[GearCharacter, ...] = () + warnings: tuple[str, ...] = field(default=()) + + @property + def cartridges(self) -> int: + return sum(item.kind == "cartridge" for item in self.items) + + @property + def modules(self) -> int: + return sum(item.kind == "module" for item in self.items) + + @property + def characters_with_loadouts(self) -> int: + return sum(bool(character.board) for character in self.characters) + + +def is_stat_name(name: str) -> bool: + return name in STAT_NAMES or (name.startswith("DamageUp") and name.endswith("Base")) + + +def bitshift(buf: bytes, shift: int) -> bytes: + if shift == 0: + return buf + return bytes( + ((buf[i] >> shift) | (buf[i + 1] << (8 - shift))) & 0xFF + for i in range(len(buf) - 1) + ) + + +class _BitAccumulator: + """Append arbitrary bit ranges; the result is a byte buffer, LSB-first.""" + + def __init__(self) -> None: + self.buf = bytearray() + self.nbits = 0 + + def append(self, data: bytes, start: int, nbits: int) -> None: + if nbits <= 0: + return + value = (int.from_bytes(data, "little") >> start) & ((1 << nbits) - 1) + offset = self.nbits % 8 + if offset: + value <<= offset + raw = value.to_bytes((offset + nbits + 7) // 8, "little") + self.buf[-1] |= raw[0] + self.buf.extend(raw[1:]) + else: + self.buf.extend(value.to_bytes((nbits + 7) // 8, "little")) + self.nbits += nbits + del self.buf[(self.nbits + 7) // 8 :] + + +def payload_end_bit(datagram: bytes) -> int: + """First bit past the payload. + + Unreal ends a packet with two 1 bits then zero-pads to a byte boundary, so + the payload stops one bit below the highest set bit of the last non-zero + byte. + """ + + i = len(datagram) - 1 + while i >= 0 and datagram[i] == 0: + i -= 1 + if i < 0: + return 0 + return i * 8 + datagram[i].bit_length() - 2 + + +def splice_datagrams(datagrams) -> bytes: + """Concatenate every datagram's payload bits with the framing removed.""" + + accumulator = _BitAccumulator() + for datagram in datagrams: + end = payload_end_bit(datagram) + if end - HEADER_BITS >= MIN_PAYLOAD_BITS: + accumulator.append(datagram, HEADER_BITS, end - HEADER_BITS) + return bytes(accumulator.buf) + + +class _BitView: + """Random access to a buffer at arbitrary bit offsets.""" + + def __init__(self, buf: bytes) -> None: + self.views = [bitshift(buf, shift) for shift in range(8)] + + def at(self, bitpos: int) -> tuple[bytes, int]: + return self.views[bitpos % 8], bitpos // 8 + + def u32(self, bitpos: int) -> int | None: + view, offset = self.at(bitpos) + if offset + 4 > len(view): + return None + return struct.unpack_from(" float | None: + view, offset = self.at(bitpos) + if offset + 4 > len(view): + return None + return struct.unpack_from(" bytes: + view, offset = self.at(bitpos) + return view[offset : offset + length] + + def text(self, bitpos: int, length: int) -> str | None: + view, offset = self.at(bitpos) + if offset + length > len(view): + return None + raw = view[offset : offset + length] + if not raw.endswith(b"\0"): + return None + try: + text = raw[:-1].decode("ascii") + except UnicodeDecodeError: + return None + return text if text.isprintable() else None + + +def _validated_frames(view: _BitView, pattern): + """Every self-validating frame, in bit order.""" + + found = [] + for shift in range(8): + haystack = view.views[shift] + for match in pattern.finditer(haystack): + body = match.group() + if match.start() < 4: + continue + if struct.unpack_from(" tuple[str, float] | None: + """Read one length-prefixed stat entry.""" + + declared = view.u32(bitpos) + if declared is None or not 4 <= declared <= 64: + return None + name = view.text(bitpos + 32, declared) + if name is None: + return None + # Layout: <4 zero bytes> + value = view.f32(bitpos + 32 + declared * 8 + 32) + if value is None: + return None + return name, round(value, 6) + + +def _chain_entries(view: _BitView, start_bit: int, needed: int): + """Walk `needed` stat entries forward from just past the item id.""" + + cursor, entries = start_bit, [] + while len(entries) < needed: + found = None + for delta in range(CHAIN_SCAN_BITS): + probe = read_entry(view, cursor + delta) + if probe and is_stat_name(probe[0]): + found = (probe, cursor + delta) + break + if not found: + return entries + entries.append(found[0]) + cursor = found[1] + MIN_ENTRY_SPACING_BITS + return entries + + +def _main_value(kind: str, stat: str, level: int, cells: int | None) -> float | None: + if kind == "module": + base = { + "HPMaxAdd": MODULE_HP_PER_CELL * (cells or 0), + "AtkAdd": MODULE_ATK_PER_CELL * (cells or 0), + }.get(stat) + else: + base = CARTRIDGE_MAIN_BASE.get(stat) + # Every elemental damage bonus shares the general bonus's base. + if base is None and stat.startswith("DamageUp") and stat.endswith("Base"): + base = CARTRIDGE_MAIN_BASE["DamageUpGeneralBase"] + if base is None: + return None + value = base * (1 + level / 5) + # Flat module stats display floored; percentages keep their precision. + return float(int(value)) if kind == "module" else round(value, 4) + + +def _item_from_frame(view: _BitView, frame) -> GearItem | None: + bitpos, match, shift, id_end_bit = frame + item_id = match.group()[:-1].decode() + is_module = item_id.startswith("cell") + needed = MODULE_ENTRIES if is_module else CARTRIDGE_ENTRIES + main_count = 2 if is_module else 1 + + entries = _chain_entries(view, id_end_bit, needed) + if len(entries) != needed: + return None + # Substats never vary, so an off-table value means the chain went astray. + if not all(value in CANONICAL_VALUES for _, value in entries): + return None + # In every record confirmed against in-game values the main stats come + # first. A main appearing later means the chain wrapped into a neighbour. + if not all(value == 0.0 for _, value in entries[:main_count]): + return None + if sum(value == 0.0 for _, value in entries) != main_count: + return None + + id_end_byte = match.end() + haystack = view.views[shift] + instance = haystack[id_end_byte + 5 : id_end_byte + 13] + if len(instance) != 8: + return None + + level_offset = id_end_byte + LEVEL_OFFSET_BYTES + if level_offset + 4 > len(haystack): + return None + level = struct.unpack_from(" MAX_ITEM_LEVEL: + level = 0 + + owner_bit = id_end_bit + OWNER_OFFSET_BITS + # Gate on the leading word: trailing bytes of the field carry unrelated data + # on some records, so testing the whole span over-reports. + owner = view.raw(owner_bit, OWNER_BYTES) if view.u32(owner_bit) else b"" + + cells = int(item_id[4]) if is_module else None + mains = tuple( + (name, _main_value("module" if is_module else "cartridge", name, level, cells)) + for name, value in entries if value == 0.0 + ) + substats = tuple((name, value) for name, value in entries if value != 0.0) + rarity = item_id.rsplit("_", 1)[1].lower() + + if is_module: + shape = "_".join(item_id.split("_")[:2]) + return GearItem( + instance=instance.hex(), kind="module", item_id=item_id, level=level, + rarity=rarity, main_stats=mains, substats=substats, + owner_group=owner.hex() or None, + shape=shape, module_type=MODULE_TYPE_NAMES.get(cells or 0), + ) + return GearItem( + instance=instance.hex(), kind="cartridge", item_id=item_id, level=level, + rarity=rarity, main_stats=mains, substats=substats, + owner_group=owner.hex() or None, + set_name=SET_NAMES.get(item_id.rsplit("_", 1)[0]), + ) + + +def _character_blocks(view: _BitView) -> list[tuple[int, str]]: + """The first ability frame of each character, in stream order.""" + + blocks: list[tuple[int, str]] = [] + for bitpos, match, _shift, _end in _validated_frames(view, ABILITY_ID): + name = match.group(1).decode() + if not blocks or blocks[-1][1] != name: + blocks.append((bitpos, name)) + return blocks + + +def _character_progress(view: _BitView, block_bit: int) -> tuple[int | None, int | None]: + """Level and breakthrough count, validated against the block's marker.""" + + if view.u32(block_bit + CHARACTER_MARKER_OFFSET_BITS) != CHARACTER_MARKER: + return None, None + level = view.u32(block_bit + CHARACTER_LEVEL_OFFSET_BITS) + breakthroughs = view.u32(block_bit + CHARACTER_BREAKTHROUGH_OFFSET_BITS) + if level is None or not 1 <= level <= MAX_CHARACTER_LEVEL: + level = None + if breakthroughs is None or breakthroughs > MAX_BREAKTHROUGHS: + breakthroughs = None + return level, breakthroughs + + +def extract_gear(datagrams) -> GearSnapshot: + """Decode every item and character loadout in a captured UDP flow.""" + + try: + buf = splice_datagrams(datagrams) + except (ValueError, MemoryError): + return GearSnapshot() + if len(buf) < 64: + return GearSnapshot() + + view = _BitView(buf) + frames = _validated_frames(view, ITEM_ID) + blocks = _character_blocks(view) + + items: dict[str, GearItem] = {} + # An id frame that chains no stat entries is a loadout reference, not a + # record: the same id quoted inside a character's equipped list. + loadouts: dict[str, list[tuple[int, int, int]]] = {} + for frame in frames: + item = _item_from_frame(view, frame) + if item is not None: + items.setdefault(item.instance, item) + continue + bitpos, _match, _shift, id_end_bit = frame + row = view.u32(id_end_bit + CELL_ROW_OFFSET_BITS) + column = view.u32(id_end_bit + CELL_COL_OFFSET_BITS) + if not row or not column: + continue + owner_index = -1 + for index, (block_bit, _name) in enumerate(blocks): + if block_bit > bitpos: + break + owner_index = index + if owner_index >= 0: + # Coordinates are doubled; a cell is one square, not four. + loadouts.setdefault(blocks[owner_index][1], []).append( + (bitpos, row // 2, column // 2) + ) + + # A character's block is transmitted more than once in a login - the active + # team appears early and again in the full roster - so merge by name and + # keep the first reading that validated. + merged: dict[str, GearCharacter] = {} + for block_bit, name in blocks: + level, breakthroughs = _character_progress(view, block_bit) + board = tuple(sorted({(r, c) for _bit, r, c in loadouts.get(name, [])})) + previous = merged.get(name) + if previous is None: + merged[name] = GearCharacter(name, level, breakthroughs, board) + continue + merged[name] = GearCharacter( + key=name, + level=previous.level if previous.level is not None else level, + breakthroughs=( + previous.breakthroughs + if previous.breakthroughs is not None + else breakthroughs + ), + board=previous.board or board, + ) + + return GearSnapshot( + items=tuple(items.values()), + characters=tuple(merged.values()), + ) diff --git a/src/nte_history_exporter/export/gear_export.py b/src/nte_history_exporter/export/gear_export.py new file mode 100644 index 0000000..26bd382 --- /dev/null +++ b/src/nte_history_exporter/export/gear_export.py @@ -0,0 +1,99 @@ +"""Gear export envelope. + +Kept out of ``json_export`` so the gear feature is a set of added files rather +than edits to upstream ones - this repo tracks Golumpa's exporter and has to be +able to merge it. +""" + +from __future__ import annotations + +from typing import Any + +from nte_history_exporter import __version__ +from nte_history_exporter.constants import EXPORTER_NAME, GAME_NAME +from nte_history_exporter.decoder.gear import GearCharacter, GearItem, GearSnapshot +from nte_history_exporter.decoder.server_region import account_region_for_server + + +def build_gear_export_json( + snapshot: GearSnapshot, + *, + source: str = "packet_capture", + capture_source: str | None = None, + user_uid: str | None = None, + server_id: str | None = None, +) -> dict[str, Any]: + scan = { + "cartridges": snapshot.cartridges, + "modules": snapshot.modules, + "characters": len(snapshot.characters), + "characters_with_loadouts": snapshot.characters_with_loadouts, + "warnings": list(snapshot.warnings), + } + + export: dict[str, Any] = { + "format": "nte-gear-export", + "format_version": 1, + "game": GAME_NAME, + "source": source, + } + if capture_source: + export["capture_source"] = capture_source + export["exporter"] = {"name": EXPORTER_NAME, "version": __version__} + export["scan"] = scan + + normalized_user_uid = user_uid.strip() if user_uid else "" + normalized_server_id = str(server_id).strip() if server_id else "" + if normalized_user_uid: + export["user_uid"] = normalized_user_uid + if normalized_server_id: + export["server_id"] = normalized_server_id + account_region = account_region_for_server(normalized_server_id) + if account_region: + export["account_region"] = account_region + + export["characters"] = [ + _gear_character_for_export(character) for character in snapshot.characters + ] + export["items"] = [_gear_item_for_export(item) for item in snapshot.items] + return export + + +def _gear_character_for_export(character: GearCharacter) -> dict[str, Any]: + record: dict[str, Any] = { + "key": character.key, + "level": character.level, + "breakthroughs": character.breakthroughs, + } + # Only emitted when a loadout was actually seen; an empty board and an + # unequipped character are different states and must not collapse. + if character.board: + record["board"] = [[row, column] for row, column in character.board] + # The capture never ties an owner group to a character name, so the app + # resolves this itself rather than being handed a guess. + record["owner_group"] = None + return record + + +def _gear_item_for_export(item: GearItem) -> dict[str, Any]: + record: dict[str, Any] = { + "instance": item.instance, + "kind": item.kind, + "item_id": item.item_id, + } + if item.shape: + record["shape"] = item.shape + if item.module_type: + record["module_type"] = item.module_type + if item.set_name: + record["set"] = item.set_name + record["level"] = item.level + record["rarity"] = item.rarity + record["main_stats"] = [ + {"stat": stat, "value": value} for stat, value in item.main_stats + ] + record["substats"] = [ + {"stat": stat, "value": value} for stat, value in item.substats + ] + record["owner_group"] = item.owner_group + return record diff --git a/src/nte_history_exporter/live_capture/gear_collector.py b/src/nte_history_exporter/live_capture/gear_collector.py new file mode 100644 index 0000000..bc9bae0 --- /dev/null +++ b/src/nte_history_exporter/live_capture/gear_collector.py @@ -0,0 +1,191 @@ +"""Gear capture, held in one object so ``runner`` only has to call it. + +This repo tracks Golumpa's exporter, so the gear feature is deliberately shaped +as added files plus a handful of one-line calls in ``runner.run_live_capture`` +rather than as edits scattered through the capture loop. Everything gear needs - +flow accumulation, the mid-capture decode, the fallback scan, the console line +and the export file - lives here. +""" + +from __future__ import annotations + +import json +import time +from datetime import datetime +from pathlib import Path + +from nte_history_exporter.console import BOLD, DIM, GREEN, style +from nte_history_exporter.decoder.gear import GearSnapshot, extract_gear +from nte_history_exporter.export.gear_export import build_gear_export_json + + +Flow = tuple[str, int, str, int] + +# The login gear burst is roughly 500 KB; the cap only exists so a long +# session on a busy flow cannot grow without bound. +MAX_UDP_FLOW_BYTES = 8 * 1024 * 1024 +# Decoding is linear in flow size, so only the few busiest are worth trying. +MAX_GEAR_FLOW_ATTEMPTS = 4 +# A gear burst arrives in a rush and then stops, so "this flow has gone quiet" +# is the only mid-stream marker there is. Attempts are gated on the flow being +# large enough to hold a burst, having been idle, and having grown since it was +# last tried, so a quiet flow is never decoded twice for the same bytes. +GEAR_LIVE_MIN_FLOW_BYTES = 64 * 1024 +GEAR_LIVE_IDLE_SECONDS = 1.0 +GEAR_LIVE_MAX_ATTEMPTS = 8 + + +def print_gear_captured( + cartridges: int, + modules: int, + characters: int, + characters_with_loadouts: int, +) -> None: + print(style(" + ", GREEN, BOLD) + "Gear captured") + print( + style( + f" Inventory {cartridges} cartridges, {modules} modules", + DIM, + ) + ) + print( + style( + f" Characters {characters} seen, " + f"{characters_with_loadouts} with loadouts", + DIM, + ) + ) + + +def gear_path(user_uid: str | None = None) -> Path: + export_dir = Path("exports") + export_dir.mkdir(parents=True, exist_ok=True) + uid_prefix = _safe_filename_part(user_uid) if user_uid else "unknown" + base = export_dir / f"{uid_prefix}_Gear_{datetime.now():%Y%m%d_%H%M%S}" + path = base.with_suffix(".json") + counter = 2 + while path.exists(): + path = export_dir / f"{base.name}_{counter}.json" + counter += 1 + return path + + +def _safe_filename_part(value: str | None) -> str: + # Deliberately a copy of runner's helper rather than an import: importing + # runner from here would be circular, and the rule is four lines long. + if not value: + return "unknown" + cleaned = "".join(ch for ch in value.strip() if ch.isalnum() or ch in ("-", "_")) + return cleaned or "unknown" + + +class GearCollector: + """Accumulates inbound UDP and decodes gear out of it. + + Call ``poll`` once per capture iteration (above any ``packet is None`` + guard - the pcap read timeout is what keeps it firing after the burst goes + quiet), ``observe`` for each packet, ``finalize`` after the loop, and + ``write`` to emit the export. + """ + + def __init__(self, local_ip: str | None) -> None: + self._local_ip = local_ip + self._flows: dict[Flow, list[bytes]] = {} + self._flow_bytes: dict[Flow, int] = {} + self._flow_last_seen: dict[Flow, float] = {} + self._attempted_bytes: dict[Flow, int] = {} + self._attempts = 0 + self.snapshot: GearSnapshot | None = None + + def observe(self, packet) -> None: + """Keep an inbound UDP payload whole. + + Datagram boundaries carry the framing, so payloads are spliced at + decode time and never concatenated here. + """ + if ( + packet is None + or packet.protocol != "udp" + or not packet.payload + or packet.dst_ip != self._local_ip + ): + return + flow: Flow = (packet.src_ip, packet.src_port, packet.dst_ip, packet.dst_port) + if self._flow_bytes.get(flow, 0) + len(packet.payload) > MAX_UDP_FLOW_BYTES: + return + self._flows.setdefault(flow, []).append(packet.payload) + self._flow_bytes[flow] = self._flow_bytes.get(flow, 0) + len(packet.payload) + self._flow_last_seen[flow] = time.monotonic() + + def poll(self) -> None: + """Decode a flow that has stopped growing, and report it immediately.""" + if self.snapshot is not None or self._attempts >= GEAR_LIVE_MAX_ATTEMPTS: + return + now = time.monotonic() + candidates = [ + flow + for flow, size in self._flow_bytes.items() + if size >= GEAR_LIVE_MIN_FLOW_BYTES + and self._attempted_bytes.get(flow) != size + and now - self._flow_last_seen.get(flow, now) >= GEAR_LIVE_IDLE_SECONDS + ] + candidates.sort(key=self._flow_bytes.__getitem__, reverse=True) + + for flow in candidates[:MAX_GEAR_FLOW_ATTEMPTS]: + self._attempted_bytes[flow] = self._flow_bytes[flow] + self._attempts += 1 + snapshot = extract_gear(self._flows[flow]) + if snapshot.items: + self._report(snapshot) + return + + def finalize(self) -> None: + """Fallback scan for a flow that never idled. + + Normally ``poll`` has already decoded the burst. Rescanning per packet + instead of only here would be quadratic in flow size. + """ + if self.snapshot is not None: + return + # The gameplay flow dwarfs the others, but history traffic can outweigh + # a short login burst, so size only sets the order - the first flow that + # actually decodes wins. + busiest = sorted(self._flow_bytes, key=self._flow_bytes.__getitem__, reverse=True) + for flow in busiest[:MAX_GEAR_FLOW_ATTEMPTS]: + snapshot = extract_gear(self._flows[flow]) + if snapshot.items: + self._report(snapshot) + return + + def write( + self, + *, + capture_source: str | None = None, + user_uid: str | None = None, + server_id: str | None = None, + source: str = "live_capture", + ) -> Path | None: + if self.snapshot is None: + return None + path = gear_path(user_uid) + export = build_gear_export_json( + self.snapshot, + source=source, + capture_source=capture_source, + user_uid=user_uid, + server_id=server_id, + ) + path.write_text( + json.dumps(export, ensure_ascii=False, indent=2) + "\n", + encoding="utf-8", + ) + return path + + def _report(self, snapshot: GearSnapshot) -> None: + self.snapshot = snapshot + print_gear_captured( + snapshot.cartridges, + snapshot.modules, + len(snapshot.characters), + snapshot.characters_with_loadouts, + ) diff --git a/src/nte_history_exporter/live_capture/runner.py b/src/nte_history_exporter/live_capture/runner.py index f36b6f8..eb6051c 100644 --- a/src/nte_history_exporter/live_capture/runner.py +++ b/src/nte_history_exporter/live_capture/runner.py @@ -21,6 +21,7 @@ from nte_history_exporter.export.json_export import ( ) from nte_history_exporter.live_capture.backends import open_capture_backend from nte_history_exporter.live_capture.diagnostics import new_diagnostics_path, write_capture_diagnostics +from nte_history_exporter.live_capture.gear_collector import GearCollector from nte_history_exporter.live_capture.session import LiveHistorySession, UdpPacket from nte_history_exporter.live_capture.stop_key import StopKeyMonitor from nte_history_exporter.live_capture.windows_raw import detect_local_ipv4 @@ -88,14 +89,17 @@ def run_live_capture( active_gap_notices: set[str] = set() tcp_segments: dict[tuple[str, int, str, int], list[tuple[int, bytes]]] = {} achievement_records = [] + gear = GearCollector(local_ip) try: with StopKeyMonitor() as stop_key: for packet in capture.packets(): if stop_key.pressed(): break + gear.poll() # above the guard: the pcap read timeout drives it if packet is None: continue + gear.observe(packet) if ( not achievement_records and packet.protocol == "tcp" @@ -176,6 +180,8 @@ def run_live_capture( stats = capture.stats() capture.close() + gear.finalize() + exports = [] achievement_path = None diagnostics_path = None @@ -183,7 +189,9 @@ def run_live_capture( diagnostics_path = new_diagnostics_path() write_capture_diagnostics(diagnostics_path, session.diagnostic_report()) resolved_user_uid = user_uid or session.user_uid - if (session.kinds_seen() or achievement_records) and not resolved_user_uid: + if ( + session.kinds_seen() or achievement_records or gear.snapshot + ) and not resolved_user_uid: resolved_user_uid = console.prompt_user_uid() if achievement_records: achievement_path = _achievement_path(resolved_user_uid) @@ -198,6 +206,11 @@ def run_live_capture( json.dumps(achievement_export, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", ) + gear_path = gear.write( + capture_source=CAPTURE_SOURCE_LABELS.get(capture.name, capture.name), + user_uid=resolved_user_uid, + server_id=session.server_id, + ) resolved_server_id = session.server_id if session.kinds_seen() and not resolved_server_id: resolved_server_id = console.prompt_server_id() @@ -251,16 +264,23 @@ def run_live_capture( console.print_note("No pull history was captured (this is fine if you only wanted achievements).") print() console.print_note(f"Export written: {achievement_path}") + elif gear_path is not None: + console.print_success("Gear export complete.") + console.print_note("No pull history was captured (this is fine if you only wanted gear).") + print() else: console.print_problem("No history pages were captured.") console.print_note("Reopen a supported history screen and scroll from page 1.") console.print_note("If no page messages appear, return to the main menu and re-enter the game.") + if gear_path is not None: + console.print_note(f"Export written: {gear_path}") if diagnostics_path is not None: console.print_note(f"Diagnostics written: {diagnostics_path}") return { "exports": [], "diagnostics_path": diagnostics_path, "achievement_path": achievement_path, + "gear_path": gear_path, } for item in exports: @@ -282,6 +302,8 @@ def run_live_capture( if achievement_path is not None: print() console.print_note(f"Export written: {achievement_path}") + if gear_path is not None: + console.print_note(f"Export written: {gear_path}") if diagnostics_path is not None: console.print_note(f"Diagnostics written: {diagnostics_path}") @@ -297,6 +319,7 @@ def run_live_capture( "exports": exports, "diagnostics_path": diagnostics_path, "achievement_path": achievement_path, + "gear_path": gear_path, } diff --git a/tests/test_export_contract.py b/tests/test_export_contract.py index 4dc7cd2..5400737 100644 --- a/tests/test_export_contract.py +++ b/tests/test_export_contract.py @@ -8,7 +8,7 @@ from nte_history_exporter.live_capture.runner import _achievement_path class ExportContractTests(unittest.TestCase): def test_public_version_references_match(self): - self.assertEqual(__version__, "0.3.0") + self.assertEqual(__version__, "0.4.0") self.assertEqual(EXPORTER_VERSION, __version__) def test_sanitized_export_omits_raw_packet_fields(self): diff --git a/tests/test_gear_decoding.py b/tests/test_gear_decoding.py new file mode 100644 index 0000000..235c65a --- /dev/null +++ b/tests/test_gear_decoding.py @@ -0,0 +1,747 @@ +"""Gear decoding, driven entirely by synthetic packets. + +The wire format is fully understood, so every packet here is constructed from +the spec rather than captured: a 120-bit header, payload bits, two 1 bits and +zero padding. No real capture is used - see tests/fixtures/README.md. +""" + +import contextlib +import io +import json +import os +import struct +import tempfile +import unittest + +from nte_history_exporter import console + +from nte_history_exporter import __version__ +from nte_history_exporter.decoder.gear import ( + HEADER_BITS, + extract_gear, + payload_end_bit, + splice_datagrams, +) +from nte_history_exporter.export.gear_export import build_gear_export_json +from nte_history_exporter.live_capture import gear_collector, runner +from nte_history_exporter.live_capture.windows_raw import ParsedIpPacket + +from tests import support + + +# --- synthetic wire construction ------------------------------------------- + + +def string_frame(text: bytes) -> bytes: + """A self-validating frame.""" + return struct.pack(" bytes: + """<4 zero bytes>.""" + return string_frame(name) + b"\0\0\0\0" + struct.pack(" None: + if len(buf) < byte_offset + len(raw): + buf.extend(b"\0" * (byte_offset + len(raw) - len(buf))) + buf[byte_offset : byte_offset + len(raw)] = raw + + +def poke_u32_at_bit(buf: bytearray, bitpos: int, value: int) -> None: + """Write a little-endian u32 at an arbitrary bit offset.""" + needed = (bitpos + 32 + 7) // 8 + 1 + if len(buf) < needed: + buf.extend(b"\0" * (needed - len(buf))) + whole = int.from_bytes(buf, "little") + whole &= ~(0xFFFFFFFF << bitpos) + whole |= value << bitpos + packed = whole.to_bytes(len(buf) + 8, "little") + buf[:] = packed[: len(buf)] + + +def item_record( + item_id: bytes, + *, + level: int = 20, + entries=(), + instance: bytes = b"\x11\x22\x33\x44\x55\x66\x77\x88", + owner: bytes | None = None, +) -> bytes: + """One item record laid out at the documented offsets past the id frame.""" + body = bytearray(b"\0" * 64) + poke(body, 5, instance) + # Level is a u32 at +38 bytes; the owner reference starts at +39 bits-wise + # (312 bits) and its first three bytes are always zero, so the two fields + # coexist exactly as they do on the wire. + poke(body, 38, struct.pack(" bytes: + """An ability frame with the progress fields at their negative offsets.""" + lead = 128 # bytes of room for the fields that sit before the frame + buf = bytearray(b"\0" * lead) + frame = string_frame(b"GA_" + name + b"_Melee") + block_bit = lead * 8 + poke_u32_at_bit(buf, block_bit - 466, 67806) + poke_u32_at_bit(buf, block_bit - 434, level) + poke_u32_at_bit(buf, block_bit - 402, breakthroughs) + return bytes(buf) + frame + + +def loadout_cell(item_id: bytes, row: int, column: int) -> bytes: + """An id frame quoted with board coordinates and no stat chain.""" + body = bytearray(b"\0" * 32) + # Coordinates sit 96 and 128 bits past the id and are stored doubled. + poke(body, 12, struct.pack("> start) & ((1 << nbits) - 1) + + header = bytearray(15) + header[0:3] = b"\xaa\xbb\xcc" + # 14-bit packet sequence at bit 24. + header[3] = seq & 0xFF + header[4] = (seq >> 8) & 0x3F + header[12:15] = b"\x64\x06\xee" + + # payload bits, then the two 1 bits Unreal terminates with. + value = int.from_bytes(header, "little") + value |= chunk << HEADER_BITS + value |= 0b11 << (HEADER_BITS + nbits) + width = (HEADER_BITS + nbits + 2 + 7) // 8 + datagrams.append(value.to_bytes(width, "little")) + + start += nbits + seq = (seq + 1) & 0x3FFF + return datagrams + + +MODULE_ENTRIES = ( + (b"HPMaxAdd", 0.0), + (b"AtkAdd", 0.0), + (b"CritDamageBase", 0.06), + (b"CritBase", 0.03), + (b"AtkUp", 0.04), + (b"HPMaxUp", 0.04), +) + +CARTRIDGE_ENTRIES = ( + (b"AtkUp", 0.0), + (b"CritDamageBase", 0.06), + (b"CritBase", 0.03), + (b"DefUp", 0.07), + (b"HPMaxUp", 0.04), +) + +OWNER = b"\0\0\0\x0d\x0d\x20\x3d\x6d\x00\x00\x00\x00" + + +class FramingTests(unittest.TestCase): + def test_payload_end_bit_finds_the_double_terminator(self): + [datagram] = pack_datagrams(b"\x01\x02\x03\x04") + self.assertEqual(payload_end_bit(datagram), HEADER_BITS + 32) + + def test_splice_round_trips_a_payload_across_datagrams(self): + payload = bytes(range(256)) * 3 + # A chunk width that is not a multiple of 8 forces the accumulator to + # rejoin the stream at a bit offset, as the real capture does. + datagrams = pack_datagrams(payload, chunk_bits=333) + self.assertGreater(len(datagrams), 1) + self.assertEqual(splice_datagrams(datagrams)[: len(payload)], payload) + + def test_pure_acks_are_skipped(self): + payload = b"\x01\x02\x03\x04\x05\x06\x07\x08" + ack = bytes(12) + datagrams = pack_datagrams(payload) + self.assertEqual(splice_datagrams([ack] + datagrams + [ack])[:8], payload) + + +class ItemRecordTests(unittest.TestCase): + def decode(self, payload: bytes, **kwargs): + return extract_gear(pack_datagrams(b"\0" * 8 + payload, **kwargs)) + + def test_module_record_decodes(self): + snapshot = self.decode( + item_record(b"cell3_style6_1_Orange", level=20, entries=MODULE_ENTRIES) + ) + [item] = snapshot.items + self.assertEqual(item.kind, "module") + self.assertEqual(item.item_id, "cell3_style6_1_Orange") + self.assertEqual(item.shape, "cell3_style6") + self.assertEqual(item.module_type, "III") + self.assertEqual(item.level, 20) + self.assertEqual(item.rarity, "orange") + self.assertEqual(item.instance, "1122334455667788") + self.assertIsNone(item.owner_group) + # Both mains are computed from the cell count, never transmitted. + self.assertEqual(dict(item.main_stats), {"HPMaxAdd": 840.0, "AtkAdd": 63.0}) + self.assertEqual(len(item.substats), 4) + + def test_cartridge_record_decodes_with_its_set(self): + snapshot = self.decode( + item_record(b"Incantation_orange", level=20, entries=CARTRIDGE_ENTRIES) + ) + [item] = snapshot.items + self.assertEqual(item.kind, "cartridge") + self.assertIsNotNone(item.set_name) + self.assertEqual(len(item.main_stats), 1) + self.assertEqual(len(item.substats), 4) + + def test_owner_reference_is_read_when_equipped(self): + snapshot = self.decode( + item_record( + b"cell3_style6_1_Orange", + level=20, + entries=MODULE_ENTRIES, + owner=OWNER, + ) + ) + [item] = snapshot.items + self.assertEqual(item.owner_group, OWNER.hex()) + # The owner field overlaps the level word without corrupting it. + self.assertEqual(item.level, 20) + + def test_record_spanning_a_datagram_boundary_decodes(self): + record = item_record( + b"cell3_style6_1_Orange", level=20, entries=MODULE_ENTRIES, owner=OWNER + ) + # Chunks far smaller than the record, at a width that is not a whole + # number of bytes: every field lands across at least one boundary. + snapshot = self.decode(record, chunk_bits=101) + [item] = snapshot.items + self.assertEqual(item.item_id, "cell3_style6_1_Orange") + self.assertEqual(item.level, 20) + self.assertEqual(item.owner_group, OWNER.hex()) + self.assertEqual(dict(item.main_stats), {"HPMaxAdd": 840.0, "AtkAdd": 63.0}) + + def test_unframed_id_is_ignored(self): + # Same bytes, wrong length prefix: the frame check must reject it. + record = bytearray( + item_record(b"cell3_style6_1_Orange", entries=MODULE_ENTRIES) + ) + record[0:4] = struct.pack("