diff --git a/pyproject.toml b/pyproject.toml index 9920b04..ecd57c0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -30,7 +30,7 @@ classifiers = [ ] dependencies = [ - "textual>=0.40.0", + "textual>=3.0", "psutil>=5.9.0", ] diff --git a/src/netshow/app.py b/src/netshow/app.py index 33a3839..48a774b 100644 --- a/src/netshow/app.py +++ b/src/netshow/app.py @@ -1,14 +1,16 @@ import os import time +from collections import deque from typing import Any, Optional, Union, cast import psutil +from rich.text import Text from textual.app import App, ComposeResult from textual.binding import Binding from textual.containers import Container, Horizontal, Vertical from textual.reactive import reactive from textual.timer import Timer -from textual.widgets import DataTable, Footer, Header, Input, Static +from textual.widgets import DataTable, Footer, Header, Input, Sparkline, Static from .detail_screen import ConnectionDetailScreen from .helpers import get_lsof_conns, get_psutil_conns @@ -24,9 +26,7 @@ class NetshowApp(App): """Network connection monitoring application using Textual TUI.""" CSS = CSS - BINDINGS = cast( - list[Union[Binding, tuple[str, str], tuple[str, str, str]]], BASIC_KEYBINDINGS - ) + BINDINGS = cast(list[Union[Binding, tuple[str, str], tuple[str, str, str]]], BASIC_KEYBINDINGS) total_connections = reactive(0) active_connections = reactive(0) @@ -36,6 +36,7 @@ class NetshowApp(App): sort_mode = reactive("default") selected_interface = reactive("all") show_emojis = reactive(True) + expand_ipv6 = reactive(False) def __init__(self, **kwargs: Any) -> None: super().__init__(**kwargs) @@ -46,6 +47,7 @@ def __init__(self, **kwargs: Any) -> None: self.title = "Netshow" # Will be updated with data source self.debounce_timer: Optional[Timer] = None self.available_interfaces = self._get_available_interfaces() + self.bandwidth_history: deque[float] = deque(maxlen=20) # Last 20 samples def _get_available_interfaces(self) -> list: """Get list of available network interfaces.""" @@ -61,18 +63,11 @@ def compose(self) -> ComposeResult: with Vertical(): with Container(id="stats_container"): with Horizontal(id="metrics_row"): - yield Static( - "πŸ“Š Connections: 0", id="conn_metric", classes="metric" - ) - yield Static("⚑ Active: 0", id="active_metric", classes="metric") - yield Static( - "πŸ‘‚ Listening: 0", id="listen_metric", classes="metric" - ) - yield Static( - "πŸ”₯ Bandwidth: 0 B/s (all)", - id="bandwidth_metric", - classes="metric", - ) + yield Static("πŸ“Š 0", id="conn_metric", classes="metric") + yield Static("⚑ 0", id="active_metric", classes="metric") + yield Static("πŸ‘‚ 0", id="listen_metric", classes="metric") + yield Static("πŸ”₯ 0 B/s", id="bandwidth_metric", classes="metric bandwidth") + yield Sparkline([], id="bandwidth_spark") with Container(id="filter_container"): yield Input( @@ -98,14 +93,18 @@ def on_mount(self) -> None: filter_container.display = False # Refresh at regular intervals - self.timer: Timer = self.set_interval( - REFRESH_INTERVAL, self.refresh_connections - ) + self.timer: Timer = self.set_interval(REFRESH_INTERVAL, self.refresh_connections) self.refresh_connections() # Focus the table for keyboard navigation table.focus() + def on_resize(self) -> None: + """Handle terminal resize - update metrics display.""" + self._update_metrics_display( + self.total_connections, self.active_connections, self.listening_connections + ) + def refresh_connections(self, sort_only: bool = False) -> None: """Refresh connection data and update the display.""" if not sort_only: @@ -163,6 +162,25 @@ def _fetch_connection_data(self) -> None: source_name = "psutil" if using_root else "lsof" self.title = f"Netshow ({source_name})" + def _truncate_addr(self, addr: str) -> str: + """Truncate IPv6 addresses for compact display.""" + if not addr or self.expand_ipv6: + return addr + # Check for IPv6 (contains multiple colons or brackets) + if addr.startswith("[") or addr.count(":") > 1: + # Extract port if present + if "]:" in addr: + ip, port = addr.rsplit(":", 1) + return f"[…]:{port}" + elif addr.startswith("["): + return "[…]" + # Bare IPv6 with port (ip:port where ip has multiple colons) + parts = addr.rsplit(":", 1) + if len(parts) == 2 and parts[0].count(":") >= 1: + return f"…:{parts[1]}" + return "…" + return addr + def _update_table_display(self) -> None: """Update the table display with current connection data.""" table = self.query_one("#connections_table", DataTable) @@ -184,18 +202,15 @@ def _update_table_display(self) -> None: if i >= table.row_count: break - status_icon = self._get_status_icon(c["status"]) + status_styled = self._get_status_styled(c["status"]) speed_indicator = self._get_speed_indicator(c) - status_text = ( - f"{status_icon} {c['status']}" if status_icon else c["status"] - ) new_row = [ c["pid"], c["friendly"], c["proc"], - c["laddr"], - c["raddr"], - status_text, + self._truncate_addr(c["laddr"]), + self._truncate_addr(c["raddr"]), + status_styled, speed_indicator, ] @@ -225,18 +240,15 @@ def _update_table_display(self) -> None: # Fall back to clear and rebuild for smaller datasets or size changes table.clear() for c in conns: - status_icon = self._get_status_icon(c["status"]) + status_styled = self._get_status_styled(c["status"]) speed_indicator = self._get_speed_indicator(c) - status_text = ( - f"{status_icon} {c['status']}" if status_icon else c["status"] - ) table.add_row( c["pid"], c["friendly"], c["proc"], - c["laddr"], - c["raddr"], - status_text, + self._truncate_addr(c["laddr"]), + self._truncate_addr(c["raddr"]), + status_styled, speed_indicator, ) @@ -265,6 +277,25 @@ def _update_table_display(self) -> None: if cursor_row < table.row_count and hasattr(table, "cursor_coordinate"): table.cursor_coordinate = (cursor_row, 0) # type: ignore + def _get_status_styled(self, status: str) -> Text: + """Get a Rich Text styled status with color coding (Solarized).""" + status_styles = { + "ESTABLISHED": ("bold #859900", "πŸš€"), # green + "LISTEN": ("bold #268bd2", "πŸ‘‚"), # blue + "TIME_WAIT": ("#b58900", "⏳"), # yellow + "CLOSE_WAIT": ("bold #dc322f", "⏸️"), # red + "SYN_SENT": ("#d33682", "πŸ“€"), # magenta + "SYN_RECV": ("#d33682", "πŸ“₯"), # magenta + "FIN_WAIT1": ("#b58900", "πŸ”„"), # yellow + "FIN_WAIT2": ("#b58900", "πŸ”"), # yellow + "CLOSING": ("#dc322f", "πŸ”š"), # red + "LAST_ACK": ("#dc322f", "🏁"), # red + } + style, icon = status_styles.get(status, ("#586e75", "❓")) # base01 + if self.show_emojis: + return Text(f"{icon} {status}", style=style) + return Text(status, style=style) + def _get_status_icon(self, status: str) -> str: """Get an appropriate icon for connection status.""" if not self.show_emojis: @@ -283,9 +314,7 @@ def _get_status_icon(self, status: str) -> str: } return status_icons.get(status, "❓") - def _get_speed_indicator( - self, connection: Union[dict[str, str], ConnectionData] - ) -> str: + def _get_speed_indicator(self, connection: Union[dict[str, str], ConnectionData]) -> str: """Generate a speed indicator based on connection characteristics.""" if not self.show_emojis: status = connection.get("status", "") @@ -311,58 +340,67 @@ def _update_metrics_display(self, total: int, active: int, listening: int) -> No active_metric = self.query_one("#active_metric", Static) listen_metric = self.query_one("#listen_metric", Static) bandwidth_metric = self.query_one("#bandwidth_metric", Static) + bandwidth_spark = self.query_one("#bandwidth_spark", Sparkline) + + # Check terminal width for adaptive display + width = self.size.width + compact = width < 100 + show_spark = width >= 80 # Emoji prefixes based on toggle state - conn_prefix = "πŸ“Š " if self.show_emojis else "" - active_prefix = "⚑ " if self.show_emojis else "" - listen_prefix = "πŸ‘‚ " if self.show_emojis else "" - bandwidth_prefix = "πŸ”₯ " if self.show_emojis else "" + conn_icon = "πŸ“Š " if self.show_emojis else "" + active_icon = "⚑ " if self.show_emojis else "" + listen_icon = "πŸ‘‚ " if self.show_emojis else "" + bw_icon = "πŸ”₯ " if self.show_emojis else "" # Get network I/O stats for bandwidth + total_bandwidth = 0.0 + interface_label = self.selected_interface try: if self.selected_interface == "all": net_io = psutil.net_io_counters() - interface_label = "all" else: net_io_per_nic = psutil.net_io_counters(pernic=True) net_io_temp = net_io_per_nic.get(self.selected_interface) - interface_label = self.selected_interface if not net_io_temp: - # Interface not found, fallback to all interface_label = "all" self.selected_interface = "all" + net_io = psutil.net_io_counters() else: net_io = net_io_temp current_time = time.time() - if ( - self.last_network_stats is not None - and self.last_stats_time is not None - ): - time_diff = max( - 0.1, current_time - self.last_stats_time - ) # Avoid division by zero - bytes_sent_diff = max( - 0, net_io.bytes_sent - self.last_network_stats.bytes_sent - ) - bytes_recv_diff = max( - 0, net_io.bytes_recv - self.last_network_stats.bytes_recv - ) + if self.last_network_stats is not None and self.last_stats_time is not None: + time_diff = max(0.1, current_time - self.last_stats_time) + bytes_sent_diff = max(0, net_io.bytes_sent - self.last_network_stats.bytes_sent) + bytes_recv_diff = max(0, net_io.bytes_recv - self.last_network_stats.bytes_recv) total_bandwidth = (bytes_sent_diff + bytes_recv_diff) / time_diff - bandwidth_text = f"{self._format_bytes(int(total_bandwidth))}/s ({interface_label})" - else: - bandwidth_text = f"0 B/s ({interface_label})" self.last_network_stats = net_io self.last_stats_time = current_time except (AttributeError, OSError): - bandwidth_text = "N/A" + pass - conn_metric.update(f"{conn_prefix}Connections: {total}") - active_metric.update(f"{active_prefix}Active: {active}") - listen_metric.update(f"{listen_prefix}Listening: {listening}") - bandwidth_metric.update(f"{bandwidth_prefix}Bandwidth: {bandwidth_text}") + # Update sparkline + self.bandwidth_history.append(total_bandwidth) + bandwidth_spark.data = list(self.bandwidth_history) + bandwidth_spark.display = show_spark + + # Adaptive labels based on width + if compact: + conn_metric.update(f"{conn_icon}{total}") + active_metric.update(f"{active_icon}{active}") + listen_metric.update(f"{listen_icon}{listening}") + bandwidth_metric.update(f"{bw_icon}{self._format_bytes(int(total_bandwidth))}/s") + else: + conn_metric.update(f"{conn_icon}Conn: {total}") + active_metric.update(f"{active_icon}Active: {active}") + listen_metric.update(f"{listen_icon}Listen: {listening}") + iface = f" ({interface_label})" if width >= 120 else "" + bandwidth_metric.update( + f"{bw_icon}{self._format_bytes(int(total_bandwidth))}/s{iface}" + ) except Exception: - pass # Gracefully handle missing widgets + pass def _format_bytes(self, bytes_val: int) -> str: """Format bytes into human readable format.""" @@ -380,11 +418,16 @@ def _format_bytes(self, bytes_val: int) -> str: def _get_selected_connection_data(self, row_data: tuple) -> ConnectionData: """Convert row data tuple to ConnectionData dict.""" - # Extract status without icon (remove first 2 characters: icon + space) - status_with_icon = row_data[5] - clean_status = ( - status_with_icon[2:] if len(status_with_icon) > 2 else status_with_icon - ) + # Extract status - handle both Rich Text objects and plain strings + status_cell = row_data[5] + if isinstance(status_cell, Text): + # Rich Text object - extract plain text and remove emoji prefix + status_text = status_cell.plain + # Remove emoji prefix if present (emoji + space = ~3 chars) + clean_status = status_text.split(" ", 1)[-1] if " " in status_text else status_text + else: + # Plain string - remove first 2 characters (icon + space) + clean_status = status_cell[2:] if len(status_cell) > 2 else status_cell return ConnectionData( pid=row_data[0], @@ -512,6 +555,11 @@ def action_toggle_emojis(self) -> None: self.total_connections, self.active_connections, self.listening_connections ) + def action_toggle_ipv6(self) -> None: + """Toggle IPv6 address expansion.""" + self.expand_ipv6 = not self.expand_ipv6 + self._update_table_display() + def _update_table_columns(self) -> None: """Update table column headers based on emoji setting.""" table = self.query_one("#connections_table", DataTable) diff --git a/src/netshow/cli.py b/src/netshow/cli.py index ae5dee5..cfd837f 100644 --- a/src/netshow/cli.py +++ b/src/netshow/cli.py @@ -1,5 +1,6 @@ """CLI entry point for NetShow.""" +import os import sys from .app import NetshowApp @@ -7,6 +8,10 @@ def main() -> None: """Main CLI entry point.""" + # Ensure truecolor support for Solarized theme + if "COLORTERM" not in os.environ: + os.environ["COLORTERM"] = "truecolor" + try: NetshowApp().run() except KeyboardInterrupt: diff --git a/src/netshow/detail_screen.py b/src/netshow/detail_screen.py index 3de2479..c7bb1ad 100644 --- a/src/netshow/detail_screen.py +++ b/src/netshow/detail_screen.py @@ -92,9 +92,7 @@ def compose(self) -> ComposeResult: local_prefix = "🏠 " if show_emojis else "" remote_prefix = "🌐 " if show_emojis else "" - yield Static( - f"{conn_prefix}Connection Info", classes="detail_title" - ) + yield Static(f"{conn_prefix}Connection Info", classes="detail_title") yield Static( f"{pid_prefix}PID: {self.connection_data['pid']}", classes="detail_item", @@ -169,9 +167,7 @@ def compose(self) -> ComposeResult: cpu_percent = self.process_info.get("cpu_percent", 0.0) if show_emojis: cpu_icon = ( - "πŸ”₯" - if cpu_percent > 50 - else "⚑" if cpu_percent > 10 else "πŸ’€" + "πŸ”₯" if cpu_percent > 50 else "⚑" if cpu_percent > 10 else "πŸ’€" ) else: cpu_icon = "" @@ -191,7 +187,9 @@ def compose(self) -> ComposeResult: memory_icon = ( "🚨" if memory_percent > 80 - else "⚠️" if memory_percent > 50 else "πŸ’Ύ" + else "⚠️" + if memory_percent > 50 + else "πŸ’Ύ" ) else: memory_icon = "" @@ -204,9 +202,7 @@ def compose(self) -> ComposeResult: # Network connections from this process connections = self.process_info.get("connections", []) if connections: - conn_count = ( - len(connections) if isinstance(connections, list) else 0 - ) + conn_count = len(connections) if isinstance(connections, list) else 0 active_conn_prefix = "🌐 " if show_emojis else "" yield Static( f"{active_conn_prefix}Active Connections: {conn_count}", diff --git a/src/netshow/helpers.py b/src/netshow/helpers.py index f36d127..c33a8f9 100644 --- a/src/netshow/helpers.py +++ b/src/netshow/helpers.py @@ -69,12 +69,16 @@ def get_psutil_conns() -> list[dict[str, str]]: "laddr": ( f"[{conn.laddr.ip}]:{conn.laddr.port}" if conn.laddr and ":" in conn.laddr.ip - else f"{conn.laddr.ip}:{conn.laddr.port}" if conn.laddr else "" + else f"{conn.laddr.ip}:{conn.laddr.port}" + if conn.laddr + else "" ), "raddr": ( f"[{conn.raddr.ip}]:{conn.raddr.port}" if conn.raddr and ":" in conn.raddr.ip - else f"{conn.raddr.ip}:{conn.raddr.port}" if conn.raddr else "" + else f"{conn.raddr.ip}:{conn.raddr.port}" + if conn.raddr + else "" ), "status": conn.status, } diff --git a/src/netshow/styles.py b/src/netshow/styles.py index d5cde1e..6d4c299 100644 --- a/src/netshow/styles.py +++ b/src/netshow/styles.py @@ -1,31 +1,24 @@ -"""Styles for the NetshowApp - Enhanced Selenized Dark Theme.""" +"""Styles for the NetshowApp - Solarized Dark Theme.""" CSS = """ -/* === SELENIZED DARK COLOR PALETTE === */ -$bg_0: #103c48; -$bg_1: #184956; -$bg_2: #2d5b69; -$dim_0: #72898f; -$fg_0: #adbcbc; -$fg_1: #cad8d9; - -$red: #fa5750; -$green: #75b938; -$yellow: #dbb32d; -$blue: #2d5b69; -$magenta: #f275be; -$cyan: #4191a5; -$orange: #ed8649; -$violet: #af88eb; - -$br_red: #ff665c; -$br_green: #84c747; -$br_yellow: #ebc13d; -$br_blue: #58a3ff; -$br_magenta: #ff84cd; -$br_cyan: #53d6c7; -$br_orange: #fd9456; -$br_violet: #bd96fa; +/* === SOLARIZED DARK COLOR PALETTE === */ +$base03: #002b36; +$base02: #073642; +$base01: #586e75; +$base00: #657b83; +$base0: #839496; +$base1: #93a1a1; +$base2: #eee8d5; +$base3: #fdf6e3; + +$yellow: #b58900; +$orange: #cb4b16; +$red: #dc322f; +$magenta: #d33682; +$violet: #6c71c4; +$blue: #268bd2; +$cyan: #2aa198; +$green: #859900; /* === METRICS ROW === */ @@ -33,38 +26,57 @@ height: 3; margin: 0; padding: 0 1; + width: 100%; + align: center middle; } .metric { - background: $bg_1; - color: $fg_1; - border: solid $cyan; + background: $base02; + color: $base1; + border: solid $base01; padding: 0 1; margin: 0; text-style: bold; - text-align: center; - width: 1fr; - min-width: 15; + content-align: center middle; + width: auto; + min-width: 8; } -/* === SPECIFIC METRIC WIDTHS === */ #conn_metric { - width: 2fr; + border: solid $blue; + color: $blue; } #active_metric { - width: 1.1fr; - min-width: 12; + border: solid $green; + color: $green; } #listen_metric { - width: 1.3fr; - min-width: 15; + border: solid $cyan; + color: $cyan; } #bandwidth_metric { - width: 2.5fr; - min-width: 25; + border: solid $orange; + color: $orange; +} + +/* === BANDWIDTH SPARKLINE === */ +#bandwidth_spark { + width: 100%; + height: 2; + background: $base02; + margin: 0; + padding: 0 1; +} + +Sparkline > .sparkline--max-color { + color: $orange; +} + +Sparkline > .sparkline--min-color { + color: $cyan; } /* === FILTER CONTAINER === */ @@ -75,38 +87,38 @@ } #filter_input { - background: $bg_1; - color: $fg_1; + background: $base02; + color: $base1; border: solid $magenta; height: 1; padding: 0 1; } #filter_input:focus { - background: $bg_0; - border: solid $br_cyan; - color: $fg_1; + background: $base03; + border: solid $cyan; + color: $base1; } /* === GLOBAL STYLES === */ Screen { - background: $bg_0; - color: $fg_0; + background: $base03; + color: $base0; } /* === HEADER & FOOTER === */ Header { - background: $bg_1; - color: $fg_1; - border-bottom: solid $blue; + background: $base02; + color: $base1; + border-bottom: solid $base01; text-style: bold; height: 2; } Footer { - background: $bg_1; - color: $fg_1; + background: $base02; + color: $base1; height: 1; } @@ -114,13 +126,13 @@ #stats_container { height: auto; margin: 0; - border: solid $dim_0; + border: solid $base01; padding: 0; } #status_bar { - background: $bg_1; - color: $fg_1; + background: $base02; + color: $base1; height: 1; padding: 0 1; border: none; @@ -141,7 +153,7 @@ /* === EDGE BORDER FIX === */ #connection_details { - border-right: #fff; + border-right: $base1; } #process_info { @@ -150,8 +162,8 @@ /* === DATA TABLE STYLING === */ DataTable { - background: $bg_0; - color: $fg_0; + background: $base03; + color: $base0; width: 100%; height: 1fr; border: none; @@ -162,34 +174,33 @@ border: none !important; } -DataTable .header { - background: $bg_1; - color: $fg_1; +DataTable > .datatable--header { + background: $base02; + color: $base1; text-style: bold; - height: 1; } -DataTable .datatable--cursor { - background: $bg_2; - color: $fg_1; +DataTable > .datatable--cursor { + background: $base01; + color: $base2; text-style: bold; } -DataTable .datatable--hover { - background: $bg_1; - color: $fg_1; +DataTable > .datatable--hover { + background: $base02; + color: $base1; } -DataTable:focus .datatable--cursor { - background: $dim_0; - color: $fg_1; +DataTable:focus > .datatable--cursor { + background: $blue; + color: $base3; text-style: bold; } /* === DETAIL SCREEN STYLING === */ #detail_title { - background: $bg_2; - color: $fg_1; + background: $base02; + color: $base1; height: 5; padding: 1 2; text-align: center; @@ -206,7 +217,7 @@ } #connection_details, #process_info { - background: $bg_1; + background: $base02; padding: 2; margin: 0; height: auto; @@ -214,8 +225,8 @@ } .section_header { - background: $bg_2; - color: $fg_1; + background: $base01; + color: $base1; padding: 1 2; text-align: center; text-style: bold; @@ -226,22 +237,22 @@ .detail_title { margin: 0 0 1 0; padding: 1 1; - color: $fg_1; + color: $base1; text-style: bold; - background: $bg_2; + background: $base01; } .detail_item { margin: 0 0 1 1; padding: 0 1; - color: $fg_1; + color: $base0; background: transparent; height: auto; } .detail_item:hover { - color: $fg_1; - background: $bg_2; + color: $base1; + background: $base02; } /* === BUTTONS === */ @@ -254,14 +265,14 @@ #back_button { background: $blue; - color: $bg_0; + color: $base3; width: 30; height: 3; text-style: bold; } #back_button:hover { - background: $br_blue; + background: $cyan; } #back_button:focus { @@ -276,9 +287,9 @@ /* === SCROLLABLE CONTAINERS === */ ScrollableContainer { background: transparent; - scrollbar-background: $bg_1; - scrollbar-color: $blue; - scrollbar-color-hover: $br_blue; + scrollbar-background: $base02; + scrollbar-color: $base01; + scrollbar-color-hover: $blue; scrollbar-color-active: $cyan; padding: 0; margin: 0; @@ -303,11 +314,4 @@ .status-CLOSE_WAIT { color: $red; } - -/* === ACCESSIBILITY === */ - -.epic-glow { - color: $br_blue; - text-style: bold; -} """ diff --git a/src/netshow/types_and_constants.py b/src/netshow/types_and_constants.py index cca0f07..de0cca8 100644 --- a/src/netshow/types_and_constants.py +++ b/src/netshow/types_and_constants.py @@ -11,6 +11,7 @@ ("p", "sort_by_process", "Sort by Process"), ("i", "toggle_interface", "Interface"), ("e", "toggle_emojis", "Emojis"), + ("v", "toggle_ipv6", "IPv6"), ("ctrl+c", "quit", "Hard Quit"), ("/", "search", "Search"), ] diff --git a/uv.lock b/uv.lock index 4e3294a..4200acd 100644 --- a/uv.lock +++ b/uv.lock @@ -214,7 +214,7 @@ wheels = [ [[package]] name = "netshow" -version = "0.2.0" +version = "0.2.2" source = { editable = "." } dependencies = [ { name = "psutil" }, @@ -242,7 +242,7 @@ requires-dist = [ { name = "psutil", specifier = ">=5.9.0" }, { name = "pytest", marker = "extra == 'dev'", specifier = ">=7.0.0" }, { name = "ruff", marker = "extra == 'dev'", specifier = ">=0.1.0" }, - { name = "textual", specifier = ">=0.40.0" }, + { name = "textual", specifier = ">=3.0" }, { name = "types-psutil", marker = "extra == 'dev'", specifier = ">=5.9.0" }, ] provides-extras = ["dev"]