Install

python -m pip install bdo-toolkit

Windows live capture needs Npcap 1.88 or later; offline replay doesn’t. See Installation & setup for the full checklist.

Workflows

Each is independent — pick by the result you need.

Start here

Install

python -m pip install bdo-toolkit

Installs the bdo_toolkit package, the bdo-toolkit command and the required dependencies.

Live-capture prerequisites

• Windows: install Npcap 1.88 or later. Domain-specific requirements live with each live API.
• Other platforms: use the packet-capture backend available to Scapy. Physical live capture is unverified outside the Windows baseline below.
• Permissions: run with permission to capture traffic. Elevate only when the driver or local security policy requires it.
• Timing: start capture before the game action: before collecting or moving an item, logging in or switching characters, or opening the Solare Leaderboard for the first time after a restart.

Network settings: PacketCaptureOptions. No game traffic: find and test capture settings.

Choose live or replay

Most applications start live. Replay helps when a representative recording already exists, or the same evidence must be analyzed again.

Supported live-capture environment

The toolkit reads inbound, unfragmented IPv4/TCP packets.

Validated Windows baseline

Windows live capture requires Npcap 1.88 or later; older versions are unsupported. As of September 3, 2026, maintainer validation covers Windows 11 25H2, Python 3.14.0, Npcap 1.88, Scapy 2.7.0 and a wired Ethernet adapter. Later Npcap releases meet the version floor but aren’t independently validated.

This is a documentation support boundary. The validated machine had WinPcap API-compatible mode on; the toolkit doesn’t know whether that mode is required and never inspects the installed Npcap version. Other operating systems and Npcap modes aren’t part of maintainer validation. Replay uses no Npcap or adapter, but the recording must still contain supported BDO traffic.

Hardware, network card and driver

The live path runs through the physical or virtual adapter, its driver and offload behavior, operating-system routing, the capture backend (Npcap on Windows), Scapy, then the toolkit. Each layer can present the same connection differently, so one known-good network card isn’t a compatibility list for every card or driver.

Choose the options object

Live APIs build default options internally. Pass an options object only when this computer needs a different adapter, destination IPv4, server-port set, filtering mode or item-event queue size.

Interfaces and VPN routing

An interface is the physical or virtual adapter the capture backend observes. Automatic selection follows the computer’s default IPv4 route with the default BDO server ports; it doesn’t inspect BDO’s own connections. Several adapters, a VPN or a local proxy can need a different interface, local IP or port set.

With a VPN, readable BDO traffic may appear on a virtual adapter. A local proxy such as ExitLag can expose the connection on loopback with a session-specific port. The toolkit can’t decode an encrypted tunnel, or traffic the capture backend can’t see.

Diagnose capture when automatic selection fails

Use this Windows walkthrough when capture starts but receives no game traffic, or after you change adapters, VPNs or routing services. It proposes settings from BDO’s current connections; it doesn’t apply them or prove that packets decode.

Inspect current connections

python -m bdo_toolkit.cli diagnose-capture

It lists grouped settings proposals: interface, local IPv4, server or proxy ports, and the underlying connections. Compatible game connections appear together, for example ports: 8884,8885,8889. Use the values from your own output; proxy ports can change after a reconnect.

Add --json for structured output. If Windows connection inspection times out, retry with --timeout 30; adapter enumeration isn’t covered by the timeout.

Review the proposals

• game_port: BDO connections on its default server ports. Keep the grouped ports together.
• local_proxy: BDO connects to a local peer. Check the peer name, then use the reported loopback interface, local IP and proxy ports.
• other_tcp: other ports, often including 443, which may carry web or authentication traffic. Don’t merge them into a game-port proposal just because the adapter and local IP match.

Several proposals can be valid observations: compare their BDO process ID, adapter, kind and connections before you pick one. A proposal with (no matching adapter) can’t be applied; check Npcap and adapter availability first.

• no_game: start BDO and retry.
• no_connections: enter a game server and retry. IPv6-only connections aren’t supported.
• unavailable: follow the reported inspection error. Discovery supports Windows only.

Apply one proposal

If your application has capture settings, copy the proposal’s interface, local IPv4 and complete port list into them, then restart its capture session. Use the adapter identifier from the output, not a guessed display name. local_ip restricts the inbound destination; it doesn’t select an adapter.

For a Python item listener, save this as capture_with_proposal.py next to a current opcodes.local and run it. It discovers settings again, asks for a proposal and listens for item activity until Ctrl+C.

from bdo_toolkit import capture_live, diagnose_capture

result = diagnose_capture()
for message in result.messages:
    print(message)
if not result.proposals:
    raise SystemExit("No settings proposals. Resolve the diagnostic above and retry.")

for number, proposal in enumerate(result.proposals, 1):
    print(number, proposal.kind, proposal.interface,
          proposal.local_ip, proposal.ports)

try:
    number = int(input("Proposal number: "))
    if not 1 <= number <= len(result.proposals):
        raise ValueError("Proposal number is out of range")
    options = result.proposals[number - 1].to_live_options()
except ValueError as error:
    raise SystemExit(str(error)) from error

try:
    for event in capture_live(
        opcode_profile="opcodes.local",
        live_options=options,
    ):
        print(event)
except KeyboardInterrupt:
    pass

For item state, calibration or Solare, build PacketCaptureOptions from the proposal’s interface, local_ip and ports and pass it as capture_options. APIs: diagnose_capture() · CaptureSettingsProposal.

Verify the selected settings

1. Start a new capture session with the selected settings, with the VPN and game connection as they were during discovery.
2. Where the API allows, use an explicit session and read its CaptureEndpoint after startup.
3. Perform one known in-game action once capture is ready, then check the workflow’s output and, where exposed, its final capture health or acquisition counters.
4. No packets: follow the acquisition checks. Packets but no item events after a known action: check filters and profile compatibility. Discovery alone can’t tell these apart.

Next: Opcode profile setup

Get a profile

Latest maintained NA/EU profile

For current NA/EU traffic, fetch the latest profile that maintainers captured, verified and promoted:

bdo-toolkit profile fetch https://ychwu.github.io/bdo-toolkit-profiles/channels/na-eu/stable.json --output opcodes.local

It downloads and verifies once; capture never updates the profile in the background. A locally calibrated profile stays on your computer and is never uploaded.

Other sources

Use it

Pass the path as opcode_profile= whenever an item workflow starts. Quickstart uses the same opcodes.local.

Verify it in game

Run Quickstart, then gather an item, collect a mob drop or withdraw something from town storage. Check that the printed item ID, quantity and source match what happened.

If a known action prints nothing, fetch the maintained profile again. If the channel isn’t current yet, follow manual calibration.

When setup doesn’t work

Capture one item event

from bdo_toolkit import LiveCaptureSession

with LiveCaptureSession(opcode_profile="opcodes.local") as session:
    print("Listening for item activity...")
    for event in session.events():
        print(event.format_human())
        break

from contextlib import closing

from bdo_toolkit import replay_pcap

with closing(replay_pcap(
    "path/to/session.pcapng",
    opcode_profile="path/to/matching-opcodes.json",
)) as events:
    for event in events:
        print(event.format_human())
        break

Live: run python quickstart.py, then gather an item, collect a mob drop or withdraw something from town storage. Replay: point both paths at a recording and the profile from its patch.

[23:06:01.245] ITEM_RECEIVED source='Gathering' item_id=7003 quantity=6 item_instance=0x6f105b8a3d7294c1

ITEM_RECEIVED means an item entered inventory, and Gathering is the in-game source; then come the raw item ID, quantity and observed instance. Other actions produce other event types and source labels.

Where to go next

Item events

Walkthrough: Item events. APIs: capture_live() · EventFilter · event writers.

Inventory & town storage

Walkthrough: Inventory & town storage. Result API: ItemStateSnapshot.

Calibration

Procedures: transfer calibration · loot preview. APIs: sync session · async session · profile writes.

Grind tracking

Reference: GrindSession · LiveXPSession · LiveAgrisSession.

Arena of Solare

Walkthrough: Arena of Solare. Everything else: browse all examples on GitHub ↗.

From a game action to Python

Item activity is a stream of changes observed after the listener starts: gathering, a monster drop, a storage withdrawal or a deposit. It doesn’t reconstruct what inventory or storage already held; for that, see Inventory & town storage.

item_receivedlive

An item receipt or inventory update decoded. source can be Gathering, Mob Drop, Storage or another known receipt context.

storage_deltalive

A live addition to town storage was proven. source names a proven origin such as Player Inventory or Worker Production, and storage_id names the destination. Withdrawals are item_received from Storage.

loot_previewlive · optional

An item was shown in a preview before collection. Not proof of receipt; needs separately calibrated preview support.

Most applications start with event_type, source, item_id and quantity. Time, enhancement, storage identity, formatting and serialization are on the same object. The toolkit returns raw game IDs; item names and market data are yours to supply.

Pick the capture control you need

EventFilter decides delivery once an event is decoded and any storage source is classified. Event writers turn a delivered event into a console line or a JSONL record; they don’t capture traffic.

Choose a walkthrough

Run the finished example

python examples/live_transfer_log.py

Wait for Listening for confirmed item activity, keep the terminal open and return to the game:

1. Gather an item or collect a monster drop.
2. Withdraw one item from town storage.
3. Optionally deposit an item, or wait for worker production to reach storage.
4. Press Ctrl+C in the terminal when you’re done.

Read the result

[09:35:41.102] ITEM_RECEIVED source='Gathering' item_id=7003 quantity=6
[09:35:48.417] ITEM_RECEIVED source='Storage' item_id=15156 quantity=1
[09:36:02.884] STORAGE_DELTA source='Player Inventory' destination='Velia' storage_id=0x00000005 item_id=7003 quantity=5
[09:36:18.513] STORAGE_DELTA source='Worker Production' destination='Heidel' storage_id=0x00000020 item_id=4003 quantity=20
[09:36:24.190] STORAGE_DELTA destination='Heidel' storage_id=0x00000020 item_id=7307 quantity=8

The last line has no source: the live addition was proven, but its origin wasn’t resolved.

The same object has timestamp_text and timestamp_iso, enhancement fields such as base_item_id and enhancement_level, format_human() for display and to_dict() for structured output.

Build it yourself

from bdo_toolkit import EventFilter, capture_live

confirmed_activity = EventFilter(
    event_types={"item_received", "storage_delta"},
)

for event in capture_live(
    opcode_profile="opcodes.local",
    event_filter=confirmed_activity,
):
    fields = [event.event_type]
    if event.source is not None:
        fields.append(event.source)
    if event.storage_id is not None:
        destination = event.storage_name or (
            f"UNKNOWN_STORAGE(0x{event.storage_id:08x})"
        )
        fields.extend((destination, f"0x{event.storage_id:08x}"))
    fields.extend((event.item_id, event.quantity))
    print(*fields)

