diff --git a/nodescraper/plugins/inband/dimm/dimmdata.py b/nodescraper/plugins/inband/dimm/dimmdata.py index b48fe730..9bd25f92 100644 --- a/nodescraper/plugins/inband/dimm/dimmdata.py +++ b/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,9 +316,17 @@ 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: @@ -333,7 +334,7 @@ def __str__(self) -> str: 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}" diff --git a/nodescraper/utils.py b/nodescraper/utils.py index f862e83c..389f27ce 100644 --- a/nodescraper/utils.py +++ b/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( diff --git a/test/unit/framework/test_utils.py b/test/unit/framework/test_utils.py index 399ae438..7dc9e28e 100644 --- a/test/unit/framework/test_utils.py +++ b/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" diff --git a/test/unit/plugin/fixtures/dmidecode_quiet.txt b/test/unit/plugin/fixtures/dmidecode_quiet.txt index ea5ef41d..45ca540a 100644 --- a/test/unit/plugin/fixtures/dmidecode_quiet.txt +++ b/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 diff --git a/test/unit/plugin/test_dimms_collector.py b/test/unit/plugin/test_dimms_collector.py index 3bcf65a6..7e9f6073 100644 --- a/test/unit/plugin/test_dimms_collector.py +++ b/test/unit/plugin/test_dimms_collector.py @@ -97,11 +97,12 @@ 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): @@ -109,7 +110,7 @@ def test_run_linux_decodes_every_field(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): diff --git a/test/unit/plugin/test_nvme_collector.py b/test/unit/plugin/test_nvme_collector.py index 1d0e819b..6cde21bd 100644 --- a/test/unit/plugin/test_nvme_collector.py +++ b/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"