mirror of
https://github.com/samjage/metro-warden.git
synced 2026-06-06 02:40:43 +00:00
108 lines
3.1 KiB
Python
108 lines
3.1 KiB
Python
"""
|
|
Lines Tab — network interfaces overview.
|
|
|
|
Displays a DataTable of all network interfaces updated live via bus subscription
|
|
to the "network.interfaces" topic.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Dict
|
|
|
|
from textual.app import ComposeResult
|
|
from textual.reactive import reactive
|
|
from textual.widget import Widget
|
|
from textual.widgets import DataTable, Label, Static
|
|
|
|
|
|
def _fmt_bytes(n: int) -> str:
|
|
"""Format a byte count as a human-readable string."""
|
|
for unit in ("B", "KB", "MB", "GB", "TB"):
|
|
if n < 1024:
|
|
return f"{n:.1f} {unit}"
|
|
n /= 1024
|
|
return f"{n:.1f} PB"
|
|
|
|
|
|
class LinesTab(Widget):
|
|
"""
|
|
Network Interfaces tab.
|
|
|
|
Subscribes to ``network.interfaces`` bus events and refreshes
|
|
the DataTable in real time.
|
|
"""
|
|
|
|
DEFAULT_CSS = """
|
|
LinesTab {
|
|
layout: vertical;
|
|
height: 1fr;
|
|
padding: 1 2;
|
|
}
|
|
LinesTab .tab-title {
|
|
color: $primary;
|
|
text-style: bold;
|
|
height: 1;
|
|
margin-bottom: 1;
|
|
}
|
|
LinesTab DataTable {
|
|
height: 1fr;
|
|
border: round $primary;
|
|
}
|
|
"""
|
|
|
|
COLUMNS = [
|
|
("Interface", 12),
|
|
("Status", 8),
|
|
("IPv4", 16),
|
|
("IPv6", 24),
|
|
("MAC", 18),
|
|
("Speed", 9),
|
|
("RX", 12),
|
|
("TX", 12),
|
|
("RX Err", 8),
|
|
("TX Err", 8),
|
|
]
|
|
|
|
def compose(self) -> ComposeResult:
|
|
yield Static("// LINES — Network Interfaces", classes="tab-title")
|
|
yield DataTable(id="lines-table", zebra_stripes=True, cursor_type="row")
|
|
|
|
def on_mount(self) -> None:
|
|
table = self.query_one("#lines-table", DataTable)
|
|
for col_label, width in self.COLUMNS:
|
|
table.add_column(col_label, width=width, key=col_label.lower().replace(" ", "_"))
|
|
|
|
# Subscribe to the bus if app exposes one
|
|
app = self.app
|
|
if hasattr(app, "bus"):
|
|
app.bus.subscribe("network.interfaces", self._on_interfaces)
|
|
# Request immediate data
|
|
if hasattr(app, "state"):
|
|
interfaces = app.state.get("network.interfaces")
|
|
if interfaces:
|
|
self._refresh_table(interfaces)
|
|
|
|
def _on_interfaces(self, topic: str, data: Any) -> None:
|
|
"""Bus handler — schedule table refresh on the UI thread."""
|
|
self.call_from_thread(self._refresh_table, data)
|
|
|
|
def _refresh_table(self, interfaces: Dict[str, Dict]) -> None:
|
|
table = self.query_one("#lines-table", DataTable)
|
|
table.clear()
|
|
for iface, info in sorted(interfaces.items()):
|
|
status_text = info.get("status", "?")
|
|
speed = info.get("speed", 0)
|
|
speed_str = f"{speed} Mb" if speed else "—"
|
|
table.add_row(
|
|
iface,
|
|
status_text,
|
|
info.get("ip4", "") or "—",
|
|
info.get("ip6", "") or "—",
|
|
info.get("mac", "") or "—",
|
|
speed_str,
|
|
_fmt_bytes(info.get("rx_bytes", 0)),
|
|
_fmt_bytes(info.get("tx_bytes", 0)),
|
|
str(info.get("rx_errors", 0)),
|
|
str(info.get("tx_errors", 0)),
|
|
)
|