1. EventFilter keeps the two confirmed activity types. Character-load records, unresolved neutral records and loot_preview stay out.
2. capture_live() starts passive capture and blocks while it waits for traffic.
3. Each pass receives one BDOEvent; your loop decides whether to print, count, store or forward it.

Run the finished example

python examples/live_mob_drops.py

1. Leave the counter running and return to the game.
2. Defeat monsters and collect their drops.
3. Press Ctrl+C when you’re done.

Read the result

Collect mob drops in game. Press Ctrl+C to stop.
item_id=1000707 quantity=1 total_this_run=1
item_id=1000707 quantity=2 total_this_run=3
item_id=44195 quantity=1 total_this_run=1

quantity is the amount in one observed receipt. total_this_run adds those receipts for this process run; it isn’t the character’s inventory quantity.

Build it yourself

from collections import Counter

from bdo_toolkit import EventFilter, capture_live

totals: Counter[int] = Counter()
mob_drops = EventFilter(
    event_types={"item_received"},
    sources={"Mob Drop"},
)

for event in capture_live(
    opcode_profile="opcodes.local",
    event_filter=mob_drops,
):
    totals[event.item_id] += event.quantity
    print(event.item_id, event.quantity, totals[event.item_id])

An event passes only if it is an item_received and its source is Mob Drop. The Counter is your application’s state; the toolkit keeps delivering ordinary BDOEvent values.

Extend it

• Show more per drop: timestamp_text, base_item_id, enhancement_level and enhancement.
• Keep only selected drops by adding item_ids to the filter.
• Keep the run by passing each event to JsonlEventWriter.

Run the finished example

python examples/live_worker_productions.py

1. In the worker window, pick an active task whose output you recognize and note its destination town.
2. Start the listener before the task finishes, then keep playing.
3. Once the production reaches town storage, read the terminal line and press Ctrl+C.

Read the result

Waiting for worker production to reach town storage. Press Ctrl+C to stop.
source='Worker Production' destination='Heidel' storage_id=0x00000020 item_id=4003 quantity=20

Read cause and destination separately:

• source names the inferred producing system. It’s classifier output, not a raw context-byte label.
• storage_id names the destination; storage_name is optional display text. An unregistered key keeps its number with storage_name=None, so you can show UNKNOWN_STORAGE(0x…) without inventing a name.
• quantity is the amount added, not the storage total. Multi-record productions can also carry record_index and record_count.

The event carries no worker identity, production node, task name or facility.

Build it yourself

from bdo_toolkit import EventFilter, capture_live

worker_productions = EventFilter(
    event_types={"storage_delta"},
    sources={"Worker Production"},
)

for event in capture_live(
    opcode_profile="opcodes.local",
    event_filter=worker_productions,
):
    destination = event.storage_name or (
        f"UNKNOWN_STORAGE(0x{event.storage_id:08x})"
        if event.storage_id is not None
        else "unknown"
    )
    print(
        event.source,
        destination,
        event.storage_id,
        event.item_id,
        event.quantity,
    )

storage_delta proves live storage activity. Requiring sources={"Worker Production"} then keeps the additions whose bounded classifier evidence points to worker production. Decoder problems still reach you through diagnostics.

Choose the stream you want

Filtering happens after decoding and storage-origin classification. Pass one filter to the capture API, and only matching events enter your loop.

Use a preset or its explicit form

Both tabs build equal immutable filters. Use the preset when its meaning is what you want; use the constructor when you add or change criteria.

from bdo_toolkit import EventFilter

activity = EventFilter.activity()

from bdo_toolkit import EventFilter

activity = EventFilter(
    event_types={"loot_preview", "item_received", "storage_delta"},
)

activity() leaves out character-load snapshot events and neutral inventory_record and storage_record values.

Combine exact criteria

from bdo_toolkit import EventFilter, capture_live

heidel_worker_items = EventFilter(
    event_types={"storage_delta"},
    sources={"Worker Production"},
    storage_ids={0x0020},
    item_ids={4607, 7003},
)

for event in capture_live(
    opcode_profile="opcodes.local",
    event_filter=heidel_worker_items,
):
    print(event.item_id, event.quantity)

Values inside one criterion are alternatives: item 4607 or 7003. Criteria add up, so the event must also be a storage_delta to Heidel with worker-production evidence.

• event_types matches exact event types.
• sources matches the inferred origin or context, such as Worker Production.
• storage_ids matches the numeric destination, such as Heidel’s 0x0020.
• item_ids matches exact raw item IDs.

Helper methods

Besides the three presets, two helpers:

• EventFilter.from_values(...) is a named constructor with the same inputs. It doesn’t combine or chain filters.
• filter.allows(event) tests an event your code already holds. Capture and replay call it for you.

from bdo_toolkit import BDOEvent, EventFilter

potato_receipts = EventFilter.from_values(
    event_types={"item_received"},
    item_ids={7003},
)


def should_keep(event: BDOEvent) -> bool:
    return potato_receipts.allows(event)

Common mistakes

• Wrap one value in a collection: storage_ids={0x0020}. A bare int, str or bytes raises TypeError.
• None leaves a criterion unrestricted; an empty iterable rejects every event.
• sources matches origin or context labels such as Worker Production and Player Inventory. For a town, use storage_ids.
• An explicit filter doesn’t inherit the activity preset: a source-only or storage-only filter can match non-activity records. Add event_types to stay activity-only.
• Filters are immutable. Build a new one when the policy changes.

Exact constructor and validation: EventFilter. The full delivery pipeline: Item-event internals.

From captured TCP to your code

TCP packets are only transport: one packet can hold several BDO messages, and one message can span several packets. The toolkit reassembles each server-to-client flow before it looks for complete length-framed messages. The selected profile then supplies the opcode and byte geometry to decode item records from a message.

Each event keeps its message’s timestamp, length, opcode and record position, plus a provisional per-flow stream position when available. None of that makes the stream a complete inventory ledger.

Framing and profile selection fail closed

Capture that attaches midstream starts unsynchronized. A generic message boundary needs framing evidence: a known opcode, an exact standalone frame or consecutive plausible lengths. Retransmission and out-of-order delivery are resolved per directional flow; bytes the capture missed can’t be rebuilt.

One explicit, immutable profile compiles into runtime layouts, and a candidate must satisfy its layout’s message length and item geometry. Receipts, loot previews and storage records need an independent top-level boundary; a nested signature alone can’t produce an event. Inventory hydration can also recover from two fully decoded, exactly adjacent wrappers on the same TCP flow generation, even across a sequence-number wrap. A differently segmented retransmission can’t give already-framed payload new boundaries.

Overlapping same-opcode loot-preview layouts and contradictory storage geometry are rejected. Where an inventory-transfer layout supports structural validation, a failed marker, instance or repeated-record check is final: those bytes aren’t retried through weaker stride decoding.

Storage records take a classification detour

Most events go straight to delivery. Storage records first pass two stateful classifiers, because the decoded item fields alone don’t prove a live addition versus existing character-load state.

The origin classifier goes first, so real deposits don’t inherit a nearby hydration epoch. Hydration then groups still-neutral records by TCP flow generation, opcode, timing and destination diversity. Neither stage infers cause from a town name or a single context byte.

EventFilter runs after both stages: an activity filter sees an evidence-promoted storage_delta, and sources sees the final label. Decoder diagnostics stay out of band and bypass the filter.

Wire time and delivery order can differ

Origin and hydration decisions use bounded lookahead. While a storage event waits for evidence, a later non-storage event can be delivered first; the storage event keeps its earlier timestamp.

Equal timestamps and separate TCP flows still have no universal total order. extra["stream_sequence"] is provisional per-flow metadata, not a cross-flow ordering key.

Current bounds and backpressure

Filtering doesn’t save capture work: candidate traffic is still framed and decoded before values are matched. The runtime can skip the origin-correlation graph when the filter can’t deliver storage events and no origin_observer is set.

Consumption, deadlines and finalization

A live capture_seconds deadline belongs to the session and keeps running while your code pauses after a yielded event. On stop, capture joins, every accepted packet ahead of the worker sentinel is decoded, TCP reassembly finishes, and pending origin and hydration decisions finalize. Shutdown-finalized events stay drainable before a retained error is raised.

Read the right reliability signal

The signals are independent: a locally clean capture can still use a stale profile, and storage_status="not_observed" proves nothing either way. Values read mid-capture are observations; use the final post-stop values when completeness matters.

From character load to one snapshot

CharacterLoadSession passively captures whatever records the game sends, and stop() assembles them into one immutable ItemStateSnapshot. It doesn’t update after capture ends.

Selection is conservative: only the latest inventory generation and storage sweep become current state. A validated empty inventory still counts as observed, so an earlier character’s items are never carried forward. The rules: Item-state internals.

Choose a walkthrough

Run the finished example

python examples/live_character_load_snapshot.py

1. Wait until the terminal says capture started.
2. Log in or switch characters for inventory, or make the first world load after a fresh launch for storage.
3. Once the character is playable and the world has settled, press Enter.

The example checks the profile path, handles Ctrl+C and prints format_item_state() for troubleshooting.

Read the result

CHARACTER LOAD SNAPSHOT DIAGNOSTIC
Storage decoder: compatible (... observed wrappers decoded)
Hydration packets detected; trigger is unclassified (initial login vs character switch).
INVENTORY SNAPSHOT
  183 serialized records: 179 occupied item stacks + 4 currency balances
STORAGE SNAPSHOT
  27/34 known destinations observed
  642 unique occupied item stacks from ... decoded snapshot records

The trigger is never classified: a login and a switch look the same on the wire. Registered towns missing from the capture are reported, not assumed empty.

Build it yourself

from bdo_toolkit.item_state import CharacterLoadSession

with CharacterLoadSession(opcode_profile="opcodes.local") as session:
    input(
        "Capture started. For storage, enter the world for the first time "
        "after a fresh game launch; a later character switch captures "
        "inventory only. Press Enter after the world has settled: "
    )
    snapshot = session.stop()

print(f"Hydration detected: {snapshot.hydration_detected}")
print(f"Inventory observed: {snapshot.inventory.hydration_observed}")
print(f"Storage decoder: {snapshot.decoder_health.storage_status}")

1. Construction loads and pins the profile for this one session.
2. Entering the block calls start(), which returns once passive capture is ready.
3. You make the load. No marker ends it, so pressing Enter is your stop decision.
4. stop() drains capture, assembles state and returns the snapshot. Context exit only cleans up.

The three prints separate any item-state evidence (hydration_detected), inventory hydration (inventory.hydration_observed) and storage-profile compatibility (decoder_health.storage_status). Application data lives in snapshot.inventory and snapshot.storages.

Run the finished example

python examples/live_inventory_contents.py

Once capture starts, log in or switch characters, wait for the playable world and press Enter.

INVENTORY CONTENTS
179 observed occupied stacks
item_id=15156 quantity=1 container_label=Main Inventory
item_id=7003 quantity=38 container_label=Main Inventory
item_id=17584 quantity=1 container_label=Pearl Inventory
… 176 more stacks …

Build it yourself

from bdo_toolkit.item_state import CharacterLoadSession

with CharacterLoadSession(opcode_profile="opcodes.local") as session:
    input("Log in or switch characters, then press Enter when settled: ")
    snapshot = session.stop()

inventory = snapshot.inventory
if not inventory.hydration_observed:
    raise RuntimeError("No character inventory was observed")
if not inventory.items:
    print("No occupied inventory stacks were observed")
for item in inventory.items:
    label = item.container_name or "Unclassified"
    print(item.item_id, item.quantity, label)

stop() returns an ItemStateSnapshot; its inventory is an InventorySnapshotSummary, and inventory.items holds one SnapshotItem per retained ordinary stack.

Pick the right query

• occupied_stacks counts records; it doesn’t sum quantities.
• hydration_observed=True with empty items means the latest generation was observed empty.
• Tuple order isn’t slot order, and known balances aren’t part of items.
• container_name is provisional; persist container_code when identity matters.
• Records without instance identity are left out, not guessed. You get item IDs, not names.

Run the finished example

Set TOWN_NAME = "Heidel" to the destination you want, then run:

python examples/live_town_storage_contents.py

Once capture starts, enter the world for the first time after the fresh launch, let it settle, and press Enter.

Read the result

HEIDEL CONTENTS
183 observed occupied stacks
item_id=<redacted> quantity=<redacted>
item_id=<redacted> quantity=<redacted>
… 178 more stacks …
item_id=<redacted> quantity=<redacted>

An empty items tuple alone doesn’t prove the town is empty. There are three distinct outcomes:

Use named() for a user-facing name (case-insensitive) and by_id() when persisting — numeric storage_id is authoritative.

Build it yourself

from bdo_toolkit.item_state import CharacterLoadSession

town_name = "Heidel"

with CharacterLoadSession(opcode_profile="opcodes.local") as session:
    input("Enter the world for the first time, then press Enter: ")
    snapshot = session.stop()

if snapshot.decoder_health.storage_status != "compatible":
    raise RuntimeError("Town-storage decoding is unavailable")

storage = snapshot.storages.named(town_name)
if storage is None:
    raise LookupError(f"{town_name} was not returned by this snapshot")
if not storage.current_state_observed:
    raise RuntimeError(f"{town_name} has no selected current state")

if storage.current_empty:
    print(f"{town_name}: explicitly observed empty")
else:
    for item in storage.items:
        print(item.item_id, item.quantity)

Run the finished example

Set TARGET_ITEM_ID = 7003 and its display label "Potato" for your application, then run:

python examples/live_item_state_search.py

Press Enter once the world settles. The label is display-only; the toolkit doesn’t translate item IDs into names.

Read the result

Potato (item_id=7003)
Character inventory: 38
Town storage: 621
  Velia: 421
  Heidel: 200
Observed total: 659

The observed total isn’t an account total. A zero means no retained match in that source, and missing coverage, unavailable storage decoding or identity-unresolved records can lower it. The example reports an unavailable source and prints the combined total only when both sources were observed.

Build it yourself

from bdo_toolkit.item_state import CharacterLoadSession

item_id = 7003

with CharacterLoadSession(opcode_profile="opcodes.local") as session:
    input("Enter the world for the first time, then press Enter when settled: ")
    snapshot = session.stop()

if not snapshot.hydration_detected:
    raise RuntimeError("No character-load item state was detected")

inventory_total = None
if snapshot.inventory.hydration_observed:
    inventory_total = snapshot.inventory.quantity_for(item_id)
    print(f"Character inventory: {inventory_total}")
else:
    print("Character inventory: unavailable")

storage_total = None
if snapshot.decoder_health.storage_status == "compatible":
    storage_total = snapshot.storages.total_quantity(item_id)
    print(f"Town storage: {storage_total}")
    for storage in snapshot.storages.locations_for(item_id):
        label = storage.name or f"UNKNOWN_STORAGE(0x{storage.storage_id:08x})"
        print(f"  {label}: {storage.quantity_for(item_id)}")
else:
    print("Town storage: unavailable")

if inventory_total is not None and storage_total is not None:
    print(f"Observed total: {inventory_total + storage_total}")

inventory.quantity_for() totals the ID across ordinary inventory stacks. storages.total_quantity() totals it across selected current town storage, and locations_for() returns the destinations that hold it.

From traffic to one snapshot

Character-load decoding reuses the item-event engine for TCP reassembly, framing, profile validation and low-level records. Inventory hydration becomes inventory_snapshot. Storage passes through neutral classification and becomes storage_snapshot only when hydration evidence is strong enough.

The assembler also reads selected raw envelopes. A valid count-zero wrapper holds no item record, so its envelope is the only evidence that an inventory group or a town is explicitly empty.

Assembly runs once, at stop() or at end of file. The result is immutable and never applies later item events.

Inventory batches are validated atomically

The profile supplies the opcode, the one-record base length and the item, quantity, context and instance geometry. For a multi-record wrapper, runtime finds a plausible declared count in the prefix and derives the stride from message length, base length and count. The division must be exact, every offset must line up, and every item ID, quantity and instance must validate.

One failing record rejects the whole wrapper. There is no valid-looking prefix and no marker-scan fallback.

Container and slot metadata are a weaker, separate inference: first a record-tail window, then a wrapper-header column, both over complete sibling groups and recognized raw codes. An ambiguous layout or unknown code withholds container metadata for that opcode but keeps valid records in unclassified_records. A known balance needs both the container code and the item ID.

Storage uses stronger finite-capture evidence

Wrapper geometry is profile-driven: the calibrated destination and count fields must agree with message length and full record validation. The numeric destination is kept even without a registry name; no other field is ever substituted.

The live item-event stream leaves an unproven storage group neutral. The item-state assembler has more evidence — a selected inventory boundary, retained count-zero envelopes and the surrounding sweep — so it can prove sparse hydration from an inventory-anchored cohort, or fold timing-split neutral groups into an already proven sweep.

That reconciliation never crosses a confirmed live storage mutation, and stays within one TCP flow generation, storage opcode, bounded inventory epoch and inferred sweep. Live item events keep their fail-neutral behavior.

How current state is selected

A selected count-zero block clears earlier items for that destination. A destination seen only in an earlier sweep stays for provenance with current_state_observed=False, and its old items are withheld — safer than carrying stale contents into a sweep that may itself be partial.

With neither an inventory record nor a count-zero inventory wrapper there is no inventory boundary. The assembler then keeps all storage diagnostics, uses only the latest sweep for contents, records generation_selection="all_observed_storage_no_inventory_boundary" and warns that the evidence may span several loads.

Instance identity controls merging

Inventory is keyed by the decoded item instance, storage by the storage instance within one selected destination block. Repeats of the same opaque identity collapse to the latest record; diagnostics keep the raw and duplicate counts.

A record without its calibrated eight-byte instance never gets a synthetic key. It shows up in missing-identity diagnostics but is left out of stacks, counts, totals, balance interpretation and duplicate inference, so a damaged layout can’t invent state.

Completion stays observational

Patch-recovery boundary

Calibration can update opcodes, base lengths, record spacing and the item, quantity, instance, destination and count positions. Runtime then derives the multi-record stride, the inventory count and container metadata from them.

A redesigned record, a new count encoding, a changed destination meaning or a changed hydration relationship needs decoder work. Ambiguity fails closed — rejected frames, missing metadata, neutral records, incompatible decoder health or no snapshot — never a guessed state.

Why calibration exists

Item decoding uses an explicit local profile that says where current-patch messages carry item IDs, quantities, instances, storage destinations and repeated records. A weekly patch can move those layouts, and the old file can still load normally while a listener quietly produces nothing.

Calibration watches a known item while you perform controlled game actions, then derives message specs from that evidence. Capture and analysis work without a usable old profile and write nothing; saving the result is a separate decision.

Choose how to recover after a patch

A maintained profile isn’t generated automatically: a maintainer captures current traffic, verifies it and promotes it. Right after maintenance, the stable channel can be missing or still point at the previous patch.

Either way, start a fresh live item-activity logger afterward and perform one known item action.

From evidence to a working profile

stop() analyzes the retained frames and returns a read-only CalibrationResult. There’s no global ok flag, because success is judged per profile family: events_found names the promoted families and specs holds their layouts.

update_profile() is the separate write step. Its ProfileUpdate reports written, added, replaced_events and backup_path. A successful write still needs a fresh runtime check.

The walkthroughs use the two-step API so you can inspect the recovered families before deciding to write.

Choose a walkthrough

Know the workflow boundary

Default writes keep unrelated families and back up an existing file before an eligible change. Calibration can recover moved opcodes and record positions, but not a fundamentally redesigned item record, count encoding, destination identity or evidence relationship.

Calibrate in three moves

Prepare five unstackable items

This guide uses five Magic Crystal of Infinity - Skill, raw item ID 15156: cheap, widely available and unstackable. Another unstackable item works too; set ITEM_ID to its raw ID.

• Use five matching copies of the item.
• Put all five in character inventory and none in the chosen storage.
• Pick one unambiguous registered town, such as Velia, Heidel, Calpheon City or Olvia.
• Don’t use Muzgar, Velandir, Yukjo Street, Godu Village or Bukpo.

Run the example and make the moves

python examples/live_calibrate_profile.py

Wait for the prompt before touching storage:

Listening in the background.
Perform the following in-game actions yourself while capture remains open.
Use one unambiguous town such as Velia, Heidel, Calpheon City, or Olvia.
Do not use Muzgar, Velandir, Yukjo Street, Godu Village, or Bukpo.
Using unstackable item 15156 (per-record quantity 1):
deposit one item, then deposit the remaining four,
then withdraw all five matching items in one action.
Press Enter when all three moves are done...

Make each move while the prompt waits

Move only these five crystals. The example keeps listening while the terminal waits for Enter.

Read the result

collected 1439 frames
scanned 1439 frames
...
detected direction(s): inventory->storage, storage->inventory
...
wrote opcodes.local
backup at opcodes_backups\opcodes.local.bak.20260823014827341818
...

This confirms both transfer directions and the local write. The backup line appears only when opcodes.local already existed. An incomplete run exits earlier with incomplete transfer profile; missing ... and writes nothing.

• The analysis result: events_found is the readiness check used here; specs holds the promoted layouts, ignored explains rejections, detected_directions() lists proven directions and retention.truncated flags a cut-off evidence window.
• The write result: written, path, added, replaced_events and backup_path.
• result.summary() and update.summary() are readable reports; neither replaces the family check.

What the sequence proves

Depositing one crystal, then four, produces two distinct repeated-record counts; together they prove the storage destination and record-count fields. Withdrawing all five proves the reverse direction. quantity=1 describes each serialized record, not the batch size.

What runtime uses

The transfer runtime needs all three families: INVENTORY_TRANSFER (the withdrawal and receipt path), STORAGE_ITEM_DELTA (destination-storage records) and SOURCE_STACK_DECREMENT (manual-deposit evidence). Missing any one makes the calibration incomplete, and the example refuses to write it. SOURCE_CONTAINER_DECREMENT and SOURCE_ITEM_REFERENCE can also appear as audit evidence, but runtime ignores their saved entries. Full contract.

These uppercase names are profile families, not lowercase BDOEvent.event_type values.

Build it yourself

from pathlib import Path

from bdo_toolkit.calibration import CalibrationSession, update_profile


profile = Path("opcodes.local")
required = {
    "INVENTORY_TRANSFER",
    "SOURCE_STACK_DECREMENT",
    "STORAGE_ITEM_DELTA",
}

with CalibrationSession(item_id=15156, quantity=1) as session:
    input("Deposit 1, deposit 4, withdraw all 5; then press Enter: ")
    result = session.stop()

missing = required - result.events_found
if missing:
    raise RuntimeError(f"incomplete calibration: {sorted(missing)}")

update = update_profile(result, profile)
print(result.summary())
print(update.summary())

1. The constructor declares the known per-record value; action="auto" is the default and analyzes both directions.
2. Entering the context starts capture; stop() returns the result.
3. The code checks events_found, because nonempty specs doesn’t prove every required family was recovered.
4. update_profile() does the separate file write and returns a ProfileUpdate.

Verify both directions

From the repository root, start a new process so it loads the updated file:

python examples/live_transfer_log.py

1. Wait until the logger says it’s listening.
2. Deposit one calibration crystal into the same allowed town.
3. Withdraw it again.

[09:42:11.038] STORAGE_DELTA source='Player Inventory' destination='Velia' storage_id=0x00000005 item_id=15156 quantity=1
[09:42:14.721] ITEM_RECEIVED source='Storage' item_id=15156 quantity=1

Both lines together verify the practical storage-addition and withdrawal paths.

What it detects

The game can show an Item List before collection, for example after gathering or when you manually loot a defeated mob. Loot-preview support recognizes the traffic that fills that window and delivers a loot_preview event when it appears. It reports the offered item, not whether the window is open.

It’s optional and doesn’t affect detecting items entering inventory, which uses the separate item_received path. Skip this page unless your application needs to know what the preview showed.

Choose a known preview

The example targets Potato, raw item ID 7003. Use a repeatable preview you know shows Potato, or set ITEM_ID and ITEM_NAME to the exact item your preview shows.

The displayed amount varies, so the example uses quantity=None and anchors on the item ID. No town storage or five-crystal sequence is involved.

Run the finished example

python examples/live_calibrate_loot_preview.py

1. Wait until the terminal says it’s listening.
2. Gather a wild potato and wait for the Potato Item List.
3. Once the window has fully appeared, return to the terminal and press Enter.

Read the result

A successful run writes the loot-preview scope to opcodes.local. If the script says loot preview was not calibrated, no update was attempted.

• The analysis: events_found is the check used here; specs and specs_by_event() show the promoted layouts, ignored explains rejections, and retention.truncated tells you when only a retained tail was scored.
• The write: written, added and backup_path.
• result.summary() and update.summary() format both stages for people.

Build it yourself

from pathlib import Path

from bdo_toolkit.calibration import CalibrationSession, update_profile


profile = Path("opcodes.local")

with CalibrationSession(
    item_id=7003,
    quantity=None,
    action="loot-preview",
) as session:
    input("Gather a wild potato; after the Item List appears, press Enter: ")
    result = session.stop()

if "LOOT_PREVIEW" not in result.events_found:
    raise RuntimeError(result.summary())

update = update_profile(result, profile, action="loot-preview")
print(result.summary())
print(update.summary())

1. item_id=7003 anchors the known Potato record.
2. quantity=None means “don’t constrain the watched quantity”, not that the message has no quantity field.
3. events_found uses the uppercase family LOOT_PREVIEW; delivered events use lowercase loot_preview.

Verify the delivered event

Start a fresh process so the decoder loads the changed profile, and request preview events explicitly:

from bdo_toolkit import EventFilter, capture_live


preview_only = EventFilter(event_types={"loot_preview"})

for event in capture_live(
    opcode_profile="opcodes.local",
    event_filter=preview_only,
):
    print(event.source, event.item_id, event.quantity)

Open the same preview again; each appearance should print one line.

Locate the failure stage

Inspect a returned result

from bdo_toolkit.calibration import CalibrationResult


def inspect_result(result: CalibrationResult) -> None:
    print(result.summary())
    print("families:", sorted(result.events_found))
    print("directions:", sorted(result.detected_directions()))

    for reason in result.ignored:
        print("ignored:", reason)

    if result.retention.truncated:
        print("Retry with a shorter, cleaner capture.")

Call inspect_result(result) right after result = session.stop() when a two-step calibration comes back incomplete.

• ignored lists why candidate layouts were rejected; it holds no application events.
• retention.truncated=True means only the newest retained tail was scored.
• Empty or partial events_found means incomplete, for any workflow whose required family is missing.

Match the message you saw

Make the retry unambiguous

• Start capture before the first storage move or before opening the preview.
• For transfers, use five matching unstackables in one allowed town, and do nothing else with items until capture stops.
• For a preview, show the exact item ID configured in the script and avoid other item windows.
• Press Enter only after the final move, or once the preview has fully loaded.

The profile updates, but verification is quiet

After transfer calibration:

1. Stop any logger that was running when the profile changed.
2. From the repository root, start python examples/live_transfer_log.py again and check the printed profile path.
3. Wait for the listening message, then repeat one deposit and one withdrawal.
4. Read any decoder diagnostic printed separately.

After loot-preview calibration: start a fresh listener with the explicit loot_preview filter from Verify the delivered event and reopen the same preview. The transfer logger deliberately excludes previews.

If storage decoding stays incompatible after another clean calibration, keep the terminal diagnostic for the report.

From captured traffic to a profile

Calibration collects complete inbound messages around the known item action. Analysis finds plausible item records, classifies their transfer direction, and proves the offsets and repeated-record shape each profile family needs. Only an unambiguous candidate becomes a MessageSpec.

The CalibrationResult is read-only. Saving is a separate update_profile() step, and a fresh decoder validates the new file again when it loads. Specs store normalized one-record geometry; the observed action totals are evidence, not runtime constants.

Transfer families and runtime use

The transfer runtime needs INVENTORY_TRANSFER, STORAGE_ITEM_DELTA and SOURCE_STACK_DECREMENT. Without any one of them, transfer calibration is incomplete.

INVENTORY_TRANSFERrequired

Storage withdrawals, item_received, and inventory hydration and state.

STORAGE_ITEM_DELTArequired

Storage deposits, destination-town identity, and storage hydration and state.

SOURCE_STACK_DECREMENTrequired

Reliably promotes a manual deposit to storage_delta with source="Player Inventory".

SOURCE_CONTAINER_DECREMENTnot used at runtime

Kept only as calibration and audit evidence. A context-after-quantity layout is allowed only when an exact receipt-instance anchor proves the companion.

SOURCE_ITEM_REFERENCEnot used at runtime

Kept only as calibration and audit evidence.

LOOT_PREVIEW is a separate, optional event-producing family. Automatic transfer calibration doesn’t look for it; the loot-preview walkthrough uses an explicit action and keeps the transfer families.

How transfer direction is classified

The default action="auto" evaluates both transfer recipes and records every candidate’s decision in DirectionEvidence. Direction comes from structural signals, not an opcode table.

How storage-wrapper geometry is proven

Before promoting STORAGE_ITEM_DELTA, calibration must uniquely identify both the destination field and the declared record-count field. Across the captured deposits it intersects registered-destination candidates with two-byte count candidates whose values fit the observed record geometry. It needs at least two distinct validated deposit counts; the one-record and four-record deposits provide that cross-check.

The excluded-town restriction applies only while learning the offset. A profile calibrated in an unambiguous town handles those destinations normally.

If either column stays ambiguous, CalibrationAuthorityError is raised before any result or write. Once both are proven, calibration records the one-record base length, context_offset and record_count_offset. Runtime derives each wrapper’s stride from its declared count, actual length and that base, and validates every declared record before emitting any neutral storage record.

How source-decrement geometry is proven

SOURCE_STACK_DECREMENT supplies the inventory-side evidence that identifies a manual storage deposit. Quantity and instance field order aren’t treated as patch constants: calibration anchors the exact source instance when available, tests non-overlapping quantity phases and measures the repeated-record period.

For a repeated decrement, only one longest field-valid shape matching the storage record count is promoted. Ambiguous phases, overlapping fields and stride aliases that skip records are withheld without discarding independently proven families. At runtime, an instance-aware spec never falls back to a coincidental quantity when its instance field is empty or invalid.

Storage-source classification is a separate runtime stage. A confirmed worker companion chain wins and assigns source="Worker Production"; otherwise calibrated decrement evidence can assign source="Player Inventory". With neither signal, proven live activity gets source=None, and a record lacking independent live or hydration proof stays storage_record.

Evidence retention and scope

Live calibration keeps the newest contiguous evidence tail under two default bounds: 50,000 frames and 64 MiB of payload. Analysis scores that retained tail, not every frame observed; CalibrationRetention reports the observed, retained, discarded and truncated counts.

A truncated result covers only the newest tail. Offline calibrate_pcap() analyzes a finite recording without the live bounds.

Patch-recovery boundary

From one leaderboard load to reusable data

The leaderboard updates daily on the game’s schedule; players can’t refresh it. Start capture before the Leaderboard tab’s first load after restarting the game. The toolkit passively observes the two responses and returns one SolareCaptureResult. It emits no BDOEvent objects and never controls the game.

Always branch on result.complete. A complete result carries one immutable SolareLeaderboardSnapshot in result.snapshot. A non-complete one keeps status, message and evidence with snapshot=None: partial rows are never shown as leaderboard state.

overall_top_100SolareOverallEntry rows

The authoritative overall ranks 1–100.

playersSolarePlayer rows

The class top-20 tables, flattened. Not a global top 620.

The two responses decode independently. A name can appear in either or both, and optional values are never copied between row types. Check overall_capabilities or class_table_capabilities before relying on Elo or performance fields.

Capture once, replay as often as needed

