Michilis/lnbits
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Shared mempool.space cache

Browse Source
This commit is contained in:
ben
2022-11-24 19:27:12 +00:00
parent b4c970a005
commit e6a3c63dd5
5 changed files with 663 additions and 404 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lnbits/extensions/gerty/helpers.py
+386 -26
View File
@@ -4,6 +4,16 @@ from datetime import datetime, timedelta
import httpx
from loguru import logger
from .crud import (
get_fees_recommended,
get_hashrate_1w,
get_hashrate_1m,
get_statistics,
get_difficulty_adjustment,
get_tip_height,
get_mempool
)
from .number_prefixer import *
@@ -65,19 +75,12 @@ def format_number(number, precision=None):
return "{:,}".format(round(number, precision))
async def get_mempool_recommended_fees(gerty):
if isinstance(gerty.mempool_endpoint, str):
async with httpx.AsyncClient() as client:
r = await client.get(gerty.mempool_endpoint + "/api/v1/fees/recommended")
return r.json()
async def get_mining_dashboard(gerty):
areas = []
if isinstance(gerty.mempool_endpoint, str):
async with httpx.AsyncClient() as client:
# current hashrate
r = await client.get(gerty.mempool_endpoint + "/api/v1/mining/hashrate/1w")
r = await get_hashrate_1w(gerty)
data = r.json()
hashrateNow = data["currentHashrate"]
hashrateOneWeekAgo = data["hashrates"][6]["avgHashrate"]
@@ -99,9 +102,7 @@ async def get_mining_dashboard(gerty):
)
areas.append(text)
r = await client.get(
gerty.mempool_endpoint + "/api/v1/difficulty-adjustment"
)
r = await get_difficulty_adjustment(gerty)
# timeAvg
text = []
@@ -131,7 +132,7 @@ async def get_mining_dashboard(gerty):
)
areas.append(text)
r = await client.get(gerty.mempool_endpoint + "/api/v1/mining/hashrate/1m")
r = await get_hashrate_1m(gerty)
data = r.json()
stat = {}
stat["current"] = data["currentDifficulty"]
@@ -141,19 +142,9 @@ async def get_mining_dashboard(gerty):
return areas
async def api_get_lightning_stats(gerty):
stat = {}
if isinstance(gerty.mempool_endpoint, str):
async with httpx.AsyncClient() as client:
r = await client.get(
gerty.mempool_endpoint + "/api/v1/lightning/statistics/latest"
)
data = r.json()
return data
async def get_lightning_stats(gerty):
data = await api_get_lightning_stats(gerty)
data = await get_statistics(gerty)
areas = []
text = []
@@ -280,16 +271,385 @@ async def api_get_mining_stat(stat_slug: str, gerty):
stat = ""
if stat_slug == "mining_current_hash_rate":
async with httpx.AsyncClient() as client:
r = await client.get(gerty.mempool_endpoint + "/api/v1/mining/hashrate/1m")
r = await get_hashrate_1m(gerty)
data = r.json()
stat = {}
stat['current'] = data['currentHashrate']
stat['1w'] = data['hashrates'][len(data['hashrates']) - 7]['avgHashrate']
elif stat_slug == "mining_current_difficulty":
async with httpx.AsyncClient() as client:
r = await client.get(gerty.mempool_endpoint + "/api/v1/mining/hashrate/1m")
r = await get_hashrate_1m(gerty)
data = r.json()
stat = {}
stat['current'] = data['currentDifficulty']
stat['previous'] = data['difficulty'][len(data['difficulty']) - 2]['difficulty']
return stat
return stat
###########################################
# Get a screen slug by its position in the screens_list
def get_screen_slug_by_index(index: int, screens_list):
logger.debug("Index: {0}".format(index))
logger.debug("len(screens_list) - 1: {0} ".format(len(screens_list) - 1))
if index <= len(screens_list) - 1:
return list(screens_list)[index - 1]
else:
return None
# Get a list of text items for the screen number
async def get_screen_data(screen_num: int, screens_list: dict, gerty):
screen_slug = get_screen_slug_by_index(screen_num, screens_list)
# first get the relevant slug from the display_preferences
logger.debug("screen_slug")
logger.debug(screen_slug)
areas = []
title = ""
if screen_slug == "dashboard":
title = gerty.name
areas = await get_dashboard(gerty)
if screen_slug == "lnbits_wallets_balance":
wallets = await get_lnbits_wallet_balances(gerty)
text = []
for wallet in wallets:
text.append(get_text_item_dict(text="{0}'s Wallet".format(wallet['name']), font_size=20,gerty_type=gerty.type))
text.append(get_text_item_dict(text="{0} sats".format(format_number(wallet['balance'])), font_size=40,gerty_type=gerty.type))
areas.append(text)
elif screen_slug == "fun_satoshi_quotes":
areas.append(await get_satoshi_quotes(gerty))
elif screen_slug == "fun_exchange_market_rate":
areas.append(await get_exchange_rate(gerty))
elif screen_slug == "onchain_difficulty_epoch_progress":
areas.append(await get_onchain_stat(screen_slug, gerty))
elif screen_slug == "onchain_block_height":
logger.debug("iam block height")
text = []
text.append(get_text_item_dict(text=format_number(await get_tip_height(gerty)), font_size=80, gerty_type=gerty.type))
areas.append(text)
elif screen_slug == "onchain_difficulty_retarget_date":
areas.append(await get_onchain_stat(screen_slug, gerty))
elif screen_slug == "onchain_difficulty_blocks_remaining":
areas.append(await get_onchain_stat(screen_slug, gerty))
elif screen_slug == "onchain_difficulty_epoch_time_remaining":
areas.append(await get_onchain_stat(screen_slug, gerty))
elif screen_slug == "dashboard_onchain":
title = "Onchain Data"
areas = await get_onchain_dashboard(gerty)
elif screen_slug == "mempool_recommended_fees":
areas.append(await get_mempool_stat(screen_slug, gerty))
elif screen_slug == "mempool_tx_count":
areas.append(await get_mempool_stat(screen_slug, gerty))
elif screen_slug == "mining_current_hash_rate":
areas.append(await get_mining_stat(screen_slug, gerty))
elif screen_slug == "mining_current_difficulty":
areas.append(await get_mining_stat(screen_slug, gerty))
elif screen_slug == "dashboard_mining":
title = "Mining Data"
areas = await get_mining_dashboard(gerty)
elif screen_slug == "lightning_dashboard":
title = "Lightning Network"
areas = await get_lightning_stats(gerty)
data = {}
data["title"] = title
data["areas"] = areas
return data
# Get the dashboard screen
async def get_dashboard(gerty):
areas = []
# XC rate
text = []
amount = await satoshis_amount_as_fiat(100000000, gerty.exchange)
text.append(get_text_item_dict(text=format_number(amount), font_size=40,gerty_type=gerty.type))
text.append(get_text_item_dict(text="BTC{0} price".format(gerty.exchange), font_size=15,gerty_type=gerty.type))
areas.append(text)
# balance
text = []
wallets = await get_lnbits_wallet_balances(gerty)
text = []
for wallet in wallets:
text.append(get_text_item_dict(text="{0}".format(wallet["name"]), font_size=15,gerty_type=gerty.type))
text.append(
get_text_item_dict(text="{0} sats".format(format_number(wallet["balance"])), font_size=20,gerty_type=gerty.type)
)
areas.append(text)
# Mempool fees
text = []
text.append(get_text_item_dict(text=format_number(await get_tip_height(gerty)), font_size=40,gerty_type=gerty.type))
text.append(get_text_item_dict(text="Current block height", font_size=15,gerty_type=gerty.type))
areas.append(text)
# difficulty adjustment time
text = []
text.append(
get_text_item_dict(
text=await get_time_remaining_next_difficulty_adjustment(gerty), font_size=15,gerty_type=gerty.type
)
)
text.append(get_text_item_dict(text="until next difficulty adjustment", font_size=12,gerty_type=gerty.type))
areas.append(text)
return areas
async def get_lnbits_wallet_balances(gerty):
# Get Wallet info
wallets = []
if gerty.lnbits_wallets != "":
for lnbits_wallet in json.loads(gerty.lnbits_wallets):
wallet = await get_wallet_for_key(key=lnbits_wallet)
logger.debug(wallet.name)
if wallet:
wallets.append(
{
"name": wallet.name,
"balance": wallet.balance_msat / 1000,
"inkey": wallet.inkey,
}
)
return wallets
async def get_placeholder_text():
return [
get_text_item_dict(text="Some placeholder text", x_pos=15, y_pos=10, font_size=50,gerty_type=gerty.type),
get_text_item_dict(text="Some placeholder text", x_pos=15, y_pos=10, font_size=50,gerty_type=gerty.type),
]
async def get_satoshi_quotes(gerty):
# Get Satoshi quotes
text = []
quote = await api_gerty_satoshi()
if quote:
if quote["text"]:
text.append(get_text_item_dict(text=quote["text"], font_size=15,gerty_type=gerty.type))
if quote["date"]:
text.append(
get_text_item_dict(text="Satoshi Nakamoto - {0}".format(quote["date"]), font_size=15,gerty_type=gerty.type)
)
return text
# Get Exchange Value
async def get_exchange_rate(gerty):
text = []
if gerty.exchange != "":
try:
amount = await satoshis_amount_as_fiat(100000000, gerty.exchange)
if amount:
price = format_number(amount)
text.append(
get_text_item_dict(
text="Current {0}/BTC price".format(gerty.exchange), font_size=15,gerty_type=gerty.type
)
)
text.append(get_text_item_dict(text=price, font_size=80,gerty_type=gerty.type))
except:
pass
return text
async def get_onchain_stat(stat_slug: str, gerty):
text = []
if (
stat_slug == "onchain_difficulty_epoch_progress" or
stat_slug == "onchain_difficulty_retarget_date" or
stat_slug == "onchain_difficulty_blocks_remaining" or
stat_slug == "onchain_difficulty_epoch_time_remaining"
):
async with httpx.AsyncClient() as client:
r = await get_difficulty_adjustment(gerty)
if stat_slug == "onchain_difficulty_epoch_progress":
stat = round(r.json()['progressPercent'])
text.append(get_text_item_dict(text="Progress through current difficulty epoch", font_size=15,gerty_type=gerty.type))
text.append(get_text_item_dict(text="{0}%".format(stat), font_size=80,gerty_type=gerty.type))
elif stat_slug == "onchain_difficulty_retarget_date":
stat = r.json()['estimatedRetargetDate']
dt = datetime.fromtimestamp(stat / 1000).strftime("%e %b %Y at %H:%M")
text.append(get_text_item_dict(text="Date of next difficulty adjustment", font_size=15,gerty_type=gerty.type))
text.append(get_text_item_dict(text=dt, font_size=40,gerty_type=gerty.type))
elif stat_slug == "onchain_difficulty_blocks_remaining":
stat = r.json()['remainingBlocks']
text.append(get_text_item_dict(text="Blocks until next difficulty adjustment", font_size=15,gerty_type=gerty.type))
text.append(get_text_item_dict(text="{0}".format(format_number(stat)), font_size=80,gerty_type=gerty.type))
elif stat_slug == "onchain_difficulty_epoch_time_remaining":
stat = r.json()['remainingTime']
text.append(get_text_item_dict(text="Time until next difficulty adjustment", font_size=15,gerty_type=gerty.type))
text.append(get_text_item_dict(text=get_time_remaining(stat / 1000, 4), font_size=20,gerty_type=gerty.type))
return text
async def get_onchain_dashboard(gerty):
areas = []
if isinstance(gerty.mempool_endpoint, str):
async with httpx.AsyncClient() as client:
r = await get_difficulty_adjustment(gerty)
text = []
stat = round(r.json()["progressPercent"])
text.append(get_text_item_dict(text="Progress through epoch", font_size=12,gerty_type=gerty.type))
text.append(get_text_item_dict(text="{0}%".format(stat), font_size=60,gerty_type=gerty.type))
areas.append(text)
text = []
stat = r.json()["estimatedRetargetDate"]
dt = datetime.fromtimestamp(stat / 1000).strftime("%e %b %Y at %H:%M")
text.append(get_text_item_dict(text="Date of next adjustment", font_size=12,gerty_type=gerty.type))
text.append(get_text_item_dict(text=dt, font_size=20,gerty_type=gerty.type))
areas.append(text)
text = []
stat = r.json()["remainingBlocks"]
text.append(get_text_item_dict(text="Blocks until adjustment", font_size=12,gerty_type=gerty.type))
text.append(get_text_item_dict(text="{0}".format(format_number(stat)), font_size=60,gerty_type=gerty.type))
areas.append(text)
text = []
stat = r.json()["remainingTime"]
text.append(get_text_item_dict(text="Time until adjustment", font_size=12,gerty_type=gerty.type))
text.append(get_text_item_dict(text=get_time_remaining(stat / 1000, 4), font_size=20,gerty_type=gerty.type))
areas.append(text)
return areas
async def get_time_remaining_next_difficulty_adjustment(gerty):
if isinstance(gerty.mempool_endpoint, str):
async with httpx.AsyncClient() as client:
r = await get_difficulty_adjustment(gerty)
stat = r.json()["remainingTime"]
time = get_time_remaining(stat / 1000, 3)
return time
async def get_mempool_stat(stat_slug: str, gerty):
text = []
if isinstance(gerty.mempool_endpoint, str):
async with httpx.AsyncClient() as client:
if stat_slug == "mempool_tx_count":
r = get_mempool(gerty)
if stat_slug == "mempool_tx_count":
stat = round(r.json()["count"])
text.append(get_text_item_dict(text="Transactions in the mempool", font_size=15,gerty_type=gerty.type))
text.append(
get_text_item_dict(text="{0}".format(format_number(stat)), font_size=80,gerty_type=gerty.type)
)
elif stat_slug == "mempool_recommended_fees":
y_offset = 60
fees = await get_fees_recommended()
pos_y = 80 + y_offset
text.append(get_text_item_dict("mempool.space", 40, 160, pos_y, gerty.type))
pos_y = 180 + y_offset
text.append(get_text_item_dict("Recommended Tx Fees", 20, 240, pos_y, gerty.type))
pos_y = 280 + y_offset
text.append(
get_text_item_dict("{0}".format("None"), 15, 30, pos_y, gerty.type)
)
text.append(
get_text_item_dict("{0}".format("Low"), 15, 235, pos_y, gerty.type)
)
text.append(
get_text_item_dict("{0}".format("Medium"), 15, 460, pos_y, gerty.type)
)
text.append(
get_text_item_dict("{0}".format("High"), 15, 750, pos_y, gerty.type)
)
pos_y = 340 + y_offset
font_size = 15
fee_append = "/vB"
fee_rate = fees["economyFee"]
text.append(
get_text_item_dict(
text="{0} {1}{2}".format(
format_number(fee_rate),
("sat" if fee_rate == 1 else "sats"),
fee_append,
),
font_size=font_size,
x_pos=30,
y_pos=pos_y,
gerty_type=gerty.type
)
)
fee_rate = fees["hourFee"]
text.append(
get_text_item_dict(
text="{0} {1}{2}".format(
format_number(fee_rate),
("sat" if fee_rate == 1 else "sats"),
fee_append,
),
font_size=font_size,
x_pos=235,
y_pos=pos_y,
gerty_type=gerty.type
)
)
fee_rate = fees["halfHourFee"]
text.append(
get_text_item_dict(
text="{0} {1}{2}".format(
format_number(fee_rate),
("sat" if fee_rate == 1 else "sats"),
fee_append,
),
font_size=font_size,
x_pos=460,
y_pos=pos_y,
gerty_type=gerty.type
)
)
fee_rate = fees["fastestFee"]
text.append(
get_text_item_dict(
text="{0} {1}{2}".format(
format_number(fee_rate),
("sat" if fee_rate == 1 else "sats"),
fee_append,
),
font_size=font_size,
x_pos=750,
y_pos=pos_y,
gerty_type=gerty.type
)
)
return text
def get_date_suffix(dayNumber):
if 4 <= dayNumber <= 20 or 24 <= dayNumber <= 30:
return "th"
else:
return ["st", "nd", "rd"][dayNumber % 10 - 1]
def get_time_remaining(seconds, granularity=2):
intervals = (
# ('weeks', 604800), # 60 * 60 * 24 * 7
('days', 86400), # 60 * 60 * 24
('hours', 3600), # 60 * 60
('minutes', 60),
('seconds', 1),
)
result = []
for name, count in intervals:
value = seconds // count
if value:
seconds -= value * count
if value == 1:
name = name.rstrip('s')
result.append("{} {}".format(round(value), name))
return ', '.join(result[:granularity])