d0s3t3ch
/
wownero-python
mirror of https://git.wownero.com/lza_menace/wownero-python.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Add classes for transactions

This commit is contained in:
Michał Sałaban 2017-12-27 01:49:59 +01:00
parent 51d1cf1b58
commit 5355824a61
9 changed files with 186 additions and 81 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
monero/account.py
View File

@ -24,11 +24,14 @@ class Account(object):
def get_addresses(self):
return self._backend.get_addresses(account=self.index)
def get_payments_in(self):
return self._backend.get_payments_in(account=self.index)
def get_payments(self, payment_id=None):
return self._backend.get_payments(account=self.index, payment_id=payment_id)
def get_payments_out(self):
return self._backend.get_payments_out(account=self.index)
def get_transactions_in(self):
return self._backend.get_transactions_in(account=self.index)
def get_transactions_out(self):
return self._backend.get_transactions_out(account=self.index)
def transfer(self, address, amount, priority=prio.NORMAL, mixin=5, unlock_time=0):
return self._backend.transfer(

13
monero/address.py
View File

@ -1,19 +1,14 @@
from binascii import hexlify, unhexlify
import re
import struct
import sys
from sha3 import keccak_256
from . import base58
from . import numbers
_ADDR_REGEX = re.compile(r'^[123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz]{95}$')
_IADDR_REGEX = re.compile(r'^[123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz]{106}$')
if sys.version_info < (3,):
_integer_types = (int, long,)
else:
_integer_types = (int,)
class Address(object):
_valid_netbytes = (18, 53)
@ -46,11 +41,7 @@ class Address(object):
return hexlify(self._decoded[1:33]).decode()
def with_payment_id(self, payment_id=0):
if isinstance(payment_id, (bytes, str)):
payment_id = int(payment_id, 16)
elif not isinstance(payment_id, _integer_types):
raise TypeError("payment_id must be either int or hexadecimal str or bytes, "
"is %r" % payment_id)
payment_id = numbers.payment_id_as_int(payment_id)
if payment_id.bit_length() > 64:
raise TypeError("Integrated payment_id cannot have more than 64 bits, "
"has %d" % payment_id.bit_length())

96
monero/backends/jsonrpc.py
View File

@ -8,12 +8,16 @@ import requests
from .. import exceptions
from ..account import Account
from ..address import address, Address
from ..numbers import from_atomic, to_atomic
from ..numbers import from_atomic, to_atomic, payment_id_as_int
from ..transaction import Transaction, Payment, Transfer
_log = logging.getLogger(__name__)
class JSONRPCWallet(object):
_master_address = None
_addresses = None
def __init__(self, protocol='http', host='127.0.0.1', port=18082, path='/json_rpc', user='', password=''):
self.url = '{protocol}://{host}:{port}/json_rpc'.format(
protocol=protocol,
@ -30,9 +34,12 @@ class JSONRPCWallet(object):
try:
_accounts = self.raw_request('get_accounts')
except MethodNotFound:
# monero <= 0.11
# monero <= 0.11 : there's only one account and one address
_lg.debug('Monero <= 0.11 found, no accounts')
self._master_address = self.get_addresses()[0]
return [Account(self, 0)]
idx = 0
self._master_address = Address(_accounts['subaddress_accounts'][0]['base_address'])
for _acc in _accounts['subaddress_accounts']:
assert idx == _acc['account_index']
accounts.append(Account(self, _acc['account_index']))
@ -43,6 +50,7 @@ class JSONRPCWallet(object):
_addresses = self.raw_request('getaddress', {'account_index': account})
if 'addresses' not in _addresses:
# monero <= 0.11
_lg.debug('Monero <= 0.11 found, assuming single address')
return [Address(_addresses['address'])]
addresses = [None] * (max(map(operator.itemgetter('address_index'), _addresses['addresses'])) + 1)
for _addr in _addresses['addresses']:
@ -53,26 +61,48 @@ class JSONRPCWallet(object):
_balance = self.raw_request('getbalance', {'account_index': account})
return (from_atomic(_balance['balance']), from_atomic(_balance['unlocked_balance']))
def get_payments_in(self, account=0):
_payments = self.raw_request('get_transfers',
def get_payments(self, account=0, payment_id=0):
payment_id = payment_id_as_int(payment_id)
_log.debug("Getting payments for account {acc}, payment_id {pid}".format(
acc=account, pid=payment_id))
if payment_id.bit_length() > 64:
_pid = '{:064x}'.format(payment_id)
else:
_pid = '{:016x}'.format(payment_id)
_payments = self.raw_request('get_payments', {
'account_index': account,
'payment_id': _pid})
pmts = []
for tx in _payments['payments']:
data = self._tx2dict(tx)
# Monero <= 0.11 : no address is passed because there's only one
data['address'] = data['address'] or self._master_address
pmts.append(Payment(**data))
return pmts
def get_transactions_in(self, account=0):
_transfers = self.raw_request('get_transfers',
{'account_index': account, 'in': True, 'out': False, 'pool': False})
return map(self._pythonify_payment, _payments.get('in', []))
return [Transaction(**self._tx2dict(tx)) for tx in _transfers.get('in', [])]
def get_payments_out(self, account=0):
_payments = self.raw_request('get_transfers',
def get_transactions_out(self, account=0):
_transfers = self.raw_request('get_transfers',
{'account_index': account, 'in': False, 'out': True, 'pool': False})
return map(self._pythonify_payment, _payments.get('out', ''))
return [Transaction(**self._tx2dict(tx)) for tx in _transfers.get('out', [])]
def _pythonify_payment(self, pm):
def _tx2dict(self, tx):
return {
'id': pm['txid'],
'timestamp': datetime.fromtimestamp(pm['timestamp']),
'amount': from_atomic(pm['amount']),
'fee': from_atomic(pm['fee']),
'height': pm['height'],
'payment_id': pm['payment_id'],
'note': pm['note'],
'subaddr': (pm['subaddr_index']['major'], pm['subaddr_index']['minor']),
'hash': tx.get('txid', tx.get('tx_hash')),
'timestamp': datetime.fromtimestamp(tx['timestamp']) if 'timestamp' in tx else None,
'amount': from_atomic(tx['amount']),
'fee': from_atomic(tx['fee']) if 'fee' in tx else None,
'height': tx.get('height', tx.get('block_height')),
'payment_id': tx['payment_id'],
'note': tx.get('note'),
# NOTE: address will be resolved only after PR#3010 has been merged to Monero
'address': address(tx['address']) if 'address' in tx else None,
'key': tx.get('key'),
'blob': tx.get('blob', None),
}
def transfer(self, destinations, priority, mixin, unlock_time, account=0):
@ -89,26 +119,18 @@ class JSONRPCWallet(object):
'new_algorithm': True,
}
_transfers = self.raw_request('transfer_split', data)
keys = ('hash', 'amount', 'fee', 'key', 'blob')
return list(map(
self._pythonify_tx,
[ dict(_tx) for _tx in map(
lambda vs: zip(keys,vs),
zip(
*[_transfers[k] for k in (
'tx_hash_list', 'amount_list', 'fee_list', 'tx_key_list', 'tx_blob_list')
]
))
]))
def _pythonify_tx(self, tx):
return {
'id': tx['hash'],
'amount': from_atomic(tx['amount']),
'fee': from_atomic(tx['fee']),
'key': tx['key'],
'blob': tx.get('blob', None),
}
keys = ('txid', 'amount', 'fee', 'key', 'blob')
return [
Transfer(**self._tx2dict(tx)) for tx in [
dict(_tx) for _tx in map(
lambda vs: zip(keys,vs),
zip(
*[_transfers[k] for k in (
'tx_hash_list', 'amount_list', 'fee_list', 'tx_key_list', 'tx_blob_list')
]
))
]
]
def raw_request(self, method, params=None):
hdr = {'Content-Type': 'application/json'}

15
monero/numbers.py
View File

@ -1,7 +1,14 @@
from decimal import Decimal
import sys
PICONERO = Decimal('0.000000000001')
if sys.version_info < (3,):
_integer_types = (int, long,)
else:
_integer_types = (int,)
def to_atomic(amount):
"""Convert Monero decimal to atomic integer of piconero."""
return int(amount * 10**12)
@ -13,3 +20,11 @@ def from_atomic(amount):
def as_monero(amount):
"""Return the amount rounded to maximal Monero precision."""
return Decimal(amount).quantize(PICONERO)
def payment_id_as_int(payment_id):
if isinstance(payment_id, (bytes, str)):
payment_id = int(payment_id, 16)
elif not isinstance(payment_id, _integer_types):
raise TypeError("payment_id must be either int or hexadecimal str or bytes, "
"is %r" % payment_id)
return payment_id

35
monero/transaction.py Normal file
View File

@ -0,0 +1,35 @@
class Transaction(object):
hash = None
height = None
timestamp = None
payment_id = '0000000000000000'
amount = None
fee = None
address = None
def __init__(self, hash=None, **kwargs):
self.hash = hash
self.height = kwargs.get('height', self.height)
self.timestamp = kwargs.get('timestamp', self.timestamp)
self.payment_id = kwargs.get('payment_id', self.payment_id)
self.amount = kwargs.get('amount', self.amount)
self.fee = kwargs.get('fee', self.fee)
self.address = kwargs.get('address', self.address)
class Payment(Transaction):
"""Incoming Transaction"""
pass
class Transfer(Transaction):
"""Outgoing Transaction"""
key = None
blob = None
note = ''
def __init__(self, **kwargs):
super(Transfer, self).__init__(**kwargs)
self.key = kwargs.get('key', self.key)
self.note = kwargs.get('note', self.note)
self.blob = kwargs.get('blob', self.blob)

11
monero/wallet.py
View File

@ -31,11 +31,14 @@ class Wallet(object):
def get_address(self, index=0):
return self.accounts[0].get_addresses()[0]
def get_payments_in(self):
return self.accounts[0].get_payments_in()
def get_payments(self, payment_id=None):
return self.accounts[0].get_payments(payment_id=payment_id)
def get_payments_out(self):
return self.accounts[0].get_payments_out()
def get_transactions_in(self):
return self.accounts[0].get_transactions_in()
def get_transactions_out(self):
return self.accounts[0].get_transactions_out()
def transfer(self, address, amount, priority=prio.NORMAL, mixin=5, unlock_time=0):
return self.accounts[0].transfer(

6
tests/numbers.py
View File

@ -1,7 +1,7 @@
from decimal import Decimal
import unittest
from monero.numbers import to_atomic, from_atomic
from monero.numbers import to_atomic, from_atomic, payment_id_as_int
class NumbersTestCase(unittest.TestCase):
def test_simple_numbers(self):
@ -14,3 +14,7 @@ class NumbersTestCase(unittest.TestCase):
def test_rounding(self):
self.assertEqual(to_atomic(Decimal('1.0000000000004')), 1000000000000)
def test_payment_id(self):
self.assertEqual(payment_id_as_int('0'), 0)
self.assertEqual(payment_id_as_int('abcdef'), 0xabcdef)

51
tests/wallet.py
View File

@ -7,6 +7,8 @@ except ImportError:
from mock import patch, Mock
from monero.wallet import Wallet
from monero.address import Address
from monero.transaction import Transaction, Payment, Transfer
from monero.backends.jsonrpc import JSONRPCWallet
class SubaddrWalletTestCase(unittest.TestCase):
@ -117,7 +119,7 @@ class SubaddrWalletTestCase(unittest.TestCase):
self.assertEqual(len(self.wallet.accounts[0].get_addresses()), 8)
@patch('monero.backends.jsonrpc.requests.post')
def test_get_payments_in(self, mock_post):
def test_get_transactions_in(self, mock_post):
mock_post.return_value.status_code = 200
mock_post.return_value.json.return_value = self.get_accounts_result
self.wallet = Wallet(JSONRPCWallet())
@ -157,14 +159,16 @@ class SubaddrWalletTestCase(unittest.TestCase):
'txid': 'd23a7d086e70df7aa0ca002361c4b35e35a272345b0a513ece4f21b773941f5e',
'type': 'in',
'unlock_time': 0}]}}
pay_in = self.wallet.get_payments_in()
pay_in = self.wallet.get_transactions_in()
self.assertEqual(len(list(pay_in)), 3)
for payment in pay_in:
self.assertIsInstance(payment['amount'], Decimal)
self.assertIsInstance(payment['fee'], Decimal)
for tx in pay_in:
self.assertIsInstance(tx, Transaction)
# self.assertIsInstance(tx.address, Address)
self.assertIsInstance(tx.amount, Decimal)
self.assertIsInstance(tx.fee, Decimal)
@patch('monero.backends.jsonrpc.requests.post')
def test_get_payments_out(self, mock_post):
def test_get_transactions_out(self, mock_post):
mock_post.return_value.status_code = 200
mock_post.return_value.json.return_value = self.get_accounts_result
self.wallet = Wallet(JSONRPCWallet())
@ -249,9 +253,34 @@ class SubaddrWalletTestCase(unittest.TestCase):
'txid': '7e3db6c59c02d870f18b37a37cfc5857eeb5412df4ea00bb1971f3095f72b0d8',
'type': 'out',
'unlock_time': 0}]}}
pay_out = self.wallet.get_payments_out()
pay_out = self.wallet.get_transactions_out()
self.assertEqual(len(list(pay_out)), 6)
for payment in pay_out:
self.assertIsInstance(payment['amount'], Decimal)
self.assertIsInstance(payment['fee'], Decimal)
self.assertIsInstance(payment['timestamp'], datetime)
for tx in pay_out:
self.assertIsInstance(tx, Transaction)
# self.assertIsInstance(tx.address, Address)
self.assertIsInstance(tx.amount, Decimal)
self.assertIsInstance(tx.fee, Decimal)
self.assertIsInstance(tx.timestamp, datetime)
@patch('monero.backends.jsonrpc.requests.post')
def test_get_payments(self, mock_post):
mock_post.return_value.status_code = 200
mock_post.return_value.json.return_value = self.get_accounts_result
self.wallet = Wallet(JSONRPCWallet())
mock_post.return_value.status_code = 200
mock_post.return_value.json.return_value = {'id': 0,
'jsonrpc': '2.0',
'result': {'payments': [{'address': 'BZ9V9tfTDgHYsnAxgeMLaGCUb6yMaGNiZJwiBWQrE23MXcRqSde9DKa9LnPw31o2G8QrdKdUNM7VWhd3dr22ivk54QGqZ6u',
'amount': 2313370000000,
'block_height': 1048268,
'payment_id': 'feedbadbeef12345',
'subaddr_index': {'major': 1, 'minor': 1},
'tx_hash': 'e84343c2ebba4d4d94764e0cd275adee07cf7b4718565513be453d3724f6174b',
'unlock_time': 0}]}}
payments = self.wallet.get_payments(payment_id=0xfeedbadbeef12345)
self.assertEqual(len(list(payments)), 1)
for payment in payments:
self.assertIsInstance(payment, Payment)
self.assertIsInstance(payment.address, Address)
self.assertIsInstance(payment.amount, Decimal)
self.assertIsInstance(payment.height, int)

