Pythonで構築する次世代TUI：Textualによる実践ターミナルUI開発ガイド

# 本体および標準Widgetのインストール
pip install textual

# 開発用ツール（デバッグコンソール等）のインストール
pip install textual-dev

textual console

textual run --dev main.py

from textual.app import App, ComposeResult
from textual.widgets import Header, Footer, Button, Static
from textual.reactive import reactive
from textual.containers import Horizontal, Vertical

class CounterWidget(Static):
    """カウンターを表示・操作するカスタムWidget"""

    # リアクティブプロパティ。値が変わると自動的にUIが更新される
    count = reactive(0)

    def watch_count(self, old_value: int, new_value: int) -> None:
        """countの値が変更された時に呼び出されるウォッチャー"""
        self.update(f"現在のカウント: [bold cyan]{new_value}[/bold cyan]")

    def compose(self) -> ComposeResult:
        """子Widgetを配置する"""
        yield Static(f"現在のカウント: {self.count}", id="display")
        yield Horizontal(
            Button("インクリメント (+1)", id="add", variant="success"),
            Button("デクリメント (-1)", id="sub", variant="error"),
            classes="button-bar"
        )

    def on_button_pressed(self, event: Button.Pressed) -> None:
        """ボタンが押された時のイベントハンドラ"""
        if event.button.id == "add":
            self.count += 1
        elif event.button.id == "sub":
            self.count -= 1


class CounterApp(App):
    """メインアプリケーションクラス"""

    # アプリ全体のキーバインド定義
    BINDINGS = [
        ("q", "quit", "終了"),
        ("r", "reset", "リセット"),
    ]

    # インラインでTCSS（Textual CSS）を記述
    CSS = """
    Screen {
        align: center middle;
        background: $surface;
    }

    CounterWidget {
        width: 60;
        height: auto;
        border: solid green;
        padding: 1 2;
        background: $panel;
    }

    .button-bar {
        margin-top: 1;
        height: auto;
        align: center middle;
    }

    Button {
        margin: 0 1;
    }
    """

    def compose(self) -> ComposeResult:
        """画面全体のレイアウト"""
        yield Header(show_clock=True)
        yield CounterWidget()
        yield Footer()

    def action_reset(self) -> None:
        """BINDINGSで定義した'reset'アクションの実装"""
        counter = self.query_one(CounterWidget)
        counter.count = 0


if __name__ == "__main__":
    app = CounterApp()
    app.run()

pip install psutil

import asyncio
import random
from datetime import datetime
import psutil

from textual.app import App, ComposeResult
from textual.containers import Container, Horizontal, Vertical, Grid
from textual.widgets import Header, Footer, Static, ProgressBar, DataTable, RichLog, Button
from textual.worker import Worker, WorkerState
from textual.reactive import reactive


class ResourceMonitor(Static):
    """CPUとメモリ使用率を表示するWidget"""

    def compose(self) -> ComposeResult:
        yield Static("[bold]CPU 使用率:[/bold]", classes="label")
        yield ProgressBar(total=100, show_percentage=True, id="cpu_bar")
        yield Static("[bold]メモリ 使用率:[/bold]", classes="label")
        yield ProgressBar(total=100, show_percentage=True, id="mem_bar")

    def update_metrics(self, cpu: float, mem: float) -> None:
        """プログレスバーの更新"""
        self.query_one("#cpu_bar", ProgressBar).progress = cpu
        self.query_one("#mem_bar", ProgressBar).progress = mem


class ProcessTable(DataTable):
    """実行中プロセスの一覧を表示するテーブルWidget"""

    def on_mount(self) -> None:
        """Widget配置時の初期設定"""
        self.cursor_type = "row"
        self.add_columns("PID", "プロセス名", "CPU (%)", "メモリ (%)")

    def update_processes() -> None:
        """プロセス情報の更新（外部から呼び出し）"""
        pass  # メインApp側で一括制御


