Linea: Real-time transaction activity monitor with Python

import asyncio, math
from web3 import AsyncHTTPProvider, AsyncWeb3
from web3.middleware import ExtraDataToPOAMiddleware
from rich.console import Console
from rich.text import Text

# Chainstack Linea mainnet endpoint
RPC = "YOUR_CHAINSTACK_ENDPOINT"

w3 = AsyncWeb3(AsyncHTTPProvider(RPC))
# Add POA middleware for Linea
w3.middleware_onion.inject(ExtraDataToPOAMiddleware, layer=0)
c  = Console()

# Tweak these bands to taste (transaction counts)
BANDS = [5, 20, 50, 100]           # quiet > busy (transactions per block)
BAR_LEN = 30                       # number of "pixels"

def mood_color(tx_count: int) -> str:
    if tx_count < BANDS[0]: return "green"
    if tx_count < BANDS[1]: return "bright_green"
    if tx_count < BANDS[2]: return "yellow"
    if tx_count < BANDS[3]: return "orange"
    return "red"

def mood_bar(tx_count: int) -> Text:
    # % of BAR_LEN that should be "hot"
    pct = min(tx_count / BANDS[-1], 1.0)
    hot = math.ceil(pct * BAR_LEN)
    bar = Text()
    for i in range(BAR_LEN):
        color = "white"
        if i < hot:
            color = mood_color(tx_count)
        bar.append("█", style=color)
    return bar

async def poll():
    # Get initial block number
    last_block_number = None
    
    while True:
        # Get latest block header
        blk = await w3.eth.get_block("latest")
        
        # Only process if it's a new block
        if last_block_number is None or blk.number > last_block_number:
            last_block_number = blk.number
            
            # Count transactions in this block
            tx_count = len(blk.transactions)
                
            line = Text.assemble(
                ("blk ", "cyan"), (str(blk.number).ljust(8), "bold cyan"),
                (" | txs ", "grey50"), (f"{tx_count}".ljust(11), "bold"),
                (" | ", "grey50"), mood_bar(tx_count)
            )
            yield line
        
        # Check for new blocks more frequently (Linea has ~2 second block times)
        await asyncio.sleep(1)

async def main():
    c.print("[dim]Powered by Chainstack—spin up your own Linea endpoint in 30s[/]")
    c.print("[bold cyan]🎯 Linea transaction activity monitor[/]")
    c.print()
    
    async for txt in poll():
        c.print(txt)

asyncio.run(main())