Save the first successful live load as PCAPNG. That one file then drives every replay and query example, without reopening the game or waiting for another load.

Choose a walkthrough

Run the finished example

From the repository root, before the first Leaderboard load of this game session:

python examples/solare_live_snapshot.py --save-pcap solare-leaderboard.pcapng

1. If the Leaderboard was already opened this game session, restart Black Desert.
2. Wait for the capture-ready message before touching Arena of Solare.
3. Open the Leaderboard tab once. The broader Arena menu isn’t enough.
4. Leave the tab visible while both leaderboard responses arrive.
5. The script stops on confirmation or after its 120-second default deadline. Ctrl+C ends it earlier.

Recording never overwrites a file, so use a new path for each attempt. Without --save-pcap, the script captures and reports the same way but writes nothing.

[capture-ready] passive Solare capture is ready ..., recording ...\solare-leaderboard.pcapng; open the Leaderboard tab for its first load after restarting the game
[snapshot-confirmed] opcode-agnostic Arena of Solare leaderboard confirmed
[capture-finished] stop reason: complete-snapshot
[saved-capture] ...\solare-leaderboard.pcapng

Once one complete load is saved, reuse that file for the query walkthroughs instead of reopening the game.

Build the capture yourself

from pathlib import Path

from bdo_toolkit.solare import capture_solare_snapshot

capture_path = Path("solare-leaderboard.pcapng")
print("Open the Arena of Solare Leaderboard once.")
result = capture_solare_snapshot(save_pcap=capture_path)

if not result.complete:
    raise RuntimeError(f"{result.status.value}: {result.message}")

snapshot = result.snapshot
assert snapshot is not None
print("complete: leaderboard snapshot ready")
print(f"saved capture: {capture_path}")
print(f"overall rows={len(snapshot.overall_top_100)}")
print(f"class-table rows={len(snapshot.players)}")

Open the Arena of Solare Leaderboard once.
complete: leaderboard snapshot ready
saved capture: solare-leaderboard.pcapng
overall rows=100
class-table rows=620

• capture_solare_snapshot() owns one bounded live run and returns one result.
• save_pcap records the accepted packets and refuses an existing destination.
• complete is the safe branch; arriving traffic alone proves nothing.
• snapshot exists only on a complete result and roots every leaderboard query.

Show progress in an application advanced

The finished example polls manually because it owns a deadline and heartbeats. When the toolkit’s milestones are enough, LiveSolareSession.updates() is simpler:

from pathlib import Path

from bdo_toolkit.solare import LiveSolareSession

with LiveSolareSession(
    save_pcap=Path("solare-leaderboard.pcapng"),
) as session:
    print("Open the Arena of Solare Leaderboard once.")

    for update in session.updates():
        print(f"[{update.kind.value}] {update.message}")

result = session.result
if result is None or not result.complete:
    raise RuntimeError("leaderboard capture did not complete")

Open the Arena of Solare Leaderboard once.
[capture-ready] passive Solare capture is ready ...
[ranked-progress] 500 unique ranked players recovered; waiting for class balance and the top-100 cross-check
[snapshot-confirmed] opcode-agnostic Arena of Solare leaderboard confirmed
[finished] Solare capture finished with status complete

updates() waits internally, yields through the terminal finished update, then ends. snapshot-confirmed is a milestone, not the end; the final update carries update.result, also kept as session.result.

Use wait() when progress doesn’t matter, poll() for your own deadlines or heartbeats, and stop() for manual shutdown. Breaking out of updates() doesn’t stop capture; exit the context or stop it yourself.

Use the terminal result

On success, continue through snapshot.overall_top_100, snapshot.players or the snapshot lookups. Otherwise print status.value and message, and read evidence only when you must tell incomplete structure from capture loss.

The same function can change its deadline, report progress through on_update or keep opaque extension bytes. None of that changes the rule: only a complete result publishes a snapshot.

Run the finished example

python examples/solare_overall_top_100.py solare-leaderboard.pcapng

OVERALL TOP 100
  1. ExamplePlayer01 | Elo 3045
  2. ExamplePlayer02 | Elo 3012
  3. ExamplePlayer03 | Elo 2988
... 97 more rows ...

Build the ranking loop yourself

from bdo_toolkit.solare import replay_solare

result = replay_solare("solare-leaderboard.pcapng")
if not result.complete or result.snapshot is None:
    raise RuntimeError(f"{result.status.value}: {result.message}")

snapshot = result.snapshot
has_elo = "elo" in snapshot.overall_capabilities

for entry in snapshot.overall_top_100:
    elo = entry.elo if has_elo else "unavailable"
    print(entry.global_rank, entry.name, elo)

1 ExamplePlayer01 3045
2 ExamplePlayer02 3012
3 ExamplePlayer03 2988

On a toolkit-built complete result, snapshot.overall_top_100 is a tuple of 100 SolareOverallEntry rows. The loop prints global_rank, name and elo.

Use the row beyond rank and name

The same row can expose direct aggregate total_wins, total_draws and total_losses; derived total_matches and total_win_rate; and its own classes_played. Read player statistics checks the capability set before using them.

Run the finished example

This command uses class code 11, Lahn. Pick another from the class codes:

python examples/solare_class_top_20.py solare-leaderboard.pcapng 11

LAHN TOP 20
  4. ExampleLahn01 | Elo 2960
 17. ExampleLahn02 | Elo 2814
 29. ExampleLahn03 | Elo 2730
... 17 more rows ...

Build the class query yourself

from bdo_toolkit.solare import replay_solare

result = replay_solare("solare-leaderboard.pcapng")
if not result.complete or result.snapshot is None:
    raise RuntimeError(f"{result.status.value}: {result.message}")

snapshot = result.snapshot
lahn_top_20 = snapshot.class_leaderboard(11)
has_elo = "elo" in snapshot.class_table_capabilities

for player in lahn_top_20:
    elo = player.elo if has_elo else "unavailable"
    print(player.global_rank, player.name, elo)

4 ExampleLahn01 2960
17 ExampleLahn02 2814
29 ExampleLahn03 2730

class_leaderboard(11) filters the class-table players by primary class and returns SolarePlayer rows in global-rank order. It never derives a class table from overall_top_100.

Read a class-table row

Each SolarePlayer owns name, global_rank, primary_class, optional elo and per-class classes_played. class_table_capabilities says which optional fields exist.

global_rank stays the global rank; the tuple position is the order within the class. Under the supported geometry, a complete snapshot returns 20 rows for every supported class.

Run the finished example

python examples/solare_find_player.py solare-leaderboard.pcapng <EXACT_PLAYER_NAME>

Use a case-sensitive name printed from your saved overall or class leaderboard.

overall top 100: rank 73
class tables: not found

This is a valid result: a place in the overall top 100 doesn’t require a place in any class top 20.

Build both lookups yourself

from bdo_toolkit.solare import replay_solare

name = "ExamplePlayer"  # Replace with an exact name from your capture.
result = replay_solare("solare-leaderboard.pcapng")
if not result.complete or result.snapshot is None:
    raise RuntimeError(f"{result.status.value}: {result.message}")

snapshot = result.snapshot
overall_entry = snapshot.get_overall_entry(name)
class_entry = snapshot.get_player(name)

print(f"overall={overall_entry is not None}")
print(f"class-table={class_entry is not None}")

overall=True
class-table=False

get_overall_entry(name) searches only overall_top_100 and returns a SolareOverallEntry or None. get_player(name) searches only the class-table players and returns a SolarePlayer or None.

Keep the two results separate

• Name matching is exact and case-sensitive.
• Either lookup can return None on its own.
• When both succeed, they’re independently decoded objects. Never copy missing Elo, classes or performance from one into the other.

Check overall_capabilities before reading optional fields from overall_entry, and class_table_capabilities before reading them from class_entry.

Run the finished example

python examples/solare_player_statistics.py solare-leaderboard.pcapng <EXACT_PLAYER_NAME>

Use a case-sensitive name from the saved overall top 100.

7. ExamplePlayer | Elo 2874
overall: 58W/2D/40L (100 matches)
Lahn (Awakening): 24W/1D/15L (40 matches)
Ranger (Succession): 18W/0D/12L (30 matches)
Sorceress (Succession): 10W/1D/5L (16 matches)

Build the statistics view yourself

from bdo_toolkit.solare import replay_solare

name = "ExamplePlayer"  # Replace with an exact overall name from your capture.
result = replay_solare("solare-leaderboard.pcapng")
if not result.complete or result.snapshot is None:
    raise RuntimeError(f"{result.status.value}: {result.message}")

snapshot = result.snapshot
entry = snapshot.get_overall_entry(name)
if entry is None:
    raise LookupError(f"{name!r} is not in the overall top 100")

required = {"elo", "aggregate_performance", "performance"}
missing = required - snapshot.overall_capabilities
if missing:
    raise RuntimeError(f"overall details unavailable: {sorted(missing)}")

print(entry.elo, entry.total_matches, entry.total_win_rate)
for record in entry.classes_played:
    class_name = record.player_class.name or record.player_class.code
    specialization = record.specialization
    specialization_name = (
        "unavailable"
        if specialization is None
        else (specialization.name or specialization.branch)
    )
    print(
        class_name,
        specialization_name,
        record.matches,
        record.wins,
        record.draws,
        record.losses,
        record.win_rate,
    )

2874 100 58.0
Lahn Awakening 40 24 1 15 60.0
Ranger Succession 30 18 0 12 60.0
Sorceress Succession 16 10 1 5 62.5

get_overall_entry() returns one SolareOverallEntry; each value in its classes_played is a SolareClassPerformance.

Separate aggregate and per-class outcomes

entry.elooverall row

Elo decoded from this overall row.

total_wins · total_draws · total_lossesdirect

The row’s own aggregate outcome tuple.

total_matches · total_win_ratederived

Computed from the complete aggregate tuple.

classes_playedper class

Zero to three detailed records from the same overall response: player_class, specialization, matches, W/D/L and win_rate.

Starting from a class-table SolarePlayer instead? Check class_table_capabilities and read that row’s elo, primary_class and classes_played. It has no overall total_* fields.

From captured traffic to a snapshot

The Solare route shares passive packet acquisition, TCP reassembly and generic message framing with the rest of the toolkit, then builds only the Solare classifier and result. It doesn’t run the item-event decoder, load an item opcode profile or emit BDOEvent values.

A snapshot needs structural confirmation and clean acquisition first. Optional detail decoding reads only the exact class-table and overall windows that discovery already selected, so it can’t promote menu-only, partial, ambiguous, gapped or dropped traffic.

What the runtime discovers and what it assumes

How leaderboard identity is proven

The scanner works per reassembled directional TCP flow, starts a new generation when a four-tuple reopens, and can synchronize after attaching midstream. Candidate families are keyed by:

