Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion nodescraper/plugins/inband/amdsmi/amdsmi_collector.py
Original file line number Diff line number Diff line change
Expand Up @@ -1312,7 +1312,7 @@ def get_static(self) -> Optional[list[AmdSmiStatic]]:
)

# NUMA
numa_node = int(numa.get("node", 0) or 0)
numa_node = numa.get("node")
affinity_raw = numa.get("affinity")
if affinity_raw in (None, "", "N/A"):
affinity_val: Union[int, str] = "N/A"
Expand Down
31 changes: 19 additions & 12 deletions nodescraper/plugins/inband/amdsmi/amdsmidata.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@
###############################################################################
import re
from enum import Enum
from typing import Any, ClassVar, Mapping, Optional, Union
from typing import Any, ClassVar, Literal, Mapping, Optional, Union

from pydantic import (
AliasChoices,
Expand All @@ -40,11 +40,13 @@
from nodescraper.models.datamodel import DataModel, FileModel
from nodescraper.utils import find_annotation_in_container

NA_STR = "N/A"
NA_LITERAL = Literal["N/A"]
_NUM_UNIT_RE = re.compile(r"^\s*([-+]?\d+(?:\.\d+)?)(?:\s*([A-Za-z%/][A-Za-z0-9%/._-]*))?\s*$")


def _value_unit_is_na(x: Any) -> bool:
return x is None or (isinstance(x, str) and x.strip().upper() in {"N/A", "NA", ""})
return x is None or (isinstance(x, str) and x.strip().upper() in {NA_STR, "NA", ""})


def _coerce_value_unit_raw(v: Any) -> Any:
Expand Down Expand Up @@ -81,15 +83,15 @@ def _coerce_value_unit_raw(v: Any) -> Any:


def na_to_none(values: Union[int, str]):
if values == "N/A":
if values == NA_STR:
return None
return values


def na_to_none_list(values: list[Union[int, str, None]]) -> list[Union[int, str, None]]:
ret_list: list[Union[int, str, None]] = values.copy()
for i in range(len(ret_list)):
if ret_list[i] == "N/A":
if ret_list[i] == NA_STR:
ret_list[i] = None
return ret_list

Expand All @@ -99,14 +101,14 @@ def na_to_none_dict(values: object) -> Optional[dict[str, Any]]:
Accepts None; returns None for 'N/A'/'NA'/'' or non-mapping inputs."""
if values is None:
return None
if isinstance(values, str) and values.strip().upper() in {"N/A", "NA", ""}:
if isinstance(values, str) and values.strip().upper() in {NA_STR, "NA", ""}:
return None
if not isinstance(values, Mapping):
return None

out: dict[str, Any] = {}
for k, v in values.items():
if isinstance(v, str) and v.strip().upper() in {"N/A", "NA", ""}:
if isinstance(v, str) and v.strip().upper() in {NA_STR, "NA", ""}:
out[k] = None
else:
out[k] = v
Expand Down Expand Up @@ -239,7 +241,7 @@ class EccState(Enum):
SING_C = "SING_C"
MULT_UC = "MULT_UC"
POISON = "POISON"
NA = "N/A"
NA = NA_STR


class ProcessListItem(BaseModel):
Expand Down Expand Up @@ -441,13 +443,13 @@ def _normalize_static_ppt_block(val: object) -> object:
"""Drop N/A-only ``ppt0``/``ppt1`` blocks; pass through already-parsed models."""
if val is None:
return None
if isinstance(val, str) and val.strip().upper() in {"N/A", "NA", ""}:
if isinstance(val, str) and val.strip().upper() in {NA_STR, "NA", ""}:
return None
if not isinstance(val, dict):
return val
out: dict[str, Any] = {}
for key, raw in val.items():
if isinstance(raw, str) and raw.strip().upper() in {"N/A", "NA", ""}:
if isinstance(raw, str) and raw.strip().upper() in {NA_STR, "NA", ""}:
continue
if raw is not None:
out[key] = raw
Expand Down Expand Up @@ -521,8 +523,12 @@ class StaticXgmiPlpd(BaseModel):


class StaticNuma(BaseModel):
node: int

model_config = ConfigDict(extra="allow")

node: Optional[int]
affinity: Union[int, str] # can be N/A
na_validator = field_validator("node", mode="before")(na_to_none)


class StaticVram(AmdSmiBaseModel):
Expand Down Expand Up @@ -1285,15 +1291,15 @@ def clock_rails(self) -> MetricClockRails:
def validate_ecc_blocks(cls, value: Union[dict[str, EccData], str]) -> dict[str, EccData]:
"""Validate the ecc_blocks field."""
if isinstance(value, str):
# If it's a string, we assume it's "N/A" and return an empty dict
# If it's a string, we assume it's NA_STR and return an empty dict
return {}
return value

@field_validator("energy", mode="before")
@classmethod
def validate_energy(cls, value: Optional[Any]) -> Optional[MetricEnergy]:
"""Validate the energy field."""
if value == "N/A" or value is None:
if value == NA_STR or value is None:
return None
return value

Expand All @@ -1306,6 +1312,7 @@ class LinkStatusTable(Enum):
DOWN = "D"
DISABLED = "X"
SELF = "SELF"
NA = NA_STR


class BiDirectionalTable(Enum):
Expand Down
7 changes: 4 additions & 3 deletions nodescraper/plugins/inband/amdsmi/amdsmidata_host.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
)

from nodescraper.plugins.inband.amdsmi.amdsmidata import (
NA_STR,
AmdSmiBaseModel,
EccData,
EccState,
Expand All @@ -35,11 +36,11 @@ def _host_na_to_none(value: Any) -> Any:
ValueUnit coercion does not collapse the dict form inside a ``ValueUnit | None``
union, so normalize both to None before field validation.
"""
if isinstance(value, str) and value.strip().upper() in ("N/A", "NA", ""):
if isinstance(value, str) and value.strip().upper() in (NA_STR, "NA", ""):
return None
if isinstance(value, dict):
inner = value.get("value")
if isinstance(inner, str) and inner.strip().upper() in ("N/A", "NA", ""):
if isinstance(inner, str) and inner.strip().upper() in (NA_STR, "NA", ""):
return None
return value

Expand Down Expand Up @@ -410,7 +411,7 @@ def validate_ecc_blocks(cls, value):
@field_validator("energy", mode="before")
@classmethod
def validate_energy(cls, value):
if value == "N/A" or value is None:
if value == NA_STR or value is None:
return None
return value

Expand Down
Loading