jlay/amdgpu_stats
Archived
1
1
Fork 0
Code Activity

move stat gathering to functions

Browse source
This commit is contained in:
Josh Lay 2023-04-25 01:20:12 -05:00
parent ad3fe3a22f
commit 4c626ef8f7
Signed by: jlay
GPG key ID: B265E45CACAD108A
4 changed files with 131 additions and 80 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

26
README.md
View file

@ -31,22 +31,20 @@ Once installed, run `amdgpu-stats` in your terminal of choice
*Rudimentary* support as a module exists; functions / variables offered can be found in `amdgpu_stats.utils`
Of most interest:
- The function `find_card` which returns a tuple; the discovered card and hwmon directory
- The variables `SRC_FILES` and `TEMP_FILES`, dictionaries of hwmon-driven statistics
Example usage:
Demonstration:
```
In [1]: from amdgpu_stats.utils import find_card, SRC_FILES, TEMP_FILES
In [1]: import amdgpu_stats.utils
In [2]: print(find_card())
('card0', '/sys/class/drm/card0/device/hwmon/hwmon9')
In [2]: print(amdgpu_stats.utils.get_core_stats())
{'sclk': 0, 'mclk': 1000000000, 'voltage': 0.01, 'util_pct': 0}
In [3]: print(SRC_FILES)
{'pwr_limit': '/sys/class/drm/card0/device/hwmon/hwmon9/power1_cap', 'pwr_average': '/sys/class/drm/card0/device/hwmon/hwmon9/power1_average',
[...]
In [3]: print(amdgpu_stats.utils.get_power_stats())
{'limit': 281, 'average': 35, 'capability': 323, 'default': 281}
In [4]: print(TEMP_FILES)
{'edge': '/sys/class/drm/card0/device/hwmon/hwmon9/temp1_input', 'junction': '/sys/class/drm/card0/device/hwmon/hwmon9/temp2_input', 'mem': '/sys/class/drm/card0/device/hwmon/hwmon9/temp3_input'}
In [4]: print(amdgpu_stats.utils.get_temp_stats())
{'edge': 33, 'junction': 36, 'mem': 42}
In [5]: print(amdgpu_stats.utils.get_fan_stats())
{'fan_rpm': 0, 'fan_rpm_target': 0}
```
_Latest_ [Source file](https://github.com/joshlay/amdgpu_stats/blob/master/src/amdgpu_stats/utils.py)
See `help('amdgpu_stats.utils')` or [the module source](https://github.com/joshlay/amdgpu_stats/blob/master/src/amdgpu_stats/utils.py) for more info

2
pyproject.toml
View file

@ -1,6 +1,6 @@
[tool.poetry]
name = "amdgpu-stats"
version = "0.1.4"
version = "0.1.5"
description = "A simple TUI (using Textual) that shows AMD GPU statistics"
authors = ["Josh Lay <pypi@jlay.io>"]
repository = "https://github.com/joshlay/amdgpu_stats"

54
src/amdgpu_stats/interface.py
View file

@ -23,7 +23,7 @@ from textual.reactive import reactive
from textual.screen import Screen
from textual.widgets import Header, Footer, Static, TextLog, Label
from .utils import CARD, SRC_FILES, TEMP_FILES, format_frequency, read_stat
from .utils import CARD, SRC_FILES, TEMP_FILES, format_frequency, get_core_stats, get_fan_stats, get_power_stats, get_temp_stats
class LogScreen(Screen):
@ -120,8 +120,7 @@ class MiscDisplay(Static):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.timer_fan = None
self.timer_temp = None
self.timer_misc = None
def compose(self) -> ComposeResult:
yield Horizontal(Label("[underline]Temperatures"),
@ -140,30 +139,14 @@ class MiscDisplay(Static):
def on_mount(self) -> None:
"""Event handler called when widget is added to the app."""
self.timer_fan = self.set_interval(1, self.update_fan_stats)
self.timer_temp = self.set_interval(1, self.update_temp_stats)
self.timer_misc = self.set_interval(1, self.update_misc_stats)
def update_fan_stats(self) -> None:
"""Method to update the 'fan' values to current measurements.
def update_misc_stats(self) -> None:
"""Method to update the temp/fan values to current measurements.
Run by a timer created 'on_mount'"""
val_update = {
"fan_rpm": read_stat(SRC_FILES['fan_rpm']),
"fan_rpm_target": read_stat(SRC_FILES['fan_rpm_target'])
}
self.fan_stats = val_update
def update_temp_stats(self) -> None:
"""Method to update the 'temperature' values to current measurements.
Run by a timer created 'on_mount'"""
val_update = {}
for temp_node, temp_file in TEMP_FILES.items():
# iterate through the discovered temperature nodes
# ... updating the dictionary with new stats
_content = f'{int(read_stat(temp_file)) / 1000:.0f}C'
val_update[temp_node] = _content
self.temp_stats = val_update
self.fan_stats = get_fan_stats()
self.temp_stats = get_temp_stats()
def watch_fan_stats(self, fan_stats: dict) -> None:
"""Called when the 'fan_stats' reactive attr changes.
@ -179,7 +162,7 @@ class MiscDisplay(Static):
# check first if the reactive object has been updated with keys
if temp_node in temp_stats:
stat_dict_item = temp_stats[temp_node]
self.query_one("#temp_" + temp_node, Static).update(stat_dict_item)
self.query_one("#temp_" + temp_node, Static).update(f'{stat_dict_item}C')
class ClockDisplay(Static):
@ -212,28 +195,20 @@ class ClockDisplay(Static):
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 = {
"sclk": format_frequency(read_stat(SRC_FILES['core_clock'])),
"mclk": format_frequency(read_stat(SRC_FILES['memory_clock'])),
"voltage": float(
f"{int(read_stat(SRC_FILES['core_voltage'])) / 1000:.2f}"
),
"util_pct": read_stat(SRC_FILES['busy_pct']),
}
self.core_vals = get_core_stats()
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"{core_vals['sclk']}")
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"{core_vals['mclk']}")
Static).update(f"{format_frequency(core_vals['mclk'])}")
class PowerDisplay(Static):
@ -271,12 +246,7 @@ class PowerDisplay(Static):
"""Method to update GPU power values to current measurements.
Run by a timer created 'on_mount'"""
self.micro_watts = {
"limit": int(int(read_stat(SRC_FILES['pwr_limit'])) / 1000000),
"average": int(int(read_stat(SRC_FILES['pwr_average'])) / 1000000),
"capability": int(int(read_stat(SRC_FILES['pwr_cap'])) / 1000000),
"default": int(int(read_stat(SRC_FILES['pwr_default'])) / 1000000),
}
self.micro_watts = get_power_stats()
def watch_micro_watts(self, micro_watts: dict) -> None:
"""Called when the micro_watts attributes change.

129
src/amdgpu_stats/utils.py
View file

@ -15,14 +15,16 @@ from typing import Tuple, Optional
from humanfriendly import format_size
# utility file -- funcs / constants, intended to provide library function
# function to find the card / hwmon_dir -- assigned to vars, informs consts
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
looking for 'amdgpu' to find a card to monitor
... looking for 'amdgpu' to find a card to monitor
returns the cardN name and hwmon directory for stats"""
Returns:
A tuple containing the 'cardN' name and hwmon directory for stats
If no AMD GPU found, this will be: (None, None)
"""
_card = None
_hwmon_dir = None
hwmon_names_glob = '/sys/class/drm/card*/device/hwmon/hwmon*/name'
@ -38,25 +40,11 @@ def find_card() -> Optional[Tuple[Optional[str], Optional[str]]]:
return _card, _hwmon_dir
def read_stat(file: str) -> str:
"""given `file`, return the contents"""
with open(file, "r", encoding="utf-8") as _fh:
data = _fh.read().strip()
return data
def format_frequency(frequency_hz) -> str:
"""takes a frequency and formats it with an appropriate Hz suffix"""
return (
format_size(int(frequency_hz), binary=False)
.replace("B", "Hz")
.replace("bytes", "Hz")
)
# globals - card, hwmon directory, and statistic file paths derived from these
# base vars: card identifier, hwmon directory for stats, then the stat dicts
CARD, hwmon_dir = find_card()
card_dir = path.join("/sys/class/drm/", CARD) # eg: /sys/class/drm/card0/
# dictionary of known source files
# 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"),
@ -70,8 +58,10 @@ SRC_FILES = {'pwr_limit': path.join(hwmon_dir, "power1_cap"),
'fan_rpm': path.join(hwmon_dir, "fan1_input"),
'fan_rpm_target': path.join(hwmon_dir, "fan1_target"),
}
# determine temperature nodes, construct a dict to store them
# interface will iterate over these, creating labels as needed
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
@ -82,3 +72,96 @@ for temp_node_label_file in temp_node_labels:
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
def read_stat(file: str) -> str:
"""Given statistic file, `file`, return the contents"""
with open(file, "r", encoding="utf-8") as _fh:
data = _fh.read()
return data.strip()
def format_frequency(frequency_hz: int) -> str:
"""Takes a frequency (in Hz) and appends it with the appropriate suffix, ie:
- Hz
- MHz
- GHz"""
return (
format_size(frequency_hz, binary=False)
.replace("B", "Hz")
.replace("bytes", "Hz")
)
def get_power_stats() -> dict:
"""
Returns:
A dictionary of current GPU *power* related statistics.
{'limit': int,
'average': int,
'capability': int,
'default': int}
"""
return {"limit": int(int(read_stat(SRC_FILES['pwr_limit'])) / 1000000),
"average": int(int(read_stat(SRC_FILES['pwr_average'])) / 1000000),
"capability": int(int(read_stat(SRC_FILES['pwr_cap'])) / 1000000),
"default": int(int(read_stat(SRC_FILES['pwr_default'])) / 1000000)}
def get_core_stats() -> dict:
"""
Returns:
A dictionary of current GPU *core/memory* related statistics.
{'sclk': int,
'mclk': int,
'voltage': float,
'util_pct': int}
Clocks are in Hz, `format_frequency` may be used to normalize
"""
return {"sclk": int(read_stat(SRC_FILES['core_clock'])),
"mclk": int(read_stat(SRC_FILES['memory_clock'])),
"voltage": float(
f"{int(read_stat(SRC_FILES['core_voltage'])) / 1000:.2f}"
),
"util_pct": int(read_stat(SRC_FILES['busy_pct']))}
def get_fan_stats() -> dict:
"""
Returns:
A dictionary of current GPU *fan* related statistics.
{'fan_rpm': int,
'fan_rpm_target': int}
"""
return {"fan_rpm": int(read_stat(SRC_FILES['fan_rpm'])),
"fan_rpm_target": int(read_stat(SRC_FILES['fan_rpm_target']))}
def get_temp_stats() -> dict:
"""
Returns:
A dictionary of current GPU *temperature* related statistics.
Keys/values are dynamically contructed based on discovered nodes
{'name_temp_node_1': int,
'name_temp_node_2': int,
'name_temp_node_3': int}
Driver provides this in millidegrees C
Returned values are converted: (floor) divided by 1000 for *proper* C
As integers for comparison
"""
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