Polkadot: Real-time network health monitoring dashboard

import time
from datetime import datetime
from typing import Dict, Optional
import signal
import sys

from substrateinterface import SubstrateInterface
from rich.console import Console
from rich.panel import Panel
from rich.layout import Layout
from rich.live import Live
from rich.align import Align
import click

console = Console()

class PolkadotHealthMonitor:  
    def __init__(self, endpoint: str):
        self.endpoint = endpoint
        self.substrate = None
        self.running = False
        self.start_time = datetime.now()
        self.blocks_processed = 0
        
    def connect(self) -> bool:
        """Connect to Polkadot network"""
        try:
            with console.status("[bold green]Connecting..."):
                self.substrate = SubstrateInterface(
                    url=self.endpoint,
                    ss58_format=0,
                    type_registry_preset='polkadot'
                )
                chain_info = self.substrate.get_chain_finalised_head()
                
            console.print(f"✅ Connected to {self.substrate.chain}")
            console.print(f"🔗 Latest finalized: {chain_info}")
            return True
            
        except Exception as e:
            console.print(f"❌ Connection failed: {e}", style="bold red")
            return False
    
    def safe_query(self, pallet: str, method: str, params: list = None) -> Optional[any]:
        """Safely execute substrate queries with error handling"""
        try:
            result = self.substrate.query(pallet, method, params or [])
            return result.value if result else None
        except:
            return None
    
    def get_all_data(self) -> Dict:
        """Fetch all network data in one consolidated method"""
        data = {'timestamp': datetime.now()}
        
        try:
            # Network info
            latest_hash = self.substrate.get_chain_head()
            finalized_hash = self.substrate.get_chain_finalised_head()
            latest_block = self.substrate.get_block(latest_hash)
            finalized_block = self.substrate.get_block(finalized_hash)
            
            latest_num = latest_block['header']['number'] if latest_block and 'header' in latest_block else 0
            finalized_num = finalized_block['header']['number'] if finalized_block and 'header' in finalized_block else 0
            
            data.update({
                'chain': str(self.substrate.chain),
                'latest_block': latest_num,
                'finalized_block': finalized_num,
                'finalization_lag': latest_num - finalized_num,
                'block_hash': str(latest_hash)[:12] + '...' if latest_hash else 'N/A'
            })
            
            # Runtime version
            try:
                runtime = self.substrate.get_runtime_version()
                data.update({
                    'runtime_version': runtime.get('specVersion', 'N/A') if runtime else 'N/A',
                    'runtime_name': runtime.get('specName', 'N/A') if runtime else 'N/A'
                })
            except Exception as e:
                # Fallback: try to get from metadata
                try:
                    metadata = self.substrate.get_metadata()
                    runtime_version = getattr(metadata, 'runtime_version', None)
                    if runtime_version:
                        data.update({
                            'runtime_version': str(runtime_version.get('spec_version', 'N/A')),
                            'runtime_name': str(runtime_version.get('spec_name', 'Polkadot'))
                        })
                    else:
                        # Fallback: show we have connection
                        data.update({'runtime_version': 'Connected', 'runtime_name': 'Polkadot'})
                except:
                    data.update({'runtime_version': 'Connected', 'runtime_name': 'Polkadot'})
            
            # Validators and staking
            validators = self.safe_query('Session', 'Validators') or []
            validator_count = len(validators)
            current_era = self.safe_query('Staking', 'CurrentEra') or 0
            total_issuance = self.safe_query('Balances', 'TotalIssuance') or 0
            active_validators = self.safe_query('Staking', 'CounterForValidators') or validator_count
            
            data.update({
                'validators': validator_count,
                'active_validators': active_validators,
                'current_era': current_era,
                'total_supply': f"{total_issuance / 10**10:,.0f} DOT" if total_issuance else "N/A"
            })
            
            # Governance
            referendum_count = (self.safe_query('Referenda', 'ReferendumCount') or 
                              self.safe_query('Democracy', 'ReferendumCount') or 0)
            council_members = self.safe_query('Council', 'Members') or []
            fellowship = self.safe_query('FellowshipCollective', 'Members') or []
            proposals = self.safe_query('Democracy', 'PublicProps') or []
            
            data.update({
                'referendums': referendum_count,
                'council_size': len(council_members) or len(fellowship),
                'proposals': len(proposals),
                'governance': 'OpenGov v2' if referendum_count > 0 else 'Legacy'
            })
            
            self.blocks_processed += 1
            
        except Exception as e:
            console.print(f"❌ Data fetch error: {e}")
            
        return data
    
    def create_panel(self, title: str, content: str, style: str = "green") -> Panel:
        """Create a formatted panel"""
        return Panel(content.strip(), title=title, border_style=style)
    
    def create_dashboard(self) -> Layout:
        """Create compact dashboard layout"""
        data = self.get_all_data()
        
        # Network panel
        network_content = f"""[bold green]Chain:[/bold green] {data.get('chain', 'N/A')}
[bold blue]Latest:[/bold blue] #{data.get('latest_block', 'N/A')}
[bold yellow]Finalized:[/bold yellow] #{data.get('finalized_block', 'N/A')}
[bold red]Lag:[/bold red] {data.get('finalization_lag', 'N/A')} blocks
[bold magenta]Runtime:[/bold magenta] {data.get('runtime_version', 'N/A')}"""
        
        # Validators panel
        validator_content = f"""[bold green]Total:[/bold green] {data.get('validators', 'N/A')}
[bold blue]Active:[/bold blue] {data.get('active_validators', 'N/A')}
[bold yellow]Supply:[/bold yellow] {data.get('total_supply', 'N/A')}"""
        
        # Staking panel
        staking_content = f"""[bold green]Era:[/bold green] {data.get('current_era', 'N/A')}
[bold blue]Duration:[/bold blue] 24 hours
[bold yellow]Validators:[/bold yellow] {data.get('validators', 'N/A')}"""
        
        # Governance panel
        gov_content = f"""[bold green]Referendums:[/bold green] {data.get('referendums', 'N/A')}
[bold blue]Council:[/bold blue] {data.get('council_size', 'N/A')}
[bold yellow]Proposals:[/bold yellow] {data.get('proposals', 'N/A')}
[bold magenta]System:[/bold magenta] {data.get('governance', 'Unknown')}"""
        
        # Status panel
        uptime = datetime.now() - self.start_time
        status_content = f"""[bold green]Status:[/bold green] {'🟢 Running' if self.running else '🔴 Stopped'}
[bold blue]Uptime:[/bold blue] {str(uptime).split('.')[0]}
[bold yellow]Blocks:[/bold yellow] {self.blocks_processed}
[bold magenta]Updated:[/bold magenta] {datetime.now().strftime('%H:%M:%S')}"""
        
        # Create layout
        layout = Layout()
        layout.split_column(
            Layout(Panel(Align.center("🟡 Polkadot Network Health Monitor 📊", style="bold yellow"), 
                        style="bold blue"), size=3),
            Layout(name="body"),
            Layout(self.create_panel("📊 Status", status_content, "cyan"), size=7)
        )
        
        layout["body"].split_row(
            Layout(name="left"),
            Layout(name="right")
        )
        
        layout["left"].split_column(
            self.create_panel("🌐 Network", network_content),
            self.create_panel("👥 Validators", validator_content, "blue")
        )
        
        layout["right"].split_column(
            self.create_panel("🏦 Staking", staking_content, "yellow"),
            self.create_panel("🗳️ Governance", gov_content, "magenta")
        )
        
        return layout
    
    def start_monitoring(self, refresh_interval: int = 10):
        """Start monitoring with graceful shutdown"""
        if not self.connect():
            return
        
        self.running = True
        signal.signal(signal.SIGINT, lambda s, f: self._shutdown())
        
        console.print(f"🚀 Starting monitor (refresh: {refresh_interval}s)")
        console.print("💡 Press Ctrl+C to stop\n")
        
        try:
            with Live(self.create_dashboard(), refresh_per_second=1/refresh_interval, screen=True) as live:
                while self.running:
                    live.update(self.create_dashboard())
                    time.sleep(refresh_interval)
        except KeyboardInterrupt:
            self._shutdown()
        except Exception as e:
            console.print(f"\n❌ Error: {e}")
        finally:
            self.running = False
    
    def _shutdown(self):
        """Graceful shutdown"""
        self.running = False
        console.print("\n👋 Monitor stopped")
        sys.exit(0)

@click.command()
@click.option('--endpoint', '-e', 
              default='YOUR_CHAINSTACK_WSS_ENDPOINT',
              help='Chainstack endpoint')
@click.option('--refresh', '-r', default=10, help='Refresh interval (seconds)')
def main(endpoint: str, refresh: int):
    """🟡 Polkadot Network Health Monitor 📊"""
    
    console.print(Panel.fit(
        """🟡 [bold yellow]Polkadot Network Health Monitor[/bold yellow] 📊

Real-time monitoring of:
• Network statistics & block production
• Validator performance metrics  
• Staking analysis & governance activity

[italic]Powered by Chainstack infrastructure[/italic]""",
        border_style="bold blue"
    ))
    
    monitor = PolkadotHealthMonitor(endpoint)
    monitor.start_monitoring(refresh_interval=refresh)

if __name__ == "__main__":
    main()