Cosmos Tx Parsing Helpers

Imports

Intent: keep parsing dependencies explicit and lightweight for reuse across cache and reward notebooks.

Event Normalization

Intent: normalize raw Cosmos event attributes into stable dictionaries.

---

extract_events

def extract_events(
    tx_response_or_like:Dict[str, Any]
)->List[Dict[str, Any]]:

---

attr_map

def attr_map(
    attrs:Any
)->Dict[str, str]:

sample = extract_events({"events": [{"type": "x", "attributes": [{"key": "a", "value": "1"}]}]})
assert sample == [{"type": "x", "attributes": {"a": "1"}}]

Node Id Extraction

Intent: normalize node id extraction across event/message payload variants.

---

node_id_from_attrs

def node_id_from_attrs(
    attrs:Dict[str, str]
)->Optional[int]:

assert node_id_from_attrs({"node_id": "14"}) == 14
assert node_id_from_attrs({"delegation_target": "2933"}) == 2933
assert node_id_from_attrs({"mix_id": "7"}) == 7
assert node_id_from_attrs({}) is None

Coin Parsing

Intent: parse Cosmos coin lists like 100unym,5ufoo into structured tuples.

---

parse_coins

def parse_coins(
    amount_str:str
)->List[Tuple[int, str]]:

assert parse_coins("100unym,5ufoo") == [(100, "unym"), (5, "ufoo")]
assert parse_coins("") == []

Message-Indexed Events

Intent: group tx events by msg_index, preferring logs and falling back to top-level events where possible.

---

events_by_msg_index

def events_by_msg_index(
    tx_resp:Dict[str, Any], prefer_logs:bool=True, logger:Optional[logging.Logger]=None, txhash:str=''
)->Dict[int, List[Dict[str, Any]]]:

tx_logs = {"logs": [{"msg_index": "0", "events": [{"type": "transfer", "attributes": []}]}]}
by_idx = events_by_msg_index(tx_logs)
assert 0 in by_idx and by_idx[0][0]["type"] == "transfer"

tx_fallback = {"events": [{"type": "transfer", "attributes": [{"key": "msg_index", "value": "2"}]}]}
by_idx_fb = events_by_msg_index(tx_fallback)
assert 2 in by_idx_fb and by_idx_fb[2][0]["type"] == "transfer"

Incoming Amount Helpers

Intent: compute incoming NYM for a specific recipient from transfer/coin_received events while avoiding double-counting mirrored events.

---

sum_incoming_nym_from_events

def sum_incoming_nym_from_events(
    events:List[Dict[str, Any]], recipient:str
)->Decimal:

---

sum_incoming_unym_from_events

def sum_incoming_unym_from_events(
    events:List[Dict[str, Any]], recipient:str
)->int:

addr = "n1abc"
mirror = [
    {"type": "transfer", "attributes": [{"key": "recipient", "value": addr}, {"key": "amount", "value": "2000000unym"}]},
    {"type": "coin_received", "attributes": [{"key": "receiver", "value": addr}, {"key": "amount", "value": "2000000unym"}]},
]
assert sum_incoming_unym_from_events(mirror, addr) == 2_000_000
assert sum_incoming_nym_from_events(mirror, addr) == Decimal("2")