Add rpc module

btc/__init__.py

@@ -4,4 +4,5 @@
 from . import config
 from . import core
 from . import ripemd160
+from . import rpc
 from . import secp256k1

btc/config.py

@@ -11,14 +11,44 @@ def __setattr__(self, name, value):
 
 develop = ObjectDict({
     'name': 'develop',
+    'rpc': ObjectDict({
+        'addr': 'http://127.0.0.1:18443',
+        'username': 'user',
+        'password': 'pass',
+    }),
+    'prefix': ObjectDict({
+        'p2pkh': 0x6f,
+        'p2sh': 0xc4,
+        'bech32': 'bcrt',
+    }),
 })
 
 mainnet = ObjectDict({
     'name': 'mainnet',
+    'rpc': ObjectDict({
+        'addr': 'http://127.0.0.1:8332',
+        'username': 'user',
+        'password': 'pass',
+    }),
+    'prefix': ObjectDict({
+        'p2pkh': 0x00,
+        'p2sh': 0x05,
+        'bech32': 'bc',
+    }),
 })
 
 testnet = ObjectDict({
     'name': 'testnet',
+    'rpc': ObjectDict({
+        'addr': 'http://127.0.0.1:18332',
+        'username': 'user',
+        'password': 'pass',
+    }),
+    'prefix': ObjectDict({
+        'p2pkh': 0x6f,
+        'p2sh': 0xc4,
+        'bech32': 'tb',
+    }),
 })
 
 current = develop

btc/core.py

@@ -8,7 +8,7 @@ def hash160(data: bytearray) -> bytearray:
     return bytearray(btc.ripemd160.ripemd160(hashlib.sha256(data).digest()).digest())
 
 
-def hash256(data: bytearray):
+def hash256(data: bytearray) -> bytearray:
     return bytearray(hashlib.sha256(hashlib.sha256(data).digest()).digest())
 
 
@@ -102,10 +102,7 @@ def sec_read(data: bytearray):
 def address_p2pkh(pubkey: PubKey):
     # Legacy
     pubkey_hash = hash160(pubkey.sec())
-    if btc.config.current == btc.config.mainnet:
-        version = bytearray([0x00])
-    else:
-        version = bytearray([0x6f])
+    version = bytearray([btc.config.current.prefix.p2pkh])
     checksum = hash256(version + pubkey_hash)
     address = btc.base58.encode(version + pubkey_hash + checksum[:4])
     return address
@@ -116,10 +113,7 @@ def address_p2sh(pubkey: PubKey):
     # See https://github.com/bitcoin/bips/blob/master/bip-0141.mediawiki
     pubkey_hash = hash160(pubkey.sec())
     redeem_hash = hash160(bytearray([0x00, 0x14]) + pubkey_hash)
-    if btc.config.current == btc.config.mainnet:
-        version = bytearray([0x05])
-    else:
-        version = bytearray([0xc4])
+    version = bytearray([btc.config.current.prefix.p2sh])
     checksum = hash256(version + redeem_hash)
     address = btc.base58.encode(version + redeem_hash + checksum[:4])
     return address
@@ -129,10 +123,7 @@ def address_p2wpkh(pubkey: PubKey):
     # Native SegWit.
     # See https://github.com/bitcoin/bips/blob/master/bip-0084.mediawiki
     pubkey_hash = hash160(pubkey.sec())
-    if btc.config.current == btc.config.mainnet:
-        return btc.bech32.encode('bc', 0, pubkey_hash)
-    else:
-        return btc.bech32.encode('tb', 0, pubkey_hash)
+    return btc.bech32.encode(btc.config.current.prefix.bech32, 0, pubkey_hash)
 
 
 def address_p2tr(pubkey: PubKey):
@@ -151,7 +142,4 @@ def address_p2tr(pubkey: PubKey):
     tweak_k = btc.secp256k1.Fr(int.from_bytes(hashlib.sha256(tweak_k_data).digest()))
     tweak_p = btc.secp256k1.G * tweak_k
     tweak_p = btc.secp256k1.Pt(btc.secp256k1.Fq(pubkey.x), btc.secp256k1.Fq(pubkey.y)) + tweak_p
-    if btc.config.current == btc.config.mainnet:
-        return btc.bech32m.encode('bc', 1, bytearray(tweak_p.x.x.to_bytes(32)))
-    else:
-        return btc.bech32m.encode('tb', 1, bytearray(tweak_p.x.x.to_bytes(32)))
+    return btc.bech32m.encode(btc.config.current.prefix.bech32, 1, bytearray(tweak_p.x.x.to_bytes(32)))

btc/rpc.py

@@ -0,0 +1,28 @@
+import btc
+import random
+import requests
+import typing
+
+
+def call(method: str, params: typing.List[typing.Any]) -> typing.Any:
+    r = requests.post(btc.config.current.rpc.addr, json={
+        'id': random.randint(0x00000000, 0xffffffff),
+        'jsonrpc': '2.0',
+        'method': method,
+        'params': params,
+    }, auth=(
+        btc.config.current.rpc.username,
+        btc.config.current.rpc.password,
+    )).json()
+    if 'error' in r and r['error']:
+        raise Exception(r['error'])
+    return r['result']
+
+
+def generate_to_address(nblocks: int, address: str) -> typing.List[bytearray]:
+    r = call('generatetoaddress', [nblocks, address])
+    return [bytearray.fromhex(e) for e in r]
+
+
+def get_block_count() -> int:
+    return call('getblockcount', [])

pyproject.toml

@@ -11,6 +11,7 @@ authors = [
 description = "Python SDK for BTC"
 readme = "README.md"
 license = { file = "LICENSE" }
+dependencies = ["requests"]
 
 [project.urls]
 homepage = "https://github.com/mohanson/pybtc"

test/test_rpc.py

@@ -0,0 +1,6 @@
+import btc
+
+
+def test_get_block_count():
+    btc.config.current = btc.config.develop
+    assert btc.rpc.get_block_count() != 0