mirror of
https://github.com/curly60e/pyblock.git
synced 2026-08-13 12:33:15 +02:00
Add enhanced block clock with 12 new features for Menu A
New clock/ package replaces the old design() polling loop with a flicker-free ANSI cursor-positioned renderer. Features include: countdown timer, epoch/halving progress bar, fee rate indicator, hashrate sparkline, matrix mining animation, odometer digit transition, heartbeat pulse, zen mode, UTC time display, fireworks on milestone blocks, generative hash art, and configurable sound modes. All features are toggleable via Settings → D (Clock Display Settings). Also fixes hardcoded config paths in menuSelection(), menuSelectionLN(), and TUI startup to use the cfg singleton instead. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
GaltRanch
parent
1fa1846790
commit
8e2737251d
10 changed files with 985 additions and 93 deletions
|
|
@ -882,70 +882,11 @@ def some_other_function():
|
|||
def execute_visualizer():
|
||||
block_visualizer.run_visualizer()
|
||||
|
||||
def artist(): # here we convert the result of the command 'getblockcount' on a random art design
|
||||
while True:
|
||||
try:
|
||||
clear()
|
||||
close()
|
||||
design()
|
||||
except KeyboardInterrupt:
|
||||
break
|
||||
except Exception as e:
|
||||
logger.debug("Loop interrupted: %s", e)
|
||||
break
|
||||
|
||||
def design():
|
||||
if os.path.isfile('config/pyblocksettingsClock.conf') or os.path.isfile('config/pyblocksettingsClock.conf'): # Check if the file 'bclock.conf' is in the same folder
|
||||
settingsv = json.load(open("config/pyblocksettingsClock.conf", "r")) # Load the file 'bclock.conf'
|
||||
settingsClock = settingsv # Copy the variable pathv to 'path'
|
||||
else:
|
||||
settingsClock = {"gradient":"", "design":"block", "colorA":"green", "colorB":"yellow"}
|
||||
with open("config/pyblocksettingsClock.conf", "w") as f: json.dump(settingsClock, f, indent=2)
|
||||
bitcoinclient = f'{path["bitcoincli"]} getblockcount'
|
||||
block = subprocess.run(str(bitcoinclient).split(), capture_output=True, text=True).stdout # 'getblockcount' convert to string
|
||||
b = block
|
||||
a = b
|
||||
output = render(str(b), colors=[settingsClock['colorA'], settingsClock['colorB']], align='center')
|
||||
print("\033[0;37;40m\x1b[?25l" + output)
|
||||
while True:
|
||||
x = a
|
||||
bitcoinclient = f'{path["bitcoincli"]} getblockcount'
|
||||
block = subprocess.run(str(bitcoinclient).split(), capture_output=True, text=True).stdout # 'getblockcount' convert to string
|
||||
b = block
|
||||
if b > a:
|
||||
clear()
|
||||
close()
|
||||
output = render(str(b), colors=[settingsClock['colorA'], settingsClock['colorB']], align='center')
|
||||
print("\a\x1b[?25l" + output)
|
||||
bitcoinclient = f'{path["bitcoincli"]} getbestblockhash'
|
||||
bb = subprocess.run(str(bitcoinclient).split(), capture_output=True, text=True).stdout
|
||||
ll = bb
|
||||
bitcoinclientgetblock = f'{path["bitcoincli"]} getblock {ll}'
|
||||
qq = subprocess.run(bitcoinclientgetblock.split(), capture_output=True, text=True).stdout
|
||||
yy = json.loads(qq)
|
||||
mm = yy
|
||||
outputsize = render(str(mm['size']) + " bytes", colors=[settingsClock['colorA'], settingsClock['colorB']], align='center', font='tiny')
|
||||
print("\x1b[?25l" + outputsize)
|
||||
outputtxs = render(str(mm['nTx']) + " txs", colors=[settingsClock['colorA'], settingsClock['colorB']], align='center', font='tiny')
|
||||
print("\x1b[?25l" + outputtxs)
|
||||
sh = int(mm['nTx']) / 4
|
||||
shq = int(sh)
|
||||
ss = str(rectangle(shq))
|
||||
print(ss.replace("None",""))
|
||||
t.sleep(10)
|
||||
txs = str(mm['nTx'])
|
||||
if txs == "1":
|
||||
try:
|
||||
p = subprocess.Popen(['curl', 'https://ascii.live/forrest'])
|
||||
p.wait(5)
|
||||
except subprocess.TimeoutExpired:
|
||||
p.kill()
|
||||
print("\033[0;37;40m\x1b[?25l")
|
||||
clear()
|
||||
close()
|
||||
a = b
|
||||
output = render(str(b), colors=[settingsClock['colorA'], settingsClock['colorB']], align='center')
|
||||
print("\x1b[?25l" + output)
|
||||
def artist():
|
||||
"""Launch the enhanced block clock."""
|
||||
from clock import run_clock
|
||||
mode = "remote" if not path.get("bitcoincli") else "local"
|
||||
run_clock(mode, path, cfg.settings_clock)
|
||||
|
||||
|
||||
#--------------------------------- Hex Block Decoder Functions -------------------------------------
|
||||
|
|
@ -3926,6 +3867,7 @@ def settings4Local():
|
|||
\u001b[38;5;27mA.\033[0;37;40m Change Logo Design
|
||||
\u001b[38;5;27mB.\033[0;37;40m Change Logo Colors
|
||||
\u001b[38;5;27mC.\033[0;37;40m Change Clock Colors
|
||||
\u001b[38;5;27mD.\033[0;37;40m Clock Display Settings
|
||||
\u001b[33;1mEnter.\033[0;37;40m Return
|
||||
\n\n\x1b[?25h""".format(n, alias['alias'], d['blocks'], version, ()))
|
||||
menuSettingsLocal(input("\033[1;32;40mSelect option: \033[0;37;40m"))
|
||||
|
|
@ -3950,6 +3892,7 @@ def settings4LocalOnchainONLY():
|
|||
\u001b[38;5;27mA.\033[0;37;40m Change Logo Design
|
||||
\u001b[38;5;27mB.\033[0;37;40m Change Logo Colors
|
||||
\u001b[38;5;27mC.\033[0;37;40m Change Clock Colors
|
||||
\u001b[38;5;27mD.\033[0;37;40m Clock Display Settings
|
||||
\u001b[33;1mEnter.\033[0;37;40m Return
|
||||
\n\n\x1b[?25h""".format(n, d['blocks'], version, ()))
|
||||
menuSettingsLocalOnchainONLY(input("\033[1;32;40mSelect option: \033[0;37;40m"))
|
||||
|
|
@ -3980,6 +3923,7 @@ def settings4Remote():
|
|||
\u001b[38;5;27mA.\033[0;37;40m Change Logo Design
|
||||
\u001b[38;5;27mB.\033[0;37;40m Change Logo Colors
|
||||
\u001b[38;5;27mC.\033[0;37;40m Change Clock Colors
|
||||
\u001b[38;5;27mD.\033[0;37;40m Clock Display Settings
|
||||
\u001b[33;1mEnter.\033[0;37;40m Return
|
||||
\n\n\x1b[?25h""".format(a, alias['alias'], d['blocks'], version, ()))
|
||||
menuSettingsRemote(input("\033[1;32;40mSelect option: \033[0;37;40m"))
|
||||
|
|
@ -5139,36 +5083,29 @@ def colorsSelectRainbowEndOnchainONLY():
|
|||
|
||||
def menuSelection():
|
||||
chln = {"fullbtclnd":"","fullbtc":"","cropped":""}
|
||||
if os.path.isfile('config/intro.conf'):
|
||||
chain = json.load(open("config/intro.conf", "r"))
|
||||
chln = chain
|
||||
if cfg.has_config('intro.conf'):
|
||||
chln = cfg.intro_mode
|
||||
print(chln + "\n")
|
||||
if chln == "B":
|
||||
path = {"ip_port":"", "rpcuser":"", "rpcpass":"", "bitcoincli":""}
|
||||
pathv = json.load(open("config/bclock.conf", "r")) # Load the file 'bclock.conf'
|
||||
path = pathv # Copy the variable pathv to 'path'
|
||||
path = cfg.path
|
||||
MainMenuLOCALChainONLY()
|
||||
elif chln == "A":
|
||||
path = {"ip_port":"", "rpcuser":"", "rpcpass":"", "bitcoincli":""}
|
||||
pathv = json.load(open("config/bclock.conf", "r")) # Load the file 'bclock.conf'
|
||||
path = pathv # Copy the variable pathv to 'path'
|
||||
path = cfg.path
|
||||
MainMenuLOCAL()
|
||||
elif chln == "C":
|
||||
from SPV.spvblock import MainMenuCROPPED as _lite_menu
|
||||
_lite_menu()
|
||||
else:
|
||||
if os.path.isfile('config/blndconnect.conf'):
|
||||
if cfg.has_config('blndconnect.conf'):
|
||||
chln['offchain'] = "offchain"
|
||||
else:
|
||||
chln['onchain'] = "onchain"
|
||||
|
||||
with open("config/selection.conf", "w") as f: json.dump(chln, f, indent=2)
|
||||
cfg.save("selection.conf", chln)
|
||||
|
||||
|
||||
def menuSelectionLN():
|
||||
lndconnectload = {"ip_port":"", "tls":"", "macaroon":"", "lncli":""}
|
||||
lndconnectData = json.load(open("config/blndconnect.conf", "r")) # Load the file 'bclock.conf'
|
||||
lndconnectload = lndconnectData # Copy the variable pathv to 'path'
|
||||
lndconnectload = cfg.lndconnectload
|
||||
if lndconnectload['ln']:
|
||||
menuLNDLOCAL()
|
||||
else:
|
||||
|
|
@ -5452,6 +5389,8 @@ def menuSettingsLocal(menuSTT):
|
|||
clear()
|
||||
blogo()
|
||||
colorsC()
|
||||
elif menuSTT in ["D", "d"]:
|
||||
clockDisplaySettings()
|
||||
|
||||
def menuSettingsLocalOnchainONLY(menuSTT):
|
||||
if menuSTT in ["A", "a"]:
|
||||
|
|
@ -5466,6 +5405,8 @@ def menuSettingsLocalOnchainONLY(menuSTT):
|
|||
clear()
|
||||
blogo()
|
||||
colorsCOnchainONLY()
|
||||
elif menuSTT in ["D", "d"]:
|
||||
clockDisplaySettings()
|
||||
|
||||
def menuSettingsRemote(menuSTT):
|
||||
if menuSTT in ["A", "a"]:
|
||||
|
|
@ -5480,6 +5421,81 @@ def menuSettingsRemote(menuSTT):
|
|||
clear()
|
||||
blogo()
|
||||
colorsCRemote()
|
||||
elif menuSTT in ["D", "d"]:
|
||||
clockDisplaySettings()
|
||||
|
||||
def clockDisplaySettings():
|
||||
"""Interactive settings menu for clock display features."""
|
||||
while True:
|
||||
try:
|
||||
clear()
|
||||
blogo()
|
||||
s = cfg.settings_clock
|
||||
|
||||
def _on_off(val):
|
||||
return "\033[1;32;40mON\033[0;37;40m" if val else "\033[1;31;40mOFF\033[0;37;40m"
|
||||
|
||||
print("""\t\t
|
||||
\033[1;37;40mClock Display Settings\033[0;37;40m
|
||||
|
||||
\u001b[38;5;27m1.\033[0;37;40m Countdown Timer {}
|
||||
\u001b[38;5;27m2.\033[0;37;40m Epoch Progress Bar {}
|
||||
\u001b[38;5;27m3.\033[0;37;40m Fee Rate Indicator {}
|
||||
\u001b[38;5;27m4.\033[0;37;40m Hashrate Sparkline {}
|
||||
\u001b[38;5;27m5.\033[0;37;40m UTC Time Display {}
|
||||
\u001b[38;5;27m6.\033[0;37;40m Zen Mode {}
|
||||
\u001b[38;5;27m7.\033[0;37;40m Heartbeat Pulse {}
|
||||
\u001b[38;5;27m8.\033[0;37;40m Generative Art {}
|
||||
\u001b[38;5;27m9.\033[0;37;40m Fireworks on Milestones {}
|
||||
|
||||
\u001b[38;5;27mA.\033[0;37;40m Animation: \033[1;33;40m{}\033[0;37;40m
|
||||
\u001b[38;5;27mS.\033[0;37;40m Sound: \033[1;33;40m{}\033[0;37;40m
|
||||
\u001b[33;1mEnter.\033[0;37;40m Return
|
||||
\n\x1b[?25h""".format(
|
||||
_on_off(s.get('show_countdown', True)),
|
||||
_on_off(s.get('show_epoch_bar', True)),
|
||||
_on_off(s.get('show_fee_rates', True)),
|
||||
_on_off(s.get('show_sparkline', False)),
|
||||
_on_off(s.get('show_utc_time', False)),
|
||||
_on_off(s.get('zen_mode', False)),
|
||||
_on_off(s.get('heartbeat', True)),
|
||||
_on_off(s.get('generative_art', False)),
|
||||
_on_off(s.get('fireworks', True)),
|
||||
s.get('animation', 'matrix'),
|
||||
s.get('sound', 'bell'),
|
||||
))
|
||||
|
||||
opt = input("\033[1;32;40mSelect option: \033[0;37;40m").strip()
|
||||
|
||||
toggles = {
|
||||
'1': 'show_countdown', '2': 'show_epoch_bar',
|
||||
'3': 'show_fee_rates', '4': 'show_sparkline',
|
||||
'5': 'show_utc_time', '6': 'zen_mode',
|
||||
'7': 'heartbeat', '8': 'generative_art',
|
||||
'9': 'fireworks',
|
||||
}
|
||||
|
||||
if opt in toggles:
|
||||
key = toggles[opt]
|
||||
s[key] = not s.get(key, False)
|
||||
cfg.save("pyblocksettingsClock.conf", s)
|
||||
elif opt in ['A', 'a']:
|
||||
modes = ['matrix', 'odometer', 'none']
|
||||
current = s.get('animation', 'matrix')
|
||||
idx = (modes.index(current) + 1) % len(modes) if current in modes else 0
|
||||
s['animation'] = modes[idx]
|
||||
cfg.save("pyblocksettingsClock.conf", s)
|
||||
elif opt in ['S', 's']:
|
||||
modes = ['bell', 'pattern', 'silent']
|
||||
current = s.get('sound', 'bell')
|
||||
idx = (modes.index(current) + 1) % len(modes) if current in modes else 0
|
||||
s['sound'] = modes[idx]
|
||||
cfg.save("pyblocksettingsClock.conf", s)
|
||||
else:
|
||||
break
|
||||
except KeyboardInterrupt:
|
||||
break
|
||||
|
||||
|
||||
def menuColors(menuCLS):
|
||||
if menuCLS in ["A", "a"]:
|
||||
|
|
@ -6085,18 +6101,9 @@ def menuWeatherOnchainONLY(menuWD):
|
|||
|
||||
def mainmenuControl(menuS, mode): #Unified execution of Main Menu options
|
||||
if menuS in ["A", "a"]:
|
||||
if mode == "remote":
|
||||
while True:
|
||||
try:
|
||||
clear()
|
||||
close()
|
||||
remotegetblock()
|
||||
tmp()
|
||||
except Exception as e:
|
||||
logger.debug("Loop interrupted: %s", e)
|
||||
break
|
||||
else:
|
||||
artist()
|
||||
from clock import run_clock
|
||||
clock_mode = "remote" if mode == "remote" else "local"
|
||||
run_clock(clock_mode, path, cfg.settings_clock)
|
||||
elif menuS in ["B", "b"]:
|
||||
if mode == "remote":
|
||||
bitcoincoremenuREMOTE()
|
||||
|
|
@ -7435,8 +7442,7 @@ if __name__ == "__main__":
|
|||
mode = "lite"
|
||||
cfg.load()
|
||||
if cfg.has_config('intro.conf'):
|
||||
with open("config/intro.conf", "r") as f:
|
||||
init_data = json.load(f)
|
||||
init_data = cfg.intro_mode
|
||||
if isinstance(init_data, str):
|
||||
if init_data == "A":
|
||||
mode = "local"
|
||||
|
|
|
|||
44
pybitblock/clock/__init__.py
Normal file
44
pybitblock/clock/__init__.py
Normal file
|
|
@ -0,0 +1,44 @@
|
|||
"""Enhanced block clock for PyBLOCK.
|
||||
|
||||
Entry point: run_clock(mode, path, settings_clock)
|
||||
"""
|
||||
|
||||
import time
|
||||
import sys
|
||||
|
||||
from . import animations
|
||||
from .data import ClockData
|
||||
from .renderer import Layout
|
||||
|
||||
|
||||
def run_clock(mode, path, settings_clock):
|
||||
"""Main clock loop with partial screen updates.
|
||||
|
||||
mode: 'local', 'remote', or 'lite'
|
||||
path: dict with bitcoincli, ip_port, rpcuser, rpcpass
|
||||
settings_clock: dict from pyblocksettingsClock.conf
|
||||
"""
|
||||
data = ClockData(mode, path)
|
||||
layout = Layout(settings_clock)
|
||||
|
||||
try:
|
||||
# Initial full fetch and render
|
||||
data.refresh()
|
||||
layout.render_full(data)
|
||||
|
||||
while True:
|
||||
time.sleep(2)
|
||||
|
||||
changed = data.poll()
|
||||
|
||||
if 'block_height' in changed:
|
||||
layout.on_new_block(data, animations)
|
||||
else:
|
||||
# Update dynamic elements
|
||||
layout.update_countdown(data)
|
||||
layout.heartbeat(data)
|
||||
|
||||
except KeyboardInterrupt:
|
||||
pass
|
||||
finally:
|
||||
layout.cleanup()
|
||||
193
pybitblock/clock/animations.py
Normal file
193
pybitblock/clock/animations.py
Normal file
|
|
@ -0,0 +1,193 @@
|
|||
"""Visual animations for the block clock.
|
||||
|
||||
Mining rain, odometer digit transition, fireworks on milestones.
|
||||
"""
|
||||
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from random import choice, randrange
|
||||
|
||||
from cfonts import render
|
||||
|
||||
# Halving blocks for milestone detection
|
||||
HALVING_BLOCKS = {210_000 * i for i in range(1, 65)}
|
||||
|
||||
|
||||
def is_milestone_block(height):
|
||||
"""Check if block is a milestone. Returns description or None."""
|
||||
if height in HALVING_BLOCKS:
|
||||
return f"HALVING #{height // 210_000}"
|
||||
if height % 100_000 == 0:
|
||||
return f"BLOCK {height:,}"
|
||||
if height % 10_000 == 0:
|
||||
return f"BLOCK {height:,}"
|
||||
return None
|
||||
|
||||
|
||||
def mining_animation(duration=3.0):
|
||||
"""Matrix-style mining rain animation for new block discovery.
|
||||
|
||||
Uses inline implementation to avoid import issues with terminal_matrix.
|
||||
"""
|
||||
cols, lines = shutil.get_terminal_size((80, 24))
|
||||
chars = [chr(i) for i in range(0x30, 0x80)]
|
||||
green = "\033[32m"
|
||||
bright_green = "\033[1;32m"
|
||||
reset = "\033[0m"
|
||||
|
||||
# Initialize cascades
|
||||
cascades = {}
|
||||
sys.stdout.write("\033[2J\033[H\x1b[?25l")
|
||||
|
||||
end_time = time.time() + duration
|
||||
while time.time() < end_time:
|
||||
# Spawn new cascades
|
||||
if len(cascades) < cols // 2:
|
||||
col = randrange(1, cols + 1)
|
||||
if col not in cascades:
|
||||
speed = randrange(1, 4)
|
||||
length = randrange(4, lines // 2)
|
||||
cascades[col] = {'row': 1, 'speed': speed, 'length': length}
|
||||
|
||||
buf = []
|
||||
to_remove = []
|
||||
for col, c in cascades.items():
|
||||
row = c['row']
|
||||
if row <= lines:
|
||||
char = choice(chars)
|
||||
buf.append(f"\033[{row};{col}H{bright_green}{char}")
|
||||
# Dim the trail
|
||||
trail_row = row - c['length']
|
||||
if 1 <= trail_row <= lines:
|
||||
buf.append(f"\033[{trail_row};{col}H{reset} ")
|
||||
c['row'] += c['speed']
|
||||
if c['row'] - c['length'] > lines:
|
||||
to_remove.append(col)
|
||||
|
||||
for col in to_remove:
|
||||
del cascades[col]
|
||||
|
||||
if buf:
|
||||
sys.stdout.write(''.join(buf))
|
||||
sys.stdout.flush()
|
||||
|
||||
time.sleep(0.03)
|
||||
|
||||
sys.stdout.write(f"\033[2J\033[H{reset}\x1b[?25l")
|
||||
sys.stdout.flush()
|
||||
|
||||
|
||||
def odometer_transition(old_height_str, new_height_str, settings, start_row):
|
||||
"""Animate changing digits like a mechanical odometer.
|
||||
|
||||
Renders intermediate digit values at the positions that changed.
|
||||
"""
|
||||
colors = [settings.get('colorA', 'green'), settings.get('colorB', 'yellow')]
|
||||
font = settings.get('design', 'block')
|
||||
|
||||
# Pad to same length
|
||||
max_len = max(len(old_height_str), len(new_height_str))
|
||||
old = old_height_str.zfill(max_len)
|
||||
new = new_height_str.zfill(max_len)
|
||||
|
||||
# Find which digits changed
|
||||
changed = [i for i in range(max_len) if old[i] != new[i]]
|
||||
|
||||
if not changed:
|
||||
return
|
||||
|
||||
# Animate: show 3 intermediate frames
|
||||
frames = 3
|
||||
for frame in range(frames):
|
||||
intermediate = list(old)
|
||||
for i in changed:
|
||||
old_d = int(old[i])
|
||||
new_d = int(new[i])
|
||||
# Roll through digits
|
||||
step = (old_d + (frame + 1) * (new_d - old_d + 10) // (frames + 1)) % 10
|
||||
if frame == frames - 1:
|
||||
step = new_d
|
||||
intermediate[i] = str(step)
|
||||
|
||||
text = ''.join(intermediate)
|
||||
output = render(text, colors=colors, align='center', font=font)
|
||||
lines = output.rstrip('\n').split('\n')
|
||||
|
||||
buf = []
|
||||
for j, line in enumerate(lines):
|
||||
buf.append(f"\033[{start_row + j};1H\033[2K{line}")
|
||||
sys.stdout.write(''.join(buf))
|
||||
sys.stdout.flush()
|
||||
time.sleep(0.12)
|
||||
|
||||
|
||||
def fireworks_animation(term_width, term_height, duration=5.0):
|
||||
"""ASCII fireworks celebration for milestone blocks."""
|
||||
colors = [
|
||||
"\033[1;31m", # red
|
||||
"\033[1;33m", # yellow
|
||||
"\033[1;32m", # green
|
||||
"\033[1;36m", # cyan
|
||||
"\033[1;35m", # magenta
|
||||
"\033[1;37m", # white
|
||||
]
|
||||
sparks = ['*', '.', '+', 'o', '\u2022', '\u2726', '\u2727', '\u2728']
|
||||
reset = "\033[0m"
|
||||
|
||||
sys.stdout.write("\033[2J\033[H\x1b[?25l")
|
||||
|
||||
end_time = time.time() + duration
|
||||
explosions = []
|
||||
|
||||
while time.time() < end_time:
|
||||
# Spawn new explosion
|
||||
if randrange(5) == 0 or not explosions:
|
||||
cx = randrange(5, term_width - 5)
|
||||
cy = randrange(3, term_height - 3)
|
||||
color = choice(colors)
|
||||
explosions.append({
|
||||
'cx': cx, 'cy': cy, 'color': color,
|
||||
'radius': 0, 'max_radius': randrange(3, 8),
|
||||
'age': 0
|
||||
})
|
||||
|
||||
buf = []
|
||||
alive = []
|
||||
for exp in explosions:
|
||||
exp['age'] += 1
|
||||
exp['radius'] = min(exp['radius'] + 1, exp['max_radius'])
|
||||
|
||||
if exp['age'] > exp['max_radius'] * 3:
|
||||
# Fade: clear spark positions
|
||||
for _ in range(8):
|
||||
dx = randrange(-exp['max_radius'], exp['max_radius'] + 1)
|
||||
dy = randrange(-exp['max_radius'] // 2, exp['max_radius'] // 2 + 1)
|
||||
x = exp['cx'] + dx
|
||||
y = exp['cy'] + dy
|
||||
if 1 <= x <= term_width and 1 <= y <= term_height:
|
||||
buf.append(f"\033[{y};{x}H ")
|
||||
continue
|
||||
|
||||
alive.append(exp)
|
||||
r = exp['radius']
|
||||
for _ in range(r * 4):
|
||||
dx = randrange(-r, r + 1)
|
||||
dy = randrange(-r // 2, r // 2 + 1)
|
||||
x = exp['cx'] + dx
|
||||
y = exp['cy'] + dy
|
||||
if 1 <= x <= term_width and 1 <= y <= term_height:
|
||||
spark = choice(sparks)
|
||||
buf.append(f"\033[{y};{x}H{exp['color']}{spark}")
|
||||
|
||||
explosions = alive
|
||||
|
||||
if buf:
|
||||
sys.stdout.write(''.join(buf) + reset)
|
||||
sys.stdout.flush()
|
||||
|
||||
time.sleep(0.08)
|
||||
|
||||
sys.stdout.write(f"\033[2J\033[H{reset}\x1b[?25l")
|
||||
sys.stdout.flush()
|
||||
200
pybitblock/clock/data.py
Normal file
200
pybitblock/clock/data.py
Normal file
|
|
@ -0,0 +1,200 @@
|
|||
"""Bitcoin data layer for the enhanced block clock.
|
||||
|
||||
Fetches block height, block details, fees, hashrate, and epoch info
|
||||
from either a local bitcoin-cli or JSON-RPC, plus mempool.space API
|
||||
for fee rates and hashrate.
|
||||
"""
|
||||
|
||||
import json
|
||||
import subprocess
|
||||
import threading
|
||||
import time
|
||||
|
||||
import requests
|
||||
|
||||
# Halving constants
|
||||
BLOCKS_PER_HALVING = 210_000
|
||||
BLOCKS_PER_EPOCH = 2016
|
||||
HALVING_BLOCKS = [BLOCKS_PER_HALVING * i for i in range(1, 65)]
|
||||
|
||||
# API endpoints
|
||||
MEMPOOL_FEES_URL = "https://mempool.space/api/v1/fees/recommended"
|
||||
MEMPOOL_HASHRATE_URL = "https://mempool.space/api/v1/mining/hashrate/3d"
|
||||
MEMPOOL_HEIGHT_URL = "https://mempool.space/api/blocks/tip/height"
|
||||
MEMPOOL_BLOCK_URL = "https://mempool.space/api/block/"
|
||||
|
||||
|
||||
class ClockData:
|
||||
"""Fetches and caches Bitcoin data for the clock display."""
|
||||
|
||||
def __init__(self, mode, path):
|
||||
"""
|
||||
mode: 'local', 'remote', or 'lite'
|
||||
path: dict with ip_port, rpcuser, rpcpass, bitcoincli
|
||||
"""
|
||||
self.mode = mode
|
||||
self.path = path
|
||||
|
||||
# Block data
|
||||
self.block_height = 0
|
||||
self.block_hash = ""
|
||||
self.block_time = 0
|
||||
self.block_size = 0
|
||||
self.block_tx_count = 0
|
||||
|
||||
# Epoch / halving
|
||||
self.epoch_progress = 0.0
|
||||
self.epoch_block = 0
|
||||
self.blocks_to_halving = 0
|
||||
self.next_halving_block = 0
|
||||
|
||||
# Fee rates (sat/vB)
|
||||
self.fee_fastest = 0
|
||||
self.fee_half_hour = 0
|
||||
self.fee_hour = 0
|
||||
|
||||
# Hashrate
|
||||
self.hashrate_current = 0.0
|
||||
self.hashrate_history = []
|
||||
self.difficulty = 0.0
|
||||
|
||||
# Internal
|
||||
self._lock = threading.Lock()
|
||||
self._bg_thread = None
|
||||
self._last_api_fetch = 0
|
||||
|
||||
# --- RPC / CLI abstraction ---
|
||||
|
||||
def _cli(self, command):
|
||||
"""Run bitcoin-cli command, return stdout string."""
|
||||
cmd = f'{self.path["bitcoincli"]} {command}'
|
||||
result = subprocess.run(cmd.split(), capture_output=True, text=True)
|
||||
return result.stdout.strip()
|
||||
|
||||
def _rpc(self, method, params=None):
|
||||
"""JSON-RPC call for remote mode."""
|
||||
payload = json.dumps({
|
||||
"jsonrpc": "2.0", "id": "clock",
|
||||
"method": method, "params": params or []
|
||||
})
|
||||
resp = requests.post(
|
||||
self.path['ip_port'],
|
||||
auth=(self.path['rpcuser'], self.path['rpcpass']),
|
||||
data=payload, timeout=10
|
||||
)
|
||||
return resp.json()['result']
|
||||
|
||||
def _get_block_count(self):
|
||||
if self.mode == 'lite':
|
||||
r = requests.get(MEMPOOL_HEIGHT_URL, timeout=10)
|
||||
return int(r.text.strip())
|
||||
elif self.mode == 'remote':
|
||||
return int(self._rpc('getblockcount'))
|
||||
else:
|
||||
return int(self._cli('getblockcount'))
|
||||
|
||||
def _get_block_details(self):
|
||||
"""Fetch full block details for current tip."""
|
||||
if self.mode == 'lite':
|
||||
r = requests.get(MEMPOOL_HEIGHT_URL, timeout=10)
|
||||
tip_hash = requests.get(
|
||||
"https://mempool.space/api/blocks/tip/hash", timeout=10
|
||||
).text.strip()
|
||||
r2 = requests.get(f"{MEMPOOL_BLOCK_URL}{tip_hash}", timeout=10)
|
||||
block = r2.json()
|
||||
self.block_hash = tip_hash
|
||||
self.block_time = block.get('timestamp', int(time.time()))
|
||||
self.block_size = block.get('size', 0)
|
||||
self.block_tx_count = block.get('tx_count', 0)
|
||||
else:
|
||||
if self.mode == 'remote':
|
||||
block_hash = self._rpc('getbestblockhash')
|
||||
block = self._rpc('getblock', [block_hash])
|
||||
else:
|
||||
block_hash = self._cli('getbestblockhash')
|
||||
raw = self._cli(f'getblock {block_hash}')
|
||||
block = json.loads(raw)
|
||||
self.block_hash = block_hash
|
||||
self.block_time = block.get('time', int(time.time()))
|
||||
self.block_size = block.get('size', 0)
|
||||
self.block_tx_count = block.get('nTx', 0)
|
||||
|
||||
def _calc_epoch(self):
|
||||
"""Calculate epoch and halving progress from block height."""
|
||||
h = self.block_height
|
||||
self.epoch_block = h % BLOCKS_PER_EPOCH
|
||||
self.epoch_progress = self.epoch_block / BLOCKS_PER_EPOCH
|
||||
|
||||
for hb in HALVING_BLOCKS:
|
||||
if h < hb:
|
||||
self.next_halving_block = hb
|
||||
self.blocks_to_halving = hb - h
|
||||
break
|
||||
else:
|
||||
self.blocks_to_halving = 0
|
||||
self.next_halving_block = 0
|
||||
|
||||
# --- API data (background thread) ---
|
||||
|
||||
def _fetch_api_data(self):
|
||||
"""Fetch fee rates and hashrate from mempool.space (non-blocking)."""
|
||||
try:
|
||||
r = requests.get(MEMPOOL_FEES_URL, timeout=10)
|
||||
fees = r.json()
|
||||
with self._lock:
|
||||
self.fee_fastest = fees.get('fastestFee', 0)
|
||||
self.fee_half_hour = fees.get('halfHourFee', 0)
|
||||
self.fee_hour = fees.get('hourFee', 0)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
try:
|
||||
r = requests.get(MEMPOOL_HASHRATE_URL, timeout=10)
|
||||
data = r.json()
|
||||
with self._lock:
|
||||
self.hashrate_current = data.get('currentHashrate', 0)
|
||||
self.difficulty = data.get('currentDifficulty', 0)
|
||||
hashrates = data.get('hashrates', [])
|
||||
self.hashrate_history = [
|
||||
h.get('avgHashrate', 0) for h in hashrates[-20:]
|
||||
]
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def _start_bg_fetch(self):
|
||||
"""Fetch API data in background thread if enough time has passed."""
|
||||
now = time.time()
|
||||
if now - self._last_api_fetch < 30:
|
||||
return
|
||||
self._last_api_fetch = now
|
||||
t = threading.Thread(target=self._fetch_api_data, daemon=True)
|
||||
t.start()
|
||||
|
||||
# --- Public API ---
|
||||
|
||||
def refresh(self):
|
||||
"""Full data fetch: block height, details, epoch, and trigger API fetch."""
|
||||
self.block_height = self._get_block_count()
|
||||
self._get_block_details()
|
||||
self._calc_epoch()
|
||||
self._start_bg_fetch()
|
||||
|
||||
def poll(self):
|
||||
"""Quick poll: just getblockcount. Returns set of changed field names."""
|
||||
changed = set()
|
||||
new_height = self._get_block_count()
|
||||
if new_height != self.block_height:
|
||||
old_height = self.block_height
|
||||
self.block_height = new_height
|
||||
self._get_block_details()
|
||||
self._calc_epoch()
|
||||
self._start_bg_fetch()
|
||||
changed.add('block_height')
|
||||
return changed
|
||||
|
||||
@property
|
||||
def seconds_since_block(self):
|
||||
"""Seconds elapsed since the last block timestamp."""
|
||||
if self.block_time == 0:
|
||||
return 0
|
||||
return int(time.time()) - self.block_time
|
||||
52
pybitblock/clock/generative.py
Normal file
52
pybitblock/clock/generative.py
Normal file
|
|
@ -0,0 +1,52 @@
|
|||
"""Block hash generative ASCII art.
|
||||
|
||||
Uses hash bytes as seeds to create a unique visual pattern per block.
|
||||
"""
|
||||
|
||||
# Characters ordered by visual density
|
||||
GLYPHS = " \u2591\u2592\u2593\u2588\u2580\u2584\u258c\u2590\u256c\u2550\u2551"
|
||||
|
||||
# 256-color ANSI foreground
|
||||
def _color256(n):
|
||||
return f"\033[38;5;{n}m"
|
||||
|
||||
RESET = "\033[0m"
|
||||
|
||||
|
||||
def hash_art(block_hash, width=40, height=6):
|
||||
"""Generate deterministic ASCII art from a block hash string.
|
||||
|
||||
Each pair of hex digits maps to a glyph and color.
|
||||
The pattern is mirrored horizontally for symmetry.
|
||||
"""
|
||||
# Convert hex hash to bytes
|
||||
raw = block_hash.strip()
|
||||
hex_pairs = [raw[i:i+2] for i in range(0, len(raw), 2)]
|
||||
values = [int(h, 16) for h in hex_pairs if len(h) == 2]
|
||||
|
||||
if not values:
|
||||
return ""
|
||||
|
||||
half_w = width // 2
|
||||
lines = []
|
||||
|
||||
for row in range(height):
|
||||
left = []
|
||||
for col in range(half_w):
|
||||
idx = (row * half_w + col) % len(values)
|
||||
val = values[idx]
|
||||
|
||||
# Glyph from lower nibble
|
||||
glyph = GLYPHS[val % len(GLYPHS)]
|
||||
# Color from upper nibble + row offset (for variety)
|
||||
color_idx = 16 + ((val + row * 7) % 216) # 216-color cube
|
||||
left.append(f"{_color256(color_idx)}{glyph}")
|
||||
|
||||
# Mirror for symmetry
|
||||
right = list(reversed(left))
|
||||
line = ''.join(left) + ''.join(right) + RESET
|
||||
# Center it
|
||||
pad = max(0, (80 - width) // 2)
|
||||
lines.append(' ' * pad + line)
|
||||
|
||||
return '\n'.join(lines)
|
||||
240
pybitblock/clock/renderer.py
Normal file
240
pybitblock/clock/renderer.py
Normal file
|
|
@ -0,0 +1,240 @@
|
|||
"""Screen layout and rendering engine for the enhanced block clock.
|
||||
|
||||
Uses ANSI cursor positioning for flicker-free partial screen updates.
|
||||
Composes cfonts output with widget overlays.
|
||||
"""
|
||||
|
||||
import shutil
|
||||
import sys
|
||||
import time
|
||||
|
||||
from cfonts import render
|
||||
|
||||
from .widgets import (
|
||||
render_countdown,
|
||||
render_epoch_bar,
|
||||
render_fees,
|
||||
render_utc_time,
|
||||
)
|
||||
from .sparkline import render_sparkline
|
||||
from .generative import hash_art
|
||||
from .sound import play_sound
|
||||
|
||||
|
||||
# ANSI helpers
|
||||
def _move(row, col=1):
|
||||
return f"\033[{row};{col}H"
|
||||
|
||||
|
||||
def _clear_line():
|
||||
return "\033[2K"
|
||||
|
||||
|
||||
def _hide_cursor():
|
||||
return "\x1b[?25l"
|
||||
|
||||
|
||||
def _show_cursor():
|
||||
return "\x1b[?25h"
|
||||
|
||||
|
||||
def _bold(text):
|
||||
return f"\033[1m{text}\033[0m"
|
||||
|
||||
|
||||
def _dim(text):
|
||||
return f"\033[2m{text}\033[0m"
|
||||
|
||||
|
||||
def _clear_screen():
|
||||
sys.stdout.write("\033[2J\033[H")
|
||||
sys.stdout.flush()
|
||||
|
||||
|
||||
def _render_block_height(height, settings):
|
||||
"""Render block height using cfonts."""
|
||||
colors = [settings.get('colorA', 'green'), settings.get('colorB', 'yellow')]
|
||||
gradient = settings.get('gradient', '')
|
||||
font = settings.get('design', 'block')
|
||||
|
||||
kwargs = {'align': 'center', 'font': font}
|
||||
if gradient == 'grd':
|
||||
kwargs['gradient'] = colors
|
||||
else:
|
||||
kwargs['colors'] = colors
|
||||
|
||||
return render(str(height), **kwargs)
|
||||
|
||||
|
||||
class Layout:
|
||||
"""Manages screen regions and partial updates."""
|
||||
|
||||
def __init__(self, settings):
|
||||
self.settings = settings
|
||||
self.term_width, self.term_height = shutil.get_terminal_size((80, 24))
|
||||
self._height_lines = 0
|
||||
self._heartbeat_step = 0
|
||||
self._last_rendered_height = None
|
||||
|
||||
def _is_zen(self):
|
||||
return self.settings.get('zen_mode', False)
|
||||
|
||||
def _write(self, text):
|
||||
sys.stdout.write(text)
|
||||
sys.stdout.flush()
|
||||
|
||||
def _render_cfonts_at(self, output, start_row):
|
||||
"""Render cfonts output at a specific row, line by line."""
|
||||
lines = output.rstrip('\n').split('\n')
|
||||
self._height_lines = len(lines)
|
||||
buf = []
|
||||
for i, line in enumerate(lines):
|
||||
buf.append(_move(start_row + i) + _clear_line() + line)
|
||||
self._write(''.join(buf))
|
||||
return start_row + len(lines)
|
||||
|
||||
def render_full(self, data):
|
||||
"""Clear screen and render all regions."""
|
||||
_clear_screen()
|
||||
self._write(_hide_cursor())
|
||||
self.term_width, self.term_height = shutil.get_terminal_size((80, 24))
|
||||
|
||||
output = _render_block_height(data.block_height, self.settings)
|
||||
self._last_rendered_height = data.block_height
|
||||
|
||||
if self._is_zen():
|
||||
# Center vertically
|
||||
lines = output.rstrip('\n').split('\n')
|
||||
start = max(1, (self.term_height - len(lines)) // 2)
|
||||
self._render_cfonts_at(output, start)
|
||||
return
|
||||
|
||||
# Region 1: Block height (row 2)
|
||||
next_row = self._render_cfonts_at(output, 2)
|
||||
|
||||
# Region 2: Block info
|
||||
next_row = self._render_info(data, next_row + 1)
|
||||
|
||||
# Region 3+: Widgets
|
||||
self._render_widgets(data, next_row + 1)
|
||||
|
||||
def _render_info(self, data, row):
|
||||
"""Render block size and tx count line."""
|
||||
if data.block_size > 0:
|
||||
size_mb = data.block_size / 1_000_000
|
||||
info = f" \033[0;37;40m{size_mb:.2f} MB · {data.block_tx_count} txs"
|
||||
center_pad = max(0, (self.term_width - len(info) + 20) // 2)
|
||||
self._write(_move(row) + _clear_line() + ' ' * center_pad + info)
|
||||
return row + 1
|
||||
return row
|
||||
|
||||
def _render_widgets(self, data, start_row):
|
||||
"""Render all enabled widget overlays."""
|
||||
row = start_row
|
||||
s = self.settings
|
||||
w = self.term_width
|
||||
|
||||
if s.get('show_countdown', True):
|
||||
text = render_countdown(data.seconds_since_block, w)
|
||||
self._write(_move(row) + _clear_line() + text)
|
||||
row += 2
|
||||
|
||||
if s.get('show_epoch_bar', True):
|
||||
text = render_epoch_bar(
|
||||
data.block_height, data.epoch_block,
|
||||
data.blocks_to_halving, data.next_halving_block, w
|
||||
)
|
||||
self._write(_move(row) + _clear_line() + text)
|
||||
row += 2
|
||||
|
||||
if s.get('show_fee_rates', True):
|
||||
text = render_fees(
|
||||
data.fee_fastest, data.fee_half_hour, data.fee_hour, w
|
||||
)
|
||||
self._write(_move(row) + _clear_line() + text)
|
||||
row += 2
|
||||
|
||||
if s.get('show_sparkline', False) and data.hashrate_history:
|
||||
text = render_sparkline(
|
||||
data.hashrate_history, data.hashrate_current, w
|
||||
)
|
||||
self._write(_move(row) + _clear_line() + text)
|
||||
row += 2
|
||||
|
||||
if s.get('show_utc_time', False):
|
||||
text = render_utc_time(
|
||||
[s.get('colorA', 'green'), s.get('colorB', 'yellow')]
|
||||
)
|
||||
lines = text.rstrip('\n').split('\n')
|
||||
for i, line in enumerate(lines):
|
||||
self._write(_move(row + i) + _clear_line() + line)
|
||||
row += len(lines) + 1
|
||||
|
||||
if s.get('generative_art', False) and data.block_hash:
|
||||
art = hash_art(data.block_hash, min(60, w - 4), 6)
|
||||
lines = art.split('\n')
|
||||
for i, line in enumerate(lines):
|
||||
self._write(_move(row + i) + _clear_line() + line)
|
||||
row += len(lines) + 1
|
||||
|
||||
return row
|
||||
|
||||
def update_countdown(self, data):
|
||||
"""Update only the countdown timer (called every poll cycle)."""
|
||||
if self._is_zen() or not self.settings.get('show_countdown', True):
|
||||
return
|
||||
# Countdown is rendered right after block height + info line
|
||||
row = 2 + self._height_lines + 2
|
||||
text = render_countdown(data.seconds_since_block, self.term_width)
|
||||
self._write(_move(row) + _clear_line() + text)
|
||||
|
||||
def heartbeat(self, data):
|
||||
"""Toggle bold/dim on block height for breathing effect."""
|
||||
if self._is_zen() or not self.settings.get('heartbeat', True):
|
||||
return
|
||||
if self._last_rendered_height is None:
|
||||
return
|
||||
|
||||
self._heartbeat_step += 1
|
||||
output = _render_block_height(data.block_height, self.settings)
|
||||
lines = output.rstrip('\n').split('\n')
|
||||
|
||||
start_row = 2
|
||||
if self._is_zen():
|
||||
start_row = max(1, (self.term_height - len(lines)) // 2)
|
||||
|
||||
# Apply dim on odd steps
|
||||
wrapper = _dim if self._heartbeat_step % 2 == 1 else lambda x: x
|
||||
buf = []
|
||||
for i, line in enumerate(lines):
|
||||
buf.append(_move(start_row + i) + _clear_line() + wrapper(line))
|
||||
self._write(''.join(buf))
|
||||
|
||||
def on_new_block(self, data, animations_mod):
|
||||
"""Handle new block arrival: sound, animation, then full re-render."""
|
||||
play_sound(self.settings.get('sound', 'bell'))
|
||||
|
||||
anim = self.settings.get('animation', 'matrix')
|
||||
|
||||
# Check for milestone fireworks first
|
||||
if self.settings.get('fireworks', True):
|
||||
milestone = animations_mod.is_milestone_block(data.block_height)
|
||||
if milestone:
|
||||
animations_mod.fireworks_animation(
|
||||
self.term_width, self.term_height, duration=5.0
|
||||
)
|
||||
|
||||
if anim == 'matrix':
|
||||
animations_mod.mining_animation(duration=3.0)
|
||||
elif anim == 'odometer' and self._last_rendered_height is not None:
|
||||
animations_mod.odometer_transition(
|
||||
str(self._last_rendered_height),
|
||||
str(data.block_height),
|
||||
self.settings, 2
|
||||
)
|
||||
|
||||
self.render_full(data)
|
||||
|
||||
def cleanup(self):
|
||||
"""Restore terminal state."""
|
||||
self._write(_show_cursor() + "\033[0m")
|
||||
23
pybitblock/clock/sound.py
Normal file
23
pybitblock/clock/sound.py
Normal file
|
|
@ -0,0 +1,23 @@
|
|||
"""Configurable sound notifications for new blocks."""
|
||||
|
||||
import sys
|
||||
import time
|
||||
|
||||
|
||||
def play_sound(mode):
|
||||
"""Play sound notification based on mode setting.
|
||||
|
||||
mode: 'bell' (single beep), 'pattern' (rhythmic), 'silent' (nothing)
|
||||
"""
|
||||
if mode == 'silent':
|
||||
return
|
||||
elif mode == 'pattern':
|
||||
# Three short beeps
|
||||
for _ in range(3):
|
||||
sys.stdout.write('\a')
|
||||
sys.stdout.flush()
|
||||
time.sleep(0.15)
|
||||
else:
|
||||
# Default: single bell
|
||||
sys.stdout.write('\a')
|
||||
sys.stdout.flush()
|
||||
48
pybitblock/clock/sparkline.py
Normal file
48
pybitblock/clock/sparkline.py
Normal file
|
|
@ -0,0 +1,48 @@
|
|||
"""Hashrate sparkline using Unicode block characters."""
|
||||
|
||||
SPARK_CHARS = " \u2581\u2582\u2583\u2584\u2585\u2586\u2587\u2588"
|
||||
|
||||
|
||||
def render_sparkline(values, current_hashrate, term_width):
|
||||
"""Render a mini sparkline graph for hashrate history.
|
||||
|
||||
values: list of hashrate floats (last N data points)
|
||||
current_hashrate: current hashrate in H/s
|
||||
term_width: terminal width for centering
|
||||
"""
|
||||
if not values:
|
||||
return ""
|
||||
|
||||
mn = min(values)
|
||||
mx = max(values)
|
||||
rng = mx - mn if mx != mn else 1
|
||||
|
||||
spark = ""
|
||||
for v in values:
|
||||
idx = int((v - mn) / rng * (len(SPARK_CHARS) - 1))
|
||||
spark += SPARK_CHARS[idx]
|
||||
|
||||
# Format hashrate in human-readable units
|
||||
hr_str = _format_hashrate(current_hashrate)
|
||||
|
||||
text = f" \033[1;33;40mHashrate\033[0;37;40m {spark} {hr_str}"
|
||||
pad = max(0, (term_width - len(spark) - len(hr_str) - 12) // 2)
|
||||
return ' ' * pad + text
|
||||
|
||||
|
||||
def _format_hashrate(h):
|
||||
"""Format hashrate in appropriate unit."""
|
||||
if h <= 0:
|
||||
return "-- H/s"
|
||||
units = [
|
||||
(1e18, "EH/s"),
|
||||
(1e15, "PH/s"),
|
||||
(1e12, "TH/s"),
|
||||
(1e9, "GH/s"),
|
||||
(1e6, "MH/s"),
|
||||
(1e3, "KH/s"),
|
||||
]
|
||||
for threshold, unit in units:
|
||||
if h >= threshold:
|
||||
return f"{h / threshold:.1f} {unit}"
|
||||
return f"{h:.0f} H/s"
|
||||
76
pybitblock/clock/widgets.py
Normal file
76
pybitblock/clock/widgets.py
Normal file
|
|
@ -0,0 +1,76 @@
|
|||
"""Clock widget components: countdown, epoch bar, fees, UTC time."""
|
||||
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from cfonts import render
|
||||
|
||||
|
||||
def render_countdown(seconds_since_block, term_width):
|
||||
"""Render time since last block with color coding.
|
||||
|
||||
Green: <600s (10min), Yellow: 600-1200s, Red: >1200s.
|
||||
"""
|
||||
mins = seconds_since_block // 60
|
||||
secs = seconds_since_block % 60
|
||||
|
||||
if seconds_since_block < 600:
|
||||
color = "\033[1;32;40m" # green
|
||||
elif seconds_since_block < 1200:
|
||||
color = "\033[1;33;40m" # yellow
|
||||
else:
|
||||
color = "\033[1;31;40m" # red
|
||||
|
||||
text = f"{color} \u23f1 {mins}m {secs:02d}s since last block\033[0;37;40m"
|
||||
pad = max(0, (term_width - 35) // 2)
|
||||
return ' ' * pad + text
|
||||
|
||||
|
||||
def render_epoch_bar(block_height, epoch_block, blocks_to_halving,
|
||||
next_halving_block, term_width):
|
||||
"""Render difficulty epoch progress bar + halving info."""
|
||||
# Difficulty adjustment progress
|
||||
epoch_pct = (epoch_block / 2016) * 100
|
||||
bar_width = min(30, term_width - 40)
|
||||
filled = int(bar_width * epoch_block / 2016)
|
||||
empty = bar_width - filled
|
||||
blocks_left = 2016 - epoch_block
|
||||
|
||||
bar = f"\033[1;36;40m\u2593" * filled + f"\033[0;37;40m\u2591" * empty
|
||||
epoch_line = (
|
||||
f" \033[1;36;40mEpoch\033[0;37;40m [{bar}\033[0;37;40m] "
|
||||
f"{epoch_block}/2016 ({epoch_pct:.1f}%) "
|
||||
f"· {blocks_left} blocks to retarget"
|
||||
)
|
||||
|
||||
# Halving progress
|
||||
if next_halving_block > 0:
|
||||
halving_num = next_halving_block // 210_000
|
||||
halving_line = (
|
||||
f" \033[1;35;40mHalving #{halving_num}\033[0;37;40m "
|
||||
f"in {blocks_to_halving:,} blocks "
|
||||
f"(block {next_halving_block:,})"
|
||||
)
|
||||
return epoch_line + "\n" + halving_line
|
||||
|
||||
return epoch_line
|
||||
|
||||
|
||||
def render_fees(fastest, half_hour, hour, term_width):
|
||||
"""Render compact fee rate display."""
|
||||
if fastest == 0 and half_hour == 0 and hour == 0:
|
||||
return ""
|
||||
|
||||
text = (
|
||||
f" \033[1;31;40m\u26a1 {fastest}\033[0;37;40m | "
|
||||
f"\033[1;33;40m\u23f3 {half_hour}\033[0;37;40m | "
|
||||
f"\033[1;32;40m\u2623 {hour}\033[0;37;40m sat/vB"
|
||||
)
|
||||
pad = max(0, (term_width - 40) // 2)
|
||||
return ' ' * pad + text
|
||||
|
||||
|
||||
def render_utc_time(colors):
|
||||
"""Render current UTC time in tiny cfonts font."""
|
||||
now = datetime.now(timezone.utc).strftime("%H:%M")
|
||||
output = render(now, colors=colors, align='center', font='tiny')
|
||||
return output
|
||||
|
|
@ -18,7 +18,17 @@ import os
|
|||
_DEFAULT_PATH = {"ip_port": "", "rpcuser": "", "rpcpass": "", "bitcoincli": ""}
|
||||
_DEFAULT_LND = {"ip_port": "", "tls": "", "macaroon": "", "ln": ""}
|
||||
_DEFAULT_SETTINGS = {"gradient": "", "design": "block", "colorA": "green", "colorB": "yellow"}
|
||||
_DEFAULT_SETTINGS_CLOCK = {"gradient": "", "colorA": "green", "colorB": "yellow"}
|
||||
_DEFAULT_SETTINGS_CLOCK = {
|
||||
"gradient": "", "colorA": "green", "colorB": "yellow",
|
||||
"show_countdown": True, "show_epoch_bar": True,
|
||||
"show_fee_rates": True, "show_sparkline": False,
|
||||
"show_utc_time": False, "zen_mode": False,
|
||||
"animation": "matrix",
|
||||
"fireworks": True,
|
||||
"generative_art": False,
|
||||
"sound": "bell",
|
||||
"heartbeat": True,
|
||||
}
|
||||
|
||||
|
||||
def _env_bitcoin_config():
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue