clightning.py (3333B)
1 import time 2 import qrcode 3 import logging 4 5 import config 6 7 # if False: # config.tor_clightningrpc_host is not None: 8 # from gateways.tor import session 9 # else: 10 # import requests 11 # 12 # session = None 13 14 15 class clightning: 16 def __init__(self, node_config): 17 from pyln.client import LightningRpc 18 19 self.config = node_config 20 self.is_onchain = False 21 22 for i in range(config.connection_attempts): 23 try: 24 if config.tunnel_host is None: 25 rpc_file = self.config['clightning_rpc_file'] 26 else: 27 rpc_file = "lightning-rpc" 28 29 logging.info( 30 "Attempting to connect to clightning with unix domain socket: {}".format( 31 rpc_file 32 ) 33 ) 34 self.clightning = LightningRpc(rpc_file) 35 36 logging.info("Getting clightning info...") 37 info = self.clightning.getinfo() 38 logging.info(info) 39 40 logging.info("Successfully connected to clightning.") 41 break 42 43 except Exception as e: 44 logging.error(e) 45 if i < 5: 46 time.sleep(2) 47 else: 48 time.sleep(60) 49 logging.info( 50 "Attempting again... {}/{}...".format( 51 i + 1, config.connection_attempts 52 ) 53 ) 54 else: 55 raise Exception( 56 "Could not connect to clightning. Check your port tunneling settings and try again." 57 ) 58 59 logging.info("Ready for payments requests.") 60 return 61 62 def create_qr(self, uuid, address, value): 63 qr_str = "{}".format(address.upper()) 64 img = qrcode.make(qr_str) 65 img.save("static/qr_codes/{}.png".format(uuid)) 66 return 67 68 def get_info(self): 69 return self.clightning.getinfo() 70 71 def get_uri(self): 72 info = self.get_info() 73 address = info["address"][0] 74 return info["id"] + "@" + address["address"] + ":" + str(address["port"]) 75 76 # Create lightning invoice 77 def create_clightning_invoice(self, btc_amount, label, expiry): 78 # Multiplying by 10^8 to convert to satoshi units 79 msats_amount = int(float(btc_amount) * 10 ** (3 + 8)) 80 lnd_invoice = self.clightning.invoice( 81 msats_amount, label, label, expiry 82 ) 83 return lnd_invoice["bolt11"], lnd_invoice["payment_hash"] 84 85 def get_address(self, amount, label, expiry): 86 address, r_hash = self.create_clightning_invoice(amount, label, expiry) 87 return address, r_hash 88 89 # Check whether the payment has been paid 90 def check_payment(self, uuid): 91 invoices = self.clightning.listinvoices(uuid)["invoices"] 92 93 if len(invoices) == 0: 94 logging.error("Could not find invoice on node. Something's wrong.") 95 return 0, 0 96 97 invoice = invoices[0] 98 99 if invoice["status"] != "paid": 100 conf_paid = 0 101 unconf_paid = 0 102 else: 103 # Store amount paid and convert to BTC units 104 conf_paid = int(invoice["msatoshi_received"]) / 10 ** (3 + 8) 105 unconf_paid = 0 106 107 return conf_paid, unconf_paid