OVOS live bus monitor utility

busmon.py

import threading

from ovos_bus_client import MessageBusClient, Message
from ovos_bus_client.session import SessionManager
from rich.console import Console
from rich.table import Table

console = Console()


def bus_monitor():
    client = MessageBusClient()
    lock = threading.Lock()

    def echo(message: str):
        with lock:
            m: Message = Message.deserialize(message)
            sess = SessionManager.get(m)

            # Display message type
            console.print(f"[bold cyan]* {m.msg_type}[/bold cyan]")

            # Display session info
            if m.context:
                console.print(f"[bold green]SESSION:[/bold green] {sess.session_id}")
                context_table = Table(title="CONTEXT", title_style="bold green")

                for k, v in m.context.items():
                    if k == "session":
                        continue
                    context_table.add_row(str(k), str(v))
                console.print(context_table)

            # Display data in a table
            if m.data:
                data_table = Table(title="DATA", title_style="bold magenta")

                for k, v in m.data.items():
                    data_table.add_row(str(k), str(v))
                console.print(data_table)

    client.on("message", echo)

    try:
        console.print("[bold green]Monitoring the message bus... Press Ctrl+C to exit.[/bold green]")
        client.run_forever()
    except KeyboardInterrupt:
        console.print("\n[bold red]Exiting bus monitor.[/bold red]")
    finally:
        client.close()


if __name__ == "__main__":
    bus_monitor()