(TCP four-tuple, connection generation, observed opcode, message length)
    → repeated candidate family

No player among global ranks 1–20 can have 20 same-class players ahead under the per-class cap, so those first 20 must appear in the class-table union. Rank 21 and later may legitimately be missing from every class top 20.

How optional details are recovered

Each value is reread from its own response. Shared-player agreement can prove or disambiguate an offset, but no overall field is filled from a class row, and no class-table field from the overall response.

Capability publication and source independence

class_table_capabilities and overall_capabilities are authoritative for their own responses, and capabilities is their intersection. Missing optional data means unavailable: never zero, a slot sum or a value copied from the other table.

Structural confirmation can be announced before optional-field finalization completes. Registered geometries stay on the fast path.

Capture integrity and bounded resources advanced

A leaderboard load can deliver more than 5 MiB in under a second. Windows live capture therefore requests a 64 MiB Npcap buffer, keeps packet callbacks minimal, and moves writing and decoding to a worker. Timeout recovery waits for capture finalization, so callback latency, native batching or worker backlog can’t make the decoder skip a prefix that’s still arriving.

With stop_on_complete=False, live capture latches the clean health of its first confirmed snapshot; replay reports health across the whole recording.

Patch-recovery boundary

A complete result with the expected capabilities proves the classifier and detail path still work. A complete rankings-only result localizes the change to optional details; a partial result points to acquisition or structural assumptions instead.

Confidence and research boundary

• Synthetic tests rewrite opcodes arbitrarily while preserving structure, and cover the exact overlap threshold, rank/name contradictions, stale prefixes, multiple refreshes, capture gaps, retention rollover and exact-50 closure.
• Six complete private captures replay across several opcode and geometry shifts. Five registered-layout captures forced through the learner reproduce every public field and opted-in raw byte, and a genuinely unregistered July 30 geometry succeeds through the ordinary ephemeral path.
• Adversarial tests corrupt or duplicate Elo, performance, aggregate, history and raw candidates one at a time, and verify that only the affected group is withheld under its path’s dependency rules.

All six real captures place every overall top-100 player somewhere in the class-table union, so the legitimate overall-only case is proven synthetically, including its directly decoded aggregate, until a divergent real capture is observed.

No public player identifier is exposed. Several nonzero cross-table eight-byte candidates look convincing, older generations use different representations, and the strongest remaining candidate still doesn’t prove account, family, character, lifetime or future stability. Identifier-like bytes stay outside learning, capabilities, serialization and snapshot identity.

Likewise, recent-result regions keep literal wire text without naming individual codes, and gear and addon regions stay opaque. Aggregate W/D/L has strong whole-table and cross-generation evidence; its exact in-client wording stays conservative until a controlled UI comparison settles it.

What asyncio integration changes

There’s no package-wide async mode. PCAP replay, CharacterLoadSession, GrindSession, LiveXPSession, LiveRawSession, profile loading and updates, formatting and serialization stay synchronous. To run one of them from asyncio, move the whole call onto a worker thread yourself.

Application-controlled item capture

A long-running application consumes events in one task, while a UI action or shutdown signal decides when capture stops:

import asyncio

from bdo_toolkit import AsyncLiveCaptureSession


async def consume(session: AsyncLiveCaptureSession) -> None:
    async for event in session.events():
        print(event.format_human())


async def main() -> None:
    stop_requested = asyncio.Event()

    async with AsyncLiveCaptureSession(opcode_profile="opcodes.local") as session:
        consumer = asyncio.create_task(consume(session))
        try:
            # Your Stop button or shutdown path sets this.
            await stop_requested.wait()
        finally:
            await session.stop()
            await consumer


asyncio.run(main())

events() ends once capture stops and buffered events drain, so stopping first, then awaiting the consumer, delivers every finalized event.

Where await belongs

awaitlifecycle · one-shot polls

session.start(), session.stop() and session.poll() can wait on capture or cleanup.

async forstreams

session.events() and Solare session.updates() are async iterators: consume them, don’t await them.

plain readsstate · models

session.running, session.result and event.to_dict() are ordinary synchronous reads.

asyncio.to_thread()blocking work

For a synchronous call too slow for the event loop, such as update_profile() after await session.stop() returns a calibration result.

Cancellation and cleanup

Cancelling an async lifecycle or consumption call is a shutdown request, not permission to abandon the capture. The facade finishes bounded cleanup for work already started before asyncio.CancelledError escapes.

• Each facade is single-use; create a new one for each capture or calibration run.
• Item capture and Agris allow one consumer. Solare allows one progress consumer plus one completion waiter.
• If cleanup_incomplete is true, retry await stop() on the same facade instead of replacing it.
• An exception already leaving an async with body stays primary while cleanup runs.

Identify the outcome first

Choose a problem

If an exception was raised

ProfileErrorprofiles

A profile failed to parse, validate, load or fetch.

CaptureIntegrityErrorcapture

Known acquisition loss: packet drops, gaps or a full queue.

CalibrationAuthorityErrorcalibration

Calibration found the target but couldn’t prove a safe decoder profile.

DirectionMismatchErrorcalibration

The capture contradicts an explicitly declared transfer direction.

XPTrackingError · AgrisDetectionErrorgrind tracking

A standalone XP or Agris reader lost authority; in GrindSession these arrive as events instead.

How the pipelines decide

When the outcome doesn’t match what you expected, the internals guides explain each decision: item events (pipeline, classification, ordering and capture bounds), calibration (evidence and runtime geometry) and Arena of Solare (structural discovery, detail inference and capture integrity).

Confirm it’s an acquisition problem

Find out what candidate traffic, if any, reached the workflow. Each API reports a signal at a different stage:

Once acquisition is established, continue with the owning workflow.

Inspect the selected interface and filters

1. Where the session exposes it, read session.endpoint after startup: its interface, local_ip and bpf_filter are fixed for the run.
2. If the interface is missing or unexpected, run scapy.all.show_interfaces() and copy the exact physical or virtual adapter name into interface=.
3. While diagnosing a pinned interface, leave local_ip=None so a wrong destination address can’t hide matching traffic.
4. Start a fresh session, wait until it’s ready, and perform one known in-game action.

Account for VPN routing

1. Stop the session, set up the VPN state and routing your application will use, then start a new session.
2. For a full tunnel, test the VPN’s virtual adapter that exposes decrypted IPv4/TCP, not the physical adapter carrying encrypted tunnel packets.
3. For split or per-application routing, check the physical adapter too: BDO can take a different route from the computer’s default.
4. If no adapter visible to the capture backend (Npcap on Windows) exposes decrypted BDO traffic, live decoding isn’t possible with that VPN setup. Offline replay works only if a compatible recording can be made.

Check the capture backend, permissions and filtering

Retry, then continue with the workflow

Once acquisition works and the workflow’s integrity or diagnostic signal shows no known loss, continue with the symptom:

Establish acquisition first

Use an explicit LiveCaptureSession and confirm that packets matching its capture filter reach the raw queue before you read anything into decoder silence.

If session.health.packets_accepted stays zero, or the selected endpoint is uncertain, finish the shared adapter, VPN, backend and filter checks first.

Ask for every event, briefly

Pass EventFilter.all() to separate capture or decoding problems from the ordinary activity preset. If records appear only then, acquisition works, and classification or your earlier filter excluded them.

inventory_record and storage_record are neutral decoded records, not confirmed activity. Their fields and decoder_health show whether calibration geometry or evidence-based promotion is missing.

Check the profile against the current patch

1. Confirm the selected file exists, is active, and has the profile families the tested action needs.
2. Perform one known action after capture is ready, and compare it with the same action from verification.
3. If traffic is healthy and all() stays quiet, fetch the maintained profile again or follow Calibration.

Storage-specific checks

Start with the two root checks

Neither is a completion marker. True doesn’t prove every container or registered town appeared.

Read coverage as a reason, not a score

Retry the load, not the queries

1. For storage, restart Black Desert and start a fresh CharacterLoadSession before the first character enters the world. A later switch only rehydrates inventory.
2. Let the world settle before stopping. Opening a town storage afterward doesn’t recreate the load window.
3. Inspect state.warnings and state.decoder_health. Use state.diagnostics only when the ordinary checks don’t explain the gap.
4. Keep the earlier result as evidence, but don’t combine snapshots as if they shared one completion boundary.

Match the returned status

no-traffic

No usable inbound game-server traffic. Next: verify the shared live-capture setup; if the Leaderboard was already opened, restart the game first.

menu-context

Solare menu or history-like traffic, but no complete Leaderboard response. Next: restart the game, start capture, then open the Leaderboard tab itself.

ranked-partial

Ranked structure found, but the class-balanced table and overall cross-check were incomplete. Next: save the evidence, then retry after a game restart.

rich-candidate

The class-balanced table was complete, but the independent overall top-100 cross-check wasn’t captured. Next: save the evidence, then retry after a game restart.

detected-incomplete

Relevant traffic recognized, but acquisition or structural checks didn’t prove a complete snapshot. Next: inspect TCP gaps, native drops, queue overflows and flow evictions, then retry after a game restart.

inconclusive

No supported complete interpretation won. Next: with clean health and a complete Leaderboard load, keep the capture and investigate patch-era layout drift.

Keep completion fail-closed

Completion currently needs the 31 × 20 class-table geometry, an authoritative overall top 100 and exact overlap checks. Early in a season, an underfilled class can make a readable response incomplete, and a new class needs reviewed geometry support once populated traffic exists.

Restart the game and retry a clean capture first.

A complete snapshot can still lack optional details

Missing Elo, aggregate outcomes, performance or raw extensions don’t invalidate confirmed rankings. Check snapshot.overall_capabilities and snapshot.class_table_capabilities; unavailable stays distinct from zero.

Start every report with the same context

import platform
import sys

import scapy

import bdo_toolkit

print(bdo_toolkit.__version__, sys.version.split()[0], scapy.VERSION, platform.platform())

1.0.8 3.14.0 2.7.0 Windows-11-10.0.26200-SP0

• Whether the run was live or replayed, and whether the problem repeats in a fresh process and capture.
• For live capture: the adapter name, whether it’s physical or virtual, and its model and driver version when available.
• On Windows: the Npcap version, and whether WinPcap API-compatible mode is on.
• With a VPN: full-tunnel or split routing, and whether its state changed after capture started.
• The in-game action taken after capture was ready, and the outcome actually received.
• Where the active opcode profile came from, and whether it was verified after the current weekly patch.

Add the workflow’s own diagnostics

Item events

The resolved session.endpoint, final session.health and session.decoder_health, diagnostic messages, and one redacted event.to_dict() if an unexpected event appeared.

Inventory & town storage

hydration_detected, identity_complete, coverage, warnings and decoder_health. Add capture diagnostics only if the ordinary schema-v5 result doesn’t explain the gap.

Calibration

The terminal message, ignored reasons, DirectionEvidence, action mode, controlled quantity and destination town; never just “calibration failed”.

Arena of Solare

Status and message, ranked, overall and cross-check counts, both capability sets, capture health and layout identifiers. The support report example prints them.

Grind tracking

Final status.to_dict() and health.to_dict(), any GrindFeatureError reasons, and which readers were enabled.

Escalate to raw data only when necessary

A structured report usually tells setup, profile, acquisition and geometry problems apart without exposing the traffic. If a maintainer needs the PCAP, agree first on the smallest capture window and a private transfer method. Never include raw session identifiers or private packet contents in a public report.

Passive and read-only boundary

The toolkit only observes local traffic. Calibration analyzes controlled traffic and writes nothing but an explicit local profile file. Optional PCAP, JSONL, item-state, Solare, raw-message and origin-candidate outputs are local writes your application asks for.

What saved outputs can contain

PCAP or PCAPNGevery save_pcap

Network endpoints, complete captured payloads, decoded and undecoded game data, and identifier-like bytes.

Item-event JSONL--jsonl · to_dict()

Timestamps, endpoints, event types, sources, item IDs, quantities, storage destinations and retained evidence.

Item-state JSONto_dict()

Observed inventory, balances, town-storage contents, warnings and decoder health. include_diagnostics=True adds capture diagnostics and the local capture_path.

Solare result JSONto_json() · CLI

Player names, rankings, Elo, aggregate outcomes, recent-result text and per-class performance. Raw sections need explicit retention and serialization.

XP and Agris readingsto_dict() · --jsonl

Your character’s level and XP, Agris balances and cap, timestamps and connection endpoints.

Raw message dumpsRawMessage.to_dict()

Every server-to-client payload byte from the selected ports, with endpoints and timestamps.

Origin-candidate JSONsave()

Structural observations, opaque IDs and short digests for review; raw shared tokens are never stored.

Share the minimum necessary

1. Start with the structured fields in Prepare a support report.
2. Read the exact file before sending it. Redact endpoints, local paths, names, item contents and identifiers the diagnosis doesn’t need.
3. Keep include_diagnostics, retain_raw_extensions and include_raw off unless the recipient needs those fields.
4. Use the shortest useful capture window, a private transfer channel, and an agreed retention and deletion plan.

replay_pcap()

Replays a .pcap or .pcapng file incrementally. With event_filter=None the stream is complete.

Example

from bdo_toolkit import replay_pcap

for event in replay_pcap(
    "path/to/session.pcapng",
    opcode_profile="path/to/matching-opcodes.json",
):
    print(event.to_dict())

{'event_type': 'item_received', 'timestamp': 1783663561.245, 'timestamp_iso': '2026-07-10T06:06:01.245Z', 'flow': {'source_ip': '203.0.113.20', 'source_port': 8889, 'destination_ip': '192.0.2.42', 'destination_port': 51824}, 'item_id': 7003, 'quantity': 6, 'source': 'Gathering', 'confidence': 'observed', ...}

Parameters

path*str | Path

PCAP or PCAPNG recording.

opcode_profile*str | Path | OpcodeProfile

Item profile from the recording’s patch era. See opcode profile setup.

portstuple[int, ...] = (8884, 8885, 8889)

Server source ports that select traffic.

event_filterEventFilter | None = None

Delivery policy. None delivers every decoded event.

origin_observerCallable[[CompanionObservation], object] | None = None

Advanced · Origin-learning callback.

on_diagnosticCallable[[DecoderDiagnostic], object] | None = None

Synchronous, out-of-band diagnostic callback.

Example

from threading import Thread

from bdo_toolkit import EventFilter, LiveCaptureSession

mob_drops = EventFilter(sources={"Mob Drop"})
session = LiveCaptureSession(
    opcode_profile="opcodes.local",
    event_filter=mob_drops,
)
worker_errors = []


def consume_events():
    try:
        for event in session.events():
            print(event.event_type, event.source, event.item_id, event.quantity)
    except BaseException as exc:
        if not worker_errors:
            worker_errors.append(exc)


worker = Thread(target=consume_events, daemon=True)


def on_start():
    session.start()
    print("Capture started")
    worker.start()


def on_stop():
    session.stop()
    worker.join()
    if worker_errors:
        raise worker_errors[0]
    session.raise_if_failed()
    print("Capture stopped:", session.stop_reason)

from bdo_toolkit import EventFilter, LiveCaptureSession

mob_drops = EventFilter(sources={"Mob Drop"})
session = LiveCaptureSession(
    opcode_profile="opcodes.local",
    event_filter=mob_drops,
)

def on_start():
    session.start()
    print("Capture started")


def on_ui_timer():
    while (event := session.poll(timeout=0)) is not None:
        print(event.event_type, event.source, event.item_id, event.quantity)


def on_stop():
    session.stop()
    on_ui_timer()  # Display events finalized during shutdown.
    print("Capture stopped:", session.stop_reason)

The worker keeps the first consumer failure, stops on the control thread, joins, then re-raises. With the UI form, call on_ui_timer() periodically and once after stop().

Capture started
item_received Mob Drop 767248 8
item_received Mob Drop 11882 1
Capture stopped: requested

Parameters

opcode_profile*str | Path | OpcodeProfile

Current local item profile, loaded when capture starts. See opcode profile setup.

live_optionsLiveCaptureOptions | None = None

Live capture and queue settings.

event_filterEventFilter | None = None

Delivery policy. None means EventFilter.activity().

origin_observerCallable[[CompanionObservation], object] | None = None

Advanced · Origin-learning callback.

on_diagnosticCallable[[DecoderDiagnostic], object] | None = None

Synchronous, out-of-band diagnostic callback.

Methods

start()→ None

Opens the adapter and returns once capture is ready; decoding continues in the background. Success makes the session single-use. After a failed start, retry start() only if cleanup_incomplete is False; otherwise retry stop().

stop()→ None

Stops intake, waits for the decoder, finalizes pending storage classification and wakes the consumer. It doesn’t drain the queue or re-raise a retained failure: keep consuming, or call raise_if_failed(). Repeat calls are no-ops.

request_stop()→ None

Requests shutdown without waiting; the stop to use inside callbacks. Keep consuming, then call stop() from control code to finish.

events()→ Iterator[BDOEvent]

Blocking iterator for the one consumer, without busy-waiting. After stop() it drains queued and shutdown-finalized events, then ends or raises the retained failure. Breaking out doesn’t stop capture.

poll(timeout=None)→ BDOEvent | None

One event, for timer-driven loops. 0 checks without waiting; None waits for an event or stop. Returns None on timeout or after the final event (check stopped), and re-raises a retained failure once buffered events drain.

raise_if_failed()→ None

Re-raises the first retained failure now, in the calling thread.

__enter__ · __exit__context protocol

Enter calls start(). Exit stops a started session that hasn’t stopped, without raise_if_failed(); the block’s exception stays primary.

Attributes

Example

import asyncio

from bdo_toolkit import AsyncLiveCaptureSession, EventFilter


async def consume(session: AsyncLiveCaptureSession) -> None:
    async for event in session:
        print(event.format_human())


async def main() -> None:
    async with AsyncLiveCaptureSession(
        opcode_profile="opcodes.local",
        event_filter=EventFilter(
            event_types={"item_received", "storage_delta"},
        ),
    ) as session:
        consumer = asyncio.create_task(consume(session))
        try:
            await asyncio.to_thread(
                input,
                "Capture started. Press Enter to stop.\n",
            )
        finally:
            await session.stop()
            await consumer


asyncio.run(main())

input() stands in for your app’s Stop control. Stopping first, then awaiting the consumer, keeps events finalized during shutdown.

Capture started. Press Enter to stop.
[14:08:31.204] ITEM_RECEIVED source='Gathering' item_id=7003 quantity=4

Parameters

opcode_profile*str | Path | OpcodeProfile

Current local item profile, loaded when capture starts. See opcode profile setup.

live_optionsLiveCaptureOptions | None = None

Live capture and queue settings.

event_filterEventFilter | None = None

Delivery policy. None means EventFilter.activity().

origin_observerCallable[[CompanionObservation], object] | None = None

Advanced · Origin-learning callback.

on_diagnosticCallable[[DecoderDiagnostic], object] | None = None

Synchronous, out-of-band diagnostic callback.

Methods

await start()→ None

Opens capture off the event loop and returns once the adapter is ready. One start attempt per session, even a failed or cancelled one; if cleanup is incomplete, keep the session and retry stop().

await stop()→ None

Finalizes without blocking the loop; concurrent callers share one shutdown. It doesn’t consume buffered events or re-raise a retained failure: keep consuming, or call raise_if_failed(). Repeat calls return.

request_stop()→ None

Requests shutdown without waiting; the stop to use inside callbacks. Keep consuming, then await stop().

events()→ AsyncIterator[BDOEvent]

Async iterator for the one consumer; waits without blocking the loop. After stop it drains queued and shutdown-finalized events, then ends or raises the retained failure. Breaking out doesn’t stop capture, and unread events stay available.

await poll(timeout=None)→ BDOEvent | None

One event, or None on timeout or after the final event. 0 checks at once. A retained failure is raised once buffered events drain. Cancelling a pending poll stops the session.

raise_if_failed()→ None

Re-raises the retained failure now, synchronously.

__aiter__async iteration

async for event in session uses events(); it isn’t a second consumer.

__aenter__ · __aexit__async context protocol

Enter awaits start(). Exit stops a session that hasn’t stopped, without raise_if_failed(); the block’s exception stays primary.

Attributes

Example

from bdo_toolkit import LiveCaptureOptions, capture_live

options = LiveCaptureOptions(
    event_queue_size=2048,
    packet_queue_size=8192,
)

for event in capture_live(
    opcode_profile="opcodes.local",
    live_options=options,
):
    print(event.format_human())

Keeps the packet defaults and doubles both queue bounds.

Parameters

interfacestr | None = None

Inherited adapter name.

local_ipstr | None = None

Inherited destination-IPv4 restriction.

portstuple[int, ...] = (8884, 8885, 8889)

Inherited server source ports.

use_bpfbool = True

Inherited backend or Python-side filtering.

auto_local_ipbool = True

Inherited default-route local-IP policy.

event_queue_sizeint = 1024

Decoded events waiting for the one consumer. When full, the decoder waits for space; shutdown-finalized events are kept separately.

packet_queue_sizeint = 4096

Captured packets waiting for decoding. The native callback never waits: an overflow is known loss, so capture stops and records CaptureIntegrityError.

A full event queue pauses decoding, so a consumer that stays slow eventually overflows the packet queue.

Accepted by

Pass it as live_options=:

