jlay/amdgpu_stats
Archived
1
1
Fork 0
Code Activity

start packaging

Browse source
This commit is contained in:
Josh Lay 2023-04-23 13:39:03 -05:00
parent cfbf5a435e
commit 10a35ae06f
Signed by: jlay
GPG key ID: B265E45CACAD108A
5 changed files with 84 additions and 78 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
.gitignore vendored Normal file
View file

@ -0,0 +1,2 @@
**/__pycache__/
/dist/

19
pyproject.toml Normal file
View file

@ -0,0 +1,19 @@
[tool.poetry]
name = "amdgpu-stats"
version = "0.1.0"
description = "A simple TUI (using Textual) that shows AMD GPU statistics"
authors = ["Josh Lay <python@jlay.io>"]
license = "MIT"
readme = "README.md"
[tool.poetry.dependencies]
python = "^3.9"
textual = ">=0.10"
humanfriendly = ">=10.0"
[tool.poetry.scripts]
amdgpu-stats = "amdgpu_stats.amdgpu_stats:main"
[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"

0
src/amdgpu_stats/__init__.py Normal file
View file

0
amdgpu_stats.css → src/amdgpu_stats/amdgpu_stats.css
View file

141
amdgpu_stats.py → src/amdgpu_stats/amdgpu_stats.py
View file

@ -7,11 +7,11 @@ TODO: restore argparse / --card, in case detection fails.
rich markup reference: rich markup reference:
https://rich.readthedocs.io/en/stable/markup.html https://rich.readthedocs.io/en/stable/markup.html
""" """
import argparse
from os import path from os import path
import glob import glob
import sys import sys
from typing import Tuple, Optional from typing import Tuple, Optional
import pkg_resources
# from textual import events # from textual import events
from textual.app import App, ComposeResult from textual.app import App, ComposeResult
@ -22,6 +22,7 @@ from textual.widgets import Header, Footer, Static, TextLog, Label
from humanfriendly import format_size from humanfriendly import format_size
# function to find the card / hwmon_dir -- assigned to vars, informs consts
def find_card() -> Optional[Tuple[Optional[str], Optional[str]]]: def find_card() -> Optional[Tuple[Optional[str], Optional[str]]]:
"""searches contents of /sys/class/drm/card*/device/hwmon/hwmon*/name """searches contents of /sys/class/drm/card*/device/hwmon/hwmon*/name
@ -43,6 +44,37 @@ def find_card() -> Optional[Tuple[Optional[str], Optional[str]]]:
return _card, _hwmon_dir return _card, _hwmon_dir
# globals - card, hwmon directory, and statistic file paths derived from these
CARD, hwmon_dir = find_card()
card_dir = path.join("/sys/class/drm/", CARD) # eg: /sys/class/drm/card0/
# ref: https://docs.kernel.org/gpu/amdgpu/thermal.html
SRC_FILES = {'pwr_limit': path.join(hwmon_dir, "power1_cap"),
'pwr_average': path.join(hwmon_dir, "power1_average"),
'pwr_cap': path.join(hwmon_dir, "power1_cap_max"),
'pwr_default': path.join(hwmon_dir, "power1_cap_default"),
'core_clock': path.join(hwmon_dir, "freq1_input"),
'core_voltage': path.join(hwmon_dir, "in0_input"),
'memory_clock': path.join(hwmon_dir, "freq2_input"),
'busy_pct': path.join(card_dir, "device/gpu_busy_percent"),
'temp_c': path.join(hwmon_dir, "temp1_input"),
'fan_rpm': path.join(hwmon_dir, "fan1_input"),
'fan_rpm_target': path.join(hwmon_dir, "fan1_target"),
}
TEMP_FILES = {}
# determine temperature nodes, construct an empty dict to store them
temp_node_labels = glob.glob(path.join(hwmon_dir, "temp*_label"))
for temp_node_label_file in temp_node_labels:
# determine the base node id, eg: temp1
# construct the path to the file that will label it. ie: edge/junction
temp_node_id = path.basename(temp_node_label_file).split('_')[0]
temp_node_value_file = path.join(hwmon_dir, f"{temp_node_id}_input")
with open(temp_node_label_file, 'r', encoding='utf-8') as _node:
temp_node_name = _node.read().strip()
print(f'found temp: {temp_node_name} (id: {temp_node_id})')
# add the node name/type and the corresponding temp file to the dict
TEMP_FILES[temp_node_name] = temp_node_value_file
def read_stat(file: str) -> str: def read_stat(file: str) -> str:
"""given `file`, return the contents""" """given `file`, return the contents"""
with open(file, "r", encoding="utf-8") as _fh: with open(file, "r", encoding="utf-8") as _fh:
@ -96,8 +128,7 @@ class GPUStats(App):
"""Textual-based tool to show AMDGPU statistics.""" """Textual-based tool to show AMDGPU statistics."""
# determine the real path of the script, to load the stylesheet # determine the real path of the script, to load the stylesheet
SCRIPT_PATH = path.dirname(path.realpath(__file__)) CSS_PATH = pkg_resources.resource_filename('amdgpu_stats', 'amdgpu_stats.css')
CSS_PATH = path.join(SCRIPT_PATH, "amdgpu_stats.css")
# initialize log screen # initialize log screen
SCREENS = {"logs": LogScreen()} SCREENS = {"logs": LogScreen()}
@ -116,9 +147,9 @@ class GPUStats(App):
yield Container(GPUStatsWidget()) yield Container(GPUStatsWidget())
self.update_log("[bold green]App started, logging begin!") self.update_log("[bold green]App started, logging begin!")
self.update_log("[bold italic]Information sources:[/]") self.update_log("[bold italic]Information sources:[/]")
for metric, source in src_files.items(): for metric, source in SRC_FILES.items():
self.update_log(f'[bold] {metric}:[/] {source}') self.update_log(f'[bold] {metric}:[/] {source}')
for metric, source in temp_files.items(): for metric, source in TEMP_FILES.items():
self.update_log(f'[bold] {metric} temperature:[/] {source}') self.update_log(f'[bold] {metric} temperature:[/] {source}')
def action_toggle_dark(self) -> None: def action_toggle_dark(self) -> None:
@ -154,7 +185,7 @@ class MiscDisplay(Static):
def compose(self) -> ComposeResult: def compose(self) -> ComposeResult:
yield Horizontal(Label("[underline]Temperatures"), yield Horizontal(Label("[underline]Temperatures"),
Label("", classes="statvalue")) Label("", classes="statvalue"))
for temp_node in temp_files: for temp_node in TEMP_FILES:
# capitalize the first letter for display # capitalize the first letter for display
caption = temp_node[0].upper() + temp_node[1:] caption = temp_node[0].upper() + temp_node[1:]
yield Horizontal(Label(f' {caption}:',), yield Horizontal(Label(f' {caption}:',),
@ -168,16 +199,16 @@ class MiscDisplay(Static):
def on_mount(self) -> None: def on_mount(self) -> None:
"""Event handler called when widget is added to the app.""" """Event handler called when widget is added to the app."""
self.timer_fan = self.set_interval(interval, self.update_fan_stats) self.timer_fan = self.set_interval(1, self.update_fan_stats)
self.timer_temp = self.set_interval(interval, self.update_temp_stats) self.timer_temp = self.set_interval(1, self.update_temp_stats)
def update_fan_stats(self) -> None: def update_fan_stats(self) -> None:
"""Method to update the 'fan' values to current measurements. """Method to update the 'fan' values to current measurements.
Run by a timer created 'on_mount'""" Run by a timer created 'on_mount'"""
val_update = { val_update = {
"fan_rpm": read_stat(src_files['fan_rpm']), "fan_rpm": read_stat(SRC_FILES['fan_rpm']),
"fan_rpm_target": read_stat(src_files['fan_rpm_target']) "fan_rpm_target": read_stat(SRC_FILES['fan_rpm_target'])
} }
self.fan_stats = val_update self.fan_stats = val_update
@ -186,7 +217,7 @@ class MiscDisplay(Static):
Run by a timer created 'on_mount'""" Run by a timer created 'on_mount'"""
val_update = {} val_update = {}
for temp_node, temp_file in temp_files.items(): for temp_node, temp_file in TEMP_FILES.items():
# iterate through the discovered temperature nodes # iterate through the discovered temperature nodes
# ... updating the dictionary with new stats # ... updating the dictionary with new stats
_content = f'{int(read_stat(temp_file)) / 1000:.0f}C' _content = f'{int(read_stat(temp_file)) / 1000:.0f}C'
@ -203,7 +234,7 @@ class MiscDisplay(Static):
def watch_temp_stats(self, temp_stats: dict) -> None: def watch_temp_stats(self, temp_stats: dict) -> None:
"""Called when the temp_stats reactive attr changes, updates labels""" """Called when the temp_stats reactive attr changes, updates labels"""
for temp_node in temp_files: for temp_node in TEMP_FILES:
# check first if the reactive object has been updated with keys # check first if the reactive object has been updated with keys
if temp_node in temp_stats: if temp_node in temp_stats:
stat_dict_item = temp_stats[temp_node] stat_dict_item = temp_stats[temp_node]
@ -235,19 +266,19 @@ class ClockDisplay(Static):
def on_mount(self) -> None: def on_mount(self) -> None:
"""Event handler called when widget is added to the app.""" """Event handler called when widget is added to the app."""
self.timer_clocks = self.set_interval(interval, self.update_core_vals) self.timer_clocks = self.set_interval(1, self.update_core_vals)
def update_core_vals(self) -> None: def update_core_vals(self) -> None:
"""Method to update GPU clock values to the current measurements. """Method to update GPU clock values to the current measurements.
Run by a timer created 'on_mount'""" Run by a timer created 'on_mount'"""
self.core_vals = { self.core_vals = {
"sclk": format_frequency(read_stat(src_files['core_clock'])), "sclk": format_frequency(read_stat(SRC_FILES['core_clock'])),
"mclk": format_frequency(read_stat(src_files['memory_clock'])), "mclk": format_frequency(read_stat(SRC_FILES['memory_clock'])),
"voltage": float( "voltage": float(
f"{int(read_stat(src_files['core_voltage'])) / 1000:.2f}" f"{int(read_stat(SRC_FILES['core_voltage'])) / 1000:.2f}"
), ),
"util_pct": read_stat(src_files['busy_pct']), "util_pct": read_stat(SRC_FILES['busy_pct']),
} }
def watch_core_vals(self, core_vals: dict) -> None: def watch_core_vals(self, core_vals: dict) -> None:
@ -289,17 +320,17 @@ class PowerDisplay(Static):
def on_mount(self) -> None: def on_mount(self) -> None:
"""Event handler called when widget is added to the app.""" """Event handler called when widget is added to the app."""
self.timer_micro_watts = self.set_interval(interval, self.update_micro_watts) self.timer_micro_watts = self.set_interval(1, self.update_micro_watts)
def update_micro_watts(self) -> None: def update_micro_watts(self) -> None:
"""Method to update GPU power values to current measurements. """Method to update GPU power values to current measurements.
Run by a timer created 'on_mount'""" Run by a timer created 'on_mount'"""
self.micro_watts = { self.micro_watts = {
"limit": int(int(read_stat(src_files['pwr_limit'])) / 1000000), "limit": int(int(read_stat(SRC_FILES['pwr_limit'])) / 1000000),
"average": int(int(read_stat(src_files['pwr_average'])) / 1000000), "average": int(int(read_stat(SRC_FILES['pwr_average'])) / 1000000),
"capability": int(int(read_stat(src_files['pwr_cap'])) / 1000000), "capability": int(int(read_stat(SRC_FILES['pwr_cap'])) / 1000000),
"default": int(int(read_stat(src_files['pwr_default'])) / 1000000), "default": int(int(read_stat(SRC_FILES['pwr_default'])) / 1000000),
} }
def watch_micro_watts(self, micro_watts: dict) -> None: def watch_micro_watts(self, micro_watts: dict) -> None:
@ -312,65 +343,19 @@ class PowerDisplay(Static):
self.query_one("#pwr_cap_val", Static).update(f"{micro_watts['capability']}W") self.query_one("#pwr_cap_val", Static).update(f"{micro_watts['capability']}W")
if __name__ == "__main__": def tui() -> None:
CARD, hwmon_dir = find_card() '''Spawns the textual UI only during CLI invocation / after argparse'''
# do the argparse dance app = GPUStats()
p = argparse.ArgumentParser( app.run()
# show the value for defaults in '-h/--help'
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
description="Show some basic AMD GPU stats -- tested on RX6xxx series",
)
# p.add_argument(
# "-c",
# "--card",
# type=str,
# default=AUTO_CARD,
# help="The GPU to inspect, see 'ls -lad /sys/class/drm/card*'",
# )
p.add_argument(
"-i",
"--interval",
type=float,
default=1.0,
help="The delay (in seconds) between polling for data",
)
p_args = p.parse_args()
interval = p_args.interval
# CARD = args.card
def main():
'''Main function, entrypoint for packaging'''
# exit if AMDGPU not found, otherwise - proceed, assigning stat files # exit if AMDGPU not found, otherwise - proceed, assigning stat files
if CARD is None: if CARD is None:
sys.exit('Could not find an AMD GPU, exiting.') sys.exit('Could not find an AMD GPU, exiting.')
tui()
card_dir = path.join("/sys/class/drm/", CARD) # eg: /sys/class/drm/card0/
# ref: https://docs.kernel.org/gpu/amdgpu/thermal.html
src_files = {'pwr_limit': path.join(hwmon_dir, "power1_cap"),
'pwr_average': path.join(hwmon_dir, "power1_average"),
'pwr_cap': path.join(hwmon_dir, "power1_cap_max"),
'pwr_default': path.join(hwmon_dir, "power1_cap_default"),
'core_clock': path.join(hwmon_dir, "freq1_input"),
'core_voltage': path.join(hwmon_dir, "in0_input"),
'memory_clock': path.join(hwmon_dir, "freq2_input"),
'busy_pct': path.join(card_dir, "device/gpu_busy_percent"),
'temp_c': path.join(hwmon_dir, "temp1_input"),
'fan_rpm': path.join(hwmon_dir, "fan1_input"),
'fan_rpm_target': path.join(hwmon_dir, "fan1_target"),
}
# determine temperature nodes, construct an empty dict to store them if __name__ == "__main__":
temp_files = {} main()
temp_node_labels = glob.glob(path.join(hwmon_dir, "temp*_label"))
for temp_node_label_file in temp_node_labels:
# determine the base node id, eg: temp1
# construct the path to the file that will label it. ie: edge/junction
temp_node_id = path.basename(temp_node_label_file).split('_')[0]
temp_node_value_file = path.join(hwmon_dir, f"{temp_node_id}_input")
with open(temp_node_label_file, 'r', encoding='utf-8') as _node:
temp_node_name = _node.read().strip()
print(f'found temp: {temp_node_name} (id: {temp_node_id})')
# add the node name/type and the corresponding temp file to the dict
temp_files[temp_node_name] = temp_node_value_file
# start the party, draw the app and start collecting metrics
app = GPUStats()
app.run()