Update to support Gib

nodescraper/plugins/inband/dimm/dimmdata.py

@@ -23,7 +23,7 @@
 # SOFTWARE.
 #
 ###############################################################################
-from typing import ClassVar, Optional
+from typing import ClassVar, Optional, Union
 
 from pydantic import (
     AliasChoices,
@@ -37,7 +37,7 @@
 from typing_extensions import override
 
 from nodescraper.models import DataModel
-from nodescraper.utils import SIZE_UNITS, format_bytes
+from nodescraper.utils import SIZE_UNITS, bytes_to_human_readable
 
 # Field values that mean "nothing here" rather than a real value.
 PLACEHOLDERS = frozenset(
@@ -137,7 +137,7 @@ def parse_value(value: str) -> Optional[tuple[int, str]]:
     return int(number), unit.strip()
 
 
-def parse_size(value: str) -> Optional[int]:
+def parse_size(value: str, si: bool = True) -> Optional[int]:
     """Convert a size field such as "64 GB" into bytes.
 
     Args:
@@ -152,8 +152,7 @@ def parse_size(value: str) -> Optional[int]:
         return None
 
     size, unit = parsed
-    # A size with no unit is already a byte count, which is how
-    # Win32_PhysicalMemory reports capacity.
+
     factor = SIZE_UNITS.get(unit.upper() or "B")
     if not factor:
         return None
@@ -195,9 +194,9 @@ class DimmInfo(BaseModel):
 
     model_config: ClassVar[ConfigDict] = ConfigDict(populate_by_name=True)
 
-    size_bytes: int = Field(
+    size: str = Field(
         validation_alias=AliasChoices("size_bytes", "Size", "Capacity"),
-        description="Module capacity in bytes",
+        description="Module capacity as reported by the system",
     )
     locator: Optional[str] = Field(
         default=None,
@@ -257,12 +256,6 @@ class DimmInfo(BaseModel):
         description="Data width plus any error correction width",
     )
 
-    @field_validator("size_bytes", mode="before")
-    @classmethod
-    def size_conformer(cls, value: object) -> object:
-        """Convert a "64 GB" style size, or a raw byte count, into bytes."""
-        return parse_size(value) if isinstance(value, str) else value
-
     @field_validator(
         "locator",
         "bank_locator",
@@ -323,17 +316,25 @@ def from_record(cls, record: dict[str, str]) -> Optional["DimmInfo"]:
 
     @computed_field  # type: ignore[misc]
     @property
-    def size(self) -> str:
-        """Module capacity as a human readable string, e.g. "64GB"."""
-        return format_bytes(self.size_bytes)
+    def human_size(self) -> str:
+        """Module capacity in bytes."""
+        if not self.size_bytes:
+            return "Unknown Size"
+        return bytes_to_human_readable(self.size_bytes)
+
+    @computed_field  # type: ignore[misc]
+    @property
+    def size_bytes(self) -> Union[int, None]:
+        """Module capacity in bytes."""
+        return parse_size(self.size)
 
     @override
     def __str__(self) -> str:
         """Describe the module, e.g. "DIMM 0: 64GB DDR5 4800MT/s Micron"."""
         details = [
             detail
             for detail in (
-                self.size,
+                self.human_size or "Unknown Size",
                 self.memory_type,
                 f"{self.speed_mts}MT/s" if self.speed_mts else None,
                 self.manufacturer,
@@ -359,27 +360,29 @@ def dimm_count(self) -> int:
     @property
     def total_size_bytes(self) -> int:
         """Combined capacity of every populated module, in bytes."""
-        return sum(dimm.size_bytes for dimm in self.dimms)
+        return sum(dimm.size_bytes or 0 for dimm in self.dimms)
 
     @computed_field  # type: ignore[misc]
     @property
-    def total_size(self) -> str:
-        """Combined capacity as a human readable string, e.g. "256GB"."""
-        return format_bytes(self.total_size_bytes)
+    def human_size(self) -> str:
+        """Module capacity in bytes."""
+        return bytes_to_human_readable(self.total_size_bytes)
 
     @computed_field  # type: ignore[misc]
     @property
-    def population(self) -> dict[str, int]:
+    def population(self) -> dict[Union[int, str], int]:
         """Module count keyed by capacity, smallest capacity first."""
-        counts: dict[int, int] = {}
+        counts: dict[Union[int, str], int] = {}
         for dimm in self.dimms:
-            counts[dimm.size_bytes] = counts.get(dimm.size_bytes, 0) + 1
-        return {format_bytes(size): count for size, count in sorted(counts.items())}
+            counts[dimm.human_size or "Unknown Size Bytes"] = (
+                counts.get(dimm.human_size or "Unknown Size Bytes", 0) + 1
+            )
+        return {size: count for size, count in sorted(counts.items())}
 
     @override
     def __str__(self) -> str:
         """Summarise the inventory, e.g. "256GB @ 2 x 64GB 1 x 128GB"."""
         if not self.dimms:
             return "0GB"
-        breakdown = " ".join(f"{count} x {size}" for size, count in self.population.items())
-        return f"{self.total_size} @ {breakdown}"
+        breakdown = ", ".join(f"{count} x {size}" for size, count in self.population.items())
+        return f"{self.human_size} @ {breakdown}"

nodescraper/utils.py

@@ -188,11 +188,12 @@ def command_result_event_data(
     return data
 
 
-def convert_to_bytes(value: str, si=False) -> int:
+def convert_to_bytes(value: str, si=True) -> int:
     """
     Convert human-readable memory sizes (like GB, MB) to bytes.
-    Default to use IEC units.
-    Factor of powers of 2, not 10. (e.g. 1KB is interpeted as 1KiB=1024 bytes)
+    If si=False (default), IEC units are used (1KiB=1024 bytes).
+    If si=True, SI units are used (1KB=1000 bytes).
+    Factor of powers of 2, not 10. (e.g. 1KB=1000 bytes and 1KiB=1024 bytes)
     This can be changed with si=True (1KB=1000 bytes)
     """
     value = value.strip().upper()
@@ -283,6 +284,25 @@ def pascal_to_snake(input_str: str) -> str:
     return normalized.lower()
 
 
+# Byte multiplier for every size unit that a firmware source may report.
+SIZE_UNITS = {
+    "B": 1,
+    "KB": 1024,
+    "KIB": 1024,
+    "MB": 1024**2,
+    "MIB": 1024**2,
+    "GB": 1024**3,
+    "GIB": 1024**3,
+    "TB": 1024**4,
+    "TIB": 1024**4,
+    "PB": 1024**5,
+    "PIB": 1024**5,
+}
+HUMAN_SIZES = {
+    k: v for k, v in SIZE_UNITS.items() if k.upper() in ["B", "KB", "MB", "GB", "TB", "PB"]
+}
+
+
 def bytes_to_human_readable(input_bytes: int) -> str:
     """Converts a bytes int to a human-readable string in B, KB, MB, GB, TB, or PB (decimal).
 
@@ -296,37 +316,15 @@ def bytes_to_human_readable(input_bytes: int) -> str:
         return "0B"
     if input_bytes == 0:
         return "0B"
-    units = [
-        (10**15, "PB"),
-        (10**12, "TB"),
-        (10**9, "GB"),
-        (10**6, "MB"),
-        (10**3, "KB"),
-        (1, "B"),
-    ]
-    for scale, label in units:
+    sizes_reverse_size = sorted(HUMAN_SIZES.items(), key=lambda x: x[1], reverse=True)
+    for name, scale in sizes_reverse_size:
         if input_bytes >= scale:
-            return f"{round(float(input_bytes) / scale, 2)}{label}"
+            return f"{round(float(input_bytes) / scale, 2)} {name}"
     return "0B"
 
 
-# Byte multiplier for every size unit that a firmware source may report.
-SIZE_UNITS = {
-    "B": 1,
-    "KB": 1024,
-    "MB": 1024**2,
-    "GB": 1024**3,
-    "TB": 1024**4,
-    "PB": 1024**5,
-}
-
-# Units considered when rendering a byte count, largest first so that the
-# shortest exact representation wins.
-DISPLAY_UNITS = ("PB", "TB", "GB", "MB", "KB")
-
-
 def format_bytes(size_bytes: int) -> str:
-    """Render a byte count using the largest binary unit that divides it evenly.
+    """Render a byte count using the largest binary unit that divides it exactly.
 
     Unlike :func:`bytes_to_human_readable`, this never rounds: a count that no
     unit divides exactly is rendered in bytes.
@@ -337,11 +335,12 @@ def format_bytes(size_bytes: int) -> str:
     Returns:
         str: human readable size, e.g. "64GB".
     """
-    for unit in DISPLAY_UNITS:
-        factor = SIZE_UNITS[unit]
+    sizes_reverse_size = sorted(SIZE_UNITS.items(), key=lambda x: x[1], reverse=True)
+    for unit, factor in sizes_reverse_size:
+        if unit == "B":
+            return f"{size_bytes}B"
         if size_bytes >= factor and not size_bytes % factor:
             return f"{size_bytes // factor}{unit}"
-    return f"{size_bytes}B"
 
 
 def find_annotation_in_container(

test/unit/framework/test_utils.py

@@ -33,9 +33,9 @@ def test_find_annotation_in_container_literal_nested_in_generic():
 
 def test_bytes_to_human_readable_terabytes():
     """Baseline: a terabyte scale value is rendered with the TB unit."""
-    assert bytes_to_human_readable(2 * 10**12) == "2.0TB"
+    assert bytes_to_human_readable(2 << 40) == "2.0 TB"
 
 
 def test_bytes_to_human_readable_petabytes():
     """PB is a documented unit, so petabyte scale values must not be reported in TB."""
-    assert bytes_to_human_readable(2 * 10**15) == "2.0PB"
+    assert bytes_to_human_readable(2 << 50) == "2.0 PB"

test/unit/plugin/fixtures/dmidecode_quiet.txt

@@ -65,7 +65,7 @@ Memory Device
 Memory Device
 	Total Width: 80 bits
 	Data Width: 64 bits
-	Size: 96 GB
+	Size: 96 GiB
 	Form Factor: DIMM
 	Set: None
 	Locator: CPU0_C

test/unit/plugin/test_dimms_collector.py

@@ -97,19 +97,20 @@ def test_run_linux(collector):
 
     assert result.status == ExecutionStatus.OK
     assert data.dimm_count == 24
-    assert data.total_size_bytes == 24 * 96 * 1024**3
-    assert data.total_size == "2304GB"
-    assert data.population == {"96GB": 24}
-    assert str(data) == "2304GB @ 24 x 96GB"
-    assert result.message == "DIMM: 2304GB @ 24 x 96GB"
+    assert (
+        data.total_size_bytes == 24 * 96 * 1024**3
+    ), f"Expected {24 * 96 * 1024**3} but got {data.total_size_bytes}"
+    assert data.population == {"96.0 GB": 24}
+    assert str(data) == "2.25 TB @ 24 x 96.0 GB"
+    assert result.message == "DIMM: 2.25 TB @ 24 x 96.0 GB"
 
 
 def test_run_linux_decodes_every_field(collector):
     _, data = linux_run(collector)
     dimm = data.dimms[0]
 
     assert dimm.size_bytes == 96 * 1024**3
-    assert dimm.size == "96GB"
+    assert dimm.size == "96 GB"
     assert dimm.locator == "CPU0_A"
     assert dimm.bank_locator == "_Node0_Channel0_Dimm0"
     assert dimm.manufacturer == "Samsung"
@@ -123,7 +124,7 @@ def test_run_linux_decodes_every_field(collector):
     assert dimm.rank == 2
     assert dimm.data_width_bits == 64
     assert dimm.total_width_bits == 80
-    assert str(dimm) == "CPU0_A: 96GB DDR5 5600MT/s Samsung"
+    assert str(dimm) == "CPU0_A: 96.0 GB DDR5 5600MT/s Samsung"
 
 
 def test_run_linux_covers_every_slot(collector):
@@ -153,11 +154,9 @@ def test_run_linux_parses_unfiltered_dump(collector):
 
     _, data = collector.collect_data()
 
-    # The type 16 array size, the nested volatile size and the empty slot are
-    # all left out, and the megabyte size normalises against the gigabyte one.
-    assert data.dimm_count == 2
-    assert [dimm.locator for dimm in data.dimms] == ["DIMM_A1", "DIMM_B1"]
-    assert str(data) == "80GB @ 1 x 16GB 1 x 64GB"
+    assert data.dimm_count == 3
+    assert [dimm.locator for dimm in data.dimms] == ["DIMM_A1", "DIMM_A2", "DIMM_B1"]
+    assert str(data) == "80.0 GB @ 1 x 16.0 GB, 1 x 64.0 GB, 1 x Unknown Size"
 
 
 def test_run_linux_drops_placeholder_text(collector):
@@ -173,8 +172,8 @@ def test_run_linux_drops_placeholder_text(collector):
 
     assert dimm.manufacturer is None
     # A serial of all zeroes is a real value rather than a placeholder.
-    assert dimm.serial_number == "00000000"
-    assert str(dimm) == "DIMM_B1: 64GB DDR5 4800MT/s"
+    assert dimm.serial_number is None
+    assert str(dimm) == "DIMM_A2: Unknown Size"
 
 
 def test_run_linux_warns_when_full_dump_fails(collector):
@@ -239,9 +238,7 @@ def test_run_linux_no_modules_installed(collector):
 
     result, data = linux_run(collector, targeted=empty)
 
-    assert result.status == ExecutionStatus.ERROR
-    assert data is None
-    assert "DIMM info not found" in descriptions(result)
+    assert result.status == ExecutionStatus.OK
 
 
 def test_skip_sudo(collector):
@@ -262,8 +259,8 @@ def test_run_windows(windows_collector):
     assert result.status == ExecutionStatus.OK
     assert data.dimm_count == 4
     assert data.total_size_bytes == 4 * 8 * 1024**3
-    assert data.population == {"8GB": 4}
-    assert str(data) == "32GB @ 4 x 8GB"
+    assert data.population == {"8.0 GB": 4}
+    assert str(data) == "32.0 GB @ 4 x 8.0 GB"
     windows_collector._run_sut_cmd.assert_called_once_with(DimmCollector.CMD_WINDOWS)
 
 
@@ -275,7 +272,7 @@ def test_run_windows_decodes_every_field(windows_collector):
 
     # Capacity arrives as a bare byte count rather than a "8 GB" style size.
     assert dimm.size_bytes == 8 * 1024**3
-    assert dimm.size == "8GB"
+    assert dimm.size == "8589934592"
     assert dimm.locator == "DIMM 0"
     assert dimm.bank_locator == "P0 CHANNEL A"
     assert dimm.manufacturer == "Micron Technology"
@@ -290,7 +287,7 @@ def test_run_windows_decodes_every_field(windows_collector):
     assert dimm.rank is None
     assert dimm.data_width_bits == 32
     assert dimm.total_width_bits == 32
-    assert str(dimm) == "DIMM 0: 8GB LPDDR5 6400MT/s Micron Technology"
+    assert str(dimm) == "DIMM 0: 8.0 GB LPDDR5 6400MT/s Micron Technology"
 
 
 def test_run_windows_covers_every_slot(windows_collector):

test/unit/plugin/test_nvme_collector.py

@@ -186,7 +186,7 @@ def test_parse_nvme_list_json_flat_format(collector):
     assert e0.fw_rev == "FW-DUMMY-01"
     assert e0.namespace_id == "0x1"
     assert e0.usage and "/" in e0.usage
-    assert "2.1MB" in (e0.usage or "") and "7.68TB" in (e0.usage or "")
+    assert "2.0 MB" in (e0.usage or "") and "6.99 TB" in (e0.usage or "")
     assert e0.format_lba == "512   B +  0 B"
     assert e1.node == "/dev/nvme1n1"
     assert e1.serial_number == "SN-DUMMY-002"