21UP/payments.py at main · Liongrass/21UP · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
# Modules
import asyncio
import json
import logging
import requests
from time import sleep
import websockets

# Functions and variables
from barometer import get_barometrics
from dispense import trigger
from display import shutdown
from screens import make_errorscreen, make_idlescreen, make_qrcode, make_success_overlay, make_failure_overlay
from var import amount, display_expiry, suceess_screen_expiry, expiry, label, lnbits_server, memo_str, pin_out, show_display, unit, x_api_key


####### VARIABLES ########

url_base = "https://" + lnbits_server + "/api/v1/payments"
ws_base = "wss://" + lnbits_server + "/api/v1/ws/"

##################################

def params(tray):
        params = {"out": False,
                  "amount": amount[tray],
                  "unit": unit[tray],
                  "memo": memo_str.format(label=label[tray],amount=amount[tray],unit=unit[tray]),
                  "expiry": expiry + display_expiry}
        logging.debug(f"Invoice parameters: {params}")
        return params

headers = {"X-Api-Key" : x_api_key,
           "Content-type" : "application/json"}

# This function generates an invoice through the LNbits API and returns only the Bolt11 invoice
def get_invoice(params, headers, tray):
    try:
        invoice_request = requests.post(url_base, json=params(tray), headers=headers)
        invoice_request.raise_for_status()
        global invoice
        invoice = invoice_request.json()
        logging.debug(f"{invoice}")
        logging.info(invoice["bolt11"])
        t = get_barometrics()
        make_qrcode(tray, t, invoice)
    except Exception as e:
        logging.debug(f"ERROR {e}")
        return

# This function connects to the LNbits websockets, checks for incoming payments and verifies whether they belong to the most recent invoice
async def listen_for_payment(ws_base, x_api_key, invoice, tray):
    async with websockets.connect(ws_base + x_api_key) as websocket:
        logging.debug(f"Connected to {ws_base}")
        logging.info(f"Waiting for payment: {invoice['amount']/1000} sat")
        while True:
            try:
                response_str = await websocket.recv()
                response = json.loads(response_str)
                if response["payment"]["payment_hash"] == invoice["payment_hash"]:
                    logging.info(f"Payment received. Dispensing {label[tray]} (tray {tray}). Payment hash: " + response['payment']['payment_hash'])
                    make_success_overlay()
                    trigger(pin_out, tray)
                    sleep(suceess_screen_expiry)
                    break
                else:
                    logging.debug(f"Ignoring incoming payment for {response['payment']['amount']/1000} sat. Payment hash does not belong to invoice")
            except websockets.exceptions.ConnectionClosed as e:
                logging.debug(f"Connection closed: {e}")
                break
            except json.JSONDecodeError as e:
                logging.debug(f"Failed to decode JSON: {e}")
                continue
            except websockets.exceptions.InvalidStatus as e:
                logging.debug(f"Failed to make connection: {e}")
                make_errorscreen()
                sleep(display_expiry)

# This is the main function. It will first get the invoice with get_invoice(), then evoke listen_for_payment(). If within sixty seconds the invoice is not paid, it will shut down.
async def payment(tray):
    logging.debug(f"Getting invoice for tray {tray} ({label[tray]})")
    get_invoice(params, headers, tray)
    try:
        invoice
        try:
            timeout = expiry + suceess_screen_expiry + display_expiry + 3
            await asyncio.wait_for(listen_for_payment(ws_base, x_api_key, invoice, tray), timeout=timeout)
        except asyncio.CancelledError:
            shutdown()
        except asyncio.TimeoutError:
            logging.info(f"Invoice expired after {expiry}s")
            logging.debug(f"Timeout reached after {timeout}s")
            make_failure_overlay()
            sleep(display_expiry)
        finally:
            logging.info("Cycle complete")
    except NameError:
        logging.error("Error obtaining invoice. Check logs for details.")
        make_errorscreen()
        sleep(display_expiry)
        return