29
utils/walletdump.py
View File

@ -36,11 +36,14 @@ _TXHDR = "timestamp height id/hash
" amount fee payment_id"
def tx2str(tx):
return "{time} {height} {fullid} {amount:17.12f} {fee:13.12f} {payment_id}".format(
time=tx['timestamp'].strftime("%d-%m-%y %H:%M:%S"),
shortid="[{}...]".format(tx['id'][:32]),
fullid=tx['id'],
**tx)
return "{time} {height} {hash} {amount:17.12f} {fee:13.12f} {payment_id} {addr}".format(
time=tx.timestamp.strftime("%d-%m-%y %H:%M:%S") if getattr(tx, 'timestamp', None) else None,
height=tx.height,
hash=tx.hash,
amount=tx.amount,
fee=tx.fee or 0,
payment_id=tx.payment_id,
addr=getattr(tx, 'receiving_address', None) or '')
w = get_wallet()
print(
@ -60,28 +63,28 @@ if len(w.accounts) > 1:
addresses = acc.get_addresses()
print("{num:2d} address(es):".format(num=len(addresses)))
print("\n".join(map(str, addresses)))
ins = acc.get_payments_in()
ins = acc.get_transactions_in()
if ins:
print("\nIncoming payments:")
print("\nIncoming transactions:")
print(_TXHDR)
for tx in ins:
print(tx2str(tx))
outs = acc.get_payments_out()
outs = acc.get_transactions_out()
if outs:
print("\nOutgoing transfers:")
print("\nOutgoing transactions:")
print(_TXHDR)
for tx in outs:
print(tx2str(tx))
else:
ins = w.get_payments_in()
ins = w.get_transactions_in()
if ins:
print("\nIncoming payments:")
print("\nIncoming transactions:")
print(_TXHDR)
for tx in ins:
print(tx2str(tx))
outs = w.get_payments_out()
outs = w.get_transactions_out()
if outs:
print("\nOutgoing transfers:")
print("\nOutgoing transactions:")
print(_TXHDR)
for tx in outs:
print(tx2str(tx))