tui: move from swarm of labels to DataTable
This commit is contained in:
parent
f323e4defe
commit
91f80e62ae
GPG key ID:
B265E45CACAD108A
3 changed files with 125 additions and 247 deletions
|
|
@ -3,36 +3,15 @@ Header {
|
|||
}
|
||||
|
||||
GPUStatsWidget {
|
||||
layout: grid;
|
||||
grid-size: 3;
|
||||
grid-gutter: 2 8;
|
||||
box-sizing: content-box;
|
||||
background: $panel;
|
||||
height: 10;
|
||||
height: 100%;
|
||||
width: 100%;
|
||||
min-width: 50;
|
||||
}
|
||||
|
||||
/* for colors, see: https://textual.textualize.io/guide/design/#theme-reference */
|
||||
|
||||
.widgetheader {
|
||||
background: $panel;
|
||||
border: solid $primary-background;
|
||||
content-align: center middle;
|
||||
}
|
||||
|
||||
.box {
|
||||
height: 100%;
|
||||
/* border: ascii $primary-background;*/
|
||||
content-align: center middle;
|
||||
padding: 0 1;
|
||||
}
|
||||
|
||||
Horizontal {
|
||||
padding: 0;
|
||||
margin: 0;
|
||||
box-sizing: content-box;
|
||||
}
|
||||
|
||||
.statvalue {
|
||||
dock: right;
|
||||
DataTable {
|
||||
width: 100%;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -22,14 +22,15 @@ import sys
|
|||
from datetime import datetime
|
||||
from os import path
|
||||
|
||||
from rich.text import Text
|
||||
from textual.binding import Binding
|
||||
from textual.app import App, ComposeResult
|
||||
from textual.containers import Container, Horizontal
|
||||
from textual.reactive import reactive
|
||||
from textual.containers import Container
|
||||
from textual.screen import Screen
|
||||
from textual.widgets import Header, Footer, Static, TextLog, Label
|
||||
from textual.widgets import Header, Footer, Static, TextLog, DataTable
|
||||
|
||||
from .utils import AMDGPU_CARDS, format_frequency, get_core_stats, get_fan_rpm, get_fan_target, get_power_stats, get_temp_stats # pylint: disable=line-too-long
|
||||
from .utils import AMDGPU_CARDS, get_fan_rpm, get_power_stats, get_temp_stat, get_clock, get_gpu_usage, get_voltage
|
||||
# pylint: disable=line-too-long
|
||||
|
||||
# rich markup reference:
|
||||
# https://rich.readthedocs.io/en/stable/markup.html
|
||||
|
|
@ -40,7 +41,7 @@ class LogScreen(Screen):
|
|||
|
||||
def __init__(self, *args, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
self.text_log = TextLog(highlight=True, markup=True)
|
||||
self.text_log = TextLog(highlight=True, markup=True, name='log_gpu')
|
||||
|
||||
def on_mount(self) -> None:
|
||||
"""Event handler called when widget is first added
|
||||
|
|
@ -48,7 +49,7 @@ class LogScreen(Screen):
|
|||
|
||||
def compose(self) -> ComposeResult:
|
||||
yield Header(show_clock=True)
|
||||
yield Container(self.text_log)
|
||||
yield self.text_log
|
||||
yield Footer()
|
||||
|
||||
# def on_key(self, event: events.Key) -> None:
|
||||
|
|
@ -59,25 +60,88 @@ class LogScreen(Screen):
|
|||
class GPUStatsWidget(Static):
|
||||
"""The main stats widget."""
|
||||
|
||||
def __init__(self, *args, card=None, **kwargs):
|
||||
columns = ["card",
|
||||
"core clock",
|
||||
"memory clock",
|
||||
"utilization",
|
||||
"voltage",
|
||||
"power usage",
|
||||
"set limit",
|
||||
"default limit",
|
||||
"capability",
|
||||
"fan rpm",
|
||||
"edge temp",
|
||||
"junction temp",
|
||||
"memory temp"]
|
||||
timer_stats = None
|
||||
table = None
|
||||
table_needs_init = True
|
||||
data = {}
|
||||
|
||||
def __init__(self, *args, cards=None, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
# Instance variables
|
||||
self.card = card
|
||||
self.hwmon_dir = AMDGPU_CARDS[self.card]
|
||||
self.cards = cards
|
||||
|
||||
async def on_mount(self) -> None:
|
||||
'''Fires when stats widget first shown'''
|
||||
self.table = self.query_one(DataTable)
|
||||
for column in self.columns:
|
||||
self.table.add_column(label=column, key=column)
|
||||
# self.table.add_columns(*self.columns)
|
||||
self.table_needs_init = True
|
||||
self.timer_stats = self.set_interval(1, self.get_stats)
|
||||
|
||||
def compose(self) -> ComposeResult:
|
||||
"""Create child widgets."""
|
||||
yield ClockDisplay(classes="box", card=self.card, hwmon_dir=self.hwmon_dir)
|
||||
yield PowerDisplay(classes="box", card=self.card, hwmon_dir=self.hwmon_dir)
|
||||
yield MiscDisplay(classes="box", card=self.card)
|
||||
_msg = f'''[bold]App:[/] creating stat widgets for [green]{self.card}[/], stats directory: {self.hwmon_dir}'''
|
||||
self.update_log(_msg)
|
||||
stats_table = DataTable(zebra_stripes=True, show_cursor=False, name='stats_table')
|
||||
yield stats_table
|
||||
self.update_log('[bold]App:[/] created stats table')
|
||||
|
||||
def update_log(self, message: str) -> None:
|
||||
"""Update the TextLog widget with a new message."""
|
||||
log_screen = AMDGPUStats.SCREENS["logs"]
|
||||
log_screen.text_log.write(message)
|
||||
|
||||
def get_stats(self):
|
||||
'''Function to fetch stats / update the table'''
|
||||
for card in self.cards:
|
||||
power_stats = get_power_stats(card=card)
|
||||
self.data = {
|
||||
"card": card,
|
||||
"core clock": get_clock('core', card=card, format_freq=True),
|
||||
"memory clock": get_clock('memory', card=card, format_freq=True),
|
||||
"utilization": f'{get_gpu_usage(card=card)}%',
|
||||
"voltage": f'{get_voltage(card=card)}V',
|
||||
"power usage": f'{power_stats["average"]}W',
|
||||
"set limit": f'{power_stats["limit"]}W',
|
||||
"default limit": f'{power_stats["default"]}W',
|
||||
"capability": f'{power_stats["capability"]}W',
|
||||
"fan rpm": f'{get_fan_rpm(card=card)}',
|
||||
"edge temp": f"{get_temp_stat(name='edge', card=card)}C",
|
||||
"junction temp": f"{get_temp_stat(name='junction', card=card)}C",
|
||||
"memory temp": f"{get_temp_stat(name='mem', card=card)}C"}
|
||||
# handle the table data appopriately
|
||||
# if needs populated anew or updated
|
||||
if self.table_needs_init:
|
||||
# Add rows for the first time
|
||||
# Adding styled and justified `Text` objects instead of plain strings.
|
||||
styled_row = [
|
||||
Text(str(cell), style="italic", justify="right") for cell in self.data.values()
|
||||
]
|
||||
self.table.add_row(*styled_row, key=card)
|
||||
hwmon_dir = AMDGPU_CARDS[card]
|
||||
self.update_log(f'[bold]stats:[/] added row for [bold green]{card}[/], info dir: {hwmon_dir}')
|
||||
else:
|
||||
# Update existing rows
|
||||
for column, value in self.data.items():
|
||||
styled_cell = Text(str(value), style="italic", justify="right")
|
||||
self.table.update_cell(card, column, styled_cell)
|
||||
if self.table_needs_init:
|
||||
# if this is the first time updating the table, mark it initialized
|
||||
self.table_needs_init = False
|
||||
self.table.refresh()
|
||||
|
||||
|
||||
class AMDGPUStats(App):
|
||||
"""Textual-based tool to show AMDGPU statistics."""
|
||||
|
|
@ -95,19 +159,17 @@ class AMDGPUStats(App):
|
|||
# Binding("l", "push_screen('logs')", "Toggle logs", priority=True),
|
||||
BINDINGS = [
|
||||
Binding("c", "custom_dark", "Colors"),
|
||||
Binding("l", "toggle_log", "Logs"),
|
||||
Binding("s", "screenshot_wrapper", "Screenshot"),
|
||||
Binding("l", "custom_log", "Logs"),
|
||||
Binding("s", "custom_screenshot", "Screenshot"),
|
||||
Binding("q", "quit", "Quit")
|
||||
]
|
||||
|
||||
def compose(self) -> ComposeResult:
|
||||
"""Create child widgets for the app."""
|
||||
yield Header(show_clock=True)
|
||||
for card in AMDGPU_CARDS:
|
||||
stat_widget_name = "stats_" + card
|
||||
yield Label(card, expand=True, classes='widgetheader')
|
||||
widget = Container(GPUStatsWidget(card=card, id=stat_widget_name))
|
||||
yield widget
|
||||
stats_widget = Container(GPUStatsWidget(cards=AMDGPU_CARDS,
|
||||
name="stats_widget"))
|
||||
yield stats_widget
|
||||
self.update_log("[bold green]App started, logging begin!")
|
||||
self.update_log(f"[bold]Discovered AMD GPUs:[/] {list(AMDGPU_CARDS)}")
|
||||
# nice-to-have: account for not storing these in dicts, but resolved in funcs
|
||||
|
|
@ -126,7 +188,7 @@ class AMDGPUStats(App):
|
|||
self.refresh()
|
||||
# self.dark = not self.dark
|
||||
|
||||
def action_screenshot_wrapper(self, screen_dir: str = '/tmp') -> None:
|
||||
def action_custom_screenshot(self, screen_dir: str = '/tmp') -> None:
|
||||
"""Action that fires when the user presses 's' for a screenshot"""
|
||||
# construct the screenshot elements + path
|
||||
timestamp = datetime.now().isoformat().replace(":", "_")
|
||||
|
|
@ -135,7 +197,7 @@ class AMDGPUStats(App):
|
|||
self.action_screenshot(path=screen_dir, filename=screen_name)
|
||||
self.update_log(f'[bold]Screenshot taken: [italic]{screen_path}')
|
||||
|
||||
def action_toggle_log(self) -> None:
|
||||
def action_custom_log(self) -> None:
|
||||
"""Toggle between the main screen and the LogScreen."""
|
||||
if isinstance(self.screen, LogScreen):
|
||||
self.pop_screen()
|
||||
|
|
@ -148,186 +210,6 @@ class AMDGPUStats(App):
|
|||
log_screen.text_log.write(message)
|
||||
|
||||
|
||||
class MiscDisplay(Static):
|
||||
"""A widget to display misc. GPU stats."""
|
||||
# construct the misc. stats dict; appended by discovered temperature nodes
|
||||
# used to make a 'reactive' object
|
||||
fan_rpm = reactive(0)
|
||||
fan_rpm_target = reactive(0)
|
||||
# do some dancing to craft the UI; initialize the reactive obj with data
|
||||
# dynamic object for temperature updates
|
||||
# populated / looped in '__init__' to get proper labels
|
||||
temp_stats = reactive({})
|
||||
# default to 'not composed', once labels are made - become true
|
||||
# avoids a race condition between discovering temperature nodes/stats
|
||||
# ... and making labels for them
|
||||
composed = False
|
||||
|
||||
def __init__(self, card: str, *args, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
self.timer_misc = None
|
||||
self.card = card
|
||||
self.temp_stats = get_temp_stats(self.card)
|
||||
|
||||
def compose(self) -> ComposeResult:
|
||||
yield Horizontal(Label("[underline]Temperatures"),
|
||||
Label("", classes="statvalue"))
|
||||
for temp_node in self.temp_stats:
|
||||
# capitalize the first letter for display
|
||||
caption = temp_node[0].upper() + temp_node[1:]
|
||||
yield Horizontal(Label(f' {caption}:',),
|
||||
Label("", id="temp_" + temp_node, classes="statvalue"))
|
||||
# padding to split groups
|
||||
yield Horizontal()
|
||||
yield Horizontal(Label("[underline]Fan RPM"),
|
||||
Label("", classes="statvalue"))
|
||||
yield Horizontal(Label(" Current:",),
|
||||
Label("", id="fan_rpm", classes="statvalue"))
|
||||
yield Horizontal(Label(" Target:",),
|
||||
Label("", id="fan_rpm_target", classes="statvalue"))
|
||||
self.composed = True
|
||||
|
||||
def on_mount(self) -> None:
|
||||
"""Event handler called when widget is added to the app."""
|
||||
self.timer_misc = self.set_interval(1, self.update_misc_stats)
|
||||
|
||||
def update_misc_stats(self) -> None:
|
||||
"""Method to update the temp/fan values to current measurements.
|
||||
|
||||
Run by a timer created 'on_mount'"""
|
||||
self.fan_rpm = get_fan_rpm(self.card)
|
||||
self.fan_rpm_target = get_fan_target(self.card)
|
||||
self.temp_stats = get_temp_stats(self.card)
|
||||
|
||||
def watch_fan_rpm(self, fan_rpm: int) -> None:
|
||||
"""Called when the 'fan_rpm' reactive attr changes.
|
||||
|
||||
- Updates label values
|
||||
- Casting inputs to string to avoid type problems w/ int/None"""
|
||||
self.query_one("#fan_rpm", Static).update(f"{fan_rpm}")
|
||||
|
||||
def watch_fan_rpm_target(self, fan_rpm_target: int) -> None:
|
||||
"""Called when the 'fan_rpm_target' reactive attr changes.
|
||||
|
||||
- Updates label values
|
||||
- Casting inputs to string to avoid type problems w/ int/None"""
|
||||
self.query_one("#fan_rpm_target", Static).update(f"{fan_rpm_target}")
|
||||
|
||||
def watch_temp_stats(self, temp_stats: dict) -> None:
|
||||
"""Called when the temp_stats reactive attr changes, updates labels"""
|
||||
# try to avoid racing
|
||||
if not self.composed:
|
||||
return
|
||||
for temp_node in temp_stats:
|
||||
item_val = self.temp_stats[temp_node]
|
||||
self.query_one("#temp_" + temp_node, Static).update(f'{item_val}C')
|
||||
|
||||
|
||||
class ClockDisplay(Static):
|
||||
"""A widget to display GPU power stats."""
|
||||
core_vals = reactive({"sclk": 0, "mclk": 0, "voltage": 0, "util_pct": 0})
|
||||
|
||||
def __init__(self, card: str, hwmon_dir: str, *args, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
self.timer_clocks = None
|
||||
self.card = card
|
||||
self.hwmon_dir = hwmon_dir
|
||||
|
||||
def compose(self) -> ComposeResult:
|
||||
yield Horizontal(Label("[underline]Performance"),
|
||||
Label("", classes="statvalue"))
|
||||
yield Horizontal(Label(" Core clock:",),
|
||||
Label("", id="clk_core_val", classes="statvalue"))
|
||||
yield Horizontal(Label(" Memory clock:"),
|
||||
Label("", id="clk_memory_val", classes="statvalue"))
|
||||
yield Horizontal(Label(" Utilization:",),
|
||||
Label("", id="util_pct", classes="statvalue"))
|
||||
yield Horizontal(Label(" Voltage:",),
|
||||
Label("", id="clk_voltage_val", classes="statvalue"))
|
||||
# padding underneath, don't let them space out vertically
|
||||
yield Horizontal()
|
||||
yield Horizontal()
|
||||
yield Horizontal()
|
||||
yield Horizontal()
|
||||
|
||||
def on_mount(self) -> None:
|
||||
"""Event handler called when widget is added to the app."""
|
||||
self.timer_clocks = self.set_interval(1, self.update_core_vals)
|
||||
|
||||
def update_core_vals(self) -> None:
|
||||
"""Method to update GPU clock values to the current measurements.
|
||||
Run by a timer created 'on_mount'"""
|
||||
self.core_vals = get_core_stats(self.card)
|
||||
|
||||
def watch_core_vals(self, core_vals: dict) -> None:
|
||||
"""Called when the clocks attribute changes
|
||||
- Updates label values
|
||||
- Casting inputs to string to avoid type problems w/ int/None"""
|
||||
self.query_one("#clk_core_val",
|
||||
Static).update(f"{format_frequency(core_vals['sclk'])}")
|
||||
self.query_one("#util_pct",
|
||||
Static).update(f"{core_vals['util_pct']}%")
|
||||
self.query_one("#clk_voltage_val",
|
||||
Static).update(f"{core_vals['voltage']}V")
|
||||
self.query_one("#clk_memory_val",
|
||||
Static).update(f"{format_frequency(core_vals['mclk'])}")
|
||||
|
||||
|
||||
class PowerDisplay(Static):
|
||||
"""A widget to display GPU power stats."""
|
||||
|
||||
watts = reactive({"limit": 0,
|
||||
"limit_pct": 0,
|
||||
"average": 0,
|
||||
"capability": 0,
|
||||
"default": 0})
|
||||
|
||||
def __init__(self, card: str, hwmon_dir: str, *args, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
self.timer_watts = None
|
||||
self.card = card
|
||||
self.hwmon_dir = hwmon_dir
|
||||
|
||||
def compose(self) -> ComposeResult:
|
||||
yield Horizontal(Label("[underline]Power"),
|
||||
Label("", classes="statvalue"))
|
||||
yield Horizontal(Label(" Usage:",),
|
||||
Label("", id="pwr_avg_val", classes="statvalue"))
|
||||
yield Horizontal(Label(" Set Limit:",),
|
||||
Label("", id="pwr_lim_val", classes="statvalue"))
|
||||
yield Horizontal(Label(" Default Limit:",),
|
||||
Label("", id="pwr_def_val", classes="statvalue"))
|
||||
yield Horizontal(Label(" Capability:",),
|
||||
Label("", id="pwr_cap_val", classes="statvalue"))
|
||||
yield Horizontal()
|
||||
yield Horizontal()
|
||||
yield Horizontal()
|
||||
yield Horizontal()
|
||||
|
||||
def on_mount(self) -> None:
|
||||
"""Event handler called when widget is added to the app."""
|
||||
self.timer_watts = self.set_interval(1, self.update_watts)
|
||||
|
||||
def update_watts(self) -> None:
|
||||
"""Method to update GPU power values to current measurements.
|
||||
|
||||
Run by a timer created 'on_mount'"""
|
||||
self.watts = get_power_stats(self.card)
|
||||
|
||||
def watch_watts(self, watts: dict) -> None:
|
||||
"""Called when the 'watts' reactive attribute (var) changes.
|
||||
- Updates label values
|
||||
- Casting inputs to string to avoid type problems w/ int/None"""
|
||||
self.query_one("#pwr_avg_val",
|
||||
Static).update(f"{watts['limit_pct']}%, {watts['average']}W")
|
||||
self.query_one("#pwr_lim_val",
|
||||
Static).update(f"{watts['limit']}W")
|
||||
self.query_one("#pwr_def_val",
|
||||
Static).update(f"{watts['default']}W")
|
||||
self.query_one("#pwr_cap_val",
|
||||
Static).update(f"{watts['capability']}W")
|
||||
|
||||
|
||||
def start() -> None:
|
||||
'''Spawns the textual UI only during CLI invocation / after argparse'''
|
||||
if len(AMDGPU_CARDS) > 0:
|
||||
|
|
|
|||
|
|
@ -276,31 +276,25 @@ def get_gpu_usage(card: Optional[str] = None) -> int:
|
|||
return int(read_stat(stat_file))
|
||||
|
||||
|
||||
def get_temp_stats(card: Optional[str] = None) -> dict:
|
||||
def get_available_temps(card: Optional[str] = None) -> dict:
|
||||
"""
|
||||
Args:
|
||||
card (str, optional): ie: `card0`. See `AMDGPU_CARDS` or `find_cards()`
|
||||
|
||||
Raises:
|
||||
ValueError: If *no* AMD cards are found, or `card` is not one of them.
|
||||
Determined with `AMDGPU_CARDS`
|
||||
|
||||
Returns:
|
||||
dict: A dictionary of current GPU *temperature* related statistics
|
||||
dict: Discovered temperature `nodes` and paths to their value files
|
||||
|
||||
If none are found, will be empty.
|
||||
|
||||
Example:
|
||||
`{'name_temp_node_1': int, 'name_temp_node_2': int, 'name_temp_node_3': int}`
|
||||
|
||||
Dictionary keys (temp nodes/names) are constructed through discovery.
|
||||
|
||||
Driver provides temperatures in *millidegrees* C
|
||||
|
||||
Returned values are converted to 'C' as integers for simple comparison
|
||||
`{'edge': '/.../temp1_input', 'junction': '/.../temp2_input', 'mem': '/.../temp3_input'}`
|
||||
"""
|
||||
card = validate_card(card)
|
||||
hwmon_dir = AMDGPU_CARDS[card]
|
||||
# determine temperature nodes, construct a dict to store them
|
||||
# interface will iterate over these, creating labels as needed
|
||||
# determine temperature nodes/types, construct a dict to store them
|
||||
temp_files = {}
|
||||
temp_node_labels = glob.glob(path.join(hwmon_dir, "temp*_label"))
|
||||
for temp_node_label_file in temp_node_labels:
|
||||
|
|
@ -312,11 +306,34 @@ def get_temp_stats(card: Optional[str] = None) -> dict:
|
|||
temp_node_name = _node.read().strip()
|
||||
# add the node name/type and the corresponding temp file to the dict
|
||||
temp_files[temp_node_name] = temp_node_value_file
|
||||
return temp_files
|
||||
|
||||
temp_update = {}
|
||||
for temp_node, temp_file in temp_files.items():
|
||||
# iterate through the discovered temperature nodes
|
||||
# ... updating the dictionary with new stats
|
||||
_temperature = int(int(read_stat(temp_file)) // 1000)
|
||||
temp_update[temp_node] = _temperature
|
||||
return temp_update
|
||||
|
||||
def get_temp_stat(name: str, card: Optional[str] = None) -> dict:
|
||||
"""
|
||||
Args:
|
||||
card (str, optional): ie: `card0`. See `AMDGPU_CARDS` or `find_cards()`
|
||||
name (str): temperature *name*, ie: `edge`, `junction`, or `mem`
|
||||
|
||||
Raises:
|
||||
ValueError: If *no* AMD cards are found, or `card` is not one of them.
|
||||
*Or* Invalid temperature name is provided.
|
||||
|
||||
Returns:
|
||||
int: Requested GPU temperature (type, by `name`).
|
||||
Either the first AMD card, or one specified with `card=`.
|
||||
|
||||
Driver provides temperatures in *millidegrees* C
|
||||
|
||||
Returned values are converted to 'C' as integers for simple comparison
|
||||
"""
|
||||
card = validate_card(card)
|
||||
# determine temperature nodes/types, construct a dict to store them
|
||||
temp_files = get_available_temps(card=card)
|
||||
|
||||
# now that we know the temperature nodes/types for 'card', check request
|
||||
if name not in temp_files:
|
||||
raise ValueError(f'{name} does not appear to be valid, temp nodes: {list(temp_files.keys())}')
|
||||
|
||||
# if the requested temperature node was found, read it / convert to C
|
||||
return int(int(read_stat(temp_files[name])) // 1000)
|
||||
|
|
|
|||
Reference in a new issue