diff --git a/nodescraper/plugins/inband/amdsmi/amdsmi_collector.py b/nodescraper/plugins/inband/amdsmi/amdsmi_collector.py index 56796ab7..67bda30b 100644 --- a/nodescraper/plugins/inband/amdsmi/amdsmi_collector.py +++ b/nodescraper/plugins/inband/amdsmi/amdsmi_collector.py @@ -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" diff --git a/nodescraper/plugins/inband/amdsmi/amdsmidata.py b/nodescraper/plugins/inband/amdsmi/amdsmidata.py index a26c49c3..41070935 100644 --- a/nodescraper/plugins/inband/amdsmi/amdsmidata.py +++ b/nodescraper/plugins/inband/amdsmi/amdsmidata.py @@ -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, @@ -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: @@ -81,7 +83,7 @@ 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 @@ -89,7 +91,7 @@ def na_to_none(values: Union[int, str]): 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 @@ -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 @@ -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): @@ -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 @@ -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): @@ -1285,7 +1291,7 @@ 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 @@ -1293,7 +1299,7 @@ def validate_ecc_blocks(cls, value: Union[dict[str, EccData], str]) -> dict[str, @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 @@ -1306,6 +1312,7 @@ class LinkStatusTable(Enum): DOWN = "D" DISABLED = "X" SELF = "SELF" + NA = NA_STR class BiDirectionalTable(Enum): diff --git a/nodescraper/plugins/inband/amdsmi/amdsmidata_host.py b/nodescraper/plugins/inband/amdsmi/amdsmidata_host.py index 6f53e6fc..a1d26b33 100644 --- a/nodescraper/plugins/inband/amdsmi/amdsmidata_host.py +++ b/nodescraper/plugins/inband/amdsmi/amdsmidata_host.py @@ -20,6 +20,7 @@ ) from nodescraper.plugins.inband.amdsmi.amdsmidata import ( + NA_STR, AmdSmiBaseModel, EccData, EccState, @@ -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 @@ -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