mirrors/lnbits-legend
1
0
Fork 0
mirror of https://github.com/lnbits/lnbits-legend.git synced 2025-01-19 05:33:47 +01:00
Code Issues Packages Projects Releases Wiki Activity

gerty mypy fixes

Browse Source
This commit is contained in:
dni ⚡ 2022-12-30 09:24:31 +01:00
parent efc156689f
commit 8e344ecf4b
7 changed files with 49 additions and 94 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
lnbits/extensions/gerty/__init__.py
View File

@ -5,11 +5,9 @@ from fastapi.staticfiles import StaticFiles
from lnbits.db import Database
from lnbits.helpers import template_renderer
from lnbits.tasks import catch_everything_and_restart
db = Database("ext_gerty")
gerty_static_files = [
{
"path": "/gerty/static",

7
lnbits/extensions/gerty/crud.py
View File

@ -55,7 +55,10 @@ async def update_gerty(gerty_id: str, **kwargs) -> Gerty:
await db.execute(
f"UPDATE gerty.gertys SET {q} WHERE id = ?", (*kwargs.values(), gerty_id)
)
return await get_gerty(gerty_id)
gerty = await get_gerty(gerty_id)
assert gerty
return gerty
async def get_gerty(gerty_id: str) -> Optional[Gerty]:
@ -82,7 +85,7 @@ async def delete_gerty(gerty_id: str) -> None:
#############MEMPOOL###########
async def get_mempool_info(endPoint: str, gerty) -> Optional[Mempool]:
async def get_mempool_info(endPoint: str, gerty) -> dict:
logger.debug(endPoint)
endpoints = MempoolEndpoint()
url = ""

93
lnbits/extensions/gerty/helpers.py
View File

@ -3,15 +3,16 @@ import os
import random
import textwrap
from datetime import datetime, timedelta
from typing import List
import httpx
from loguru import logger
from lnbits.core.crud import get_user, get_wallet_for_key
from lnbits.core.crud import get_wallet_for_key
from lnbits.settings import settings
from lnbits.utils.exchange_rates import satoshis_amount_as_fiat
from .crud import get_gerty, get_mempool_info
from .crud import get_mempool_info
from .number_prefixer import *
@ -24,8 +25,8 @@ def get_percent_difference(current, previous, precision=3):
def get_text_item_dict(
text: str,
font_size: int,
x_pos: int = None,
y_pos: int = None,
x_pos: int = -1,
y_pos: int = -1,
gerty_type: str = "Gerty",
):
# Get line size by font size
@ -63,12 +64,12 @@ def get_text_item_dict(
# logger.debug('multilineText')
# logger.debug(multilineText)
text = {"value": multilineText, "size": font_size}
if x_pos is None and y_pos is None:
text["position"] = "center"
data_text = {"value": multilineText, "size": font_size}
if x_pos == -1 and y_pos == -1:
data_text["position"] = "center"
else:
text["x"] = x_pos
text["y"] = y_pos
data_text["x"] = x_pos if x_pos > 0 else 0
data_text["y"] = y_pos if x_pos > 0 else 0
return text
@ -293,8 +294,7 @@ def get_next_update_time(sleep_time_seconds: int = 0, utc_offset: int = 0):
def gerty_should_sleep(utc_offset: int = 0):
utc_now = datetime.utcnow()
local_time = utc_now + timedelta(hours=utc_offset)
hours = local_time.strftime("%H")
hours = int(hours)
hours = int(local_time.strftime("%H"))
if hours >= 22 and hours <= 23:
return True
else:
@ -380,23 +380,19 @@ async def get_mining_stat(stat_slug: str, gerty):
async def api_get_mining_stat(stat_slug: str, gerty):
stat = ""
stat = {}
if stat_slug == "mining_current_hash_rate":
async with httpx.AsyncClient() as client:
r = await get_mempool_info("hashrate_1m", gerty)
data = r
stat = {}
stat["current"] = data["currentHashrate"]
stat["1w"] = data["hashrates"][len(data["hashrates"]) - 7]["avgHashrate"]
r = await get_mempool_info("hashrate_1m", gerty)
data = r
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 get_mempool_info("hashrate_1m", gerty)
data = r
stat = {}
stat["current"] = data["currentDifficulty"]
stat["previous"] = data["difficulty"][len(data["difficulty"]) - 2][
"difficulty"
]
r = await get_mempool_info("hashrate_1m", gerty)
data = r
stat["current"] = data["currentDifficulty"]
stat["previous"] = data["difficulty"][len(data["difficulty"]) - 2][
"difficulty"
]
return stat
@ -427,10 +423,10 @@ def get_screen_slug_by_index(index: int, screens_list):
# Get a list of text items for the screen number
async def get_screen_data(screen_num: int, screens_list: dict, gerty):
async def get_screen_data(screen_num: int, screens_list: list, gerty):
screen_slug = get_screen_slug_by_index(screen_num, screens_list)
# first get the relevant slug from the display_preferences
areas = []
areas: List = []
title = ""
if screen_slug == "dashboard":
@ -533,9 +529,10 @@ async def get_screen_data(screen_num: int, screens_list: dict, gerty):
title = "Lightning Network"
areas = await get_lightning_stats(gerty)
data = {}
data["title"] = title
data["areas"] = areas
data = {
"title": title,
"areas": areas,
}
return data
@ -598,7 +595,7 @@ async def get_dashboard(gerty):
text = []
text.append(
get_text_item_dict(
text=await get_time_remaining_next_difficulty_adjustment(gerty),
text=await get_time_remaining_next_difficulty_adjustment(gerty) or "0",
font_size=15,
gerty_type=gerty.type,
)
@ -630,7 +627,7 @@ async def get_lnbits_wallet_balances(gerty):
return wallets
async def get_placeholder_text():
async def get_placeholder_text(gerty):
return [
get_text_item_dict(
text="Some placeholder text",
@ -838,14 +835,14 @@ async def get_time_remaining_next_difficulty_adjustment(gerty):
r = await get_mempool_info("difficulty_adjustment", gerty)
stat = r["remainingTime"]
time = get_time_remaining(stat / 1000, 3)
return time
return time
async def get_mempool_stat(stat_slug: str, gerty):
text = []
if isinstance(gerty.mempool_endpoint, str):
if stat_slug == "mempool_tx_count":
r = get_mempool_info("mempool", gerty)
r = await get_mempool_info("mempool", gerty)
if stat_slug == "mempool_tx_count":
stat = round(r["count"])
text.append(
@ -951,29 +948,3 @@ async def get_mempool_stat(stat_slug: str, gerty):
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])

1
lnbits/extensions/gerty/models.py
View File

@ -1,5 +1,4 @@
from sqlite3 import Row
from typing import Optional
from fastapi import Query
from pydantic import BaseModel

6
lnbits/extensions/gerty/views.py
View File

@ -1,10 +1,7 @@
import json
from http import HTTPStatus
from fastapi import Request
from fastapi.params import Depends
from fastapi import Request, Depends
from fastapi.templating import Jinja2Templates
from loguru import logger
from starlette.exceptions import HTTPException
from starlette.responses import HTMLResponse
@ -13,7 +10,6 @@ from lnbits.decorators import check_user_exists
from . import gerty_ext, gerty_renderer
from .crud import get_gerty
from .views_api import api_gerty_json
templates = Jinja2Templates(directory="templates")

33
lnbits/extensions/gerty/views_api.py
View File

@ -1,24 +1,12 @@
import json
import math
import os
import random
import time
from datetime import datetime
from http import HTTPStatus
import httpx
from fastapi import Query
from fastapi.params import Depends
from fastapi.templating import Jinja2Templates
from lnurl import decode as decode_lnurl
from fastapi import Query, Depends
from loguru import logger
from starlette.exceptions import HTTPException
from lnbits.core.crud import get_user, get_wallet_for_key
from lnbits.core.services import create_invoice
from lnbits.core.views.api import api_payment, api_wallet
from lnbits.core.crud import get_user
from lnbits.decorators import WalletTypeInfo, get_key_type, require_admin_key
from lnbits.utils.exchange_rates import satoshis_amount_as_fiat
from . import gerty_ext
from .crud import (
@ -29,8 +17,9 @@ from .crud import (
get_mempool_info,
update_gerty,
)
from .helpers import *
from .models import Gerty, MempoolEndpoint
from .models import Gerty
@gerty_ext.get("/api/v1/gerty", status_code=HTTPStatus.OK)
@ -39,7 +28,8 @@ async def api_gertys(
):
wallet_ids = [wallet.wallet.id]
if all_wallets:
wallet_ids = (await get_user(wallet.wallet.user)).wallet_ids
user = await get_user(wallet.wallet.user)
wallet_ids = user.wallet_ids if user else []
return [gerty.dict() for gerty in await get_gertys(wallet_ids)]
@ -51,7 +41,6 @@ async def api_link_create_or_update(
wallet: WalletTypeInfo = Depends(get_key_type),
gerty_id: str = Query(None),
):
logger.debug(data)
if gerty_id:
gerty = await get_gerty(gerty_id)
if not gerty:
@ -91,13 +80,13 @@ async def api_gerty_delete(
raise HTTPException(status_code=HTTPStatus.NO_CONTENT)
@gerty_ext.get("/api/v1/gerty/satoshiquote", status_code=HTTPStatus.OK)
async def api_gerty_satoshi():
return await get_satoshi
# @gerty_ext.get("/api/v1/gerty/satoshiquote", status_code=HTTPStatus.OK)
# async def api_gerty_satoshi():
# return await get_satoshi
@gerty_ext.get("/api/v1/gerty/pages/{gerty_id}/{p}")
async def api_gerty_json(gerty_id: str, p: int = None): # page number
async def api_gerty_json(gerty_id: str, p: int = 0): # page number
gerty = await get_gerty(gerty_id)
if not gerty:
@ -117,7 +106,7 @@ async def api_gerty_json(gerty_id: str, p: int = None): # page number
enabled_screen_count += 1
enabled_screens.append(screen_slug)
logger.debug("Screeens " + str(enabled_screens))
logger.debug("Screens " + str(enabled_screens))
data = await get_screen_data(p, enabled_screens, gerty)
next_screen_number = 0 if ((p + 1) >= enabled_screen_count) else p + 1

1
pyproject.toml
View File

@ -93,7 +93,6 @@ exclude = """(?x)(
| ^lnbits/extensions/boltz.
| ^lnbits/extensions/boltcards.
| ^lnbits/extensions/events.
| ^lnbits/extensions/gerty.
| ^lnbits/extensions/hivemind.
| ^lnbits/extensions/invoices.
| ^lnbits/extensions/livestream.