Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Update PET4L to Python 3.10 #22

Open
wants to merge 11 commits into
base: master
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 5 additions & 5 deletions .github/workflows/main.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@ jobs:
- name: Initialize Python
uses: actions/setup-python@v5
with:
python-version: '3.10'
python-version: '3.10.6'

- name: Install Dependencies
run: |
Expand All @@ -26,9 +26,9 @@ jobs:
pip install setuptools==53.0.0
pip install -r requirements.txt
pip install importlib-metadata==4.13.0
pip install flake8==3.8.4
pip install mypy==0.781
pip install vulture==2.3
pip install flake8==4.0.0
pip install mypy==0.920
pip install vulture==2.4

- name: Lint
run: |
Expand Down Expand Up @@ -66,7 +66,7 @@ jobs:
- name: Setup Python 3.10
uses: actions/setup-python@v5
with:
python-version: '3.10'
python-version: '3.10.6'
- name: Setup pip cache
uses: actions/cache@v4
with:
Expand Down
2 changes: 2 additions & 0 deletions requirements.txt
Original file line number Diff line number Diff line change
Expand Up @@ -8,3 +8,5 @@ PyQt5>=5.15.10
requests==2.32.2
simplejson==3.19.2
ecdsa==0.19.0
types-requests>=2.28.11,<3.0
types-simplejson>=3.18.0,<4.0
7 changes: 4 additions & 3 deletions src/apiClient.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,9 +31,10 @@ def process_api_exceptions_int(*args, **kwargs):

class ApiClient:

def __init__(self, isTestnet=False):
self.isTestnet = isTestnet
self.api = BlockBookClient(isTestnet)
def __init__(self, main_wnd):
self.main_wnd = main_wnd
self.isTestnet = main_wnd.isTestnetRPC
self.api = BlockBookClient(main_wnd, self.isTestnet)

@process_api_exceptions
def getAddressUtxos(self, address):
Expand Down
24 changes: 9 additions & 15 deletions src/blockbookClient.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,11 +15,8 @@ def process_blockbook_exceptions_int(*args, **kwargs):
try:
return func(*args, **kwargs)
except Exception as e:
if client.isTestnet:
new_url = "https://testnet.fuzzbawls.pw"
else:
new_url = "https://zkbitcoin.com/"
message = "BlockBook Client exception on %s\nTrying backup server %s" % (client.url, new_url)
new_url = "https://testnet.fuzzbawls.pw" if client.isTestnet else "https://zkbitcoin.com/"
message = f"BlockBook Client exception on {client.url}\nTrying backup server {new_url}"
printException(getCallerName(True), getFunctionName(True), message, str(e))
try:
client.url = new_url
Expand All @@ -33,21 +30,18 @@ def process_blockbook_exceptions_int(*args, **kwargs):

class BlockBookClient:

def __init__(self, isTestnet=False):
def __init__(self, main_wnd, isTestnet=False):
self.main_wnd = main_wnd
self.isTestnet = isTestnet
if isTestnet:
self.url = "https://testnet.rockdev.org/"
else:
self.url = "https://explorer.rockdev.org/"
self.url = "https://testnet.rockdev.org/" if isTestnet else "https://explorer.rockdev.org/"

def checkResponse(self, method, param=""):
url = self.url + "/api/%s" % method
if param != "":
url += "/%s" % param
url = f"{self.url}/api/{method}"
if param:
url += f"/{param}"
resp = requests.get(url, data={}, verify=True)
if resp.status_code == 200:
data = resp.json()
return data
return resp.json()
raise Exception("Invalid response")

@process_blockbook_exceptions
Expand Down
8 changes: 4 additions & 4 deletions src/constants.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@
import os
from queue import Queue

wqueue = Queue() # type: Queue[str]
wqueue: Queue[str] = Queue()

APPDATA_DIRNAME = ".PET4L-DATA"

Expand All @@ -21,7 +21,7 @@
TESTNET_MAGIC_BYTE = 139
TESTNET_STAKE_MAGIC_BYTE = 73
DEFAULT_PROTOCOL_VERSION = 70915
MINIMUM_FEE = 0.0001 # minimum PIV/kB
MINIMUM_FEE = 0.0001 # minimum PIV/kB
SECONDS_IN_2_MONTHS = 60 * 24 * 60 * 60
MAX_INPUTS_NO_WARNING = 75
starting_width = 1033
Expand Down Expand Up @@ -50,8 +50,8 @@
trusted_RPC_Servers = [
["https", "lithuania.fuzzbawls.pw:8080", "spmtUser", "WUss6sr8956S5Paex254"],
["https", "latvia.fuzzbawls.pw:8080", "spmtUser", "8X88u7TuefPm7mQaJY52"],
["https", "charlotte.fuzzbawls.pw:8080", "spmtUser", "ZyD936tm9dvqmMP8A777"]]

["https", "charlotte.fuzzbawls.pw:8080", "spmtUser", "ZyD936tm9dvqmMP8A777"]
]

