Decode reward keys and use mapping files
Add decoding for Monopoly reward keys and switch reward metadata to be looked up by decoded id. Implement decode_reward_key() and infer_reward_type(); replace the previous raw-key KNOWN_REWARDS approach with REWARDS_BY_ID built from mappings/*.json. Add mappings/items.json and expand mappings/arcs.json. Update decoder to produce reward_id, use REWARDS_BY_ID, and change guess_quantity() to accept reward_id. Refactor pool_mappings to reuse the new mapping loader, and update tests and docs to validate mapping files and document the reward key encoding/behavior.
This commit is contained in:
Golumpa
parent
401df44ecd
commit
354b0e9232
8 changed files with 197 additions and 113 deletions
|
|
@ -15,7 +15,52 @@ from nte_history_exporter.constants import (
|
|||
TIMESTAMP_TICKS_PER_SECOND,
|
||||
VALID_DICE_FIELDS,
|
||||
)
|
||||
from nte_history_exporter.mappings import KNOWN_REWARDS
|
||||
from nte_history_exporter.mappings import REWARDS_BY_ID
|
||||
|
||||
REWARD_ID_CHARS = frozenset("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789_")
|
||||
|
||||
|
||||
def decode_reward_key(raw: bytes) -> str:
|
||||
"""Decode a Monopoly reward key into its reward id string.
|
||||
|
||||
Keys encode the id one character per byte as ASCII*4 with carry chaining
|
||||
(Arc history uses the same scheme at ASCII*2). The final byte is either the
|
||||
pending carry (0 or 1) acting as a terminator, or a regular character byte.
|
||||
The carry bit makes some bytes ambiguous, so decode with backtracking and
|
||||
accept only parses made of identifier characters.
|
||||
"""
|
||||
results: list[str] = []
|
||||
|
||||
def walk(index: int, carry: int, acc: str) -> None:
|
||||
if results:
|
||||
return
|
||||
if index == len(raw):
|
||||
if carry == 0:
|
||||
results.append(acc)
|
||||
return
|
||||
byte = raw[index]
|
||||
if index == len(raw) - 1 and byte == carry:
|
||||
results.append(acc)
|
||||
return
|
||||
for carry_out in (0, 1):
|
||||
value = byte - carry + 256 * carry_out
|
||||
if value % 4 == 0 and chr(value // 4) in REWARD_ID_CHARS:
|
||||
walk(index + 1, carry_out, acc + chr(value // 4))
|
||||
|
||||
walk(0, 0, "")
|
||||
return results[0] if results else ""
|
||||
|
||||
|
||||
def infer_reward_type(reward_id: str) -> str:
|
||||
if not reward_id:
|
||||
return ""
|
||||
if reward_id.startswith("fork_"):
|
||||
return "arc"
|
||||
if reward_id.isdigit():
|
||||
return "character"
|
||||
if reward_id.startswith("Fashion_"):
|
||||
return "cosmetic"
|
||||
return "item"
|
||||
|
||||
|
||||
def decode_history_timestamp(raw8: bytes) -> tuple[int, float, str]:
|
||||
|
|
@ -98,20 +143,20 @@ def classify_result_type(
|
|||
return "dice", None
|
||||
|
||||
|
||||
def guess_quantity(chunk_hex: str, key_hex: str, result_type: str | None = None) -> int | None:
|
||||
if key_hex == "10a58d957dd1a58dad95d17dc1c400":
|
||||
def guess_quantity(chunk_hex: str, reward_id: str, result_type: str | None = None) -> int | None:
|
||||
if reward_id == "Dice_ticket_01":
|
||||
if result_type == "chase_reward":
|
||||
return 30
|
||||
return 4
|
||||
if key_hex == "10a58d9539bdc9b585b101":
|
||||
if reward_id == "DiceNormal":
|
||||
return 1
|
||||
if key_hex == "10a58d957dd1a58dad95d17dc1c800":
|
||||
if reward_id == "Dice_ticket_02":
|
||||
if "c8b0d4c0" in chunk_hex:
|
||||
return 50
|
||||
if "c8b0ccc0" in chunk_hex:
|
||||
return 30
|
||||
return None
|
||||
if key_hex:
|
||||
if reward_id:
|
||||
return 1
|
||||
return None
|
||||
|
||||
|
|
@ -142,7 +187,8 @@ def decode_response_records(response_content: bytes) -> list[dict[str, Any]]:
|
|||
dice = -4
|
||||
dice_raw = -4
|
||||
key_hex = extract_key(chunk)
|
||||
reward = KNOWN_REWARDS.get(key_hex, {})
|
||||
reward_id = decode_reward_key(bytes.fromhex(key_hex)) if key_hex else ""
|
||||
reward = REWARDS_BY_ID.get(reward_id, {})
|
||||
timestamp_raw = response_content[marker_offset + len(marker) : marker_offset + len(marker) + 8]
|
||||
timestamp_ticks, timestamp_unix, timestamp_decoded = decode_history_timestamp(timestamp_raw)
|
||||
chunk_hex = chunk.hex()
|
||||
|
|
@ -163,11 +209,11 @@ def decode_response_records(response_content: bytes) -> list[dict[str, Any]]:
|
|||
"dice_raw_u32": dice_raw,
|
||||
"dice_offset_in_record": dice_offset,
|
||||
"reward_key_hex": key_hex,
|
||||
"reward_type": reward.get("type", ""),
|
||||
"reward_id": reward.get("id", ""),
|
||||
"reward_type": reward.get("type") or infer_reward_type(reward_id),
|
||||
"reward_id": reward_id,
|
||||
"reward_name": reward.get("name", ""),
|
||||
"reward_rank": reward.get("rank"),
|
||||
"quantity": guess_quantity(chunk_hex, key_hex, result_type),
|
||||
"quantity": guess_quantity(chunk_hex, reward_id, result_type),
|
||||
"timestamp_raw_hex": timestamp_raw.hex(),
|
||||
"timestamp_ticks": timestamp_ticks,
|
||||
"timestamp_unix": f"{timestamp_unix:.6f}",
|
||||
|
|
|
|||
|
|
@ -1,92 +1,33 @@
|
|||
KNOWN_REWARDS = {
|
||||
"98bdc9ad7d91d5cdd189a5b901": {"type": "arc", "id": "fork_dustbin", "name": "Dangerous Game", "rank": "B"},
|
||||
"98bdc9ad7dd9a5b99501": {"type": "arc", "id": "fork_vine", "name": "Be Happy", "rank": "B"},
|
||||
"98bdc9ad7db9bdb9bdcd01": {"type": "arc", "id": "fork_nonos", "name": "First Step to Success", "rank": "B"},
|
||||
"98bdc9ad7d85c1c1b1a585b98d9501": {"type": "arc", "id": "fork_appliance", "name": "\"Real Music\"", "rank": "B"},
|
||||
"98bdc9ad7d41c9bdad85c9e5bdb901": {"type": "arc", "id": "fork_Prokaryon", "name": "Us.", "rank": "B"},
|
||||
"98bdc9ad7d4185c195c941b185b99501": {"type": "arc", "id": "fork_PaperPlane", "name": "Clear Skies", "rank": "A"},
|
||||
"98bdc9ad7ddda1d585add585b99d01": {"type": "arc", "id": "fork_wuhuakuang", "name": "The Forgotten", "rank": "A"},
|
||||
"98bdc9ad7dddd5a1d585add585b99d01": {"type": "arc", "id": "fork_wuhuakuang", "name": "The Forgotten", "rank": "A"},
|
||||
"98bdc9ad7d2da5d19501": {"type": "arc", "id": "fork_Kite", "name": "Watch Your Heads!", "rank": "A"},
|
||||
"98bdc9ad7de5d5c995b901": {"type": "arc", "id": "fork_yuren", "name": "Umbrella", "rank": "A"},
|
||||
"98bdc9ad7de585bd9185bd01": {"type": "arc", "id": "fork_yaodao", "name": "Drawn Blade", "rank": "A"},
|
||||
"10a58d9539bdc9b585b101": {"type": "item", "id": "DiceNormal", "name": "Fabricated Dice", "rank": None},
|
||||
"10a58d957dd1a58dad95d17dc1c400": {"type": "item", "id": "Dice_ticket_01", "name": "Warp Piece", "rank": None},
|
||||
"10a58d957dd1a58dad95d17dc1c800": {"type": "item", "id": "Dice_ticket_02", "name": "Lost Piece", "rank": None},
|
||||
"10a58d95b1a5b5a5d19501": {"type": "item", "id": "Dicelimite", "name": "Solid Dice", "rank": None},
|
||||
"1885cda1a5bdb97dd995a1a58db1957dc5c0c4c07c59c1c0e000": {
|
||||
"type": "cosmetic",
|
||||
"id": "Fashion_vehicle_1010_V008",
|
||||
"name": "Fashion Vehicle Skin 1010 V008",
|
||||
"rank": None,
|
||||
},
|
||||
"1885cda1a5bdb97d1db1a591957dc5c0c4c000": {
|
||||
"type": "cosmetic",
|
||||
"id": "Fashion_glide_1010",
|
||||
"name": "Fashion Glider 1010",
|
||||
"rank": None,
|
||||
},
|
||||
"c4c0cccc00": {"type": "character", "id": "1033", "name": "Adler", "rank": "A"},
|
||||
"c4c0c8c000": {"type": "character", "id": "1020", "name": "Haniel", "rank": "A"},
|
||||
"c4c0c8cc00": {"type": "character", "id": "1023", "name": "Baicang", "rank": "S"},
|
||||
"c4c0c8d400": {"type": "character", "id": "1025", "name": "Hathor", "rank": "S"},
|
||||
"c4c0c4e400": {"type": "character", "id": "1019", "name": "Mint", "rank": "A"},
|
||||
"c4c0c0e000": {"type": "character", "id": "1008", "name": "Skia", "rank": "A"},
|
||||
"c4c0dcc0": {"type": "character", "id": "1070", "name": "Aurelia", "rank": "A"},
|
||||
"c4c0dcc000": {"type": "character", "id": "1070", "name": "Aurelia", "rank": "A"},
|
||||
"c4c0c8c4": {"type": "character", "id": "1021", "name": "Edgar", "rank": "A"},
|
||||
"c4c0c8c400": {"type": "character", "id": "1021", "name": "Edgar", "rank": "A"},
|
||||
}
|
||||
from __future__ import annotations
|
||||
|
||||
ARC_META = {
|
||||
"fork_nonos": {"name": "First Step to Success", "rank": "B"},
|
||||
"fork_Prokaryon": {"name": "Us.", "rank": "B"},
|
||||
"fork_appliance": {"name": "\"Real Music\"", "rank": "B"},
|
||||
"fork_PaperPlane": {"name": "Clear Skies", "rank": "A"},
|
||||
"fork_mofeikesi": {"name": "Good Boy's Grand Adventure", "rank": "S"},
|
||||
"fork_dustbin": {"name": "Dangerous Game", "rank": "B"},
|
||||
"fork_vine": {"name": "Be Happy", "rank": "B"},
|
||||
"fork_Kite": {"name": "Watch Your Heads!", "rank": "A"},
|
||||
"fork_wuhuakuang": {"name": "The Forgotten", "rank": "A"},
|
||||
"fork_jingmotingyuan": {"name": "Camellia Society", "rank": "S"},
|
||||
"fork_yuren": {"name": "Umbrella", "rank": "A"},
|
||||
"fork_yaodao": {"name": "Drawn Blade", "rank": "A"},
|
||||
}
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
CHARACTERS = {
|
||||
"1003": {"name": "Sakiri", "rank": "S"},
|
||||
"1004": {"name": "Lacrimosa", "rank": "S"},
|
||||
"1008": {"name": "Skia", "rank": "A"},
|
||||
"1010": {"name": "Nanally", "rank": "S"},
|
||||
"1019": {"name": "Mint", "rank": "A"},
|
||||
"1020": {"name": "Haniel", "rank": "A"},
|
||||
"1021": {"name": "Edgar", "rank": "A"},
|
||||
"1023": {"name": "Baicang", "rank": "S"},
|
||||
"1025": {"name": "Hathor", "rank": "S"},
|
||||
"1033": {"name": "Adler", "rank": "A"},
|
||||
"1039": {"name": "Fadia", "rank": "S"},
|
||||
"1046": {"name": "Zero", "rank": "S"},
|
||||
"1051": {"name": "Zero", "rank": "S"},
|
||||
"1052": {"name": "Hotori", "rank": "S"},
|
||||
"1054": {"name": "Daffodill", "rank": "S"},
|
||||
"1055": {"name": "Jiuyuan", "rank": "S"},
|
||||
"1056": {"name": "Lacrimosa", "rank": "S"},
|
||||
"1070": {"name": "Aurelia", "rank": "A"},
|
||||
"1073": {"name": "Chiz", "rank": "S"},
|
||||
}
|
||||
PROJECT_ROOT = Path(__file__).resolve().parents[2]
|
||||
MAPPINGS_DIR = PROJECT_ROOT / "mappings"
|
||||
|
||||
|
||||
def character_id_to_key(character_id: str) -> str:
|
||||
return "".join(f"{0xC0 + int(digit) * 4:02x}" for digit in character_id) + "00"
|
||||
def load_mapping_file(filename: str) -> dict[str, Any]:
|
||||
return json.loads((MAPPINGS_DIR / filename).read_text(encoding="utf-8"))
|
||||
|
||||
|
||||
for _character_id, _info in CHARACTERS.items():
|
||||
KNOWN_REWARDS.setdefault(
|
||||
character_id_to_key(_character_id),
|
||||
{
|
||||
"type": "character",
|
||||
"id": _character_id,
|
||||
"name": _info["name"],
|
||||
"rank": _info["rank"],
|
||||
},
|
||||
)
|
||||
ARC_META: dict[str, dict[str, Any]] = load_mapping_file("arcs.json")
|
||||
CHARACTERS: dict[str, dict[str, Any]] = load_mapping_file("characters.json")
|
||||
ITEMS: dict[str, dict[str, Any]] = load_mapping_file("items.json")
|
||||
|
||||
# Reward keys in Monopoly history packets are the reward id string itself
|
||||
# (ASCII*4 with carry, see decoder.protocol.decode_reward_key), so all reward
|
||||
# metadata is looked up by decoded id rather than by raw key bytes.
|
||||
REWARDS_BY_ID: dict[str, dict[str, Any]] = {}
|
||||
for _arc_id, _meta in ARC_META.items():
|
||||
REWARDS_BY_ID[_arc_id] = {"type": "arc", "id": _arc_id, "name": _meta["name"], "rank": _meta["rank"]}
|
||||
for _character_id, _meta in CHARACTERS.items():
|
||||
REWARDS_BY_ID[_character_id] = {
|
||||
"type": "character",
|
||||
"id": _character_id,
|
||||
"name": _meta["name"],
|
||||
"rank": _meta["rank"],
|
||||
}
|
||||
for _item_id, _meta in ITEMS.items():
|
||||
REWARDS_BY_ID[_item_id] = {"type": _meta["type"], "id": _item_id, "name": _meta["name"], "rank": _meta.get("rank")}
|
||||
|
|
|
|||
|
|
@ -1,11 +1,8 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
PROJECT_ROOT = Path(__file__).resolve().parents[2]
|
||||
MAPPINGS_DIR = PROJECT_ROOT / "mappings"
|
||||
from nte_history_exporter.mappings import load_mapping_file
|
||||
|
||||
POOL_MAPPING_FILES = {
|
||||
"permanent": "permanent_board.json",
|
||||
|
|
@ -15,8 +12,7 @@ POOL_MAPPING_FILES = {
|
|||
|
||||
|
||||
def load_pool_mapping(pool_key: str) -> dict[str, Any]:
|
||||
filename = POOL_MAPPING_FILES[pool_key]
|
||||
return json.loads((MAPPINGS_DIR / filename).read_text(encoding="utf-8"))
|
||||
return load_mapping_file(POOL_MAPPING_FILES[pool_key])
|
||||
|
||||
|
||||
def load_pool_mappings() -> dict[str, dict[str, Any]]:
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue