/
transaction_record.py
136 lines (116 loc) · 5.36 KB
/
transaction_record.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
from __future__ import annotations
from dataclasses import dataclass
from typing import Dict, Generic, List, Optional, Tuple, Type, TypeVar
from chia.consensus.coinbase import farmer_parent_id, pool_parent_id
from chia.types.blockchain_format.coin import Coin
from chia.types.blockchain_format.sized_bytes import bytes32
from chia.types.mempool_inclusion_status import MempoolInclusionStatus
from chia.types.spend_bundle import SpendBundle
from chia.util.bech32m import decode_puzzle_hash, encode_puzzle_hash
from chia.util.errors import Err
from chia.util.ints import uint8, uint32, uint64
from chia.util.streamable import Streamable, streamable
from chia.wallet.conditions import ConditionValidTimes
from chia.wallet.util.transaction_type import TransactionType
T = TypeVar("T")
_T_TransactionRecord = TypeVar("_T_TransactionRecord", bound="TransactionRecordOld")
minimum_send_attempts = 6
@dataclass
class ItemAndTransactionRecords(Generic[T]):
item: T
transaction_records: List[TransactionRecord]
@streamable
@dataclass(frozen=True)
class TransactionRecordOld(Streamable):
"""
Used for storing transaction data and status in wallets.
"""
confirmed_at_height: uint32
created_at_time: uint64
to_puzzle_hash: bytes32
amount: uint64
fee_amount: uint64
confirmed: bool
sent: uint32
spend_bundle: Optional[SpendBundle]
additions: List[Coin]
removals: List[Coin]
wallet_id: uint32
# Represents the list of peers that we sent the transaction to, whether each one
# included it in the mempool, and what the error message (if any) was
sent_to: List[Tuple[str, uint8, Optional[str]]]
trade_id: Optional[bytes32]
type: uint32 # TransactionType
# name is also called bundle_id and tx_id
name: bytes32
memos: List[Tuple[bytes32, List[bytes]]]
def is_in_mempool(self) -> bool:
# If one of the nodes we sent it to responded with success or pending, we return True
for _, mis, _ in self.sent_to:
if MempoolInclusionStatus(mis) in (MempoolInclusionStatus.SUCCESS, MempoolInclusionStatus.PENDING):
return True
return False
def height_farmed(self, genesis_challenge: bytes32) -> Optional[uint32]:
if not self.confirmed:
return None
if self.type == TransactionType.FEE_REWARD or self.type == TransactionType.COINBASE_REWARD:
for block_index in range(self.confirmed_at_height, self.confirmed_at_height - 100, -1):
if block_index < 0:
return None
pool_parent = pool_parent_id(uint32(block_index), genesis_challenge)
farmer_parent = farmer_parent_id(uint32(block_index), genesis_challenge)
if pool_parent == self.additions[0].parent_coin_info:
return uint32(block_index)
if farmer_parent == self.additions[0].parent_coin_info:
return uint32(block_index)
return None
def get_memos(self) -> Dict[bytes32, List[bytes]]:
return {coin_id: ms for coin_id, ms in self.memos}
@classmethod
def from_json_dict_convenience(cls: Type[_T_TransactionRecord], modified_tx_input: Dict) -> _T_TransactionRecord:
modified_tx = modified_tx_input.copy()
if "to_address" in modified_tx:
modified_tx["to_puzzle_hash"] = decode_puzzle_hash(modified_tx["to_address"]).hex()
if "to_address" in modified_tx:
del modified_tx["to_address"]
# Converts memos from a flat dict into a nested list
memos_dict: Dict[str, List[str]] = {}
memos_list: List = []
if "memos" in modified_tx:
for coin_id, memo in modified_tx["memos"].items():
if coin_id not in memos_dict:
memos_dict[coin_id] = []
memos_dict[coin_id].append(memo)
for coin_id, memos in memos_dict.items():
memos_list.append((coin_id, memos))
modified_tx["memos"] = memos_list
return cls.from_json_dict(modified_tx)
def to_json_dict_convenience(self, config: Dict) -> Dict:
selected = config["selected_network"]
prefix = config["network_overrides"]["config"][selected]["address_prefix"]
formatted = self.to_json_dict()
formatted["to_address"] = encode_puzzle_hash(self.to_puzzle_hash, prefix)
formatted["memos"] = {
coin_id.hex(): memo.hex()
for coin_id, memos in self.get_memos().items()
for memo in memos
if memo is not None
}
return formatted
def is_valid(self) -> bool:
if len(self.sent_to) < minimum_send_attempts:
# we haven't tried enough peers yet
return True
if any(x[1] == MempoolInclusionStatus.SUCCESS for x in self.sent_to):
# we managed to push it to mempool at least once
return True
if any(x[2] in (Err.INVALID_FEE_LOW_FEE.name, Err.INVALID_FEE_TOO_CLOSE_TO_ZERO.name) for x in self.sent_to):
# we tried to push it to mempool and got a fee error so it's a temporary error
return True
return False
def hint_dict(self) -> Dict[bytes32, bytes32]:
return {coin_id: bytes32(memos[0]) for coin_id, memos in self.memos if len(memos) > 0 and len(memos[0]) == 32}
@streamable
@dataclass(frozen=True)
class TransactionRecord(TransactionRecordOld):
valid_times: ConditionValidTimes