• Item events: capture_live(), LiveCaptureSession, AsyncLiveCaptureSession
• Grind tracking: GrindSession, LiveXPSession, LiveAgrisSession, AsyncLiveAgrisSession, calibrate_agris_live(), calibrate_agris_and_update()

Example

When automatic selection picks the wrong adapter, list the interfaces and pass the exact Scapy name, not an IP address, as interface=.

from bdo_toolkit import PacketCaptureOptions
from scapy.all import show_interfaces

show_interfaces()
options = PacketCaptureOptions(
    interface="Ethernet",  # Replace with the displayed name
)

Parameters

interfacestr | None = None

Capture adapter name. None tries the generic default route, not BDO’s own route. Availability is checked at start.

local_ipstr | None = None

Destination IPv4 restriction, not an adapter selector; validated and normalized at construction. With an explicit interface, None leaves it unrestricted.

portstuple[int, ...] = (8884, 8885, 8889)

Server source ports treated as BDO traffic: at least one int from 1 to 65,535. Duplicates are dropped; first-seen order is kept.

use_bpfbool = True

Filters in the capture backend. False brings more traffic into Python and applies the same source-port and destination checks there; the adapter doesn’t change.

auto_local_ipbool = True

With interface and local_ip both omitted, also restricts capture to the default route’s local IPv4, if discovery finds one. False keeps the default-route adapter without that restriction.

How selection resolves

Automatic selection follows the computer’s generic default IPv4 route, not the route of one BDO connection. With several adapters or a VPN, pin the adapter.

Accepted by

Pass it as capture_options=. Continuing live APIs take the LiveCaptureOptions subclass; offline replay has its own ports.

• Item state: CharacterLoadSession
• Calibration: calibrate_live(), calibrate_and_update() in live mode, CalibrationSession, AsyncCalibrationSession
• Arena of Solare: capture_solare_snapshot(), LiveSolareSession, AsyncLiveSolareSession

Example

import time

from bdo_toolkit import LiveCaptureSession

with LiveCaptureSession(opcode_profile="opcodes.local") as session:
    deadline = time.monotonic() + 60
    while time.monotonic() < deadline:
        session.poll(timeout=1.0)  # handle events here
    session.stop()
    while session.poll(timeout=0) is not None:
        pass  # drain events finalized during shutdown

health = session.health
decoder = session.decoder_health
print("filter:", session.endpoint.bpf_filter)
print("capture clean:", health.capture_is_clean)
print("storage decoder:", decoder.storage_status)
if not health.capture_is_clean:
    print(health.to_dict())
if decoder.storage_is_compatible is False:
    print(decoder.to_dict())

filter: tcp and (src port 8884 or src port 8885 or src port 8889) and dst host 192.0.2.42
capture clean: True
storage decoder: compatible

Example

from bdo_toolkit import replay_pcap

for event in replay_pcap(
    "path/to/session.pcapng",
    opcode_profile="path/to/matching-opcodes.json",
):
    print(event.event_type, event.source, event.item_id, event.quantity)
    print(event.format_human())
    break

item_received Gathering 7003 6
[23:06:01.245] ITEM_RECEIVED source='Gathering' item_id=7003 quantity=6

Which optional attributes are set depends on the event type and the profile.

Attributes

One BDOEvent type; the groups aren’t separate models.

Core

event_typestr

Normalized category. See event types.

timestampfloat

Epoch time of the decoded network message.

flowFlow

Server-to-client network flow; not the gameplay context in source.

item_idint

Raw item ID from the wire.

quantityint

Positive amount. inventory_snapshot keeps the full uint64 wire value; ordinary live item and storage events use uint32. A zero wire value never becomes an item event.

extraMapping[str, Any]

Immutable extension metadata; empty on a hand-built event unless supplied. See the provisional keys.

Optional

sourcestr | None

Known origin, producing system or context; None when unresolved or inapplicable. See source labels.

base_item_idint | None

item_id with detected enhancement bits removed.

enhancement_levelint | None

Detected enhancement level.

enhancementstr | None

Label for a recognized level, such as PRI or PEN.

inventory_slotint | None

Decoded inventory slot.

item_instancestr | None

Hex per-item instance ID.

storage_instancestr | None

Opaque hex storage-record value; not the destination.

storage_idint | None

Numeric storage destination: the value to persist and filter on. See the registry.

storage_namestr | None

Current registry display name for storage_id; None when unregistered.

storage_name_confidencestr | None

observed, inferred or probable confidence in that name.

Decoder metadata

raw_contextstr | None

Hex context bytes, when the specification exposes them; None when the source came from a default or no context.

opcodeint | None

Opcode of the containing message. to_dict() writes it as uppercase hex.

message_lengthint | None

Byte length of the containing message.

record_indexint | None

One-based position in a multi-record message; otherwise None.

record_countint | None

Record slots decoded from a multi-record message; otherwise None.

record_offsetint | None

Byte offset of this record’s item_id in the message.

confidencestr | None

observed on events from replay or live capture.

Derived timestamps

timestamp_textstr

Local HH:MM:SS.mmm, for display.

timestamp_isostr

Deterministic UTC ISO 8601, such as 2026-07-10T06:06:01.245Z.

Methods

format_human()→ str

One terminal line in host-local time with the fields this event has. Storage lines read destination='Heidel' storage_id=0x00000020; an unregistered ID may show as UNKNOWN_STORAGE(0xdeadbeef).

to_dict()→ dict[str, Any]

Mutable dict without None optionals, plus timestamp_iso. Toolkit values are JSON-safe; arbitrary objects in extra may not be.

to_dict() conversion

Event types

Under each name: its kind (live activity, state reported at character load, or unclassified) and the delivery preset that includes it. A filter that names the exact type selects any of them.

loot_previewlive · activity()

Shown in a gathering or loot-preview window before receipt. Not proof it reached inventory.

item_receivedlive · activity()

Inventory receipt or update: a mob drop, gathering reward or storage withdrawal (source="Storage"). quantity is the record amount, not a proven stack total.

storage_deltalive · activity()

Proven positive addition to town storage; quantity is the amount added. source names a confirmed manual or worker origin when known. Withdrawals are item_received.

inventory_snapshotexisting state · snapshot_records()

Inventory amount held at login or a character switch, with source="Character Load". Unsigned 64-bit; zero stays unresolved. Not an acquisition or the whole inventory.

storage_snapshotexisting state · snapshot_records()

Town-storage amount held at character load, with source=None. Not a deposit or the whole town.

inventory_recordunclassified · all()

Decoded, but the profile lacks the context field to prove receipt or existing state.

storage_recordunclassified · all()

Decoded with source=None; evidence proved neither a live addition nor existing state.

Known source labels

source is the known origin, producing system or context. Each label below is one exact, case-sensitive string; None means unresolved or not applicable.

An unresolved live storage addition has source=None, as do storage hydration and unresolved storage records:

def is_unresolved_deposit(event):
    return event.event_type == "storage_delta" and event.source is None

source="Storage" can still appear on the item_received from a storage withdrawal.

Unknown values

UNKNOWN_STORAGE(0xdeadbeef)display text

Shown for a decoded nonzero storage_id missing from the location registry; no town is guessed or stored. The event keeps the numeric storage_id with storage_name=None, and source stays the separately classified origin, or None. See the unknown-key contract.

UNKNOWN(0x…)source value

An unrecognized generic non-storage context. raw_context keeps the bytes when the profile exposes them.

Example

from bdo_toolkit import EventFilter, capture_live

heidel_worker_production = EventFilter(
    event_types={"storage_delta"},
    sources={"Worker Production"},
    storage_ids={0x0020},
    item_ids={7003, 4607},
)

for event in capture_live(
    opcode_profile="opcodes.local",
    event_filter=heidel_worker_production,
):
    print(
        event.source,
        event.storage_name,
        f"0x{event.storage_id:08x}",
        event.item_id,
        event.quantity,
    )

Worker Production Heidel 0x00000020 4607 18

All four criteria must match. source names the producing system and storage_id the destination; show storage_name, but persist the numeric ID.

To keep every confirmed live addition to Heidel, drop the source and item criteria:

from bdo_toolkit import EventFilter

heidel_activity = EventFilter(
    event_types={"storage_delta"},
    storage_ids={0x0020},
)

Parameters

event_typesIterable[str] | None = None

Exact event type, read from event.event_type. Stored as frozenset[str].

sourcesIterable[str] | None = None

Exact, case-sensitive source label, read from event.source. Stored as frozenset[str].

storage_idsIterable[int] | None = None

Exact numeric storage destination, read from event.storage_id. Stored as frozenset[int].

item_idsIterable[int] | None = None

Exact raw item ID, read from event.item_id. Stored as frozenset[int].

Matching rules

Delivery presets

EventFilter.activity()loot_preview · item_received · storage_delta

Ordinary live item activity; the default when a live API gets no filter.

EventFilter.snapshot_records()inventory_snapshot · storage_snapshot

Proven character-load records, without neutral ones. Not an assembled item-state snapshot.

EventFilter.all()no restriction

Everything, neutral inventory_record and storage_record included.

Methods

EventFilter.activity()→ EventFilter

The live-activity preset.

EventFilter.snapshot_records()→ EventFilter

The proven character-load record preset.

EventFilter.all()→ EventFilter

An unrestricted filter, equal to EventFilter().

EventFilter.from_values(*, event_types=None, sources=None, storage_ids=None, item_ids=None)→ EventFilter

Named constructor with the same inputs and normalization.

allows(event)→ bool

Whether event passes every configured criterion.

Example

from bdo_toolkit import ConsoleEventWriter, replay_pcap

writer = ConsoleEventWriter()
for event in replay_pcap(
    "path/to/session.pcapng",
    opcode_profile="path/to/matching-opcodes.json",
):
    writer.write(event)

[23:06:01.245] ITEM_RECEIVED source='Gathering' item_id=7003 quantity=6 item_instance=0x6f105b8a3d7294c1

from bdo_toolkit import JsonlEventWriter, replay_pcap

writer = JsonlEventWriter()
for event in replay_pcap(
    "path/to/session.pcapng",
    opcode_profile="path/to/matching-opcodes.json",
):
    writer.write(event)

{"confidence": "observed", "event_type": "item_received", "extra": {"stream_sequence": 58210}, "flow": {"destination_ip": "192.0.2.42", "destination_port": 51824, "source_ip": "203.0.113.20", "source_port": 8889}, "item_id": 7003, "item_instance": "0x6f105b8a3d7294c1", "message_length": 255, "opcode": "0x0F16", "quantity": 6, "raw_context": "0x0471ee0e", "record_offset": 33, "source": "Gathering", "timestamp": 1783663561.245, "timestamp_iso": "2026-07-10T06:06:01.245Z"}

Stream behavior