class SystemMonitorApp(App):
    """システム監視ダッシュボード・メインアプリ"""

    TITLE = "OKPy System Monitor & Log Analyzer"
    SUB_TITLE = "Textual Practical Example"

    BINDINGS = [
        ("q", "quit", "終了"),
        ("c", "clear_logs", "ログ消去"),
        ("p", "toggle_pause", "一時停止/再開"),
    ]

    CSS = """
    Screen {
        layout: grid;
        grid-size: 2 2;
        grid-rows: 1fr 2fr;
        grid-columns: 1fr 1fr;
        padding: 1;
        grid-gutter: 1;
    }

    #left_top {
        border: round $accent;
        padding: 1;
        background: $panel;
    }

    #right_top {
        border: round $success;
        padding: 1;
        background: $panel;
    }

    #bottom_panel {
        column-span: 2;
        border: round $primary;
        background: $panel;
    }

    .label {
        margin-top: 1;
        margin-bottom: 0;
    }

    ProgressBar {
        width: 100%;
    }

    DataTable {
        height: 100%;
    }

    RichLog {
        height: 100%;
        background: $surface-down;
    }
    """

    is_paused = reactive(False)

    def compose(self) -> ComposeResult:
        yield Header(show_clock=True)

        # 左上：リソースプログレスバー
        with Vertical(id="left_top"):
            yield Static("[bold cyan]■ リソース使用状況[/bold cyan]")
            yield ResourceMonitor(id="resource_monitor")

        # 右上：クイックアクションとステータス
        with Vertical(id="right_top"):
            yield Static("[bold green]■ 制御・ステータス[/bold green]")
            yield Static("ステータス: [bold green]監視中[/bold green]", id="status_text")
            yield Horizontal(
                Button("一時停止 / 再開", id="btn_pause", variant="primary"),
                Button("ログクリア", id="btn_clear", variant="warning"),
                classes="label"
            )

        # 下部：プロセス一覧とログビューアを横並び
        with Horizontal(id="bottom_panel"):
            with Vertical(id="table_wrapper"):
                yield Static("[bold yellow]■ 上位プロセス (CPU順)[/bold yellow]")
                yield ProcessTable(id="process_table")
            with Vertical(id="log_wrapper"):
                yield Static("[bold magenta]■ システムログ・ストリーム[/bold magenta]")
                yield RichLog(highlight=True, markup=True, id="sys_log")

        yield Footer()

    def on_mount(self) -> None:
        """アプリ起動時の初期化処理（定期タイマーの設定）"""
        # 1秒ごとにシステムメトリクスを更新
        self.set_interval(1.0, self.refresh_system_data)
        # 1.5秒ごとに疑似ログを出力
        self.set_interval(1.5, self.generate_dummy_log)

    def refresh_system_data(self) -> None:
        """システム情報の取得と画面更新（タイマーコールバック）"""
        if self.is_paused:
            return

        # 1. CPU・メモリの更新
        cpu_percent = psutil.cpu_percent()
        mem_percent = psutil.virtual_memory().percent

        res_mon = self.query_one("#resource_monitor", ResourceMonitor)
        res_mon.update_metrics(cpu_percent, mem_percent)

        # 2. プロセス一覧の更新
        table = self.query_one("#process_table", ProcessTable)
        table.clear()

        processes = []
        for proc in psutil.process_iter(['pid', 'name', 'cpu_percent', 'memory_percent']):
            try:
                processes.append(proc.info)
            except (psutil.NoSuchProcess, psutil.AccessDenied):
                pass

        # CPU使用率順にソートして上位5件を表示
        top_processes = sorted(processes, key=lambda p: p['cpu_percent'] or 0, reverse=True)[:5]

        for p in top_processes:
            table.add_row(
                str(p['pid']),
                p['name'][:15],
                f"{p['cpu_percent']:.1f}",
                f"{p['memory_percent']:.1f}"
            )

    def generate_dummy_log(self) -> None:
        """疑似ログの追記"""
        if self.is_paused:
            return

        log_widget = self.query_one("#sys_log", RichLog)
        now = datetime.now().strftime("%H:%M:%S")

        log_levels = [
            ("[green]INFO[/green]", "データベースへの接続が正常に完了しました。"),
            ("[green]INFO[/green]", "APIリクエストを受信: GET /api/v1/status (200 OK)"),
            ("[yellow]WARN[/yellow]", "メモリ使用量が80%を超過しつつあります。"),
            ("[red]ERROR[/red]", "外部決済APIへの接続タイムアウトが発生しました (再試行中)"),
            ("[green]INFO[/green]", "バックグラウンドクリーンアップジョブが終了しました。")
        ]

        level, msg = random.choice(log_levels)
        log_widget.write(f"[{now}] {level} - {msg}")

    def action_toggle_pause(self) -> None:
        """監視のポーズ／再開切り替え"""
        self.is_paused = not self.is_paused
        status_text = self.query_one("#status_text", Static)

        if self.is_paused:
            status_text.update("ステータス: [bold yellow]一時停止中[/bold yellow]")
            self.notify("監視を一時停止しました", severity="warning")
        else:
            status_text.update("ステータス: [bold green]監視中[/bold green]")
            self.notify("監視を再開しました", severity="information")

    def action_clear_logs(self) -> None:
        """ログビューアのクリア"""
        log_widget = self.query_one("#sys_log", RichLog)
        log_widget.clear()
        self.notify("ログを消去しました")

    def on_button_pressed(self, event: Button.Pressed) -> None:
        """ボタン押下時の処理"""
        if event.button.id == "btn_pause":
            self.action_toggle_pause()
        elif event.button.id == "btn_clear":
            self.action_clear_logs()


if __name__ == "__main__":
    app = SystemMonitorApp()
    app.run()

from textual.worker import work

class MyApp(App):
    @work(exclusive=True, thread=True)
    def fetch_heavy_data(self) -> dict:
        # 重い同期処理（例: requestsを使った通信や巨大ファイルの読み込み）
        import requests
        response = requests.get("https://api.example.com/large-data")
        return response.json()

    async def on_button_pressed(self, event: Button.Pressed) -> None:
        # バックグラウンド処理を開始（UIはブロックされない）
        self.fetch_heavy_data()

pip install textual-web
textual-web --config config.toml

import pytest
from my_app import CounterApp

@pytest.mark.asyncio
async def test_counter_app():
    app = CounterApp()
    # run_test()でパイロットモードを起動
    async with app.run_test() as pilot:
        # ボタン押下をシミュレート
        await pilot.click("#add")
        # リアクティブ変数の検証
        counter = app.query_one("CounterWidget")
        assert counter.count == 1