diff --git a/.github/workflows/main.yaml b/.github/workflows/main.yaml
index 939734a..c697947 100644
--- a/.github/workflows/main.yaml
+++ b/.github/workflows/main.yaml
@@ -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: |
@@ -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: |
@@ -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:
diff --git a/requirements.txt b/requirements.txt
index 340430b..6c1baca 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -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
diff --git a/src/apiClient.py b/src/apiClient.py
index 2205db8..a54e903 100644
--- a/src/apiClient.py
+++ b/src/apiClient.py
@@ -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):
diff --git a/src/blockbookClient.py b/src/blockbookClient.py
index 4348542..d6d55b7 100644
--- a/src/blockbookClient.py
+++ b/src/blockbookClient.py
@@ -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
@@ -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
diff --git a/src/constants.py b/src/constants.py
index 222b648..bc38d39 100644
--- a/src/constants.py
+++ b/src/constants.py
@@ -7,7 +7,7 @@
import os
from queue import Queue
-wqueue = Queue() # type: Queue[str]
+wqueue: Queue[str] = Queue()
APPDATA_DIRNAME = ".PET4L-DATA"
@@ -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
@@ -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)
diff --git a/src/cryptoIDClient.py b/src/cryptoIDClient.py
index 45fb180..fbfef04 100644
--- a/src/cryptoIDClient.py
+++ b/src/cryptoIDClient.py
@@ -11,7 +11,6 @@
api_keys = ["b62b40b5091e", "f1d66708a077", "ed85c85c0126", "ccc60d06f737"]
-
def process_cryptoID_exceptions(func):
def process_cryptoID_exceptions_int(*args, **kwargs):
try:
@@ -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")
@@ -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)
diff --git a/src/hwdevice.py b/src/hwdevice.py
index 0b270e3..b93af8e 100644
--- a/src/hwdevice.py
+++ b/src/hwdevice.py
@@ -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")
@@ -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):
@@ -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)
@@ -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:
@@ -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
diff --git a/src/ledgerClient.py b/src/ledgerClient.py
index e0c46bb..ac71283 100644
--- a/src/ledgerClient.py
+++ b/src/ledgerClient.py
@@ -29,10 +29,10 @@ def process_ledger_exceptions_int(*args, **kwargs):
except BTChipException as e:
printDbg('Error while communicating with Ledger hardware wallet.')
e.message = 'Error while communicating with Ledger hardware wallet.'
- if (e.sw in (0x6f01, 0x6d00, 0x6700, 0x6faa)):
+ if e.sw in (0x6f01, 0x6d00, 0x6700, 0x6faa):
e.message = 'Make sure the PIVX app is open on your Ledger device.'
e.message += '<br>If there is a program (such as Ledger Bitcoin Wallet) interfering with the USB communication, close it first.'
- elif (e.sw == 0x6982):
+ elif e.sw == 0x6982:
e.message = 'Enter the PIN on your Ledger device.'
printException(getCallerName(True), getFunctionName(True), e.message, e.args)
raise DisconnectedException(e.message, hwDevice)
@@ -63,7 +63,7 @@ class LedgerApi(QObject):
sigTxdone = pyqtSignal(bytearray, str)
# signal: sigtx (thread) is done (aborted) - emitted by signTxFinish
sigTxabort = pyqtSignal()
- # signal: tx_progress percent - emitted by perepare_transfer_tx_bulk
+ # signal: tx_progress percent - emitted by prepare_transfer_tx_bulk
tx_progress = pyqtSignal(int)
# signal: sig_progress percent - emitted by signTxSign
sig_progress = pyqtSignal(int)
@@ -71,8 +71,8 @@ class LedgerApi(QObject):
sig_disconnected = pyqtSignal(str)
def __init__(self, main_wnd, *args, **kwargs):
+ super().__init__(*args, **kwargs)
self.main_wnd = main_wnd
- QObject.__init__(self, *args, **kwargs)
self.model = [x[0] for x in HW_devices].index("LEDGER Nano")
self.messages = [
'Device not initialized.',
@@ -96,9 +96,9 @@ def initDevice(self):
printDbg("Ledger Initialized")
self.status = 1
ver = self.chip.getFirmwareVersion()
- printOK("Ledger HW device connected [v. %s]" % str(ver.get('version')))
+ printOK(f"Ledger HW device connected [v. {ver.get('version')}]")
# Check device is unlocked
- bip32_path = MPATH + "%d'/0/%d" % (0, 0)
+ bip32_path = f"{MPATH}{0}'/0/{0}"
_ = self.chip.getWalletPublicKey(bip32_path)
self.status = 2
self.sig_progress.connect(self.updateSigProgress)
@@ -125,8 +125,7 @@ def append_inputs_to_TX(self, utxo, bip32_path):
utxo_tx_index = utxo['vout']
if utxo_tx_index < 0 or utxo_tx_index > len(prev_transaction.outputs):
- raise Exception('Incorrect value of outputIndex for UTXO %s-%d' %
- (utxo['txid'], utxo['vout']))
+ raise Exception(f'Incorrect value of outputIndex for UTXO {utxo["txid"]}-{utxo["vout"]}')
trusted_input = self.chip.getTrustedInput(prev_transaction, utxo_tx_index)
self.trusted_inputs.append(trusted_input)
@@ -136,10 +135,10 @@ def append_inputs_to_TX(self, utxo, bip32_path):
pubkey_hash = bin_hash160(curr_pubkey)
pubkey_hash_from_script = extract_pkh_from_locking_script(prev_transaction.outputs[utxo_tx_index].script)
if pubkey_hash != pubkey_hash_from_script:
- text = "Error: The hashes for the public key for the BIP32 path (%s), and the UTXO locking script do not match." % bip32_path
+ text = f"Error: The hashes for the public key for the BIP32 path ({bip32_path}), and the UTXO locking script do not match."
text += "Your signed transaction will not be validated by the network.\n"
- text += "pubkey_hash: %s\n" % pubkey_hash.hex()
- text += "pubkey_hash_from_script: %s\n" % pubkey_hash_from_script.hex()
+ text += f"pubkey_hash: {pubkey_hash.hex()}\n"
+ text += f"pubkey_hash_from_script: {pubkey_hash_from_script.hex()}\n"
printDbg(text)
self.arg_inputs.append({
@@ -156,18 +155,15 @@ def prepare_transfer_tx_bulk(self, caller, rewardsArray, dest_address, tx_fee, i
with self.lock:
# For each UTXO create a Ledger 'trusted input'
self.trusted_inputs = []
- # https://klmoney.wordpress.com/bitcoin-dissecting-transactions-part-2-building-a-transaction-by-hand)
+ # https://klmoney.wordpress.com/bitcoin-dissecting-transactions-part-2-building-a-transaction-by-hand
self.arg_inputs = []
self.amount = 0
- num_of_sigs = sum([len(mnode['utxos']) for mnode in rewardsArray])
+ num_of_sigs = sum(len(mnode['utxos']) for mnode in rewardsArray)
curr_utxo_checked = 0
for mnode in rewardsArray:
# Add proper HW path (for current device) on each utxo
- if isTestnet:
- mnode['path'] = MPATH_TESTNET + mnode['path']
- else:
- mnode['path'] = MPATH + mnode['path']
+ mnode['path'] = (MPATH_TESTNET if isTestnet else MPATH) + mnode['path']
# Create a TX input with each utxo
for utxo in mnode['utxos']:
@@ -198,10 +194,10 @@ def prepare_transfer_tx_bulk(self, caller, rewardsArray, dest_address, tx_fee, i
self.mBox2 = QMessageBox(caller)
self.messageText = "<p>Confirm transaction on your device, with the following details:</p>"
- # messageText += "From bip32_path: <b>%s</b><br><br>" % str(bip32_path)
- self.messageText += "<p>Payment to:<br><b>%s</b></p>" % dest_address
- self.messageText += "<p>Net amount:<br><b>%s</b> PIV</p>" % str(round(self.amount / 1e8, 8))
- self.messageText += "<p>Fees:<br><b>%s</b> PIV<p>" % str(round(int(tx_fee) / 1e8, 8))
+ # messageText += f"From bip32_path: <b>{bip32_path}</b><br><br>"
+ self.messageText += f"<p>Payment to:<br><b>{dest_address}</b></p>"
+ self.messageText += f"<p>Net amount:<br><b>{round(self.amount / 1e8, 8)}</b> PIV</p>"
+ self.messageText += f"<p>Fees:<br><b>{round(int(tx_fee) / 1e8, 8)}</b> PIV<p>"
messageText = self.messageText + "Signature Progress: 0 %"
self.mBox2.setText(messageText)
self.mBox2.setIconPixmap(caller.ledgerImg.scaledToHeight(200, Qt.SmoothTransformation))
@@ -215,23 +211,15 @@ def prepare_transfer_tx_bulk(self, caller, rewardsArray, dest_address, tx_fee, i
@process_ledger_exceptions
def scanForAddress(self, hwAcc, spath, intExt=0, isTestnet=False):
with self.lock:
- if not isTestnet:
- curr_path = MPATH + "%d'/%d/%d" % (hwAcc, intExt, spath)
- curr_addr = self.chip.getWalletPublicKey(curr_path).get('address')[12:-2]
- else:
- curr_path = MPATH_TESTNET + "%d'/%d/%d" % (hwAcc, intExt, spath)
- pubkey = compress_public_key(self.chip.getWalletPublicKey(curr_path).get('publicKey')).hex()
- curr_addr = pubkey_to_address(pubkey, isTestnet)
+ curr_path = (MPATH_TESTNET if isTestnet else MPATH) + f"{hwAcc}'/{intExt}/{spath}"
+ curr_addr = self.chip.getWalletPublicKey(curr_path).get('address')[12:-2] if not isTestnet else pubkey_to_address(compress_public_key(self.chip.getWalletPublicKey(curr_path).get('publicKey')).hex(), isTestnet)
return curr_addr
@process_ledger_exceptions
def scanForPubKey(self, account, spath, isTestnet=False):
- hwpath = "%d'/0/%d" % (account, spath)
- if isTestnet:
- curr_path = MPATH_TESTNET + hwpath
- else:
- curr_path = MPATH + hwpath
+ hwpath = f"{account}'/0/{spath}"
+ curr_path = (MPATH_TESTNET if isTestnet else MPATH) + hwpath
with self.lock:
nodeData = self.chip.getWalletPublicKey(curr_path)
@@ -240,10 +228,7 @@ def scanForPubKey(self, account, spath, isTestnet=False):
@process_ledger_exceptions
def signMess(self, caller, hwpath, message, isTestnet=False):
- if isTestnet:
- path = MPATH_TESTNET + hwpath
- else:
- path = MPATH + hwpath
+ path = (MPATH_TESTNET if isTestnet else MPATH) + hwpath
# Ledger doesn't accept characters other that ascii printable:
# https://ledgerhq.github.io/btchip-doc/bitcoin-technical.html#_sign_message
message = message.encode('ascii', 'ignore')
@@ -251,9 +236,11 @@ def signMess(self, caller, hwpath, message, isTestnet=False):
# Connection pop-up
mBox = QMessageBox(caller)
- warningText = "Another application (such as Ledger Wallet app) has probably taken over "
- warningText += "the communication with the Ledger device.<br><br>To continue, close that application and "
- warningText += "click the <b>Retry</b> button.\nTo cancel, click the <b>Abort</b> button"
+ warningText = (
+ "Another application (such as Ledger Wallet app) has probably taken over "
+ "the communication with the Ledger device.<br><br>To continue, close that application and "
+ "click the <b>Retry</b> button.\nTo cancel, click the <b>Abort</b> button"
+ )
mBox.setText(warningText)
mBox.setWindowTitle("WARNING")
mBox.setStandardButtons(QMessageBox.Retry | QMessageBox.Abort)
@@ -274,8 +261,7 @@ def signMess(self, caller, hwpath, message, isTestnet=False):
printOK('Signing Message')
self.mBox = QMessageBox(caller.ui)
- messageText = "Check display of your hardware device\n\n- message hash:\n\n%s\n\n-path:\t%s\n" % (
- message_sha, path)
+ messageText = f"Check display of your hardware device\n\n- message hash:\n\n{message_sha}\n\n-path:\t{path}\n"
self.mBox.setText(messageText)
self.mBox.setIconPixmap(caller.ledgerImg.scaledToHeight(200, Qt.SmoothTransformation))
self.mBox.setWindowTitle("CHECK YOUR LEDGER")
@@ -314,13 +300,13 @@ def signMessageFinish(self):
printOK("Message signed")
sig1 = work.hex()
else:
- printDbg('client.signMessageSign() returned invalid response (code 3): ' + self.signature.hex())
+ printDbg(f'client.signMessageSign() returned invalid response (code 3): {self.signature.hex()}')
sig1 = "None"
else:
- printDbg('client.signMessageSign() returned invalid response (code 2): ' + self.signature.hex())
+ printDbg(f'client.signMessageSign() returned invalid response (code 2): {self.signature.hex()}')
sig1 = "None"
else:
- printDbg('client.signMessageSign() returned invalid response (code 1): ' + self.signature.hex())
+ printDbg(f'client.signMessageSign() returned invalid response (code 1): {self.signature.hex()}')
sig1 = "None"
else:
printOK("Signature refused by the user")
@@ -377,13 +363,13 @@ def signTxFinish(self):
self.mBox2.accept()
if self.tx_raw is not None:
- # Signal to be catched by FinishSend on TabRewards / dlg_sewwpAll
+ # Signal to be caught by FinishSend on TabRewards / dlg_sewwpAll
self.sigTxdone.emit(self.tx_raw, str(round(self.amount / 1e8, 8)))
else:
printOK("Transaction refused by the user")
self.sigTxabort.emit()
def updateSigProgress(self, percent):
- messageText = self.messageText + "Signature Progress: <b style='color:red'>" + str(percent) + " %</b>"
+ messageText = f"{self.messageText}Signature Progress: <b style='color:red'>{percent} %</b>"
self.mBox2.setText(messageText)
QApplication.processEvents()
diff --git a/src/mainApp.py b/src/mainApp.py
index e0b9419..de9c74b 100644
--- a/src/mainApp.py
+++ b/src/mainApp.py
@@ -17,8 +17,8 @@
from misc import printDbg, initLogs, saveCacheSettings, readCacheSettings, getVersion
from mainWindow import MainWindow
from constants import user_dir, SECONDS_IN_2_MONTHS
-from qt.dlg_configureRPCservers import ConfigureRPCservers_dlg
-from qt.dlg_signmessage import SignMessage_dlg
+from qt.dlg_configureRPCservers import ConfigureRPCserversDlg
+from qt.dlg_signmessage import SignMessageDlg
class ServiceExit(Exception):
@@ -30,7 +30,7 @@ class ServiceExit(Exception):
def service_shutdown(signum, frame):
- print('Caught signal %d' % signum)
+ print(f'Caught signal {signum}')
raise ServiceExit
@@ -54,7 +54,7 @@ def __init__(self, imgDir, app, start_args):
# Get version and title
self.version = getVersion()
- self.title = 'PET4L - PIVX Emergency Tool For Ledger - v.%s-%s' % (self.version['number'], self.version['tag'])
+ self.title = f'PET4L - PIVX Emergency Tool For Ledger - v.{self.version["number"]}-{self.version["tag"]}'
# Open database
self.db = Database(self)
@@ -107,7 +107,7 @@ def initUI(self, imgDir):
self.show()
self.activateWindow()
- def closeEvent(self, *args, **kwargs):
+ def closeEvent(self, event):
# Terminate the running threads.
# Set the shutdown flag on each thread to trigger a clean shutdown of each thread.
self.mainWindow.myRpcWd.shutdown_flag.set()
@@ -133,16 +133,16 @@ def closeEvent(self, *args, **kwargs):
# Adios
print("Bye Bye.")
- return QMainWindow.closeEvent(self, *args, **kwargs)
+ return super().closeEvent(event)
def onEditRPCServer(self):
# Create Dialog
- ui = ConfigureRPCservers_dlg(self)
+ ui = ConfigureRPCserversDlg(self)
if ui.exec():
printDbg("Configuring RPC Servers...")
def onSignVerifyMessage(self):
# Create Dialog
- ui = SignMessage_dlg(self.mainWindow)
+ ui = SignMessageDlg(self.mainWindow)
if ui.exec():
printDbg("Sign/Verify message...")
diff --git a/src/mainWindow.py b/src/mainWindow.py
index 727b793..87ae2fe 100644
--- a/src/mainWindow.py
+++ b/src/mainWindow.py
@@ -41,9 +41,8 @@ class MainWindow(QWidget):
# signal: UTXO list loading percent (emitted by load_utxos_thread in tabRewards)
sig_UTXOsLoading = pyqtSignal(int)
-
def __init__(self, parent, imgDir):
- super(QWidget, self).__init__(parent)
+ super().__init__(parent)
self.parent = parent
self.imgDir = imgDir
self.runInThread = ThreadFuns.runInThread
@@ -79,7 +78,7 @@ def __init__(self, parent, imgDir):
self.hwdevice = HWdevice(self)
# -- init Api Client
- self.apiClient = ApiClient(self.isTestnetRPC)
+ self.apiClient = ApiClient(self) # Pass 'self' as main_wnd reference
# -- Create Queue to redirect stdout
self.queue = wqueue
@@ -257,7 +256,7 @@ def checkVersion(self, ctrl):
(remote_version[0] == local_version[0] and remote_version[1] > local_version[1]) or \
(remote_version[0] == local_version[0] and remote_version[1] == local_version[1] and remote_version[2] >
local_version[2]):
- self.versionMess = '<b style="color:red">New Version Available:</b> %s ' % (self.gitVersion)
+ self.versionMess = f'<b style="color:red">New Version Available:</b> {self.gitVersion} '
self.versionMess += '(<a href="https://github.com/PIVX-Project/PET4L/releases/">download</a>)'
else:
self.versionMess = "You have the latest version of PET4L"
@@ -265,7 +264,7 @@ def checkVersion(self, ctrl):
def updateVersion(self):
if self.versionMess is not None:
self.versionLabel.setText(self.versionMess)
- printOK("Remote version: %s" % str(self.gitVersion))
+ printOK(f"Remote version: {self.gitVersion}")
def onChangeSelectedHW(self, i):
# Clear status
@@ -288,14 +287,13 @@ def onSaveConsole(self):
timestamp = strftime('%Y-%m-%d_%H-%M-%S', gmtime(now()))
options = QFileDialog.Options()
options |= QFileDialog.DontUseNativeDialog
- fileName, _ = QFileDialog.getSaveFileName(self, "Save Logs to file", "PET4L_Logs_%s.txt" % timestamp, "All Files (*);; Text Files (*.txt)", options=options)
+ fileName, _ = QFileDialog.getSaveFileName(self, f"Save Logs to file PET4L_Logs_{timestamp}.txt", "All Files (*);; Text Files (*.txt)", options=options)
try:
if fileName:
- printOK("Saving logs to %s" % fileName)
- log_file = open(fileName, 'w+', encoding="utf-8")
- log_text = self.consoleArea.toPlainText()
- log_file.write(log_text)
- log_file.close()
+ printOK(f"Saving logs to {fileName}")
+ with open(fileName, 'w+', encoding="utf-8") as log_file:
+ log_text = self.consoleArea.toPlainText()
+ log_file.write(log_text)
except Exception as e:
err_msg = "error writing Log file"
@@ -315,14 +313,14 @@ def onToggleConsole(self):
def showHWstatus(self):
self.updateHWleds()
- myPopUp_sb(self, "info", 'PET4L - hw check', "%s" % self.hwStatusMess)
+ myPopUp_sb(self, "info", 'PET4L - hw check', f"{self.hwStatusMess}")
def showRPCstatus(self, server_index, fDebug):
# Update displayed status only if selected server is not changed
if server_index == self.header.rpcClientsBox.currentIndex():
self.updateRPCled(fDebug)
if fDebug:
- myPopUp_sb(self, "info", 'PET4L - rpc check', "%s" % self.rpcStatusMess)
+ myPopUp_sb(self, "info", 'PET4L - rpc check', f"{self.rpcStatusMess}")
def updateHWleds(self):
if self.hwStatus == 1:
@@ -342,7 +340,7 @@ def updateHWstatus(self, ctrl):
printDbg(str(e))
pass
- printDbg("status:%s - mess: %s" % (self.hwStatus, self.hwStatusMess))
+ printDbg(f"status:{self.hwStatus} - mess: {self.hwStatusMess}")
def updateLastBlockLabel(self):
text = '--'
@@ -370,9 +368,9 @@ def updateLastBlockPing(self):
color = "green"
self.header.lastPingIcon.setPixmap(self.connGreen_icon)
if self.rpcResponseTime is not None:
- self.header.responseTimeLabel.setText("%.3f" % self.rpcResponseTime)
- self.header.responseTimeLabel.setStyleSheet("color: %s" % color)
- self.header.lastPingIcon.setStyleSheet("color: %s" % color)
+ self.header.responseTimeLabel.setText(f"{self.rpcResponseTime:.3f}")
+ self.header.responseTimeLabel.setStyleSheet(f"color: {color}")
+ self.header.lastPingIcon.setStyleSheet(f"color: {color}")
def updateRPCled(self, fDebug=False):
if self.rpcConnected:
@@ -403,7 +401,7 @@ def updateRPClist(self):
# Add public servers (italics)
italicsFont = QFont("Times", italic=True)
for s in public_servers:
- url = s["protocol"] + "://" + s["host"].split(':')[0]
+ url = f"{s['protocol']}://{s['host'].split(':')[0]}"
self.header.rpcClientsBox.addItem(url, s)
self.header.rpcClientsBox.setItemData(self.getServerListIndex(s), italicsFont, Qt.FontRole)
# Add Local Wallet (bold)
@@ -413,7 +411,7 @@ def updateRPClist(self):
self.header.rpcClientsBox.setItemData(self.getServerListIndex(custom_servers[0]), boldFont, Qt.FontRole)
# Add custom servers
for s in custom_servers[1:]:
- url = s["protocol"] + "://" + s["host"].split(':')[0]
+ url = f"{s['protocol']}://{s['host'].split(':')[0]}"
self.header.rpcClientsBox.addItem(url, s)
# reset index
if self.parent.cache['selectedRPC_index'] >= self.header.rpcClientsBox.count():
@@ -428,7 +426,7 @@ def updateRPClist(self):
def updateRPCstatus(self, ctrl, fDebug=False):
rpc_index, rpc_protocol, rpc_host, rpc_user, rpc_password = self.getRPCserver()
if fDebug:
- printDbg("Trying to connect to RPC %s://%s..." % (rpc_protocol, rpc_host))
+ printDbg(f"Trying to connect to RPC {rpc_protocol}://{rpc_host}...")
try:
rpcClient = RpcClient(rpc_protocol, rpc_host, rpc_user, rpc_password)
diff --git a/src/misc.py b/src/misc.py
index 6e377ec..9fc962a 100644
--- a/src/misc.py
+++ b/src/misc.py
@@ -8,24 +8,26 @@
import os
import sys
import time
+import io
from contextlib import redirect_stdout
from ipaddress import ip_address
from urllib.parse import urlparse
+from typing import Any, Callable, Dict, Optional, Tuple
import simplejson as json
from PyQt5.QtCore import QObject, pyqtSignal, QSettings
from PyQt5.QtWidgets import QMessageBox
-from constants import log_File, DefaultCache, wqueue, MAX_INPUTS_NO_WARNING
+from constants import log_File, DefaultCache, MAX_INPUTS_NO_WARNING
-def add_defaultKeys_to_dict(dictObj, defaultObj):
+def add_defaultKeys_to_dict(dictObj: Dict[str, Any], defaultObj: Dict[str, Any]) -> None:
for key in defaultObj:
if key not in dictObj:
dictObj[key] = defaultObj[key]
-QT_MESSAGE_TYPE = {
+QT_MESSAGE_TYPE: Dict[str, QMessageBox.Icon] = {
"info": QMessageBox.Information,
"warn": QMessageBox.Warning,
"crit": QMessageBox.Critical,
@@ -33,71 +35,69 @@ def add_defaultKeys_to_dict(dictObj, defaultObj):
}
-def checkRPCstring(urlstring):
+def checkRPCstring(urlstring: str) -> bool:
try:
o = urlparse(urlstring)
- if o.scheme is None or o.scheme == '':
- raise Exception("Wrong protocol. Set either http or https.")
- if o.netloc is None or o.netloc == '':
- raise Exception("Malformed host network location part.")
- if o.port is None or o.port == '':
- raise Exception("Wrong IP port number")
- if o.username is None:
- raise Exception("Malformed username")
- if o.password is None:
- raise Exception("Malformed password")
+ if not o.scheme:
+ raise ValueError("Wrong protocol. Set either http or https.")
+ if not o.netloc:
+ raise ValueError("Malformed host network location part.")
+ if not o.port:
+ raise ValueError("Wrong IP port number")
+ if not o.username:
+ raise ValueError("Malformed username")
+ if not o.password:
+ raise ValueError("Malformed password")
return True
-
except Exception as e:
error_msg = "Unable to parse URL"
printException(getCallerName(), getFunctionName(), error_msg, e)
return False
-def checkTxInputs(parentWindow, num_of_inputs):
+def checkTxInputs(parentWindow: Any, num_of_inputs: int) -> Optional[int]:
if num_of_inputs == 0:
myPopUp_sb(parentWindow, "warn", 'Transaction NOT sent', "No UTXO to send")
return None
if num_of_inputs > MAX_INPUTS_NO_WARNING:
- warning = "Warning: Trying to spend %d inputs.\nA few minutes could be required " \
- "for the transaction to be prepared and signed.\n\nThe hardware device must remain unlocked " \
- "during the whole time (it's advised to disable the auto-lock feature)\n\n" \
- "Do you wish to proceed?" % num_of_inputs
- title = "PET4L - spending more than %d inputs" % MAX_INPUTS_NO_WARNING
+ warning = (f"Warning: Trying to spend {num_of_inputs} inputs.\n"
+ "A few minutes could be required for the transaction to be prepared and signed.\n\n"
+ "The hardware device must remain unlocked during the whole time "
+ "(it's advised to disable the auto-lock feature)\n\n"
+ "Do you wish to proceed?")
+ title = f"PET4L - spending more than {MAX_INPUTS_NO_WARNING} inputs"
return myPopUp(parentWindow, "warn", title, warning)
return QMessageBox.Yes
-def clean_for_html(text):
+def clean_for_html(text: Optional[str]) -> str:
if text is None:
return ""
return text.replace("<", "{").replace(">", "}")
-def clear_screen():
+def clear_screen() -> None:
os.system('clear')
-def getCallerName(inDecorator=False):
+def getCallerName(inDecorator: bool = False) -> Optional[str]:
try:
- if inDecorator:
- return sys._getframe(3).f_code.co_name
- return sys._getframe(2).f_code.co_name
+ frame = sys._getframe(3 if inDecorator else 2)
+ return frame.f_code.co_name
except Exception:
return None
-def getFunctionName(inDecorator=False):
+def getFunctionName(inDecorator: bool = False) -> Optional[str]:
try:
- if inDecorator:
- return sys._getframe(2).f_code.co_name
- return sys._getframe(1).f_code.co_name
+ frame = sys._getframe(2 if inDecorator else 1)
+ return frame.f_code.co_name
except Exception:
return None
-def getRemotePET4Lversion():
+def getRemotePET4Lversion() -> str:
import requests
try:
resp = requests.get("https://raw.githubusercontent.com/PIVX-Project/PET4L/master/src/version.txt")
@@ -105,78 +105,65 @@ def getRemotePET4Lversion():
data = resp.json()
return data['number']
else:
- raise Exception
-
+ raise ValueError("Invalid response from GitHub")
except Exception:
redirect_print("Invalid response getting version from GitHub\n")
return "0.0.0"
-def getVersion():
- version_file = os.path.join(
- os.path.dirname(os.path.abspath(__file__)), 'version.txt')
+def getVersion() -> Dict[str, Any]:
+ version_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'version.txt')
with open(version_file, encoding="utf-8") as data_file:
data = json.load(data_file)
-
return data
-def getTxidTxidn(txid, txidn):
+def getTxidTxidn(txid: Optional[str], txidn: Optional[int]) -> Optional[str]:
if txid is None or txidn is None:
return None
else:
- return txid + '-' + str(txidn)
+ return f"{txid}-{txidn}"
-def initLogs():
+def initLogs() -> None:
filename = log_File
filemode = 'w'
format = '%(asctime)s - %(levelname)s - %(threadName)s | %(message)s'
level = logging.DEBUG
- logging.basicConfig(filename=filename,
- filemode=filemode,
- format=format,
- level=level
- )
+ logging.basicConfig(filename=filename, filemode=filemode, format=format, level=level)
-def ipport(ip, port):
+def ipport(ip: Optional[str], port: Optional[str]) -> Optional[str]:
if ip is None or port is None:
return None
elif ip.endswith('.onion'):
- return ip + ':' + port
+ return f"{ip}:{port}"
else:
ipAddr = ip_address(ip)
if ipAddr.version == 4:
- return ip + ':' + port
+ return f"{ip}:{port}"
elif ipAddr.version == 6:
- return "[" + ip + "]:" + port
+ return f"[{ip}]:{port}"
else:
- raise Exception("invalid IP version number")
+ raise ValueError("Invalid IP version number")
-def myPopUp(parentWindow, messType, messTitle, messText, defaultButton=QMessageBox.No):
- if messType in QT_MESSAGE_TYPE:
- type = QT_MESSAGE_TYPE[messType]
- else:
- type = QMessageBox.Question
- mess = QMessageBox(type, messTitle, messText, defaultButton, parent=parentWindow)
+def myPopUp(parentWindow: Any, messType: str, messTitle: str, messText: str, defaultButton: QMessageBox.StandardButton = QMessageBox.No) -> int:
+ message_type = QT_MESSAGE_TYPE.get(messType, QMessageBox.Question)
+ mess = QMessageBox(message_type, messTitle, messText, defaultButton, parent=parentWindow)
mess.setStandardButtons(QMessageBox.Yes | QMessageBox.No)
mess.setDefaultButton(defaultButton)
return mess.exec_()
-def myPopUp_sb(parentWindow, messType, messTitle, messText, singleButton=QMessageBox.Ok):
- if messType in QT_MESSAGE_TYPE:
- type = QT_MESSAGE_TYPE[messType]
- else:
- type = QMessageBox.Information
- mess = QMessageBox(type, messTitle, messText, singleButton, parent=parentWindow)
- mess.setStandardButtons(singleButton | singleButton)
+def myPopUp_sb(parentWindow: Any, messType: str, messTitle: str, messText: str, singleButton: QMessageBox.StandardButton = QMessageBox.Ok) -> int:
+ message_type = QT_MESSAGE_TYPE.get(messType, QMessageBox.Information)
+ mess = QMessageBox(message_type, messTitle, messText, singleButton, parent=parentWindow)
+ mess.setStandardButtons(singleButton)
return mess.exec_()
-def is_hex(s):
+def is_hex(s: str) -> bool:
try:
int(s, 16)
return True
@@ -184,122 +171,100 @@ def is_hex(s):
return False
-def now():
+def now() -> int:
return int(time.time())
-def persistCacheSetting(cache_key, cache_value):
+def persistCacheSetting(cache_key: str, cache_value: Any) -> Any:
settings = QSettings('PIVX', 'PET4L')
if not settings.contains(cache_key):
- printDbg("Cache key %s not found" % str(cache_key))
+ printDbg(f"Cache key {cache_key} not found")
printOK("Adding new cache key to settings...")
-
- if type(cache_value) in [list, dict]:
- settings.setValue(cache_key, json.dumps(cache_value))
- else:
- settings.setValue(cache_key, cache_value)
-
+ settings.setValue(cache_key, json.dumps(cache_value) if isinstance(cache_value, (list, dict)) else cache_value)
return cache_value
-def printDbg(what):
+def printDbg(what: str) -> None:
logging.info(what)
log_line = printDbg_msg(what)
redirect_print(log_line)
-def printDbg_msg(what):
+def printDbg_msg(what: str) -> str:
what = clean_for_html(what)
timestamp = time.strftime('%Y-%m-%d %H:%M:%S', time.gmtime(now()))
- log_line = '<b style="color: yellow">{}</b> : {}<br>'.format(timestamp, what)
+ log_line = f'<b style="color: yellow">{timestamp}</b> : {what}<br>'
return log_line
-def printError(
- caller_name,
- function_name,
- what
-):
- logging.error("%s | %s | %s" % (caller_name, function_name, what))
+def printError(caller_name: Optional[str], function_name: Optional[str], what: str) -> None:
+ logging.error(f"{caller_name} | {function_name} | {what}")
log_line = printException_msg(caller_name, function_name, what, None, True)
redirect_print(log_line)
-def printException(
- caller_name,
- function_name,
- err_msg,
- errargs=None
-):
+def printException(caller_name: Optional[str], function_name: Optional[str], err_msg: str, errargs: Optional[Any] = None) -> None:
what = err_msg
if errargs is not None:
- what += " ==> %s" % str(errargs)
- logging.warning("%s | %s | %s" % (caller_name, function_name, what))
+ what += f" ==> {errargs}"
+ logging.warning(f"{caller_name} | {function_name} | {what}")
text = printException_msg(caller_name, function_name, err_msg, errargs)
redirect_print(text)
-def printException_msg(
- caller_name,
- function_name,
- err_msg,
- errargs=None,
- is_error=False
-):
- if is_error:
- msg = '<b style="color: red">ERROR</b><br>'
- else:
- msg = '<b style="color: red">EXCEPTION</b><br>'
- msg += '<span style="color:white">caller</span> : %s<br>' % caller_name
- msg += '<span style="color:white">function</span> : %s<br>' % function_name
+def printException_msg(caller_name: Optional[str], function_name: Optional[str], err_msg: str, errargs: Optional[Any] = None, is_error: bool = False) -> str:
+ msg = '<b style="color: red">ERROR</b><br>' if is_error else '<b style="color: red">EXCEPTION</b><br>'
+ msg += f'<span style="color:white">caller</span> : {caller_name}<br>'
+ msg += f'<span style="color:white">function</span> : {function_name}<br>'
msg += '<span style="color:red">'
if errargs:
- msg += 'err: %s<br>' % str(errargs)
-
- msg += '===> %s</span><br>' % err_msg
+ msg += f'err: {errargs}<br>'
+ msg += f'===> {err_msg}</span><br>'
return msg
-def printOK(what):
+def printOK(what: str) -> None:
logging.debug(what)
- msg = '<b style="color: #cc33ff">===> ' + what + '</b><br>'
+ msg = f'<b style="color: #cc33ff">===> {what}</b><br>'
redirect_print(msg)
-def splitString(text, n):
+def splitString(text: str, n: int) -> str:
arr = [text[i:i + n] for i in range(0, len(text), n)]
return '\n'.join(arr)
-def readCacheSettings():
+def readCacheSettings() -> Dict[str, Any]:
settings = QSettings('PIVX', 'PET4L')
try:
- cache = {}
- cache["lastAddress"] = settings.value('cache_lastAddress', DefaultCache["lastAddress"], type=str)
- cache["window_width"] = settings.value('cache_winWidth', DefaultCache["window_width"], type=int)
- cache["window_height"] = settings.value('cache_winHeight', DefaultCache["window_height"], type=int)
- cache["splitter_x"] = settings.value('cache_splitterX', DefaultCache["splitter_x"], type=int)
- cache["splitter_y"] = settings.value('cache_splitterY', DefaultCache["splitter_y"], type=int)
- cache["console_hidden"] = settings.value('cache_consoleHidden', DefaultCache["console_hidden"], type=bool)
- cache["selectedHW_index"] = settings.value('cache_HWindex', DefaultCache["selectedHW_index"], type=int)
- cache["selectedRPC_index"] = settings.value('cache_RPCindex', DefaultCache["selectedRPC_index"], type=int)
- cache["isTestnetRPC"] = settings.value('cache_isTestnetRPC', DefaultCache["isTestnetRPC"], type=bool)
- cache["hwAcc"] = settings.value('cache_hwAcc', DefaultCache["hwAcc"], type=int)
- cache["spathFrom"] = settings.value('cache_spathFrom', DefaultCache["spathFrom"], type=int)
- cache["spathTo"] = settings.value('cache_spathTo', DefaultCache["spathTo"], type=int)
- cache["intExt"] = settings.value('cache_intExt', DefaultCache["intExt"], type=int)
+ cache = {
+ "lastAddress": settings.value('cache_lastAddress', DefaultCache["lastAddress"], type=str),
+ "window_width": settings.value('cache_winWidth', DefaultCache["window_width"], type=int),
+ "window_height": settings.value('cache_winHeight', DefaultCache["window_height"], type=int),
+ "splitter_x": settings.value('cache_splitterX', DefaultCache["splitter_x"], type=int),
+ "splitter_y": settings.value('cache_splitterY', DefaultCache["splitter_y"], type=int),
+ "console_hidden": settings.value('cache_consoleHidden', DefaultCache["console_hidden"], type=bool),
+ "selectedHW_index": settings.value('cache_HWindex', DefaultCache["selectedHW_index"], type=int),
+ "selectedRPC_index": settings.value('cache_RPCindex', DefaultCache["selectedRPC_index"], type=int),
+ "isTestnetRPC": settings.value('cache_isTestnetRPC', DefaultCache["isTestnetRPC"], type=bool),
+ "hwAcc": settings.value('cache_hwAcc', DefaultCache["hwAcc"], type=int),
+ "spathFrom": settings.value('cache_spathFrom', DefaultCache["spathFrom"], type=int),
+ "spathTo": settings.value('cache_spathTo', DefaultCache["spathTo"], type=int),
+ "intExt": settings.value('cache_intExt', DefaultCache["intExt"], type=int)
+ }
add_defaultKeys_to_dict(cache, DefaultCache)
return cache
- except:
+ except Exception:
return DefaultCache
-def redirect_print(what):
- with redirect_stdout(WriteStream(wqueue)):
- print(what)
+def redirect_print(what: str) -> None:
+ with io.StringIO() as buffer:
+ with redirect_stdout(buffer):
+ print(what)
-def saveCacheSettings(cache):
+def saveCacheSettings(cache: Dict[str, Any]) -> None:
settings = QSettings('PIVX', 'PET4L')
settings.setValue('cache_lastAddress', cache.get('lastAddress'))
settings.setValue('cache_winWidth', cache.get('window_width'))
@@ -316,28 +281,25 @@ def saveCacheSettings(cache):
settings.setValue('cache_intExt', cache.get('intExt'))
-def sec_to_time(seconds):
- days = seconds // 86400
- seconds -= days * 86400
- hrs = seconds // 3600
- seconds -= hrs * 3600
- mins = seconds // 60
- seconds -= mins * 60
- return "{} days, {} hrs, {} mins, {} secs".format(days, hrs, mins, seconds)
+def sec_to_time(seconds: int) -> str:
+ days, seconds = divmod(seconds, 86400)
+ hrs, seconds = divmod(seconds, 3600)
+ mins, seconds = divmod(seconds, 60)
+ return f"{days} days, {hrs} hrs, {mins} mins, {seconds} secs"
-def timeThis(function, *args):
+def timeThis(function: Callable[..., Any], *args: Any) -> Tuple[Optional[Any], Optional[float]]:
try:
- start = time.clock()
+ start = time.perf_counter()
val = function(*args)
- end = time.clock()
+ end = time.perf_counter()
return val, (end - start)
except Exception:
return None, None
class DisconnectedException(Exception):
- def __init__(self, message, hwDevice):
+ def __init__(self, message: str, hwDevice: Any):
# Call the base class constructor
super().__init__(message)
# clear device
@@ -345,27 +307,39 @@ def __init__(self, message, hwDevice):
# Stream object to redirect sys.stdout and sys.stderr to a queue
-class WriteStream(object):
- def __init__(self, queue):
+class WriteStream:
+ def __init__(self, queue: Any):
self.queue = queue
- def write(self, text):
+ def write(self, text: str) -> None:
self.queue.put(text)
- def flush(self):
+ def flush(self) -> None:
pass
+class WrapperWriteStream:
+ def __init__(self, write_stream: WriteStream):
+ self.write_stream = write_stream
+
+ def write(self, message: str) -> int:
+ self.write_stream.write(message)
+ return len(message)
+
+ def flush(self) -> None:
+ self.write_stream.flush()
+
+
# QObject (to be run in QThread) that blocks until data is available
# and then emits a QtSignal to the main thread.
class WriteStreamReceiver(QObject):
mysignal = pyqtSignal(str)
- def __init__(self, queue, *args, **kwargs):
- QObject.__init__(self, *args, **kwargs)
+ def __init__(self, queue: Any, *args: Any, **kwargs: Any):
+ super().__init__(*args, **kwargs)
self.queue = queue
- def run(self):
+ def run(self) -> None:
while True:
text = self.queue.get()
self.mysignal.emit(text)
diff --git a/src/pivx_b58.py b/src/pivx_b58.py
index 1518dda..093aa23 100644
--- a/src/pivx_b58.py
+++ b/src/pivx_b58.py
@@ -1,5 +1,5 @@
#!/usr/bin/env python3
-# -*- coding: utf-8 -*-import sys
+# -*- coding: utf-8 -*-
# Copyright (c) 2017-2019 Random.Zebra (https://github.com/random-zebra/)
# Distributed under the MIT software license, see the accompanying
# file LICENSE.txt or http://www.opensource.org/licenses/mit-license.php.
@@ -8,17 +8,16 @@
__b58base = len(__b58chars)
b58chars = __b58chars
-long = int
_bchr = lambda x: bytes([x])
_bord = lambda x: x
-def b58encode(v):
+def b58encode(v: bytes) -> str:
"""
encode v, which is a string of bytes, to base58.
"""
long_value = 0
- for (i, c) in enumerate(v[::-1]):
+ for i, c in enumerate(v[::-1]):
long_value += (256 ** i) * _bord(c)
result = ''
@@ -40,11 +39,10 @@ def b58encode(v):
return (__b58chars[0] * nPad) + result
-def b58decode(v, length=None):
- """ decode v into a string of len bytes
- """
+def b58decode(v: str, length: int = None) -> bytes:
+ """ decode v into a string of len bytes """
long_value = 0
- for (i, c) in enumerate(v[::-1]):
+ for i, c in enumerate(v[::-1]):
long_value += __b58chars.find(c) * (__b58base ** i)
result = bytes()
@@ -63,6 +61,6 @@ def b58decode(v, length=None):
result = _bchr(0) * nPad + result
if length is not None and len(result) != length:
- return None
+ return b''
return result
diff --git a/src/pivx_hashlib.py b/src/pivx_hashlib.py
index 628d7f2..58164d0 100644
--- a/src/pivx_hashlib.py
+++ b/src/pivx_hashlib.py
@@ -12,20 +12,20 @@
from pivx_b58 import b58encode, b58decode
-def double_sha256(data):
+def double_sha256(data: bytes) -> bytes:
return hashlib.sha256(hashlib.sha256(data).digest()).digest()
-def single_sha256(data):
+def single_sha256(data: bytes) -> bytes:
return hashlib.sha256(data).digest()
-def generate_privkey(isTestnet=False):
+def generate_privkey(isTestnet: bool = False) -> str:
"""
Based on Andreas Antonopolous work from 'Mastering Bitcoin'.
"""
valid = False
- privkey = 0
+ privkey = ""
while not valid:
privkey = bitcoin.random_key()
decoded_private_key = bitcoin.decode_privkey(privkey, 'hex')
@@ -33,20 +33,20 @@ def generate_privkey(isTestnet=False):
return base58fromhex(privkey, isTestnet)
-def base58fromhex(hexstr, isTestnet):
+def base58fromhex(hexstr: str, isTestnet: bool) -> str:
base58_secret = TESTNET_WIF_PREFIX if isTestnet else WIF_PREFIX
data = bytes([base58_secret]) + bytes.fromhex(hexstr)
checksum = bitcoin.bin_dbl_sha256(data)[0:4]
return b58encode(data + checksum)
-def pubkey_to_address(pubkey, isTestnet=False, isCold=False):
+def pubkey_to_address(pubkey: str, isTestnet: bool = False, isCold: bool = False) -> str:
pubkey_bin = bytes.fromhex(pubkey)
pkey_hash = bitcoin.bin_hash160(pubkey_bin)
return pubkeyhash_to_address(pkey_hash, isTestnet, isCold)
-def pubkeyhash_to_address(pkey_hash, isTestnet=False, isCold=False):
+def pubkeyhash_to_address(pkey_hash: bytes, isTestnet: bool = False, isCold: bool = False) -> str:
if isCold:
base58_secret = TESTNET_STAKE_MAGIC_BYTE if isTestnet else STAKE_MAGIC_BYTE
else:
@@ -56,24 +56,22 @@ def pubkeyhash_to_address(pkey_hash, isTestnet=False, isCold=False):
return b58encode(data + checksum)
-def wif_to_privkey(string):
- wif_compressed = 52 == len(string)
+def wif_to_privkey(string: str) -> str | None:
+ wif_compressed = len(string) == 52
pvkeyencoded = b58decode(string).hex()
wifversion = pvkeyencoded[:2]
checksum = pvkeyencoded[-8:]
vs = bytes.fromhex(pvkeyencoded[:-8])
check = double_sha256(vs)[0:4]
- if (wifversion == WIF_PREFIX.to_bytes(1, byteorder='big').hex() and checksum == check.hex()) \
- or (wifversion == TESTNET_WIF_PREFIX.to_bytes(1, byteorder='big').hex() and checksum == check.hex()):
+ if (wifversion == WIF_PREFIX.to_bytes(1, byteorder='big').hex() and checksum == check.hex()) or \
+ (wifversion == TESTNET_WIF_PREFIX.to_bytes(1, byteorder='big').hex() and checksum == check.hex()):
if wif_compressed:
privkey = pvkeyencoded[2:-10]
-
else:
privkey = pvkeyencoded[2:-8]
return privkey
- else:
- return None
+ return None
diff --git a/src/pivx_parser.py b/src/pivx_parser.py
index ce95c88..d970c28 100644
--- a/src/pivx_parser.py
+++ b/src/pivx_parser.py
@@ -4,17 +4,18 @@
# Distributed under the MIT software license, see the accompanying
# file LICENSE.txt or http://www.opensource.org/licenses/mit-license.php.
+from typing import Dict, Tuple, List, Literal, Any, TypedDict
from misc import getCallerName, getFunctionName, printException
import utils
from pivx_hashlib import pubkeyhash_to_address
class HexParser:
- def __init__(self, hex_str):
+ def __init__(self, hex_str: str):
self.cursor = 0
self.hex_str = hex_str
- def readInt(self, nbytes, byteorder="big", signed=False):
+ def readInt(self, nbytes: int, byteorder: Literal['little', 'big'] = 'big', signed: bool = False) -> int:
if self.cursor + nbytes * 2 > len(self.hex_str):
raise Exception("HexParser range error")
b = bytes.fromhex(self.hex_str[self.cursor:self.cursor + nbytes * 2])
@@ -22,7 +23,7 @@ def readInt(self, nbytes, byteorder="big", signed=False):
self.cursor += nbytes * 2
return res
- def readVarInt(self):
+ def readVarInt(self) -> int:
r = self.readInt(1)
if r == 253:
return self.readInt(2, "little")
@@ -32,7 +33,7 @@ def readVarInt(self):
return self.readInt(8, "little")
return r
- def readString(self, nbytes, byteorder="big"):
+ def readString(self, nbytes: int, byteorder: str = "big") -> str:
if self.cursor + nbytes * 2 > len(self.hex_str):
raise Exception("HexParser range error")
res = self.hex_str[self.cursor:self.cursor + nbytes * 2]
@@ -43,18 +44,39 @@ def readString(self, nbytes, byteorder="big"):
return res
-def IsCoinBase(vin):
+class VinType(TypedDict, total=False):
+ txid: str
+ vout: int
+ scriptSig: Dict[str, str]
+ sequence: int
+ coinbase: str
+
+
+class VoutType(TypedDict):
+ value: int
+ scriptPubKey: Dict[str, Any]
+
+
+class TxType(TypedDict):
+ version: int
+ vin: List[VinType]
+ vout: List[VoutType]
+ locktime: int
+
+
+def IsCoinBase(vin: VinType) -> bool:
return vin["txid"] == "0" * 64 and vin["vout"] == 4294967295 and vin["scriptSig"]["hex"][:2] != "c2"
-def ParseTxInput(p):
- vin = {}
- vin["txid"] = p.readString(32, "little")
- vin["vout"] = p.readInt(4, "little")
- script_len = p.readVarInt()
- vin["scriptSig"] = {}
- vin["scriptSig"]["hex"] = p.readString(script_len, "big")
- vin["sequence"] = p.readInt(4, "little")
+def ParseTxInput(p: HexParser) -> VinType:
+ vin: VinType = {
+ "txid": p.readString(32, "little"),
+ "vout": p.readInt(4, "little"),
+ "scriptSig": {
+ "hex": p.readString(p.readVarInt(), "big")
+ },
+ "sequence": p.readInt(4, "little")
+ }
if IsCoinBase(vin):
del vin["txid"]
del vin["vout"]
@@ -64,16 +86,16 @@ def ParseTxInput(p):
return vin
-def ParseTxOutput(p, isTestnet=False):
- vout = {}
- vout["value"] = p.readInt(8, "little")
- script_len = p.readVarInt()
- vout["scriptPubKey"] = {}
- vout["scriptPubKey"]["hex"] = p.readString(script_len, "big")
- vout["scriptPubKey"]["addresses"] = []
+def ParseTxOutput(p: HexParser, isTestnet: bool = False) -> VoutType:
+ vout: VoutType = {
+ "value": p.readInt(8, "little"),
+ "scriptPubKey": {
+ "hex": p.readString(p.readVarInt(), "big"),
+ "addresses": []
+ }
+ }
try:
locking_script = bytes.fromhex(vout["scriptPubKey"]["hex"])
- # add addresses only if P2PKH, P2PK or P2CS
if len(locking_script) in [25, 35, 51]:
add_bytes = utils.extract_pkh_from_locking_script(locking_script)
address = pubkeyhash_to_address(add_bytes, isTestnet)
@@ -83,37 +105,28 @@ def ParseTxOutput(p, isTestnet=False):
return vout
-def ParseTx(hex_string, isTestnet=False):
+def ParseTx(hex_string: str, isTestnet: bool = False) -> TxType:
p = HexParser(hex_string)
- tx = {}
-
- tx["version"] = p.readInt(4, "little")
-
- num_of_inputs = p.readVarInt()
- tx["vin"] = []
- for i in range(num_of_inputs):
- tx["vin"].append(ParseTxInput(p))
-
- num_of_outputs = p.readVarInt()
- tx["vout"] = []
- for i in range(num_of_outputs):
- tx["vout"].append(ParseTxOutput(p, isTestnet))
-
- tx["locktime"] = p.readInt(4, "little")
+ tx: TxType = {
+ "version": p.readInt(4, "little"),
+ "vin": [ParseTxInput(p) for _ in range(p.readVarInt())],
+ "vout": [ParseTxOutput(p, isTestnet) for _ in range(p.readVarInt())],
+ "locktime": p.readInt(4, "little")
+ }
return tx
-def IsPayToColdStaking(rawtx, out_n):
+def IsPayToColdStaking(rawtx: str, out_n: int) -> Tuple[bool, bool]:
tx = ParseTx(rawtx)
script = tx['vout'][out_n]["scriptPubKey"]["hex"]
return utils.IsPayToColdStaking(bytes.fromhex(script)), IsCoinStake(tx)
-def IsCoinStake(json_tx):
+def IsCoinStake(json_tx: TxType) -> bool:
return json_tx['vout'][0]["scriptPubKey"]["hex"] == ""
-def GetDelegatedStaker(rawtx, out_n, isTestnet):
+def GetDelegatedStaker(rawtx: str, out_n: int, isTestnet: bool) -> str:
tx = ParseTx(rawtx)
script = tx['vout'][out_n]["scriptPubKey"]["hex"]
if not utils.IsPayToColdStaking(bytes.fromhex(script)):
diff --git a/src/qt/dlg_configureRPCservers.py b/src/qt/dlg_configureRPCservers.py
index 94e7ede..81a76eb 100644
--- a/src/qt/dlg_configureRPCservers.py
+++ b/src/qt/dlg_configureRPCservers.py
@@ -4,17 +4,17 @@
# Distributed under the MIT software license, see the accompanying
# file LICENSE.txt or http://www.opensource.org/licenses/mit-license.php.
-from PyQt5.QtWidgets import QDialog, QHBoxLayout, QVBoxLayout, QLabel, \
- QListWidget, QFrame, QFormLayout, QComboBox, QLineEdit, QListWidgetItem, \
- QWidget, QPushButton, QMessageBox
-
+from PyQt5.QtWidgets import (
+ QDialog, QHBoxLayout, QVBoxLayout, QLabel, QListWidget, QFrame, QFormLayout,
+ QComboBox, QLineEdit, QListWidgetItem, QWidget, QPushButton, QMessageBox
+)
from misc import myPopUp, checkRPCstring
from threads import ThreadFuns
-class ConfigureRPCservers_dlg(QDialog):
+class ConfigureRPCserversDlg(QDialog):
def __init__(self, main_wnd):
- QDialog.__init__(self, parent=main_wnd)
+ super().__init__(parent=main_wnd)
self.main_wnd = main_wnd
self.setWindowTitle('RPC Servers Configuration')
self.changing_index = None
@@ -29,49 +29,48 @@ def clearEditFrame(self):
self.ui.host_edt.clear()
def initUI(self):
- self.ui = Ui_ConfigureRPCserversDlg()
+ self.ui = UiConfigureRPCserversDlg()
self.ui.setupUi(self)
def insert_server_list(self, server):
- id = server['id']
+ server_id = server['id']
index = self.main_wnd.mainWindow.getServerListIndex(server)
server_line = QWidget()
server_row = QHBoxLayout()
- server_text = "%s://%s" % (server['protocol'], server['host'])
- if server['id'] == 0 and server['isCustom']:
- # Local Wallet
- server_text = server_text + " <b>Local Wallet</b>"
+ server_text = f"{server['protocol']}://{server['host']}"
+ if server_id == 0 and server['isCustom']:
+ server_text += " <b>Local Wallet</b>"
elif not server['isCustom']:
- server_text = "<em style='color: purple'>%s</em>" % server_text
+ server_text = f"<em style='color: purple'>{server_text}</em>"
server_row.addWidget(QLabel(server_text))
server_row.addStretch(1)
- # -- Edit button
- editBtn = QPushButton()
- editBtn.setIcon(self.main_wnd.mainWindow.editMN_icon)
- editBtn.setToolTip("Edit server configuration")
+
+ edit_btn = QPushButton()
+ edit_btn.setIcon(self.main_wnd.mainWindow.editMN_icon)
+ edit_btn.setToolTip("Edit server configuration")
if not server['isCustom']:
- editBtn.setDisabled(True)
- editBtn.setToolTip('Default servers are not editable')
- editBtn.clicked.connect(lambda: self.onAddServer(index))
- server_row.addWidget(editBtn)
- # -- Remove button
- removeBtn = QPushButton()
- removeBtn.setIcon(self.main_wnd.mainWindow.removeMN_icon)
- removeBtn.setToolTip("Remove server configuration")
- if id == 0:
- removeBtn.setDisabled(True)
- removeBtn.setToolTip('Cannot remove local wallet')
+ edit_btn.setDisabled(True)
+ edit_btn.setToolTip('Default servers are not editable')
+ edit_btn.clicked.connect(lambda: self.onAddServer(index))
+ server_row.addWidget(edit_btn)
+
+ remove_btn = QPushButton()
+ remove_btn.setIcon(self.main_wnd.mainWindow.removeMN_icon)
+ remove_btn.setToolTip("Remove server configuration")
+ if server_id == 0:
+ remove_btn.setDisabled(True)
+ remove_btn.setToolTip('Cannot remove local wallet')
if not server['isCustom']:
- removeBtn.setDisabled(True)
- removeBtn.setToolTip('Cannot remove default servers')
- removeBtn.clicked.connect(lambda: self.onRemoveServer(index))
- server_row.addWidget(removeBtn)
- # --
+ remove_btn.setDisabled(True)
+ remove_btn.setToolTip('Cannot remove default servers')
+ remove_btn.clicked.connect(lambda: self.onRemoveServer(index))
+ server_row.addWidget(remove_btn)
+
server_line.setLayout(server_row)
- self.serverItems[id] = QListWidgetItem()
- self.serverItems[id].setSizeHint(server_line.sizeHint())
- self.ui.serversBox.addItem(self.serverItems[id])
- self.ui.serversBox.setItemWidget(self.serverItems[id], server_line)
+ self.serverItems[server_id] = QListWidgetItem()
+ self.serverItems[server_id].setSizeHint(server_line.sizeHint())
+ self.ui.serversBox.addItem(self.serverItems[server_id])
+ self.ui.serversBox.setItemWidget(self.serverItems[server_id], server_line)
def loadServers(self):
# Clear serversBox
@@ -85,10 +84,7 @@ def loadEditFrame(self, index):
server = self.main_wnd.mainWindow.rpcServersList[index]
self.ui.user_edt.setText(server['user'])
self.ui.passwd_edt.setText(server['password'])
- if server['protocol'] == 'https':
- self.ui.protocol_select.setCurrentIndex(1)
- else:
- self.ui.protocol_select.setCurrentIndex(0)
+ self.ui.protocol_select.setCurrentIndex(1 if server['protocol'] == 'https' else 0)
self.ui.host_edt.setText(server['host'])
def onAddServer(self, index=None):
@@ -122,13 +118,11 @@ def onClose(self):
self.close()
def onRemoveServer(self, index):
- mess = "Are you sure you want to remove server with index %d (%s) from list?" % (
- index, self.main_wnd.mainWindow.rpcServersList[index].get('host'))
+ mess = f"Are you sure you want to remove server with index {index} ({self.main_wnd.mainWindow.rpcServersList[index].get('host')}) from list?"
ans = myPopUp(self, QMessageBox.Question, 'PET4L - remove server', mess)
if ans == QMessageBox.Yes:
- # Remove entry from database
- id = self.main_wnd.mainWindow.rpcServersList[index].get('id')
- self.main_wnd.db.removeRPCServer(id)
+ server_id = self.main_wnd.mainWindow.rpcServersList[index].get('id')
+ self.main_wnd.db.removeRPCServer(server_id)
def onSave(self):
# Get new config data
@@ -136,27 +130,24 @@ def onSave(self):
host = self.ui.host_edt.text()
user = self.ui.user_edt.text()
passwd = self.ui.passwd_edt.text()
- # Check malformed URL
- url_string = "%s://%s:%s@%s" % (protocol, user, passwd, host)
+ url_string = f"{protocol}://{user}:{passwd}@{host}"
if checkRPCstring(url_string):
if self.changing_index is None:
# Save new entry in DB.
self.main_wnd.db.addRPCServer(protocol, host, user, passwd)
else:
- # Edit existing entry to DB.
- id = self.main_wnd.mainWindow.rpcServersList[self.changing_index].get('id')
- self.main_wnd.db.editRPCServer(protocol, host, user, passwd, id)
- # If this was previously selected in mainWindow, update status
+ server_id = self.main_wnd.mainWindow.rpcServersList[self.changing_index].get('id')
+ self.main_wnd.db.editRPCServer(protocol, host, user, passwd, server_id)
clients = self.main_wnd.mainWindow.header.rpcClientsBox
data = clients.itemData(clients.currentIndex())
- if data.get('id') == id and data.get('isCustom'):
- ThreadFuns.runInThread(self.main_wnd.mainWindow.updateRPCstatus, (True,), )
+ if data.get('id') == server_id and data.get('isCustom'):
+ ThreadFuns.runInThread(self.main_wnd.mainWindow.updateRPCstatus, (True,))
# call onCancel
self.onCancel()
-class Ui_ConfigureRPCserversDlg(object):
+class UiConfigureRPCserversDlg:
def setupUi(self, ConfigureRPCserversDlg):
ConfigureRPCserversDlg.setModal(True)
# -- Layout
diff --git a/src/qt/dlg_pinMatrix.py b/src/qt/dlg_pinMatrix.py
index fbb9656..c5670b4 100644
--- a/src/qt/dlg_pinMatrix.py
+++ b/src/qt/dlg_pinMatrix.py
@@ -10,10 +10,10 @@
from misc import myPopUp_sb
-class PinMatrix_dlg(QDialog):
+class PinMatrixDlg(QDialog):
def __init__(self, text='', title='Enter PIN', fHideBtns=True):
- QDialog.__init__(self)
- self.text = text if text != '' else "Check device and enter PIN"
+ super().__init__()
+ self.text = text if text else "Check device and enter PIN"
self.hideBtns = fHideBtns
self.pin = ''
self.setWindowTitle(title)
@@ -46,7 +46,7 @@ def onOK(self):
myPopUp_sb(self, "warn", 'Wrong PIN!', text)
def setupUI(self):
- Ui_pinMatrixDlg.setupUi(self, self)
+ UiPinMatrixDlg.setupUi(self, self)
# Connect buttons matrix
for i in range(9):
self.btn[i].clicked.connect(lambda _, b=i + 1: self.btn_clicked(str(b)))
@@ -56,14 +56,14 @@ def setupUI(self):
self.btn_cancel.clicked.connect(self.onCancel)
-class Ui_pinMatrixDlg(object):
+class UiPinMatrixDlg:
def setupUi(self, PinMatrixDlg):
PinMatrixDlg.setModal(True)
layout = QVBoxLayout(PinMatrixDlg)
layout.setContentsMargins(10, 8, 10, 10)
# Header
- title = QLabel("<b>%s</b>" % PinMatrixDlg.text)
+ title = QLabel(f"<b>{PinMatrixDlg.text}</b>")
title.setAlignment(Qt.AlignCenter)
layout.addWidget(title)
diff --git a/src/qt/dlg_signmessage.py b/src/qt/dlg_signmessage.py
index 2a9b55d..8dbf7bb 100644
--- a/src/qt/dlg_signmessage.py
+++ b/src/qt/dlg_signmessage.py
@@ -32,18 +32,18 @@
from utils import checkPivxAddr, ecdsa_verify_addr
-class SignMessage_dlg(QDialog):
+class SignMessageDlg(QDialog):
def __init__(self, main_wnd):
- QDialog.__init__(self, parent=main_wnd)
+ super().__init__(parent=main_wnd)
self.setWindowTitle('Sign/Verify Message')
self.initUI(main_wnd)
def initUI(self, main_wnd):
- self.ui = Ui_SignMessageDlg()
+ self.ui = UiSignMessageDlg()
self.ui.setupUi(self, main_wnd)
-class Ui_SignMessageDlg(object):
+class UiSignMessageDlg:
def setupUi(self, SignMessageDlg, main_wnd):
SignMessageDlg.setModal(True)
SignMessageDlg.setMinimumWidth(600)
@@ -61,7 +61,7 @@ def setupUi(self, SignMessageDlg, main_wnd):
class TabSign:
def __init__(self, main_wnd):
self.main_wnd = main_wnd
- self.ui = TabSign_gui()
+ self.ui = TabSignGui()
self.loadAddressComboBox()
# connect signals/buttons
self.ui.addressComboBox.currentIndexChanged.connect(lambda: self.onChangeSelectedAddress())
@@ -102,9 +102,9 @@ def findPubKey(self):
pk = device.scanForPubKey(self.hwAcc, self.spath, self.currIsTestnet)
if pk is None:
- mess = "Unable to find public key. The action was refused on the device or another application "
- mess += "might have taken over the USB communication with the device.<br><br>"
- mess += "The operation was canceled."
+ mess = ("Unable to find public key. The action was refused on the device or another application "
+ "might have taken over the USB communication with the device.<br><br>"
+ "The operation was canceled.")
myPopUp_sb(self.main_wnd, QMessageBox.Critical, 'PET4L - PK not found', mess)
return
self.updateGenericAddress(pk)
@@ -124,10 +124,9 @@ def findSpath_done(self):
addy = self.ui.addressLineEdit.text().strip()
starting_spath = self.curr_starting_spath
spath_count = self.curr_spath_count
- mess = "Scanned addresses <b>%d</b> to <b>%d</b> of HW account <b>%d</b>.<br>" % (
- starting_spath, starting_spath + spath_count - 1, self.hwAcc)
- mess += "Unable to find the address <i>%s</i>.<br>Maybe it's on a different account.<br><br>" % addy
- mess += "Do you want to scan %d more addresses of account n.<b>%d</b> ?" % (spath_count, self.hwAcc)
+ mess = (f"Scanned addresses <b>{starting_spath}</b> to <b>{starting_spath + spath_count - 1}</b> of HW account <b>{self.hwAcc}</b>.<br>"
+ f"Unable to find the address <i>{addy}</i>.<br>Maybe it's on a different account.<br><br>"
+ f"Do you want to scan {spath_count} more addresses of account n.<b>{self.hwAcc}</b> ?")
ans = myPopUp(self.main_wnd, QMessageBox.Question, 'PET4L - spath search', mess)
if ans == QMessageBox.Yes:
# Look for 10 more addresses
@@ -165,9 +164,8 @@ def onSave(self):
"All Files (*);; Text Files (*.txt)", options=options)
try:
if fileName:
- save_file = open(fileName, 'w', encoding="utf-8")
- save_file.write(self.ui.signatureTextEdt.toPlainText())
- save_file.close()
+ with open(fileName, 'w', encoding="utf-8") as save_file:
+ save_file.write(self.ui.signatureTextEdt.toPlainText())
myPopUp_sb(self.main_wnd, QMessageBox.Information, 'PET4L - saved', "Signature saved to file")
return
except Exception as e:
@@ -188,7 +186,7 @@ def onSearchPK(self):
if not checkPivxAddr(addy, self.currIsTestnet):
net = "testnet" if self.currIsTestnet else "mainnet"
- mess = "PIVX address not valid. Insert valid PIVX %s address" % net
+ mess = f"PIVX address not valid. Insert valid PIVX {net} address"
myPopUp_sb(self.main_wnd, QMessageBox.Warning, 'PET4L - invalid address', mess)
return
@@ -251,9 +249,9 @@ def setSignEnabled(self, enabled):
self.ui.hiddenLine.setVisible(not enabled)
tooltip = ""
if not enabled:
- tooltip = "You need to find the address PK in your hardware device first.\n" \
- "Insert the account number (usually 0) and either a PIVX address\n" \
- "or the spath_id (address number) and click 'Search HW'."
+ tooltip = ("You need to find the address PK in your hardware device first.\n"
+ "Insert the account number (usually 0) and either a PIVX address\n"
+ "or the spath_id (address number) and click 'Search HW'.")
self.ui.addressLabel.setText("")
self.ui.signBtn.setToolTip(tooltip)
@@ -263,14 +261,14 @@ def updateGenericAddress(self, pk):
# double check address
addy = self.ui.addressLineEdit.text().strip()
if addy != genericAddy:
- mess = "Error! retrieved address (%s) different from input (%s)" % (genericAddy, addy)
+ mess = f"Error! retrieved address ({genericAddy}) different from input ({addy})"
myPopUp_sb(self.main_wnd, QMessageBox.Critical, 'PET4L - address mismatch', mess)
self.ui.addressLabel.setText("")
return
# update generic address
self.setSignEnabled(True)
self.currAddress = genericAddy
- self.currHwPath = "%d'/0/%d" % (self.hwAcc, self.spath)
+ self.currHwPath = f"{self.hwAcc}'/0/{self.spath}"
self.ui.addressLabel.setText(self.currAddress)
self.ui.editBtn.setVisible(True)
@@ -278,7 +276,7 @@ def updateGenericAddress(self, pk):
class TabVerify:
def __init__(self, main_wnd):
self.main_wnd = main_wnd
- self.ui = TabVerify_gui()
+ self.ui = TabVerifyGui()
# connect signals/buttons
self.ui.verifyBtn.clicked.connect(lambda: self.onVerify())
@@ -307,21 +305,21 @@ def onVerify(self):
self.ui.signatureTextEdt.toPlainText(),
self.ui.addressLineEdit.text().strip())
except Exception as e:
- mess = "Error decoding signature:\n" + str(e)
+ mess = f"Error decoding signature:\n{e}"
myPopUp_sb(self.main_wnd, QMessageBox.Warning, 'PET4L - invalid signature', mess)
ok = False
if ok:
mess = "<span style='color: green'>Signature OK"
else:
mess = "<span style='color: red'>Signature doesn't verify"
- mess = "<b>" + mess + "</span></b>"
+ mess = f"<b>{mess}</span></b>"
self.ui.resultLabel.setText(mess)
self.ui.resultLabel.setVisible(True)
-class TabSign_gui(QWidget):
+class TabSignGui(QWidget):
def __init__(self):
- QWidget.__init__(self)
+ super().__init__()
layout = QVBoxLayout()
layout.setContentsMargins(10, 10, 10, 10)
layout.setSpacing(13)
@@ -426,7 +424,7 @@ def __init__(self):
self.copyBtn.setVisible(False)
row5.addWidget(self.copyBtn)
self.saveBtn = QPushButton("Save")
- self.saveBtn.setToolTip("Save signature to ca file")
+ self.saveBtn.setToolTip("Save signature to a file")
self.saveBtn.setVisible(False)
row5.addWidget(self.saveBtn)
layout.addLayout(row5)
@@ -434,9 +432,9 @@ def __init__(self):
self.setLayout(layout)
-class TabVerify_gui(QWidget):
+class TabVerifyGui(QWidget):
def __init__(self):
- QWidget.__init__(self)
+ super().__init__()
layout = QVBoxLayout()
layout.setContentsMargins(10, 10, 10, 10)
layout.setSpacing(13)
diff --git a/src/qt/guiHeader.py b/src/qt/guiHeader.py
index fdaf92e..75f2b80 100644
--- a/src/qt/guiHeader.py
+++ b/src/qt/guiHeader.py
@@ -12,7 +12,7 @@
class GuiHeader(QWidget):
def __init__(self, caller, *args, **kwargs):
- QWidget.__init__(self)
+ super().__init__(*args, **kwargs)
layout = QHBoxLayout()
layout.setContentsMargins(0, 0, 0, 0)
# --- 1) Check Box
@@ -28,7 +28,7 @@ def __init__(self, caller, *args, **kwargs):
self.button_checkRpc.setToolTip("try to connect to RPC server")
self.centralBox.addWidget(self.button_checkRpc, 0, 2)
self.rpcLed = QLabel()
- self.rpcLed.setToolTip("%s" % caller.rpcStatusMess)
+ self.rpcLed.setToolTip(f"{caller.rpcStatusMess}")
self.rpcLed.setPixmap(caller.ledGrayH_icon)
self.centralBox.addWidget(self.rpcLed, 0, 3)
self.lastPingBox = QWidget()
@@ -65,7 +65,7 @@ def __init__(self, caller, *args, **kwargs):
self.button_checkHw.setToolTip("try to connect to Hardware Wallet")
self.centralBox.addWidget(self.button_checkHw, 1, 2)
self.hwLed = QLabel()
- self.hwLed.setToolTip("status: %s" % caller.hwStatusMess)
+ self.hwLed.setToolTip(f"status: {caller.hwStatusMess}")
self.hwLed.setPixmap(caller.ledGrayH_icon)
self.centralBox.addWidget(self.hwLed, 1, 3)
layout.addLayout(self.centralBox)
diff --git a/src/qt/gui_tabRewards.py b/src/qt/gui_tabRewards.py
index 31c1da5..91cc2d0 100644
--- a/src/qt/gui_tabRewards.py
+++ b/src/qt/gui_tabRewards.py
@@ -8,16 +8,18 @@
import os.path
from PyQt5.QtCore import Qt
-from PyQt5.Qt import QLabel, QFormLayout, QDoubleSpinBox, QTableWidget, QTableWidgetItem, QAbstractItemView, QHeaderView, QSpinBox
-from PyQt5.QtWidgets import QWidget, QPushButton, QHBoxLayout, QGroupBox, QVBoxLayout
-from PyQt5.QtWidgets import QLineEdit, QComboBox, QProgressBar
+from PyQt5.QtWidgets import (
+ QWidget, QPushButton, QHBoxLayout, QGroupBox, QVBoxLayout, QLineEdit, QComboBox,
+ QProgressBar, QLabel, QFormLayout, QDoubleSpinBox, QTableWidget, QTableWidgetItem,
+ QAbstractItemView, QHeaderView, QSpinBox
+)
sys.path.append(os.path.join(os.path.dirname(__file__), 'src'))
-class TabRewards_gui(QWidget):
+class TabRewardsGui(QWidget):
def __init__(self, imgDir, *args, **kwargs):
- QWidget.__init__(self)
+ super().__init__(*args, **kwargs)
self.imgDir = imgDir
self.initRewardsForm()
mainVertical = QVBoxLayout()
@@ -34,6 +36,7 @@ def initRewardsForm(self):
layout.setContentsMargins(10, 10, 10, 10)
layout.setSpacing(13)
layout.setFieldGrowthPolicy(QFormLayout.AllNonFixedFieldsGrow)
+
# --- ROW 1
line1 = QHBoxLayout()
line1.addWidget(QLabel("Account HW"))
@@ -69,6 +72,7 @@ def initRewardsForm(self):
self.btn_reload.setToolTip("Reload data from ledger device")
line1.addWidget(self.btn_reload)
layout.addRow(line1)
+
# --- ROW 2: address and copy btn
hBox = QHBoxLayout()
self.addySelect = QComboBox()
@@ -79,6 +83,7 @@ def initRewardsForm(self):
self.btn_Copy.setToolTip("Copy address to clipboard")
hBox.addWidget(self.btn_Copy)
layout.addRow(hBox)
+
# --- ROW 3: UTXOs
self.rewardsList = QVBoxLayout()
self.rewardsList.statusLabel = QLabel('<b style="color:red">Reload Rewards</b>')
@@ -86,7 +91,6 @@ def initRewardsForm(self):
self.rewardsList.addWidget(self.rewardsList.statusLabel)
self.rewardsList.box = QTableWidget()
self.rewardsList.box.setMinimumHeight(230)
- # self.rewardsList.box.setMaximumHeight(140)
self.rewardsList.box.setHorizontalScrollBarPolicy(Qt.ScrollBarAlwaysOff)
self.rewardsList.box.setSelectionMode(QAbstractItemView.MultiSelection)
self.rewardsList.box.setSelectionBehavior(QAbstractItemView.SelectRows)
@@ -95,6 +99,7 @@ def initRewardsForm(self):
self.rewardsList.box.setRowCount(0)
self.rewardsList.box.horizontalHeader().setSectionResizeMode(2, QHeaderView.Stretch)
self.rewardsList.box.verticalHeader().hide()
+
item = QTableWidgetItem()
item.setText("PIVs")
item.setTextAlignment(Qt.AlignCenter)
@@ -111,8 +116,10 @@ def initRewardsForm(self):
item.setText("TX Output N")
item.setTextAlignment(Qt.AlignCenter)
self.rewardsList.box.setHorizontalHeaderItem(3, item)
+
self.rewardsList.addWidget(self.rewardsList.box)
layout.addRow(self.rewardsList)
+
# --- ROW 3
hBox2 = QHBoxLayout()
self.btn_selectAllRewards = QPushButton("Select All")
@@ -129,6 +136,7 @@ def initRewardsForm(self):
hBox2.addWidget(self.selectedRewardsLine)
hBox2.addStretch(1)
layout.addRow(hBox2)
+
# --- ROW 4
hBox3 = QHBoxLayout()
self.destinationLine = QLineEdit()
@@ -145,6 +153,7 @@ def initRewardsForm(self):
self.btn_sendRewards = QPushButton("Send")
hBox3.addWidget(self.btn_sendRewards)
layout.addRow(QLabel("Destination Address"), hBox3)
+
hBox4 = QHBoxLayout()
hBox4.addStretch(1)
self.loadingLine = QLabel("<b style='color:red'>Preparing TX.</b> Completed: ")
@@ -157,8 +166,10 @@ def initRewardsForm(self):
self.loadingLine.hide()
self.loadingLinePercent.hide()
layout.addRow(hBox4)
+
# --- Set Layout
self.rewardsForm.setLayout(layout)
+
# --- ROW 5
self.btn_Cancel = QPushButton("Clear")
diff --git a/src/rpcClient.py b/src/rpcClient.py
index 64e8db6..11b8456 100644
--- a/src/rpcClient.py
+++ b/src/rpcClient.py
@@ -9,6 +9,7 @@
import http.client as httplib
import ssl
import threading
+from typing import Union
from constants import DEFAULT_PROTOCOL_VERSION, MINIMUM_FEE
from misc import getCallerName, getFunctionName, printException, printDbg, now, timeThis
@@ -19,7 +20,6 @@ def process_RPC_exceptions_int(*args, **kwargs):
try:
args[0].httpConnection.connect()
return func(*args, **kwargs)
-
except Exception as e:
message = "Exception in RPC client"
printException(getCallerName(True), getFunctionName(True), message, str(e))
@@ -35,13 +35,24 @@ def process_RPC_exceptions_int(*args, **kwargs):
class RpcClient:
- def __init__(self, rpc_protocol, rpc_host, rpc_user, rpc_password):
+ def __init__(self, rpc_protocol: str, rpc_host: str, rpc_user: str, rpc_password: str):
# Lock for threads
self.lock = threading.RLock()
- self.rpc_url = "%s://%s:%s@%s" % (rpc_protocol, rpc_user, rpc_password, rpc_host)
+ self.rpc_url = f"{rpc_protocol}://{rpc_user}:{rpc_password}@{rpc_host}"
+
+ if ":" in rpc_host:
+ host, port_str = rpc_host.split(":")
+ try:
+ port = int(port_str)
+ except ValueError:
+ raise ValueError(f"Invalid port specified: {port_str}")
+ else:
+ host = rpc_host
+ port = None
+
+ self.httpConnection: Union[httplib.HTTPConnection, httplib.HTTPSConnection]
- host, port = rpc_host.split(":")
if rpc_protocol == "https":
self.httpConnection = httplib.HTTPSConnection(host, port, timeout=20, context=ssl._create_unverified_context())
else:
@@ -50,65 +61,47 @@ def __init__(self, rpc_protocol, rpc_host, rpc_user, rpc_password):
self.conn = AuthServiceProxy(self.rpc_url, timeout=1000, connection=self.httpConnection)
@process_RPC_exceptions
- def getBlockCount(self):
- n = 0
+ def getBlockCount(self) -> int:
with self.lock:
- n = self.conn.getblockcount()
-
- return n
+ return self.conn.getblockcount()
@process_RPC_exceptions
- def getBlockHash(self, blockNum):
- h = None
+ def getBlockHash(self, blockNum: int) -> str:
with self.lock:
- h = self.conn.getblockhash(blockNum)
-
- return h
+ return self.conn.getblockhash(blockNum)
@process_RPC_exceptions
- def getBudgetVotes(self, proposal):
- votes = {}
+ def getBudgetVotes(self, proposal: str) -> dict:
with self.lock:
- votes = self.conn.getbudgetvotes(proposal)
-
- return votes
+ return self.conn.getbudgetvotes(proposal)
@process_RPC_exceptions
- def getFeePerKb(self):
- res = MINIMUM_FEE
+ def getFeePerKb(self) -> float:
with self.lock:
# get transaction data from last 200 blocks
feePerKb = float(self.conn.getfeeinfo(200)['feeperkb'])
- res = (feePerKb if feePerKb > MINIMUM_FEE else MINIMUM_FEE)
-
- return res
+ return feePerKb if feePerKb > MINIMUM_FEE else MINIMUM_FEE
@process_RPC_exceptions
- def getMNStatus(self, address):
- mnStatus = None
+ def getMNStatus(self, address: str) -> dict:
with self.lock:
mnStatusList = self.conn.listmasternodes(address)
if not mnStatusList:
- return None
+ return {}
mnStatus = mnStatusList[0]
mnStatus['mnCount'] = self.conn.getmasternodecount()['enabled']
-
- return mnStatus
+ return mnStatus
@process_RPC_exceptions
- def getMasternodeCount(self):
- ans = None
+ def getMasternodeCount(self) -> dict:
with self.lock:
- ans = self.conn.getmasternodecount()
-
- return ans
+ return self.conn.getmasternodecount()
@process_RPC_exceptions
- def getMasternodes(self):
+ def getMasternodes(self) -> dict:
printDbg("RPC: Getting masternode list...")
mnList = {}
score = []
- masternodes = []
with self.lock:
masternodes = self.conn.listmasternodes()
@@ -120,7 +113,6 @@ def getMasternodes(self):
else:
lastpaid_ago = now() - mn.get('lastpaid')
mn['score'] = min(lastpaid_ago, mn.get('activetime'))
-
else:
mn['score'] = 0
@@ -138,17 +130,13 @@ def getMasternodes(self):
return mnList
@process_RPC_exceptions
- def getNextSuperBlock(self):
- n = 0
+ def getNextSuperBlock(self) -> int:
with self.lock:
- n = self.conn.getnextsuperblock()
-
- return n
+ return self.conn.getnextsuperblock()
@process_RPC_exceptions
- def getProposalsProjection(self):
+ def getProposalsProjection(self) -> list:
printDbg("RPC: Getting proposals projection...")
- data = []
proposals = []
with self.lock:
# get budget projection JSON data
@@ -156,11 +144,12 @@ def getProposalsProjection(self):
for p in data:
# create proposal-projection dictionary
- new_proposal = {}
- new_proposal['Name'] = p.get('Name')
- new_proposal['Allotted'] = float(p.get("Alloted"))
- new_proposal['Votes'] = p.get('Yeas') - p.get('Nays')
- new_proposal['Total_Allotted'] = float(p.get('TotalBudgetAlloted'))
+ new_proposal = {
+ 'Name': p.get('Name'),
+ 'Allotted': float(p.get("Alloted")),
+ 'Votes': p.get('Yeas') - p.get('Nays'),
+ 'Total_Allotted': float(p.get('TotalBudgetAlloted'))
+ }
# append dictionary to list
proposals.append(new_proposal)
@@ -168,27 +157,20 @@ def getProposalsProjection(self):
return proposals
@process_RPC_exceptions
- def getProtocolVersion(self):
- res = DEFAULT_PROTOCOL_VERSION
+ def getProtocolVersion(self) -> int:
with self.lock:
prot_version = self.conn.getinfo().get('protocolversion')
- res = int(prot_version)
-
- return res
+ return int(prot_version) if prot_version else DEFAULT_PROTOCOL_VERSION
@process_RPC_exceptions
- def getRawTransaction(self, txid):
- res = None
+ def getRawTransaction(self, txid: str) -> str:
with self.lock:
- res = self.conn.getrawtransaction(txid)
-
- return res
+ return self.conn.getrawtransaction(txid)
@process_RPC_exceptions
- def getStatus(self):
+ def getStatus(self) -> tuple[bool, str, int, float, bool]:
status = False
- statusMess = "Unable to connect to a PIVX RPC server.\n"
- statusMess += "Either the local PIVX wallet is not open, or the remote RPC server is not responding."
+ statusMess = "Unable to connect to a PIVX RPC server.\nEither the local PIVX wallet is not open, or the remote RPC server is not responding."
n = 0
response_time = None
with self.lock:
@@ -204,58 +186,38 @@ def getStatus(self):
return status, statusMess, n, response_time, isTestnet
@process_RPC_exceptions
- def isBlockchainSynced(self):
- res = False
- response_time = None
+ def isBlockchainSynced(self) -> tuple[bool, float]:
with self.lock:
status, response_time = timeThis(self.conn.mnsync, 'status')
if status is not None:
- res = status.get("IsBlockchainSynced")
-
- return res, response_time
+ return status.get("IsBlockchainSynced"), response_time
+ return False, response_time
@process_RPC_exceptions
- def mnBudgetRawVote(self, mn_tx_hash, mn_tx_index, proposal_hash, vote, time, vote_sig):
- res = None
+ def mnBudgetRawVote(self, mn_tx_hash: str, mn_tx_index: int, proposal_hash: str, vote: str, time: int, vote_sig: str) -> str:
with self.lock:
- res = self.conn.mnbudgetrawvote(mn_tx_hash, mn_tx_index, proposal_hash, vote, time, vote_sig)
-
- return res
+ return self.conn.mnbudgetrawvote(mn_tx_hash, mn_tx_index, proposal_hash, vote, time, vote_sig)
@process_RPC_exceptions
- def decodemasternodebroadcast(self, work):
+ def decodemasternodebroadcast(self, work: str) -> str:
printDbg("RPC: Decoding masternode broadcast...")
- res = ""
with self.lock:
- res = self.conn.decodemasternodebroadcast(work.strip())
-
- return res
+ return self.conn.decodemasternodebroadcast(work.strip())
@process_RPC_exceptions
- def relaymasternodebroadcast(self, work):
+ def relaymasternodebroadcast(self, work: str) -> str:
printDbg("RPC: Relaying masternode broadcast...")
- res = ""
with self.lock:
- res = self.conn.relaymasternodebroadcast(work.strip())
-
- return res
+ return self.conn.relaymasternodebroadcast(work.strip())
@process_RPC_exceptions
- def sendRawTransaction(self, tx_hex):
- dbg_mess = "RPC: Sending raw transaction"
- dbg_mess += "..."
- printDbg(dbg_mess)
- tx_id = None
+ def sendRawTransaction(self, tx_hex: str) -> str:
+ printDbg("RPC: Sending raw transaction...")
with self.lock:
- tx_id = self.conn.sendrawtransaction(tx_hex, True)
-
- return tx_id
+ return self.conn.sendrawtransaction(tx_hex, True)
@process_RPC_exceptions
- def verifyMessage(self, pivxaddress, signature, message):
+ def verifyMessage(self, pivxaddress: str, signature: str, message: str) -> bool:
printDbg("RPC: Verifying message...")
- res = False
with self.lock:
- res = self.conn.verifymessage(pivxaddress, signature, message)
-
- return res
+ return self.conn.verifymessage(pivxaddress, signature, message)
diff --git a/src/tabRewards.py b/src/tabRewards.py
index 4e6c1c7..48023f2 100644
--- a/src/tabRewards.py
+++ b/src/tabRewards.py
@@ -7,15 +7,14 @@
import threading
import simplejson as json
-from PyQt5.Qt import QApplication
-from PyQt5.QtCore import Qt
+from PyQt5.Qt import QApplication, Qt
from PyQt5.QtWidgets import QMessageBox, QTableWidgetItem, QHeaderView
from constants import MINIMUM_FEE
from misc import printDbg, printError, printException, getCallerName, getFunctionName, \
persistCacheSetting, myPopUp, myPopUp_sb, DisconnectedException, checkTxInputs
from pivx_parser import ParseTx, IsPayToColdStaking, GetDelegatedStaker
-from qt.gui_tabRewards import TabRewards_gui
+from qt.gui_tabRewards import TabRewardsGui
from threads import ThreadFuns
from txCache import TxCache
from utils import checkPivxAddr
@@ -34,7 +33,7 @@ def __init__(self, caller):
self.suggestedFee = MINIMUM_FEE
# --- Initialize GUI
- self.ui = TabRewards_gui(self.caller.imgDir)
+ self.ui = TabRewardsGui(self.caller.imgDir)
self.caller.tabRewards = self.ui
self.ui.btn_Copy.setIcon(self.caller.copy_icon)
@@ -48,14 +47,14 @@ def __init__(self, caller):
self.updateFee()
# Connect GUI buttons
- self.ui.addySelect.currentIndexChanged.connect(lambda: self.onChangeSelected())
- self.ui.rewardsList.box.itemClicked.connect(lambda: self.updateSelection())
- self.ui.btn_reload.clicked.connect(lambda: self.loadSelection())
- self.ui.btn_selectAllRewards.clicked.connect(lambda: self.onSelectAllRewards())
- self.ui.btn_deselectAllRewards.clicked.connect(lambda: self.onDeselectAllRewards())
- self.ui.btn_sendRewards.clicked.connect(lambda: self.onSendRewards())
- self.ui.btn_Cancel.clicked.connect(lambda: self.onCancel())
- self.ui.btn_Copy.clicked.connect(lambda: self.onCopy())
+ self.ui.addySelect.currentIndexChanged.connect(self.onChangeSelected)
+ self.ui.rewardsList.box.itemClicked.connect(self.updateSelection)
+ self.ui.btn_reload.clicked.connect(self.loadSelection)
+ self.ui.btn_selectAllRewards.clicked.connect(self.onSelectAllRewards)
+ self.ui.btn_deselectAllRewards.clicked.connect(self.onDeselectAllRewards)
+ self.ui.btn_sendRewards.clicked.connect(self.onSendRewards)
+ self.ui.btn_Cancel.clicked.connect(self.onCancel)
+ self.ui.btn_Copy.clicked.connect(self.onCopy)
# Connect Signals
self.caller.sig_UTXOsLoading.connect(self.update_loading_utxos)
@@ -72,10 +71,10 @@ def display_utxos(self):
rewards = self.caller.parent.db.getRewardsList(self.curr_addr)
if rewards is not None:
- def item(value):
+ def item(value: str) -> QTableWidgetItem:
item = QTableWidgetItem(value)
item.setTextAlignment(Qt.AlignCenter)
- item.setFlags(Qt.ItemIsEnabled | Qt.ItemIsSelectable)
+ item.setFlags(item.flags() | Qt.ItemIsEnabled | Qt.ItemIsSelectable)
return item
# Clear up old list
@@ -93,17 +92,17 @@ def item(value):
self.ui.rewardsList.box.showRow(row)
if utxo['staker'] != "":
self.ui.rewardsList.box.item(row, 2).setIcon(self.caller.coldStaking_icon)
- self.ui.rewardsList.box.item(row, 2).setToolTip("Staked by <b>%s</b>" % utxo['staker'])
+ self.ui.rewardsList.box.item(row, 2).setToolTip(f"Staked by <b>{utxo['staker']}</b>")
# make immature rewards unselectable
if utxo['coinstake']:
required = 16 if self.caller.isTestnetRPC else 101
if utxo['confirmations'] < required:
- for i in range(0, 4):
+ for i in range(4):
self.ui.rewardsList.box.item(row, i).setFlags(Qt.NoItemFlags)
ttip = self.ui.rewardsList.box.item(row, i).toolTip()
self.ui.rewardsList.box.item(row, i).setToolTip(
- ttip + "\n(Immature - %d confirmations required)" % required)
+ ttip + f"\n(Immature - {required} confirmations required)")
self.ui.rewardsList.box.resizeColumnsToContents()
@@ -115,15 +114,12 @@ def item(value):
if not self.caller.rpcConnected:
self.ui.resetStatusLabel('<b style="color:red">PIVX wallet not connected</b>')
else:
- self.ui.resetStatusLabel('<b style="color:red">Found no Rewards for %s</b>' % self.curr_addr)
+ self.ui.resetStatusLabel(f'<b style="color:red">Found no Rewards for {self.curr_addr}</b>')
- def getSelection(self):
+ def getSelection(self) -> list:
# Get selected rows indexes
items = self.ui.rewardsList.box.selectedItems()
- rows = set()
- for i in range(0, len(items)):
- row = items[i].row()
- rows.add(row)
+ rows = {item.row() for item in items}
indexes = list(rows)
# Get UTXO info from DB for each
selection = []
@@ -139,7 +135,7 @@ def loadSelection(self):
printDbg("Checking HW device")
if self.caller.hwStatus != 2:
myPopUp_sb(self.caller, "crit", 'PET4L - hw device check', "Connect to HW device first")
- printDbg("Unable to connect - hw status: %d" % self.caller.hwStatus)
+ printDbg(f"Unable to connect - hw status: {self.caller.hwStatus}")
return None
self.ui.addySelect.clear()
@@ -159,7 +155,7 @@ def loadSelection_thread(self, ctrl):
self.caller.parent.cache["intExt"] = persistCacheSetting('cache_intExt', intExt)
for i in range(spathFrom, spathTo + 1):
- path = "%d'/%d/%d" % (hwAcc, intExt, i)
+ path = f"{hwAcc}'/{intExt}/{i}"
address = self.caller.hwdevice.scanForAddress(hwAcc, i, intExt, isTestnet)
try:
balance = self.caller.apiClient.getBalance(address)
@@ -167,9 +163,9 @@ def loadSelection_thread(self, ctrl):
print(e)
balance = 0
- itemLine = "%s -- %s" % (path, address)
- if (balance):
- itemLine += " [%s PIV]" % str(balance)
+ itemLine = f"{path} -- {address}"
+ if balance:
+ itemLine += f" [{balance} PIV]"
self.ui.addySelect.addItem(itemLine, [path, address, balance])
@@ -192,7 +188,7 @@ def load_utxos_thread(self, ctrl):
# Get raw tx
u['rawtx'] = TxCache(self.caller)[u['txid']]
if u['rawtx'] is None:
- printDbg("Unable to get raw TX with hash=%s from RPC server." % u['txid'])
+ printDbg(f"Unable to get raw TX with hash={u['txid']} from RPC server.")
# Don't save UTXO if raw TX is unavailable
utxos.remove(u)
u['staker'] = ""
@@ -263,7 +259,7 @@ def onSendRewards(self):
self.caller.onCheckHw()
if self.caller.hwStatus != 2:
myPopUp_sb(self.caller, "crit", 'PET4L - hw device check', "Connect to HW device first")
- printDbg("Unable to connect to hardware device. The device status is: %d" % self.caller.hwStatus)
+ printDbg(f"Unable to connect to hardware device. The device status is: {self.caller.hwStatus}")
return None
# SEND
@@ -277,15 +273,15 @@ def SendRewards(self, inputs=None, gui=None):
# re-connect signals
try:
self.caller.hwdevice.api.sigTxdone.disconnect()
- except:
+ except Exception:
pass
try:
self.caller.hwdevice.api.sigTxabort.disconnect()
- except:
+ except Exception:
pass
try:
self.caller.hwdevice.api.tx_progress.disconnect()
- except:
+ except Exception:
pass
self.caller.hwdevice.api.sigTxdone.connect(gui.FinishSend)
self.caller.hwdevice.api.sigTxabort.connect(gui.AbortSend)
@@ -301,7 +297,7 @@ def SendRewards(self, inputs=None, gui=None):
num_of_inputs = len(self.selectedRewards)
else:
# bulk send
- num_of_inputs = sum([len(x['utxos']) for x in inputs])
+ num_of_inputs = sum(len(x['utxos']) for x in inputs)
ans = checkTxInputs(self.caller, num_of_inputs)
if ans is None or ans == QMessageBox.No:
# emit sigTxAbort and return
@@ -310,9 +306,9 @@ def SendRewards(self, inputs=None, gui=None):
# LET'S GO
if inputs is None:
- printDbg("Sending from PIVX address %s to PIVX address %s " % (self.curr_addr, self.dest_addr))
+ printDbg(f"Sending from PIVX address {self.curr_addr} to PIVX address {self.dest_addr} ")
else:
- printDbg("Sweeping rewards to PIVX address %s " % self.dest_addr)
+ printDbg(f"Sweeping rewards to PIVX address {self.dest_addr} ")
self.ui.rewardsList.statusLabel.hide()
printDbg("Preparing transaction. Please wait...")
self.ui.loadingLine.show()
@@ -374,10 +370,9 @@ def FinishSend(self, serialized_tx, amount_to_send):
decodedTx = ParseTx(tx_hex, self.caller.isTestnetRPC)
destination = decodedTx.get("vout")[0].get("scriptPubKey").get("addresses")[0]
amount = decodedTx.get("vout")[0].get("value")
- message = '<p>Broadcast signed transaction?</p><p>Destination address:<br><b>%s</b></p>' % destination
- message += '<p>Amount: <b>%s</b> PIV<br>' % str(round(amount / 1e8, 8))
- message += 'Fees: <b>%s</b> PIV <br>Size: <b>%d</b> Bytes</p>' % (
- str(round(self.currFee / 1e8, 8)), len(tx_hex) / 2)
+ message = f'<p>Broadcast signed transaction?</p><p>Destination address:<br><b>{destination}</b></p>'
+ message += f'<p>Amount: <b>{round(amount / 1e8, 8)}</b> PIV<br>'
+ message += f'Fees: <b>{round(self.currFee / 1e8, 8)}</b> PIV <br>Size: <b>{len(tx_hex) / 2}</b> Bytes</p>'
except Exception as e:
printException(getCallerName(), getFunctionName(), "decoding exception", str(e))
message = '<p>Unable to decode TX- Broadcast anyway?</p>'
@@ -392,7 +387,7 @@ def FinishSend(self, serialized_tx, amount_to_send):
txid = self.caller.rpcClient.sendRawTransaction(tx_hex)
if txid is None:
raise Exception("Unable to send TX - connection to RPC server lost.")
- printDbg("Transaction sent. ID: %s" % txid)
+ printDbg(f"Transaction sent. ID: {txid}")
mess2_text = "<p>Transaction successfully sent.</p>"
mess2 = QMessageBox(QMessageBox.Information, 'transaction Sent', mess2_text)
mess2.setDetailedText(txid)
@@ -431,15 +426,15 @@ def updateSelection(self, clicked_item=None):
self.selectedRewards = self.getSelection()
numOfInputs = len(self.selectedRewards)
if numOfInputs:
- for i in range(0, numOfInputs):
- total += int(self.selectedRewards[i].get('satoshis'))
+ for reward in self.selectedRewards:
+ total += int(reward.get('satoshis'))
# update suggested fee and selected rewards
estimatedTxSize = (44 + numOfInputs * 148) * 1.0 / 1000 # kB
feePerKb = self.caller.rpcClient.getFeePerKb()
self.suggestedFee = round(feePerKb * estimatedTxSize, 8)
- printDbg("estimatedTxSize is %s kB" % str(estimatedTxSize))
- printDbg("suggested fee is %s PIV (%s PIV/kB)" % (str(self.suggestedFee), str(feePerKb)))
+ printDbg(f"estimatedTxSize is {estimatedTxSize} kB")
+ printDbg(f"suggested fee is {self.suggestedFee} PIV ({feePerKb} PIV/kB)")
self.ui.selectedRewardsLine.setText(str(round(total / 1e8, 8)))
@@ -450,8 +445,6 @@ def updateSelection(self, clicked_item=None):
def update_loading_utxos(self, percent):
if percent < 100:
- self.ui.resetStatusLabel('<em><b style="color:purple">Checking explorer... %d%%</b></em>' % percent)
+ self.ui.resetStatusLabel(f'<em><b style="color:purple">Checking explorer... {percent}%</b></em>')
else:
self.display_utxos()
-
-
diff --git a/src/threads.py b/src/threads.py
index 8ba0225..d0c3113 100644
--- a/src/threads.py
+++ b/src/threads.py
@@ -8,28 +8,41 @@
Based on project:
https://github.com/Bertrand256/dash-masternode-tool
"""
+
import threading
import traceback
from functools import partial
from workerThread import WorkerThread
+from typing import Callable, Any, Optional
class ThreadFuns:
@staticmethod
- def runInThread(worker_fun, worker_fun_args, on_thread_finish=None, on_thread_exception=None,
- skip_raise_exception=False):
+ def runInThread(
+ worker_fun: Callable[..., Any],
+ worker_fun_args: tuple,
+ on_thread_finish: Optional[Callable[[], None]] = None,
+ on_thread_exception: Optional[Callable[[Exception], None]] = None,
+ skip_raise_exception: bool = False
+ ) -> threading.Thread:
"""
Run a function inside a thread.
:param worker_fun: reference to function to be executed inside a thread
:param worker_fun_args: arguments passed to a thread function
:param on_thread_finish: function to be called after thread finishes its execution
+ :param on_thread_exception: function to be called if an exception occurs in the thread
:param skip_raise_exception: Exception raised inside the 'worker_fun' will be passed to the calling thread if:
- on_thread_exception is a valid function (it's exception handler)
- skip_raise_exception is False
:return: reference to a thread object
"""
- def on_thread_finished_int(thread_arg, on_thread_finish_arg, skip_raise_exception_arg, on_thread_exception_arg):
+ def on_thread_finished_int(
+ thread_arg: WorkerThread,
+ on_thread_finish_arg: Optional[Callable[[], None]],
+ skip_raise_exception_arg: bool,
+ on_thread_exception_arg: Optional[Callable[[Exception], None]]
+ ):
if thread_arg.worker_exception:
if on_thread_exception_arg:
on_thread_exception_arg(thread_arg.worker_exception)
@@ -44,14 +57,13 @@ def on_thread_finished_int(thread_arg, on_thread_finish_arg, skip_raise_exceptio
# starting thread from another thread causes an issue of not passing arguments'
# values to on_thread_finished_int function, so on_thread_finish is not called
st = traceback.format_stack()
- print('Running thread from inside another thread. Stack: \n' + ''.join(st))
+ print(f'Running thread from inside another thread. Stack: \n{"".join(st)}')
thread = WorkerThread(worker_fun=worker_fun, worker_fun_args=worker_fun_args)
# in Python 3.5 local variables sometimes are removed before calling on_thread_finished_int
# so we have to bind that variables with the function ref
- bound_on_thread_finished = partial(on_thread_finished_int, thread, on_thread_finish, skip_raise_exception,
- on_thread_exception)
+ bound_on_thread_finished = partial(on_thread_finished_int, thread, on_thread_finish, skip_raise_exception, on_thread_exception)
thread.finished.connect(bound_on_thread_finished)
thread.daemon = True
diff --git a/src/trezorClient.py b/src/trezorClient.py
index cef6ee9..4eaa4e7 100644
--- a/src/trezorClient.py
+++ b/src/trezorClient.py
@@ -23,7 +23,7 @@
from threads import ThreadFuns
from txCache import TxCache
-from qt.dlg_pinMatrix import PinMatrix_dlg
+from qt.dlg_pinMatrix import PinMatrixDlg
def process_trezor_exceptions(func):
@@ -47,21 +47,15 @@ def process_trezor_exceptions_int(*args, **kwargs):
class TrezorApi(QObject):
- # signal: sig1 (thread) is done - emitted by signMessageFinish
sig1done = pyqtSignal(str)
- # signal: sigtx (thread) is done - emitted by signTxFinish
sigTxdone = pyqtSignal(bytearray, str)
- # signal: sigtx (thread) is done (aborted) - emitted by signTxFinish
sigTxabort = pyqtSignal()
- # signal: tx_progress percent - emitted by perepare_transfer_tx_bulk
tx_progress = pyqtSignal(int)
- # signal: sig_progress percent - emitted by signTxSign
sig_progress = pyqtSignal(int)
- # signal: sig_disconnected -emitted with DisconnectedException
sig_disconnected = pyqtSignal(str)
def __init__(self, model, main_wnd, *args, **kwargs):
- QObject.__init__(self, *args, **kwargs)
+ super().__init__(*args, **kwargs)
self.model = model # index of HW_devices
self.main_wnd = main_wnd
self.messages = [
@@ -71,7 +65,6 @@ def __init__(self, model, main_wnd, *args, **kwargs):
"Wrong device model detected.",
"Wrong PIN inserted"
]
- # Device Lock for threads
self.lock = threading.RLock()
self.status = 0
self.client = None
@@ -83,9 +76,7 @@ def append_inputs_to_TX(self, utxo, bip32_path, inputs):
if utxo['staker'] != "":
printException(getCallerName(), getFunctionName(), "Unable to sing P2CS on Trezor", "")
return
- # Update amount
self.amount += int(utxo['satoshis'])
- # Add input
address_n = parse_path(bip32_path)
prev_hash = binascii.unhexlify(utxo['txid'])
it = trezor_proto.TxInputType(
@@ -120,9 +111,7 @@ def initDevice(self):
self.status = 0
devices = enumerate_devices()
if not len(devices):
- # No device connected
return
- # Use the first device for now
d = devices[0]
ui = TrezorUi()
try:
@@ -138,13 +127,10 @@ def initDevice(self):
model = self.client.features.model or "1"
if not self.checkModel(model):
self.status = 3
- self.messages[3] = "Wrong device model (%s) detected.\nLooking for model %s." % (
- HW_devices[self.model][0], model
- )
+ self.messages[3] = f"Wrong device model ({model}) detected.\nLooking for model {HW_devices[self.model][0]}."
return
required_version = MINIMUM_FIRMWARE_VERSION[model]
- printDbg("Current version is %s (minimum required: %s)" % (str(self.client.version), str(required_version)))
- # Check device is unlocked
+ printDbg(f"Current version is {str(self.client.version)} (minimum required: {str(required_version)})")
bip32_path = parse_path(MPATH + "%d'/0/%d" % (0, 0))
_ = btc.get_address(self.client, 'PIVX', bip32_path, False)
self.status = 2
@@ -161,7 +147,6 @@ def load_prev_txes(self, rewardsArray):
json_tx = ParseTx(raw_tx)
txes[prev_hash] = self.json_to_tx(json_tx)
- # completion percent emitted
curr_utxo_checked += 1
completion = int(95 * curr_utxo_checked / num_of_txes)
self.tx_progress.emit(completion)
@@ -176,7 +161,6 @@ def json_to_tx(self, jtx):
t.bin_outputs = [self.json_to_bin_output(output) for output in jtx["vout"]]
return t
-
def json_to_input(self, input):
i = trezor_proto.TxInputType()
if "coinbase" in input:
@@ -190,7 +174,6 @@ def json_to_input(self, input):
i.sequence = input["sequence"]
return i
-
def json_to_bin_output(self, output):
o = trezor_proto.TxOutputBinType()
o.amount = int(output["value"])
@@ -204,13 +187,11 @@ def prepare_transfer_tx_bulk(self, caller, rewardsArray, dest_address, tx_fee, i
self.amount = 0
for mnode in rewardsArray:
- # Add proper HW path (for current device) on each utxo
if isTestnet:
mnode['path'] = MPATH_TESTNET + mnode['path']
else:
mnode['path'] = MPATH + mnode['path']
- # Create a TX input with each utxo
for utxo in mnode['utxos']:
self.append_inputs_to_TX(utxo, mnode['path'], inputs)
@@ -230,10 +211,9 @@ def prepare_transfer_tx_bulk(self, caller, rewardsArray, dest_address, tx_fee, i
self.mBox2 = QMessageBox(caller)
self.messageText = "<p>Signing transaction...</p>"
- # messageText += "From bip32_path: <b>%s</b><br><br>" % str(bip32_path)
- self.messageText += "<p>Payment to:<br><b>%s</b></p>" % dest_address
- self.messageText += "<p>Net amount:<br><b>%s</b> PIV</p>" % str(round(self.amount / 1e8, 8))
- self.messageText += "<p>Fees:<br><b>%s</b> PIV<p>" % str(round(int(tx_fee) / 1e8, 8))
+ self.messageText += f"<p>Payment to:<br><b>{dest_address}</b></p>"
+ self.messageText += f"<p>Net amount:<br><b>{str(round(self.amount / 1e8, 8))}</b> PIV</p>"
+ self.messageText += f"<p>Fees:<br><b>{str(round(int(tx_fee) / 1e8, 8))}</b> PIV<p>"
messageText = self.messageText + "Signature Progress: 0 %"
self.mBox2.setText(messageText)
self.setBoxIcon(self.mBox2, caller)
@@ -281,17 +261,14 @@ def signMess(self, caller, hwpath, message, isTestnet=False):
path = MPATH_TESTNET + hwpath
else:
path = MPATH + hwpath
- # Connection pop-up
self.mBox = QMessageBox(caller)
- messageText = "Check display of your hardware device\n\n- message:\n\n%s\n\n-path:\t%s\n" % (
- splitString(message, 32), path)
+ messageText = f"Check display of your hardware device\n\n- message:\n\n{splitString(message, 32)}\n\n-path:\t{path}\n"
self.mBox.setText(messageText)
self.setBoxIcon(self.mBox, caller)
self.mBox.setWindowTitle("CHECK YOUR TREZOR")
self.mBox.setStandardButtons(QMessageBox.NoButton)
self.mBox.show()
- # Sign message
ThreadFuns.runInThread(self.signMessageSign, (path, message, isTestnet), self.signMessageFinish)
@process_trezor_exceptions
@@ -331,29 +308,23 @@ def signTxSign(self, ctrl, inputs, outputs, txes, isTestnet=False):
def signTxFinish(self):
self.mBox2.accept()
if self.tx_raw is not None:
- # Signal to be catched by FinishSend on TabRewards / dlg_sewwpAll
self.sigTxdone.emit(self.tx_raw, str(round(self.amount / 1e8, 8)))
-
else:
printOK("Transaction refused by the user")
self.sigTxabort.emit()
def updateSigProgress(self, percent):
- # -1 simply adds a waiting message to the actual progress
if percent == -1:
t = self.mBox2.text()
messageText = t + "<br>Please confirm action on your Trezor device..."
else:
- messageText = self.messageText + "Signature Progress: <b style='color:red'>" + str(percent) + " %</b>"
+ messageText = self.messageText + f"Signature Progress: <b style='color:red'>{percent} %</b>"
self.mBox2.setText(messageText)
QApplication.processEvents()
-# From trezorlib.btc
def sign_tx(sig_percent, client, coin_name, inputs, outputs, details=None, prev_txes=None):
- # set up a transactions dict
txes = {None: trezor_proto.TransactionType(inputs=inputs, outputs=outputs)}
- # preload all relevant transactions ahead of time
for inp in inputs:
if inp.script_type not in (
trezor_proto.InputScriptType.SPENDP2SHWITNESS,
@@ -379,13 +350,11 @@ def sign_tx(sig_percent, client, coin_name, inputs, outputs, details=None, prev_
res = client.call(signtx)
- # Prepare structure for signatures
signatures = [None] * len(inputs)
serialized_tx = b""
def copy_tx_meta(tx):
tx_copy = trezor_proto.TransactionType(**tx)
- # clear fields
tx_copy.inputs_cnt = len(tx.inputs)
tx_copy.inputs = []
tx_copy.outputs_cnt = len(tx.bin_outputs or tx.outputs)
@@ -397,10 +366,9 @@ def copy_tx_meta(tx):
R = trezor_proto.RequestType
- percent = 0 # Used for signaling progress. 1-10 for inputs/outputs, 10-100 for sigs.
+ percent = 0
sig_percent.emit(percent)
while isinstance(res, trezor_proto.TxRequest):
- # If there's some part of signed transaction, let's add it
if res.serialized:
if res.serialized.serialized_tx:
serialized_tx += res.serialized.serialized_tx
@@ -409,16 +377,14 @@ def copy_tx_meta(tx):
idx = res.serialized.signature_index
sig = res.serialized.signature
if signatures[idx] is not None:
- raise ValueError("Signature for index %d already filled" % idx)
+ raise ValueError(f"Signature for index {idx} already filled")
signatures[idx] = sig
- # emit completion percent
percent = 10 + int(90 * (idx + 1) / len(signatures))
sig_percent.emit(percent)
if res.request_type == R.TXFINISHED:
break
- # Device asked for one more information, let's process it.
current_tx = txes[res.details.tx_hash]
if res.request_type == R.TXMETA:
@@ -434,7 +400,6 @@ def copy_tx_meta(tx):
res = client.call(trezor_proto.TxAck(tx=msg))
elif res.request_type == R.TXOUTPUT:
- # Update just one percent then display additional waiting message (emitting -1)
if percent == 9:
percent += 1
sig_percent.emit(percent)
@@ -466,10 +431,9 @@ def copy_tx_meta(tx):
return signatures, serialized_tx
-class TrezorUi(object):
+class TrezorUi:
def __init__(self):
self.prompt_shown = False
- pass
def get_pin(self, code=None) -> str:
if code == PIN_CURRENT:
@@ -481,7 +445,7 @@ def get_pin(self, code=None) -> str:
else:
desc = "PIN"
- pin = ask_for_pin_callback("Please enter {}".format(desc))
+ pin = ask_for_pin_callback(f"Please enter {desc}")
if pin is None:
raise exceptions.Cancelled
return pin
@@ -500,7 +464,7 @@ def button_request(self, msg_code):
def ask_for_pin_callback(msg, hide_numbers=True):
- dlg = PinMatrix_dlg(title=msg, fHideBtns=hide_numbers)
+ dlg = PinMatrixDlg(title=msg, fHideBtns=hide_numbers)
if dlg.exec_():
return dlg.getPin()
else:
diff --git a/src/utils.py b/src/utils.py
index 911ebb7..d7c16e0 100644
--- a/src/utils.py
+++ b/src/utils.py
@@ -35,13 +35,13 @@ def b64encode(text):
def checkPivxAddr(address, isTestnet=False):
try:
- # check leading char 'D' or (for testnet) 'x' or 'y'
+ # Check leading char 'D' or (for testnet) 'x' or 'y'
if isTestnet and address[0] not in P2PKH_PREFIXES_TNET + P2SH_PREFIXES_TNET:
return False
if not isTestnet and address[0] not in P2PKH_PREFIXES + P2SH_PREFIXES:
return False
- # decode and verify checksum
+ # Decode and verify checksum
addr_bin = bytes.fromhex(b58decode(address).hex())
addr_bin_check = bin_dbl_sha256(addr_bin[0:-4])[0:4]
if addr_bin[-4:] != addr_bin_check:
@@ -58,13 +58,13 @@ def compose_tx_locking_script(dest_address, isTestnet=False):
:param dest_address: destination address in Base58Check format
:return: sequence of opcodes and its arguments, defining logic of the locking script
"""
- pubkey_hash = bytearray.fromhex(b58check_to_hex(dest_address)) # convert address to a public key hash
+ pubkey_hash = bytearray.fromhex(b58check_to_hex(dest_address)) # Convert address to a public key hash
if len(pubkey_hash) != 20:
raise Exception('Invalid length of the public key hash: ' + str(len(pubkey_hash)))
if (((not isTestnet) and (dest_address[0] in P2PKH_PREFIXES))
or (isTestnet and (dest_address[0] in P2PKH_PREFIXES_TNET))):
- # sequence of opcodes/arguments for p2pkh (pay-to-public-key-hash)
+ # Sequence of opcodes/arguments for p2pkh (pay-to-public-key-hash)
scr = OP_DUP + \
OP_HASH160 + \
int.to_bytes(len(pubkey_hash), 1, byteorder='little') + \
@@ -73,7 +73,7 @@ def compose_tx_locking_script(dest_address, isTestnet=False):
OP_CHECKSIG
elif (((not isTestnet) and (dest_address[0] in P2SH_PREFIXES))
or (isTestnet and (dest_address[0] in P2SH_PREFIXES_TNET))):
- # sequence of opcodes/arguments for p2sh (pay-to-script-hash)
+ # Sequence of opcodes/arguments for p2sh (pay-to-script-hash)
scr = OP_HASH160 + \
int.to_bytes(len(pubkey_hash), 1, byteorder='little') + \
pubkey_hash + \
@@ -182,7 +182,7 @@ def ipmap(ip, port):
ipv6map += a
else:
- raise Exception("invalid version number (%d)" % ipAddr.version)
+ raise Exception("Invalid version number (%d)" % ipAddr.version)
ipv6map += int(port).to_bytes(2, byteorder='big').hex()
if len(ipv6map) != 36:
@@ -210,16 +210,16 @@ def num_to_varint(a):
def read_varint(buffer, offset):
- if (buffer[offset] < 0xfd):
+ if buffer[offset] < 0xfd:
value_size = 1
value = buffer[offset]
- elif (buffer[offset] == 0xfd):
+ elif buffer[offset] == 0xfd:
value_size = 3
value = int.from_bytes(buffer[offset + 1: offset + 3], byteorder='little')
- elif (buffer[offset] == 0xfe):
+ elif buffer[offset] == 0xfe:
value_size = 5
value = int.from_bytes(buffer[offset + 1: offset + 5], byteorder='little')
- elif (buffer[offset] == 0xff):
+ elif buffer[offset] == 0xff:
value_size = 9
value = int.from_bytes(buffer[offset + 1: offset + 9], byteorder='little')
else:
@@ -236,7 +236,6 @@ def serialize_input_str(tx, prevout_n, sequence, script_sig):
s.append(', ')
if tx == '00' * 32 and prevout_n == 0xffffffff:
s.append('coinbase %s' % script_sig)
-
else:
script_sig2 = script_sig
if len(script_sig2) > 24:
diff --git a/src/watchdogThreads.py b/src/watchdogThreads.py
index 574c866..9bc016e 100644
--- a/src/watchdogThreads.py
+++ b/src/watchdogThreads.py
@@ -12,13 +12,13 @@
from misc import printOK
-class CtrlObject(object):
+class CtrlObject:
pass
class RpcWatchdog(QObject):
def __init__(self, control_tab, timer_off=10, timer_on=120, *args, **kwargs):
- QObject.__init__(self, *args, **kwargs)
+ super().__init__(*args, **kwargs)
self.firstLoop = True
self.shutdown_flag = Event()
self.control_tab = control_tab
diff --git a/src/workerThread.py b/src/workerThread.py
index 6b07dd2..0f8ba50 100644
--- a/src/workerThread.py
+++ b/src/workerThread.py
@@ -13,7 +13,7 @@
from misc import printError
-class CtrlObject(object):
+class CtrlObject:
pass
@@ -23,7 +23,7 @@ class WorkerThread(QThread):
"""
def __init__(self, worker_fun, worker_fun_args):
- QThread.__init__(self)
+ super().__init__()
self.worker_fun = worker_fun
self.worker_fun_args = worker_fun_args
# prepare control object passed to external thread function
@@ -44,4 +44,5 @@ def run(self):
self.worker_result = self.worker_fun(self.ctrl_obj, *self.worker_fun_args)
except Exception as e:
printError("worker thread", "run", str(e))
+ self.worker_exception = e
self.stop()