jlay/amdgpu_stats
Archived
1
1
Fork 0
Code Activity

tabular TUI updates

Browse source 
TUI updates
This commit is contained in:
Josh Lay 2023-05-01 22:02:09 -05:00 committed by GitHub
commit 22bbabc034
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
7 changed files with 258 additions and 253 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
README.md
View file

@ -2,9 +2,7 @@
A Python module/TUI for AMD GPU statistics
![Screenshot of main screen](https://raw.githubusercontent.com/joshlay/amdgpu_stats/master/screens/main.png "Main screen")
Supported GPUs and temperature nodes (`edge`/`junction`/etc.) are discovered automatically.
![Screenshot of main screen](https://raw.githubusercontent.com/joshlay/amdgpu_stats/master/screens/main.svg "Main screen")
Tested _only_ on `RX6000` series cards; APUs and more _may_ be supported. Please file an issue if finding incompatibility!

4
pyproject.toml
View file

@ -1,6 +1,6 @@
[tool.poetry]
name = "amdgpu-stats"
version = "0.1.11"
version = "0.1.12"
description = "A module/TUI for AMD GPU statistics"
authors = ["Josh Lay <pypi@jlay.io>"]
repository = "https://github.com/joshlay/amdgpu_stats"
@ -10,7 +10,7 @@ documentation = "https://amdgpu-stats.readthedocs.io/en/latest/"
[tool.poetry.dependencies]
python = "^3.9"
textual = ">=0.10"
textual = ">=0.11"
humanfriendly = ">=10.0"
[tool.poetry.scripts]

BIN
screens/main.png
View file

Binary file not shown.
Side by side Swipe Overlay

Before

Width:  |  Height:  |  Size: 32 KiB

After

Width:  |  Height:  |  Size: 22 KiB

123
screens/main.svg Normal file
View file

File diff suppressed because one or more lines are too long
Side by side

After

Width:  |  Height:  |  Size: 20 KiB

29
src/amdgpu_stats/style.css
View file

@ -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%;
}

287
src/amdgpu_stats/tui.py
View file

@ -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,185 +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,
"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['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:

64
src/amdgpu_stats/utils.py
View file

@ -132,12 +132,17 @@ def get_power_stats(card: Optional[str] = None) -> dict:
"""
card = validate_card(card)
hwmon_dir = AMDGPU_CARDS[card]
return {"limit": read_stat(path.join(hwmon_dir, "power1_cap"), stat_type='power'),
_pwr = {"limit": read_stat(path.join(hwmon_dir, "power1_cap"), stat_type='power'),
"limit_pct": 0,
"average": read_stat(path.join(hwmon_dir, "power1_average"), stat_type='power'),
"capability": read_stat(path.join(hwmon_dir, "power1_cap_max"), stat_type='power'),
"default": read_stat(path.join(hwmon_dir, "power1_cap_default"), stat_type='power')}
if _pwr['limit'] != 0:
_pwr['limit_pct'] = round((_pwr['average'] / _pwr['limit']) * 100, 1)
return _pwr
def get_core_stats(card: Optional[str] = None) -> dict:
"""
@ -271,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:
@ -307,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)