HW_devices = [
# (model name, api index)
Expand Down
39 changes: 17 additions & 22 deletions src/cryptoIDClient.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,6 @@

api_keys = ["b62b40b5091e", "f1d66708a077", "ed85c85c0126", "ccc60d06f737"]


def process_cryptoID_exceptions(func):
def process_cryptoID_exceptions_int(*args, **kwargs):
try:
Expand All @@ -20,27 +19,23 @@ def process_cryptoID_exceptions_int(*args, **kwargs):
message = "CryptoID Client exception"
printException(getCallerName(True), getFunctionName(True), message, str(e))
return None

return process_cryptoID_exceptions_int


def UTXOS_cryptoID_to_trezor(utxos):
# convert JSON labels
new_utxos = []
for u in utxos:
new_u = {}
new_u["txid"] = u["tx_hash"]
new_u["vout"] = u["tx_ouput_n"]
new_u["satoshis"] = u["value"]
new_u["confirmations"] = u["confirmations"]
new_u["script"] = u["script"]
new_u = {
"txid": u["tx_hash"],
"vout": u["tx_ouput_n"],
"satoshis": u["value"],
"confirmations": u["confirmations"],
"script": u["script"]
}
new_utxos.append(new_u)

return new_utxos


class CryptoIDClient:

def __init__(self, isTestnet=False):
if isTestnet:
raise Exception("\nNo CryptoID Testnet server\n")
Expand All @@ -53,24 +48,24 @@ def checkResponse(self, parameters):
parameters['key'] = key
resp = requests.get(self.url, params=parameters)
if resp.status_code == 200:
data = resp.json()
return data
return resp.json()
return None

@process_cryptoID_exceptions
def getAddressUtxos(self, address):
self.parameters = {}
self.parameters['q'] = 'unspent'
self.parameters['active'] = address
self.parameters = {
'q': 'unspent',
'active': address
}
res = self.checkResponse(self.parameters)
if res is None:
return None
else:
return UTXOS_cryptoID_to_trezor(res['unspent_outputs'])
return UTXOS_cryptoID_to_trezor(res['unspent_outputs'])

@process_cryptoID_exceptions
def getBalance(self, address):
self.parameters = {}
self.parameters['q'] = 'getbalance'
self.parameters['a'] = address
self.parameters = {
'q': 'getbalance',
'a': address
}
return self.checkResponse(self.parameters)
32 changes: 15 additions & 17 deletions src/hwdevice.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,31 +14,28 @@
from time import sleep
from trezorClient import TrezorApi


def check_api_init(func):
def func_int(*args, **kwargs):
hwDevice = args[0]
if hwDevice.api is None:
logging.warning("%s: hwDevice.api is None" % func.__name__)
logging.warning(f"{func.__name__}: hwDevice.api is None")
raise Exception("HW device: client not initialized")
return func(*args, **kwargs)

return func_int


class HWdevice(QObject):
# signal: sig1 (thread) is done - emitted by signMessageFinish
sig1done = pyqtSignal(str)

def __init__(self, main_wnd, *args, **kwargs):
printDbg("HW: Initializing Class...")
QObject.__init__(self, *args, **kwargs)
super().__init__(*args, **kwargs)
self.main_wnd = main_wnd
self.api = None
printOK("HW: Class initialized")

def initDevice(self, hw_index):
printDbg("HW: initializing hw device with index %d" % hw_index)
printDbg(f"HW: initializing hw device with index {hw_index}")
if hw_index >= len(HW_devices):
raise Exception("Invalid HW index")

Expand All @@ -53,7 +50,7 @@ def initDevice(self, hw_index):
self.api.initDevice()
self.sig1done = self.api.sig1done
self.api.sig_disconnected.connect(self.main_wnd.clearHWstatus)
printOK("HW: hw device with index %d initialized" % hw_index)
printOK(f"HW: hw device with index {hw_index} initialized")

@check_api_init
def clearDevice(self):
Expand All @@ -68,14 +65,15 @@ def clearDevice(self):
@check_api_init
def getStatus(self):
printDbg("HW: checking device status...")
printOK("Status: %d" % self.api.status)
printOK(f"Status: {self.api.status}")
return self.api.model, self.api.status, self.api.messages[self.api.status]

def prepare_transfer_tx(self, caller, bip32_path, utxos_to_spend, dest_address, tx_fee, isTestnet=False):
def prepare_transfer_tx(self, caller, bip32_path, utxos_to_spend, dest_address, tx_fee, isTestnet=False):
rewardsArray = []
mnode = {}
mnode['path'] = bip32_path
mnode['utxos'] = utxos_to_spend
mnode = {
'path': bip32_path,
'utxos': utxos_to_spend
}
rewardsArray.append(mnode)
self.prepare_transfer_tx_bulk(caller, rewardsArray, dest_address, tx_fee, isTestnet)

Expand All @@ -86,17 +84,17 @@ def prepare_transfer_tx_bulk(self, caller, rewardsArray, dest_address, tx_fee, i

@check_api_init
def scanForAddress(self, hwAcc, spath, intExt=0, isTestnet=False):
printOK("HW: Scanning for Address n. %d on account n. %d" % (spath, hwAcc))
printOK(f"HW: Scanning for Address n. {spath} on account n. {hwAcc}")
return self.api.scanForAddress(hwAcc, spath, intExt, isTestnet)

@check_api_init
def scanForBip32(self, account, address, starting_spath=0, spath_count=10, isTestnet=False):
printOK("HW: Scanning for Bip32 path of address: %s" % address)
printOK(f"HW: Scanning for Bip32 path of address: {address}")
found = False
spath = -1

for i in range(starting_spath, starting_spath + spath_count):
printDbg("HW: checking path... %d'/0/%d" % (account, i))
printDbg(f"HW: checking path... {account}'/0/{i}")
curr_addr = self.api.scanForAddress(account, i, isTestnet)

if curr_addr == address:
Expand All @@ -106,11 +104,11 @@ def scanForBip32(self, account, address, starting_spath=0, spath_count=10, isTes

sleep(0.01)

return (found, spath)
return found, spath

@check_api_init
def scanForPubKey(self, account, spath, isTestnet=False):
printOK("HW: Scanning for PubKey of address n. %d on account n. %d" % (spath, account))
printOK(f"HW: Scanning for PubKey of address n. {spath} on account n. {account}")
return self.api.scanForPubKey(account, spath, isTestnet)

@check_api_init
Expand Down
Loading
Loading