Python Rich — smukke CLI-tools til IoT-platform administration

rich giver professionelt formateret terminal-output til CLI-tools — fra farverige tabeller til live progress bars. Her er implementeringen til IoT-platform-administration.

---

Rich fundamentals

# pip install rich typer

from rich.console import Console
from rich.table import Table
from rich.panel import Panel
from rich.progress import Progress, SpinnerColumn, TimeElapsedColumn
from rich.syntax import Syntax
from rich import print as rprint

console = Console()

# Simpel formateret output:
console.print("[bold green]Gateway GW-0001[/bold green] online", style="")
console.print("[red]FEJL:[/red] AES-nøgle mangler for meter 12345678")
console.print(Panel("Gateway status rapport", style="cyan"))

# Rich markup i f-strings:
gateway_id = "GW-0001"
rssi = -87
color = "green" if rssi > -95 else "yellow" if rssi > -105 else "red"
rprint(f"[bold]{gateway_id}[/bold] RSSI: [{color}]{rssi} dBm[/{color}]")

---

Gateway-status tabel

# server/scripts/show_gateway_status.py

import asyncio
from rich.table import Table
from rich.console import Console
from datetime import datetime, timezone

console = Console()


async def show_gateway_table():
    """Vis alle gateways med status i pæn tabel."""
    gateways = await fetch_all_gateways()

    table = Table(
        title="Gateway Fleet Status",
        show_header=True,
        header_style="bold magenta",
    )

    table.add_column("Gateway ID", style="cyan", no_wrap=True)
    table.add_column("Ejendom")
    table.add_column("Sidst set", justify="right")
    table.add_column("Firmware", justify="center")
    table.add_column("Målere", justify="right")
    table.add_column("Status", justify="center")

    for gw in gateways:
        now = datetime.now(timezone.utc)
        hours_ago = (now - gw.last_seen_at).total_seconds() / 3600 if gw.last_seen_at else 9999

        if hours_ago < 1:
            status = "[green]● Online[/green]"
            last_seen = f"{int(hours_ago * 60)} min siden"
        elif hours_ago < 36:
            status = "[yellow]◐ Stale[/yellow]"
            last_seen = f"{int(hours_ago)} t siden"
        else:
            status = "[red]○ Offline[/red]"
            last_seen = f"{int(hours_ago / 24)} dage siden"

        table.add_row(
            gw.gateway_id,
            gw.property_name or "–",
            last_seen,
            gw.firmware_version or "ukendt",
            str(gw.meter_count),
            status,
        )

    console.print(table)
    console.print(f"\n[dim]Total: {len(gateways)} gateways[/dim]")


asyncio.run(show_gateway_table())

---

Typer CLI + Rich

# server/scripts/mbus_admin.py
# Kør: python -m server.scripts.mbus_admin --help

import typer
from rich.console import Console
from rich.prompt import Confirm, Prompt
from typing import Optional

app = typer.Typer(help="M-Bus Gateway administrationstool")
console = Console()


@app.command()
def gateway_status(
    tenant_id: Optional[str] = typer.Option(None, help="Filtrer på tenant"),
    show_offline: bool = typer.Option(False, "--offline", help="Kun offline gateways"),
):
    """Vis gateway-fleet status."""
    import asyncio
    asyncio.run(_gateway_status(tenant_id, show_offline))


@app.command()
def create_superadmin(
    email: str = typer.Argument(..., help="Super admin email"),
    name: str = typer.Option(..., prompt=True, help="Navn"),
):
    """Opret ny super admin bruger."""
    password = Prompt.ask("Adgangskode", password=True)
    confirm = Prompt.ask("Bekræft adgangskode", password=True)

    if password != confirm:
        console.print("[red]Adgangskoder matcher ikke[/red]")
        raise typer.Exit(1)

    if not Confirm.ask(f"Opret super admin '{email}'?"):
        raise typer.Abort()

    # ... opret bruger
    console.print(f"[green]✓[/green] Super admin [bold]{email}[/bold] oprettet")


@app.command()
def send_ota(
    gateway_id: str = typer.Argument(...),
    version: str = typer.Argument(...),
    dry_run: bool = typer.Option(False, "--dry-run"),
):
    """Send OTA-opdatering til gateway."""
    console.print(Panel(
        f"[bold]OTA Update[/bold]\nGateway: {gateway_id}\nVersion: {version}",
        style="yellow"
    ))

    if not dry_run:
        if not Confirm.ask("Send OTA?", default=False):
            raise typer.Abort()
        # ... send OTA
        console.print("[green]✓[/green] OTA-kommando sendt")
    else:
        console.print("[dim]Dry run — ingen kommando sendt[/dim]")


if __name__ == "__main__":
    app()

---

Progress bar til batch-operationer

# Batch-afregning med live progress:

from rich.progress import (
    Progress, BarColumn, TextColumn,
    TimeElapsedColumn, TimeRemainingColumn,
    SpinnerColumn,
)

async def generate_all_settlements(property_ids: list[str]) -> dict:
    results = {"success": 0, "failed": 0, "errors": []}

    with Progress(
        SpinnerColumn(),
        TextColumn("[progress.description]{task.description}"),
        BarColumn(),
        TextColumn("[progress.percentage]{task.percentage:>3.0f}%"),
        TimeElapsedColumn(),
        TimeRemainingColumn(),
        console=console,
    ) as progress:
        task = progress.add_task(
            f"Genererer afregninger...",
            total=len(property_ids),
        )

        for pid in property_ids:
            progress.update(task, description=f"Ejendom {pid[:8]}...")
            try:
                await generate_settlement(pid)
                results["success"] += 1
            except Exception as e:
                results["failed"] += 1
                results["errors"].append(f"{pid}: {e}")
            finally:
                progress.advance(task)

    # Sammenfatning:
    console.print(f"\n[green]✓ {results['success']} afregninger genereret[/green]")
    if results["failed"]:
        console.print(f"[red]✗ {results['failed']} fejlede[/red]")
        for err in results["errors"]:
            console.print(f"  [dim]{err}[/dim]")

    return results

---

JSON syntax highlighting

# Vis gateway-konfiguration med syntax highlighting:

import json
from rich.syntax import Syntax

async def show_gateway_config(gateway_id: str):
    config = await fetch_gateway_config(gateway_id)
    config_json = json.dumps(config, indent=2, ensure_ascii=False)

    syntax = Syntax(config_json, "json", theme="monokai", line_numbers=True)
    console.print(Panel(syntax, title=f"Config: {gateway_id}"))


# MQTT-telegram debugging:
async def debug_telegram(raw_hex: str):
    parsed = parse_wmbus_telegram(raw_hex)
    syntax = Syntax(
        json.dumps(parsed, indent=2, default=str),
        "json",
        theme="github-dark",
    )
    console.print(syntax)

---

Konklusion

rich + typer er standardkombinationen til professionelle Python CLI-tools. Farverige tabeller med gateway-status, live progress bars til batch-afregning og JSON syntax highlighting til debugging giver en markant bedre admin-oplevelse end rå print. Prompt.ask og Confirm.ask erstatter input() med validering. Scripts køres direkte med python -m server.scripts.mbus_admin og kan pakkes som uv tool install.

Se GitHub Actions CI guide eller FastAPI testing guide.