diff --git a/.clang-format b/.clang-format new file mode 100644 index 0000000..3be2907 --- /dev/null +++ b/.clang-format @@ -0,0 +1,17 @@ +Language: Cpp +BreakBeforeBraces: Stroustrup +PointerAlignment: Right +IndentWidth: 2 +AccessModifierOffset: 0 +ColumnLimit: 80 +NamespaceIndentation: All +AlignTrailingComments: true +AllowAllParametersOfDeclarationOnNextLine: true +AlwaysBreakTemplateDeclarations: true +AlignAfterOpenBracket: AlwaysBreak +UseTab: Never +IncludeBlocks: Preserve +AlignConsecutiveDeclarations: true +AlignConsecutiveAssignments: true +SpacesInParentheses: false +SpaceBeforeParens: ControlStatements diff --git a/MANIFEST.in b/MANIFEST.in index 024d041..2ec4679 100644 --- a/MANIFEST.in +++ b/MANIFEST.in @@ -1,4 +1,5 @@ include LICENSE include README.md -recursive-include src/hwprobe/interops/win/bindings *.dll -recursive-include src/hwprobe/interops/mac/bindings *.dylib +recursive-include src/hwprobe/interops/win/bindings *.dll +recursive-include src/hwprobe/interops/mac/bindings *.dylib +recursive-include src/hwprobe/interops/linux/bindings *.so diff --git a/pyproject.toml b/pyproject.toml index a0b72fe..7a4669b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -31,13 +31,19 @@ dependencies = [ where = ["src"] [tool.setuptools.package-data] -"hwprobe.interops.win.bindings" = ["*.dll"] -"hwprobe.interops.mac.bindings" = ["*.dylib"] +"hwprobe.interops.win.bindings" = ["*.dll"] +"hwprobe.interops.mac.bindings" = ["*.dylib"] +"hwprobe.interops.linux.bindings" = ["*.so"] [project.urls] Homepage = "https://github.com/Mahasvan/HWProbe" Issues = "https://github.com/Mahasvan/HWProbe/issues" +[dependency-groups] +dev = [ + "ruff>=0.16.3", +] + [tool.ruff] target-version = "py39" line-length = 120 diff --git a/src/hwprobe/core/common/edid.py b/src/hwprobe/core/common/edid.py index d6bfee0..027f51a 100644 --- a/src/hwprobe/core/common/edid.py +++ b/src/hwprobe/core/common/edid.py @@ -1,3 +1,5 @@ +from typing import Optional + from hwprobe.models.display_models import DisplayModuleInfo, ResolutionInfo BIT_DEPTH_ENUM = {1: 6, 2: 8, 3: 10, 4: 12, 5: 14, 6: 16} @@ -17,6 +19,14 @@ 0xFC: "Display Product Name", } +CTA_EXTENSION_TAG = 0x02 +DISPLAYID_EXTENSION_TAG = 0x70 +DISPLAYID_TYPE_I_TIMING_TAG = 0x03 + +# (width, height, refresh_rate) +ResolutionCandidate = tuple[int, int, float] +_NO_RESOLUTION: ResolutionCandidate = (0, 0, 0) + def _get_bits(data: bytes, start_bit: int, end_bit: int) -> int: # Get the bit values in an offset, given a bytes object. @@ -39,80 +49,201 @@ def _get_bits(data: bytes, start_bit: int, end_bit: int) -> int: return (value >> shift) & ((1 << length) - 1) -def parse_edid(edid_data: bytes) -> DisplayModuleInfo: - # todo: Parse EDID v1.2 and v1.3. This will work for v1.4, but need to verify on the older versions. - module = DisplayModuleInfo() +def _detailed_timing(descriptor: bytes) -> Optional[ResolutionCandidate]: + # Parse an 18-byte EDID Detailed Timing Descriptor into (width, height, refresh_rate). + # Returns None if this slot isn't a timing descriptor (e.g. all-zero padding/tag block). + if len(descriptor) < 18 or descriptor[:2] == b"\x00\x00": + return None - edid_version = (edid_data[0x12], edid_data[0x13]) + pixel_clock_hz = (descriptor[0] | (descriptor[1] << 8)) * 10_000 - module.year = edid_data[0x11] + 1990 + horiz = ((descriptor[4] & 0xF0) << 4) | descriptor[2] + vert = ((descriptor[7] & 0xF0) << 4) | descriptor[5] - manuf_bits = int.from_bytes(edid_data[0x08:0x0A], byteorder="big") - char1 = chr(((manuf_bits >> 10) & 0x1F) + 64) - char2 = chr(((manuf_bits >> 5) & 0x1F) + 64) - char3 = chr((manuf_bits & 0x1F) + 64) - manuf_string = char1 + char2 + char3 + h_blank = ((descriptor[4] & 0x0F) << 8) | descriptor[3] + v_blank = ((descriptor[7] & 0x0F) << 8) | descriptor[6] - module.manufacturer_code = manuf_string + total_h = horiz + h_blank + total_v = vert + v_blank + if total_h == 0 or total_v == 0: + return None - product_code = edid_data[0x0A:0x0C].hex().upper() + refresh_rate = pixel_clock_hz / (total_h * total_v) + return horiz, vert, round(refresh_rate, 2) - id_serial_number = "0x" + edid_data[0x0C:0x10].hex().upper() - input_type = edid_data[0x14] - module.resolution = ResolutionInfo() +def _type_i_timing(entry: bytes) -> Optional[ResolutionCandidate]: + # Parse a 20-byte DisplayID Type I Timing entry into (width, height, refresh_rate). + # Layout verified byte-for-byte against a real DisplayID extension block: pixel clock + # is a 24-bit LE value in units of 10 kHz (stored as actual-1), and active/blank pixel + # counts are 16-bit LE (also stored as actual-1) - unlike the base/CTA DTD format. + if len(entry) < 20: + return None + + pixel_clock_hz = ((entry[0] | (entry[1] << 8) | (entry[2] << 16)) + 1) * 10_000 + + h_active = (entry[4] | (entry[5] << 8)) + 1 + h_blank = (entry[6] | (entry[7] << 8)) + 1 + v_active = (entry[12] | (entry[13] << 8)) + 1 + v_blank = (entry[14] | (entry[15] << 8)) + 1 + + total_h = h_active + h_blank + total_v = v_active + v_blank + if total_h == 0 or total_v == 0: + return None + + refresh_rate = pixel_clock_hz / (total_h * total_v) + return h_active, v_active, round(refresh_rate, 2) - if input_type >> 7 == 1: # MSB is 1 => Digital output - if edid_version >= (1, 4): - module.resolution.bit_depth = BIT_DEPTH_ENUM.get( - _get_bits(input_type.to_bytes(1, byteorder="little"), 1, 4), 0 - ) - module.interface = INTERFACE_ENUM.get(input_type & 7, "Unknown") - else: - module.interface = "Analog" - resolution = (0, 0, 0) # Width, Height, Refresh Rate - # We will use this tuple to find the max resolution and refresh rate, and update it in `module.resolution`. +def _better_resolution( + current: ResolutionCandidate, candidate: Optional[ResolutionCandidate] +) -> ResolutionCandidate: + """Keep whichever of current/candidate has the larger area, breaking ties on refresh rate.""" + if candidate is None: + return current + return max(current, candidate, key=lambda r: (r[0] * r[1], r[2])) + + +def _parse_manufacturer_code(manuf_bytes: bytes) -> str: + """Decode the 3-letter PNP manufacturer ID packed into bytes 0x08-0x0A (5 bits/letter, offset from 'A'-1).""" + manuf_bits = int.from_bytes(manuf_bytes, byteorder="big") + return "".join(chr(((manuf_bits >> shift) & 0x1F) + 64) for shift in (10, 5, 0)) + + +def _parse_video_input( + input_type: int, edid_version: tuple[int, int] +) -> tuple[Optional[int], Optional[str]]: + """ + Decode the video input definition byte (offset 0x14) into (bit_depth, interface). + + Either value may come back None if it isn't determinable - this matches the original + behavior where pre-1.4 digital displays don't report bit depth/interface here at all. + """ + if input_type >> 7 != 1: + return None, "Analog" + + if edid_version < (1, 4): + return None, None + + bit_depth = BIT_DEPTH_ENUM.get(_get_bits(input_type.to_bytes(1, byteorder="little"), 1, 4), 0) + interface = INTERFACE_ENUM.get(input_type & 7, "Unknown") + return bit_depth, interface + + +def _process_display_descriptors( + edid_data: bytes, +) -> tuple[Optional[str], Optional[str], ResolutionCandidate]: + """ + Walk the four 18-byte descriptor blocks in the base EDID (offset 0x36-0x6C). + + Each block is either a Detailed Timing Descriptor, or a display descriptor + (serial number / product name / other, tagged by DESCRIPTOR_TAG_ENUM). + Returns (serial_number, name, best_resolution_found). + """ + serial_number = None + name = None + resolution = _NO_RESOLUTION for block_start in range(0x36, 0x6D, 18): block = edid_data[block_start : block_start + 18] - if block[:2] == b"\x00\x00": - tag = block[3] - if tag in DESCRIPTOR_TAG_ENUM: - # Refer to DESCRIPTOR_TAG_ENUM for valid block type codes - if tag == 0xFF: - # todo: test if this works - module.serial_number = block[5:].decode("ascii").strip() - elif tag == 0xFC: - module.name = block[5:].decode("ascii").strip() - - else: - if not module.resolution: - continue - - pixel_clock_hz = (block[0] | (block[1] << 8)) * 10_000 - - horiz = ((block[4] & 0xF0) << 4) | block[2] - vert = ((block[7] & 0xF0) << 4) | block[5] - - h_blank = ((block[4] & 0x0F) << 8) | block[3] - v_blank = ((block[7] & 0x0F) << 8) | block[6] - refresh_rate = pixel_clock_hz / ((horiz + h_blank) * (vert + v_blank)) - - resolution = max(resolution, (horiz, vert, round(refresh_rate, 2)), key=lambda x: (x[0] * x[1], x[2])) - - if resolution != (0, 0, 0): - if not module.resolution: - module.resolution = ResolutionInfo() - module.resolution.width = resolution[0] - module.resolution.height = resolution[1] - module.resolution.refresh_rate = resolution[2] - # print("\nRaw EDID:") - # for byte in edid_data: - # print(f"{byte:02X}", end=" ") + if block[:2] != b"\x00\x00": + resolution = _better_resolution(resolution, _detailed_timing(block)) + continue - return module + tag = block[3] + if tag == 0xFF: + # todo: test if this works + serial_number = block[5:].decode("ascii").strip() + elif tag == 0xFC: + name = block[5:].decode("ascii").strip() + return serial_number, name, resolution + + +def _process_cta_extension(ext_block: bytes, resolution: ResolutionCandidate) -> ResolutionCandidate: + """Scan a CTA-861 extension block for additional Detailed Timing Descriptors.""" + dtd_offset = ext_block[2] + if dtd_offset < 4: + # 0 means this extension carries no Detailed Timing Descriptors + return resolution + + for descriptor_start in range(dtd_offset, 127, 18): + timing = _detailed_timing(ext_block[descriptor_start : descriptor_start + 18]) + resolution = _better_resolution(resolution, timing) + + return resolution + + +def _process_displayid_extension(ext_block: bytes, resolution: ResolutionCandidate) -> ResolutionCandidate: + """Scan a DisplayID extension block for Type I Timing data blocks. + + Structure: [tag, version, section_size, product_type, ext_count], followed by a + sequence of data blocks: [tag, revision, payload_len, *payload]. + """ + section_end = min(5 + ext_block[2], 127) + offset = 5 + + while offset + 3 <= section_end: + block_tag = ext_block[offset] + payload_len = ext_block[offset + 2] + payload_start = offset + 3 + + if payload_start + payload_len > section_end: + break + + if block_tag == DISPLAYID_TYPE_I_TIMING_TAG: + for entry_start in range(payload_start, payload_start + payload_len, 20): + timing = _type_i_timing(ext_block[entry_start : entry_start + 20]) + resolution = _better_resolution(resolution, timing) + + offset = payload_start + payload_len + + return resolution + + +def parse_edid(edid_data: bytes) -> DisplayModuleInfo: + # todo: Parse EDID v1.2 and v1.3. This will work for v1.4, but need to verify on the older versions. + if len(edid_data) < 128: + raise ValueError(f"EDID data too short: expected at least 128 bytes, got {len(edid_data)}") + + module = DisplayModuleInfo() + module.resolution = ResolutionInfo() + + edid_version = (edid_data[0x12], edid_data[0x13]) + module.year = edid_data[0x11] + 1990 + module.manufacturer_code = _parse_manufacturer_code(edid_data[0x08:0x0A]) + + bit_depth, interface = _parse_video_input(edid_data[0x14], edid_version) + if bit_depth is not None: + module.resolution.bit_depth = bit_depth + if interface is not None: + module.interface = interface + + serial_number, name, resolution = _process_display_descriptors(edid_data) + if serial_number is not None: + module.serial_number = serial_number + if name is not None: + module.name = name + + # High-refresh-rate timings (e.g. 4K@120Hz+) are frequently declared only in a + # CTA-861 or DisplayID extension block rather than the base 128 bytes, so scan those too. + num_extensions = edid_data[0x7E] if len(edid_data) > 0x7E else 0 + for i in range(num_extensions): + ext_start = 128 + i * 128 + ext_block = edid_data[ext_start : ext_start + 128] + if len(ext_block) < 128: + continue + + if ext_block[0] == CTA_EXTENSION_TAG: + resolution = _process_cta_extension(ext_block, resolution) + elif ext_block[0] == DISPLAYID_EXTENSION_TAG: + resolution = _process_displayid_extension(ext_block, resolution) + + if resolution != _NO_RESOLUTION: + module.resolution.width = resolution[0] + module.resolution.height = resolution[1] + module.resolution.refresh_rate = resolution[2] -# todo: parse extension blocks + return module \ No newline at end of file diff --git a/src/hwprobe/core/linux/common.py b/src/hwprobe/core/linux/common.py index bf6e279..a7d8bf3 100644 --- a/src/hwprobe/core/linux/common.py +++ b/src/hwprobe/core/linux/common.py @@ -1,50 +1,130 @@ -# Source: https://github.com/KernelWanderers/OCSysInfo/blob/main/src/util/pci_root.py - import posixpath import re +from typing import Optional +PCI_ROOT_PATH = "/sys/bus/pci/devices/" _PCI_BDF_PATTERN = re.compile(r"^[0-9a-fA-F]{4}:[0-9a-fA-F]{2}:[0-9a-fA-F]{2}\.[0-7]$") +_PCI_BDF_PATH_PATTERN = re.compile(r"(pci([a-f\d]{4}):([a-f\d]{2}))|(([a-f\d]{4}):([a-f\d]{2}):([a-f\d]{2})\.([a-f\d]{1}))", re.IGNORECASE) - -def pci_path_linux(device_slot: str): +def _resolve_acpi_path(device_bdf: str) -> tuple[Optional[str], bool]: """ - :param device_slot: format: ::. - :return: PCI path, e.g. PciRoot(0x0)/Pci(0x2,0x0) + Resolve the ACPI path for a given PCI device. + + :param device_bdf: The BDF identifier (Bus:Device.Function) of the device + + :return: The resolved ACPI path if found, else None. + + The way this function works is: + 1. It first checks if the device has a direct ACPI path in its sysfs entry. + 2. If not, it checks the parent devices recursively until it finds an ACPI path or reaches the root. """ - try: - domain = int(device_slot.split(":")[0], 16) - except (IndexError, ValueError): - return None + ret_val = None, False + device_path = posixpath.join(PCI_ROOT_PATH, device_bdf) + if not posixpath.exists(device_path): + return ret_val - slots = _resolve_device_chain_from_sysfs(device_slot) or [device_slot] - pci_components = [_format_pci_component(s) for s in slots] - pci_suffix = "".join(f"/Pci({c})" for c in pci_components if c) - return f"PciRoot({hex(domain)}){pci_suffix}" + acpi_path = _read_from_sysfs(device_path, "firmware_node", "path") + if acpi_path: + return acpi_path, True + device_path = posixpath.realpath(device_path) -def _format_pci_component(slot_name: str): - """Return 'slot,func' as hex string, e.g. '0x1f,0x3', or None.""" + # Parent directory should be something like RRRR:BB:DD.F try: - device_func = slot_name.split(":")[-1] - slot, func = device_func.split(".") - return f"{hex(int(slot, 16))},{hex(int(func, 16))}" - except (ValueError, IndexError, AttributeError): + while (acpi_path := _read_from_sysfs(device_path, "firmware_node", "path")) is None: + if device_path == posixpath.dirname(device_path): + # We've reached the root of the filesystem without finding an ACPI path + return ret_val + + device_path = posixpath.dirname(device_path) + + except Exception: + return ret_val + + # Device has no ACPI path, it is found via PCI enumeration + # Return parent ACPI path instead. + return acpi_path, False + + +# Linux implemented this very annoyingly +# - https://tldp.org/LDP/tlk/dd/pci.html +# - https://wiki.osdev.org/PCI +def pci_path_linux(device_slot: str) -> Optional[str]: + """ + :param device_slot: format: ::. + :return: UEFI-style PCI path, e.g. PciRoot(0x0)/Pci(0x2,0x0) + """ + # Invalid fallback value + if not device_slot or not _PCI_BDF_PATTERN.match(device_slot): + return None + + raw_path = f"/sys/bus/pci/devices/{device_slot}/" + + path = posixpath.realpath(raw_path) + if not path: return None + pci_root = "" + pci_segments = [] + + r""" + On Linux, PCI devices are described directly in their sysfs path structure. + I.e: /sys/devices////... + + We are trying to extract the PCI root value, and the PCI function segments from this path. + This is done by finding all matches of the PCI BDF PATH PATTERN in the path, + and using the match groups to determine if it's a root or segment path. + + The base regex contains two distinct capture groups: + pci([0-9a-fA-F]{4}):([0-9a-fA-F]{2})) + ([0-9a-fA-F]{4}):([0-9a-fA-F]{2}):([0-9a-fA-F]{2})\.([0-9a-fA-F]{1})) + + This will then allow us to determine if a match is a root or segment path + by checking match[0], if it's an empty string: it's a segment path, otherwise it's a root path. + + Given a path: '/sys/devices/pci0000:00/0000:00:03.1/0000:09:00.0' + + This would make the PCI hierarchy: PciRoot(0x0)/Pci(0x3,0x1)/Pci(0x0,0x0) + + Then, when matching, we'd get a result like so: + matches = [ + ('pci0000:00', '0000', '00', '', '', '', '', ''), + ('', '', '', '0000:00:03.1', '0000', '00', '03', '1'), + ('', '', '', '0000:09:00.0', '0000', '09', '00', '0')P + ] -def _resolve_device_chain_from_sysfs(device_slot: str): - """Return ordered PCI BDFs from root bridge to endpoint for a device.""" - sysfs_path = posixpath.realpath(f"/sys/bus/pci/devices/{device_slot}") - if not sysfs_path: + Structure of a single match: + + PCI Root PCI function (segment) + _ _ _ _ _|_ _ _ _ _ _ _ _ _ _ _ _ | _ _ _ _ _ _ _ + | | | | | | | | + ( 'pciXXXX:YY', 'XXXX', 'YY', 'XXXX:YY:ZZ.F', 'XXXX', 'YY', 'ZZ', 'F' ) + """ + matches = _PCI_BDF_PATH_PATTERN.findall(path) + for match in matches: + if not match: + continue + + if match[0]: + pci_root = f"PciRoot(0x{int(match[1], 16):x})" + elif match[4]: + pci_segments.append(f"Pci(0x{int(match[6], 16):x},0x{int(match[7], 16):x})") + + if not pci_root or not pci_segments: return None - bdfs = [p for p in sysfs_path.split(posixpath.sep) if _PCI_BDF_PATTERN.match(p)] - if not bdfs: + return f"{pci_root}/{'/'.join(pci_segments)}" + + +def _read_from_sysfs(base: str, *paths) -> Optional[str]: + """Read a string from a sysfs file, return None if not found.""" + path = posixpath.join(base, *paths) + + if not posixpath.exists(path): return None try: - end = next(i for i, b in enumerate(bdfs) if b.lower() == device_slot.lower()) - except StopIteration: + with open(path) as f: + return f.read().strip() + except Exception: return None - - return bdfs[: end + 1] diff --git a/src/hwprobe/core/linux/display.py b/src/hwprobe/core/linux/display.py index 9ec4cff..6acdd06 100644 --- a/src/hwprobe/core/linux/display.py +++ b/src/hwprobe/core/linux/display.py @@ -6,6 +6,7 @@ from hwprobe.core.common.edid import INTERFACE_ENUM, parse_edid from hwprobe.core.linux.common import pci_path_linux from hwprobe.models.display_models import DisplayInfo, DisplayModuleInfo +from hwprobe.models.gpu_models import GPUInfo from hwprobe.models.status_models import StatusType _PCI_BDF_PATTERN = re.compile(r"^[0-9a-fA-F]{4}:[0-9a-fA-F]{2}:[0-9a-fA-F]{2}\.[0-7]$") @@ -23,6 +24,21 @@ "DSI": "DSI", } +def _resolve_parent_gpu_by_bdf(pci_bdf: str, gpu_devices: list[GPUInfo]) -> Optional[str]: + """ + Given a PCI BDF (::.) of a display device, + find the parent GPU in the list of GPUInfo objects. + """ + if pci_bdf is None: + return None + + for gpu in gpu_devices: + path = pci_path_linux(pci_bdf) + if path and gpu.pci_path == path: + return gpu.name + + return None + def _extract_pci_bdf_from_sysfs_path(path: str) -> Optional[str]: """Extract the endpoint PCI BDF from a resolved sysfs path.""" @@ -40,14 +56,13 @@ def _parse_connector_type(device_path: str) -> Optional[str]: return DRM_CONNECTOR_TYPE.get(m.group(1)) -def _fetch_individual_monitor_info(device_path: str) -> Optional[DisplayModuleInfo]: +def _fetch_individual_monitor_info( + device_path: str, + gpu_devices: list[GPUInfo] +) -> Optional[DisplayModuleInfo]: edid_path = posixpath.join(device_path, "edid") if not posixpath.exists(edid_path): return None - parent_path = posixpath.join(device_path, "device") - - # todo: populate parent graphics card info - # we have vendor and device ids of the parent gpu. When PCI-IDs integration is done, use it to get name with open(edid_path, "rb") as f: edid_data = f.read() @@ -59,10 +74,13 @@ def _fetch_individual_monitor_info(device_path: str) -> Optional[DisplayModuleIn if connector_type := _parse_connector_type(device_path): monitor_data.interface = connector_type - pci_path_full = posixpath.realpath(parent_path) - pci_bdf = _extract_pci_bdf_from_sysfs_path(pci_path_full) - if pci_bdf: - monitor_data.pci_path = pci_path_linux(pci_bdf) + parent_path = posixpath.realpath(device_path) + if parent_path != device_path: + pci_bdf = _extract_pci_bdf_from_sysfs_path(posixpath.realpath(parent_path)) + + # Resolve parent GPU based on the PCI BDF + if pci_bdf is not None and (parent_gpu := _resolve_parent_gpu_by_bdf(pci_bdf, gpu_devices)) is not None: + monitor_data.gpu_name = parent_gpu acpi_file = posixpath.join(device_path, "firmware_node", "path") if posixpath.exists(acpi_file): @@ -72,7 +90,9 @@ def _fetch_individual_monitor_info(device_path: str) -> Optional[DisplayModuleIn return monitor_data -def fetch_display_info(): +def fetch_display_info( + gpu_devices: list[GPUInfo] +): display_info = DisplayInfo() pattern = re.compile(r"^card\d+$") root_path = "/sys/class/drm" @@ -89,7 +109,7 @@ def fetch_display_info(): children = [x for x in os.listdir(parent_path) if x.startswith("card")] for child in children: try: - response = _fetch_individual_monitor_info(posixpath.join(parent_path, child)) + response = _fetch_individual_monitor_info(posixpath.join(parent_path, child), gpu_devices) if response: display_info.modules.append(response) except Exception as e: diff --git a/src/hwprobe/core/linux/graphics.py b/src/hwprobe/core/linux/graphics.py index 36d6963..e472640 100644 --- a/src/hwprobe/core/linux/graphics.py +++ b/src/hwprobe/core/linux/graphics.py @@ -1,116 +1,54 @@ -import glob import os import posixpath -import subprocess from typing import Optional -from hwprobe.core.linux.common import pci_path_linux +from hwprobe.core.linux.common import PCI_ROOT_PATH, _read_from_sysfs, _resolve_acpi_path, pci_path_linux +from hwprobe.interops.linux.bindings.gpu_info import GPUInfoQueryStatus from hwprobe.models.gpu_models import GPUInfo, GraphicsInfo from hwprobe.models.size_models import Megabyte from hwprobe.models.status_models import StatusType -from hwprobe.util.nvidia import fetch_gpu_details_nvidia -# Currently, the info in /sys/class/drm/cardX is being used. -# todo: Check if lspci and lshw -c display can be used -# https://unix.stackexchange.com/questions/393/how-to-check-how-many-lanes-are-used-by-the-pcie-card +# Try to import native C library bindings +try: + from hwprobe.interops.linux.bindings import gpu_info as native_gpu -PCI_ROOT_PATH = "/sys/bus/pci/devices/" + NATIVE_AVAILABLE = native_gpu.is_available() +except (ImportError, RuntimeError): + NATIVE_AVAILABLE = False +DISPLAY_CONTROLLER_CLASS = 0x03 # Display Controller class code in PCI -def _vram_amd(device: str) -> Optional[int]: - ROOT_PATH = "/sys/bus/pci/devices/" - vram_files = posixpath.join(*[ROOT_PATH, device, "drm", "card*", "device", "mem_info_vram_total"]) - try: - drm_files = glob.glob(vram_files) - if drm_files: - with open(drm_files[0]) as f: - vram_bits = int(f.read().strip()) - vram_mb = int(vram_bits / 1024 / 1024) - return vram_mb - return None - except Exception: - return None - +def _pcie_gen(raw_speed: Optional[str]) -> Optional[int]: + # Path example: /sys/bus/pci/devices/0000:03:00.0/max_link_speed -def _pcie_gen(device: str) -> Optional[int]: - # Path example: /sys/bus/pci/devices/0000:03:00.0/current_link_speed - path = f"/sys/bus/pci/devices/{device}/current_link_speed" - - if not posixpath.exists(path): + if not raw_speed: return None - try: - with open(path) as f: - raw_speed = f.read().strip() # e.g., "16.0 GT/s" - - # Mapping Dictionary - speed_to_gen = {"2.5 GT/s": 1, "5.0 GT/s": 2, "8.0 GT/s": 3, "16.0 GT/s": 4, "32.0 GT/s": 5, "64.0 GT/s": 6} + # Mapping Dictionary + speed_to_gen = {"2.5 GT/s": 1, "5.0 GT/s": 2, "8.0 GT/s": 3, "16.0 GT/s": 4, "32.0 GT/s": 5, "64.0 GT/s": 6} - for k, v in speed_to_gen.items(): - """ `8.0 GT/s PCIe` may be a possible candidate, so we dont use direct matching""" - if k in raw_speed: - return v + for k, v in speed_to_gen.items(): + """ `8.0 GT/s PCIe` may be a possible candidate, so we dont use direct matching""" + if k in raw_speed: + return v - return None - - except Exception: - return None + return None def _check_gpu_class(device: str) -> bool: - path = posixpath.join(PCI_ROOT_PATH, device) - with open(posixpath.join(path, "class")) as f: - device_class = f.read().strip() """ The class code is three hex-bytes, where the leftmost hex-byte is the base class We want the devices of base class 0x03, which denotes a Display Controller. """ - class_code = int(device_class, base=16) - base_class = class_code >> 16 + device_class = _read_from_sysfs(PCI_ROOT_PATH, device, "class") - return base_class == 3 + if device_class is None: + return False + class_code = int(device_class, base=16) + base_class = class_code >> 16 -def _populate_amd_info(gpu: GPUInfo, device: str) -> GPUInfo: - # get VRAM for AMD GPUs - vram_capacity = _vram_amd(device) - if vram_capacity is not None: - gpu.vram = Megabyte(capacity=vram_capacity) - return gpu - - -def _populate_nvidia_info(gpu: GPUInfo, device: str) -> GPUInfo: - gpu_name, pcie_width, pcie_gen, vram_total = fetch_gpu_details_nvidia(device) - if gpu_name: - gpu.name = gpu_name - if pcie_width: - gpu.pcie_width = pcie_width - if pcie_gen: - gpu.pcie_gen = pcie_gen - if vram_total: - gpu.vram = Megabyte(capacity=vram_total) - - return gpu - - -def _populate_lspci_info(gpu: GPUInfo, device: str) -> GPUInfo: - lspci_output = subprocess.run(["lspci", "-s", device, "-vmm"], capture_output=True, text=True, check=True).stdout - # We gather all data here and parse whatever data we have. Subsystem data may not be returned. - # If LSPCI not found, check=True ensures error is thrown - - data = {} - for line in lspci_output.splitlines(): - if ":" in line: - key, value = line.split(":", maxsplit=1) - data[key.strip()] = value.strip() - - gpu.manufacturer = data.get("Vendor") - gpu.name = data.get("Device") - gpu.subsystem_manufacturer = data.get("SVendor") - gpu.subsystem_model = data.get("SDevice") - - return gpu - + return base_class == DISPLAY_CONTROLLER_CLASS def fetch_graphics_info() -> GraphicsInfo: graphics_info = GraphicsInfo() @@ -121,66 +59,83 @@ def fetch_graphics_info() -> GraphicsInfo: return graphics_info for device in os.listdir(PCI_ROOT_PATH): - # print("Found device: ", device) - try: - if not _check_gpu_class(device): - continue - except Exception as e: - graphics_info.status.type = StatusType.PARTIAL - graphics_info.status.messages.append(f"Could not open file for {device}: {e}") + if not _check_gpu_class(device): continue gpu = GPUInfo() gpu_path = posixpath.join(PCI_ROOT_PATH, device) - try: - with open(posixpath.join(gpu_path, "vendor")) as f: - gpu.vendor_id = f.read().strip() - with open(posixpath.join(gpu_path, "device")) as f: - gpu.device_id = f.read().strip() - with open(posixpath.join(gpu_path, "current_link_width")) as f: - width = f.read().strip() - if width.isnumeric() and int(width) > 0: - gpu.pcie_width = int(width) - except Exception as e: - graphics_info.status.type = StatusType.PARTIAL - graphics_info.status.messages.append(f"Could not get GPU properties: {e}") - try: - with open(posixpath.join(gpu_path, "firmware_node", "path")) as f: - acpi_path = f.read().strip() + if (vendor_id := _read_from_sysfs(gpu_path, "vendor")) is not None: + gpu.vendor_id = vendor_id + else: + graphics_info.status.make_partial(f"Could not read vendor ID for {device}") + + + if (device_id := _read_from_sysfs(gpu_path, "device")) is not None: + gpu.device_id = device_id + else: + graphics_info.status.make_partial(f"Could not read device ID for {device}") + + + if (cur_width := _read_from_sysfs(gpu_path, "current_link_width")) is not None: + if cur_width.isnumeric() and int(cur_width) > 0: + cur_width = int(cur_width) + else: + graphics_info.status.make_partial(f"Could not read current link width for {device}") + + if (cur_pcie_speed := _read_from_sysfs(gpu_path, "current_link_speed")) is not None: + if cur_pcie_speed: + cur_pcie_speed = _pcie_gen(cur_pcie_speed) + else: + graphics_info.status.make_partial(f"Could not read current link speed for {device}") + + + acpi_path, result = _resolve_acpi_path(device) + if acpi_path is not None: gpu.acpi_path = acpi_path - except Exception as e: - graphics_info.status.type = StatusType.PARTIAL - graphics_info.status.messages.append(f"Could not get ACPI path: {e}") - try: - pci_path = pci_path_linux(device) - gpu.pci_path = pci_path - except Exception as e: - graphics_info.status.type = StatusType.PARTIAL - graphics_info.status.messages.append(f"Could not get PCI path: {e}") - if pcie_gen := _pcie_gen(device): - gpu.pcie_gen = pcie_gen + if not result: + graphics_info.status.messages.append(f"ACPI path for {device} was inferred through parent device, device itself was likely found via PCI enumeration") + else: + graphics_info.status.make_partial(f"Could not read ACPI path for {device}") + + + if (pci_path := pci_path_linux(device)) is not None: + gpu.pci_path = pci_path + else: + graphics_info.status.make_partial(f"Could not resolve PCI path for {device}") + + if not NATIVE_AVAILABLE: + graphics_info.status.make_partial(f"Native GPU info library not available, cannot fetch GPU name or VRAM for {device}") + elif vendor_id is None: + graphics_info.status.make_partial(f"Vendor ID not available, cannot fetch GPU name or VRAM for {device}") + elif (ret_native := native_gpu.get_gpu_info(device, int(vendor_id, 16))) is None: # pyright: ignore[reportPossiblyUnboundVariable] + graphics_info.status.make_partial(f"Native GPU info library could not fetch GPU name or VRAM for {device} with vendor ID {vendor_id}") else: - graphics_info.status.type = StatusType.PARTIAL - graphics_info.status.messages.append("Could not get PCI gen") - - if gpu.vendor_id == "0x1002": - gpu = _populate_amd_info(gpu, device) - elif gpu.vendor_id and gpu.vendor_id.lower() == "0x10de": - # get VRAM for Nvidia GPUs - try: - gpu = _populate_nvidia_info(gpu, device) - except Exception as e: - graphics_info.status.type = StatusType.PARTIAL - graphics_info.status.messages.append(f"Could not get additional GPU info for NVIDIA GPU {device}: {e}") - - try: - gpu = _populate_lspci_info(gpu, device) - except Exception as e: - graphics_info.status.type = StatusType.PARTIAL - graphics_info.status.messages.append(f"Could not parse LSPCI output for GPU {device}: {e}") + ret, native = ret_native + if ret == GPUInfoQueryStatus.FAILURE: + graphics_info.status.make_partial(f"Native GPU info library could not fetch GPU name or VRAM for {device} with vendor ID {vendor_id}") + else: + if ret & GPUInfoQueryStatus.VULKAN_NAME_FALLBACK: + graphics_info.status.make_partial(f"Native GPU info library used Vulkan fallback to fetch GPU name for {device} with vendor ID {vendor_id}") + gpu.name = native.name + if native.vram_total_mb > 0: + gpu.vram = Megabyte(capacity=int(native.vram_total_mb)) + if ret & GPUInfoQueryStatus.VULKAN_VRAM_FALLBACK: + graphics_info.status.make_partial(f"Native GPU info library used Vulkan fallback to fetch VRAM for {device} with vendor ID {vendor_id}") + else: + graphics_info.status.make_partial(f"Native GPU info library returned VRAM size of 0 for {device} with vendor ID {vendor_id}") + + if isinstance(cur_width, int): + gpu.pcie_width = cur_width + + if isinstance(cur_pcie_speed, int): + gpu.pcie_gen = cur_pcie_speed graphics_info.modules.append(gpu) + if not graphics_info.modules: + graphics_info.status.type = StatusType.FAILED + graphics_info.status.messages.append("No GPU modules found") + return graphics_info diff --git a/src/hwprobe/core/linux/manager.py b/src/hwprobe/core/linux/manager.py index 54f05c5..a1eaca4 100644 --- a/src/hwprobe/core/linux/manager.py +++ b/src/hwprobe/core/linux/manager.py @@ -4,7 +4,6 @@ from hwprobe.core.linux.memory import fetch_memory_info from hwprobe.core.linux.network import fetch_network_info from hwprobe.core.linux.storage import fetch_storage_info -from hwprobe.models.display_models import DisplayInfo from hwprobe.models.gpu_models import GraphicsInfo from hwprobe.models.info_models import ( CPUInfo, @@ -12,6 +11,7 @@ HardwareManagerInterface, LinuxHardwareInfo, MemoryInfo, + DisplayInfo, ) from hwprobe.models.network_models import NetworkInfo from hwprobe.models.storage_models import StorageInfo @@ -47,6 +47,17 @@ def fetch_graphics_info(self) -> GraphicsInfo: self.info.graphics = fetch_graphics_info() return self.info.graphics + def fetch_display_info(self) -> DisplayInfo: + if not self.info.graphics or not len(self.info.graphics.modules): + self.fetch_graphics_info() + + # Both 'ty' and 'pyright' complain for two reasons: + # 1. The attribute 'graphics' is optional and may not be present + # 2. The attribute 'modules' is attached to the 'graphics' attribute, which is optional + # + # However, with all of our checks in this method, we know that 'graphics' is present and 'modules' is not empty. + return fetch_display_info(self.info.graphics.modules) # ty: ignore[unresolved-attribute] # pyright: ignore[reportOptionalMemberAccess] + def fetch_hardware_info(self) -> HardwareInfo: self.fetch_cpu_info() self.fetch_graphics_info() @@ -55,8 +66,5 @@ def fetch_hardware_info(self) -> HardwareInfo: self.fetch_storage_info() return self.info - def fetch_display_info(self) -> DisplayInfo: - return fetch_display_info() - def fetch_network_info(self) -> NetworkInfo: return fetch_network_info() diff --git a/src/hwprobe/interops/linux/CMakeLists.txt b/src/hwprobe/interops/linux/CMakeLists.txt new file mode 100644 index 0000000..832a193 --- /dev/null +++ b/src/hwprobe/interops/linux/CMakeLists.txt @@ -0,0 +1,85 @@ +cmake_minimum_required(VERSION 3.21) +project(LinuxDeviceInfo LANGUAGES C CXX) + +set(CMAKE_C_STANDARD 11) +set(CMAKE_C_STANDARD_REQUIRED ON) + +set(CMAKE_CXX_STANDARD 17) +set(CMAKE_CXX_STANDARD_REQUIRED ON) +set(CMAKE_CXX_EXTENSIONS OFF) +set(CMAKE_EXPORT_COMPILE_COMMANDS ON) + +if(NOT CMAKE_BUILD_TYPE) + set(CMAKE_BUILD_TYPE Release CACHE STRING "Build type" FORCE) +endif() + +find_package(PkgConfig REQUIRED) +find_package(Vulkan REQUIRED) +pkg_check_modules(LIBDRM REQUIRED libdrm) + +# --------------------------------------------------------------------- +# Shared library +# --------------------------------------------------------------------- + +add_library( + device_info + SHARED + src/gpu_info.cpp +) + +target_include_directories( + device_info + PUBLIC + ${CMAKE_CURRENT_SOURCE_DIR}/include + PRIVATE + ${LIBDRM_INCLUDE_DIRS} +) + +# Link against libdrm and the Vulkan loader. We no longer need libdl for dlopen/dlsym. +if(TARGET Vulkan::Vulkan) + target_link_libraries( + device_info + PRIVATE + ${LIBDRM_LIBRARIES} + Vulkan::Vulkan + ) +else() + target_link_libraries( + device_info + PRIVATE + ${LIBDRM_LIBRARIES} + ${Vulkan_LIBRARY} + ) + target_include_directories( + device_info + PRIVATE + ${Vulkan_INCLUDE_DIRS} + ) +endif() + +target_compile_options( + device_info + PRIVATE + ${LIBDRM_CFLAGS_OTHER} + -Wall + -Werror +) + +if(CMAKE_BUILD_TYPE STREQUAL "Release") + target_compile_options( + device_info + PRIVATE + -g + -fno-omit-frame-pointer + ) +endif() + +add_executable(LinuxDeviceInfo main.c) +target_link_libraries(LinuxDeviceInfo PRIVATE device_info) + +set_target_properties( + device_info + PROPERTIES + LIBRARY_OUTPUT_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}/bindings + OUTPUT_NAME device_info +) diff --git a/src/hwprobe/interops/linux/README.md b/src/hwprobe/interops/linux/README.md new file mode 100644 index 0000000..7b19712 --- /dev/null +++ b/src/hwprobe/interops/linux/README.md @@ -0,0 +1,113 @@ +# LinuxDeviceInfo + +A tiny Linux utility and shared library that enumerates GPUs using DRM ioctls (vendor-specific) with a Vulkan +fallback (universal), reporting model name and VRAM without relying on external tools like `lspci` or `nvidia-smi`. + +The native library lives in `src/` and `include/`, and is exposed via a command-line tester (`main.c`). +Also powers a thin Python `ctypes` binding in `bindings/gpu_info.py`. + +This is intended to be used via each hardware component's respective python interface, like `gpu_info.py`. The CLI tool +is primarily for testing and demonstration purposes, but it can be used directly if desired. + +Full disclosure: A big part of this C++ connector was written by Claude. +If you are someone with more know-how, and find lapses in this code, we'd be more than happy to welcome Pull Requests. + +## Vendor Coverage + +| Vendor | VRAM Source | Notes | +| ---------- | ----------------------------------------- | --------------------------------------- | +| **AMD** | `AMDGPU_INFO_VRAM_GTT` ioctl | Direct kernel interface | +| **Intel** | `DRM_I915_QUERY_MEMORY_REGIONS` ioctl | Intel Arc + integrated | +| **NVIDIA** | Nouveau DRM ioctl + Vulkan fallback | Vulkan preferred for proprietary driver | +| **Any** | Vulkan `VkPhysicalDeviceMemoryProperties` | Universal fallback via `libvulkan.so.1` | + +## Requirements + +- Linux with a populated `/sys/class/drm` (kernel 5.16+ for Intel Arc VRAM queries) +- GCC/Clang with C++17 support +- CMake 3.21+ +- `libdrm` development headers (linked at build time) +- Python 3.7+ (for the `gpu_info.py` binding) - Assuming you want to compile this to use with HWProbe. +- `libvulkan.so.1` at runtime and other Vulkan-related packages + +```bash +# Debian/Ubuntu +sudo apt install build-essential cmake pkg-config libdrm-dev libvulkan-dev vulkan-headers + +# Fedora/RHEL +sudo dnf install gcc-c++ cmake pkgconf-pkg-config libdrm-devel vulkan-loader-devel vulkan-headers + +# Arch +sudo pacman -S base-devel cmake pkgconf libdrm vulkan-headers vulkan-icd-loader +``` + +## Build + +```sh +cmake -S . -B build +cmake --build build +``` + +- `LinuxDeviceInfo` (the CLI tool) is emitted to `build/LinuxDeviceInfo`. +- `libdevice_info.so` is copied automatically into `bindings/` for the Python binding. +- The default build type is **Release**, compiled with `-g -fno-omit-frame-pointer` so native profilers (perf, + `py-spy --native`) can still unwind and symbolize the library. + +## CLI Usage + +```sh +./build/LinuxDeviceInfo +``` + +Sample output: + +``` +Vulkan fallback: VRAM total not detected for GPU 0000:09:00.0 +GPU at 0000:09:00.0: + Name: NVIDIA GeForce RTX 5070 Ti + VRAM Total: 16303 MB + VRAM Used: 2721 MB +``` + +The tool exits with code `0` when enumeration succeeds, or `1` if the underlying DRM/Vulkan query fails. + +## Python Binding + +After building the project once (so that `bindings/libdevice_info.so` exists), you can inspect GPUs from Python: + +```python +from gpu_info import get_gpu_info + +gpu = get_gpu_info("0000:09:00.0", 0x10DE) +print(gpu) +``` + +On import, the script loads the colocated `libdevice_info.so`; ensure you rebuild the CMake project whenever you make +changes to the native code. + +```python +from hwprobe.core.linux.graphics import fetch_graphics_info + +info = fetch_graphics_info() +for gpu in info.modules: + print(f"{gpu.name}: {gpu.vram.capacity}{gpu.vram.unit} VRAM") +``` + +## Why C++ Instead of Pure Python? + +1. **No external dependencies** — `lspci`, `nvidia-smi`, `rocm-smi` may not be installed +2. **Efficient VRAM detection** — DRM ioctls work for all vendors without proprietary tools +3. **Faster** — Direct kernel interface, no subprocess overhead +4. **Unified Vulkan fallback** + +## Limitations + +- **Intel Arc VRAM**: Requires kernel 5.16+ for `DRM_I915_QUERY_MEMORY_REGIONS` +- **Nouveau VRAM**: Requires open Nouveau driver (proprietary NVIDIA driver uses Vulkan path) + +## Troubleshooting + +- **`libdevice_info.so not found`**: run the CMake build so the shared library is (re)generated in `bindings/`. +- **`get_gpu_info` returns -1**: verify that `/sys/class/drm` is populated and that DRM/Vulkan drivers are installed. +- **VRAM shows 0 MB**: the Vulkan fallback requires `libvulkan.so.1` to be installed; without it, VRAM is only + reported for vendors with a supported DRM ioctl (AMD, Intel, open-source Nouveau). diff --git a/src/hwprobe/interops/linux/bindings/gpu_info.py b/src/hwprobe/interops/linux/bindings/gpu_info.py new file mode 100644 index 0000000..5765be9 --- /dev/null +++ b/src/hwprobe/interops/linux/bindings/gpu_info.py @@ -0,0 +1,94 @@ +""" +Python bindings for Linux GPU info via C library. +Uses ctypes to interface with libdevice_info.so +""" + +import ctypes +from dataclasses import dataclass +from enum import IntEnum +from pathlib import Path +from typing import Optional + + +class GPUInfoQueryStatus(IntEnum): + FAILURE = -1 + DRM_SUCCESS = 0 + VULKAN_VRAM_FALLBACK = 1 + VULKAN_NAME_FALLBACK = 2 + + +@dataclass +class GPUProperties: + """Python representation of the C GPUProperties struct""" + + name: str + vram_total_mb: int + vram_used_mb: int + + +class _CGPUProperties(ctypes.Structure): + """C struct layout matching gpu_info.h""" + + _fields_ = [ + ("name", ctypes.c_char * 256), + ("vram_total_mb", ctypes.c_uint64), + ("vram_used_mb", ctypes.c_uint64), + ] + + +def _find_library() -> Optional[ctypes.CDLL]: + """Locate and load libdevice_info.so""" + # Try relative to this file first + _HERE = Path(__file__).parent + _LIB_PATH = _HERE / "libdevice_info.so" + + if not _LIB_PATH.exists(): + return None + + try: + lib = ctypes.CDLL(str(_LIB_PATH)) + # Configure function signature + lib.get_gpu_info.argtypes = [ctypes.c_char_p, ctypes.c_uint32, ctypes.POINTER(_CGPUProperties)] + lib.get_gpu_info.restype = GPUInfoQueryStatus + return lib + except (OSError, AttributeError): + return None + + +# Module-level library handle +_lib = _find_library() + + +def is_available() -> bool: + """Check if the native library is available""" + return _lib is not None + + +def get_gpu_info(bdf: str, vendor_id: int) -> tuple[GPUInfoQueryStatus, GPUProperties]: + """ + Query VRAM and driver name for the GPU at the given PCI BDF address. + + Args: + bdf: PCI address in ``domain:bus:device.function`` form, e.g. ``0000:01:00.0`` + vendor_id: PCI vendor ID (e.g. ``0x1002`` for AMD, ``0x8086`` for Intel) + + Returns: + A GPUProperties object with VRAM figures and DRM driver name + + Raises: + RuntimeError: If the library is not available or the underlying C call fails + """ + if not _lib: + raise RuntimeError("libdevice_info.so not found. Build the C library first.") + + c_gpu = _CGPUProperties() + ret = _lib.get_gpu_info(bdf.encode(), vendor_id, ctypes.byref(c_gpu)) + + return ( + ret, + GPUProperties( + name=c_gpu.name.decode("utf-8", errors="replace").strip(), + vram_total_mb=c_gpu.vram_total_mb, + vram_used_mb=c_gpu.vram_used_mb, + ), + ) diff --git a/src/hwprobe/interops/linux/bindings/libdevice_info.so b/src/hwprobe/interops/linux/bindings/libdevice_info.so new file mode 100755 index 0000000..df6d06f Binary files /dev/null and b/src/hwprobe/interops/linux/bindings/libdevice_info.so differ diff --git a/src/hwprobe/interops/linux/include/gpu_info.h b/src/hwprobe/interops/linux/include/gpu_info.h new file mode 100644 index 0000000..6475587 --- /dev/null +++ b/src/hwprobe/interops/linux/include/gpu_info.h @@ -0,0 +1,42 @@ +#pragma once + +#include + +#ifdef __cplusplus +extern "C" { +#endif + +#define AMD_VENDOR_ID 0x1002 +#define INTEL_VENDOR_ID 0x8086 +#define NVIDIA_VENDOR_ID 0x10DE + +#define MAX_GPU_CARDS (16u) +#define BYTES_PER_MB (1024 * 1024) + +typedef enum { + FAILURE = -1, + DRM_SUCCESS = 0, + VULKAN_VRAM_FALLBACK, + VULKAN_NAME_FALLBACK, +} GPUInfoQueryStatus; + +typedef struct { + uint16_t domain : 16u; //!< PCI domain number + uint8_t bus : 8u; //!< PCI bus number + uint8_t device : 5u; //!< PCI device number + uint8_t function : 3u; //!< PCI function number +} PCIAddress; + +typedef struct { + char name[256]; //!< GPU name or description, if available. + uint64_t vram_total_mb; //!< Total VRAM capacity in MB: 0 if unavailable, + //!< greater than 0 otherwise. + uint64_t vram_used_mb; //!< Total VRAM used by all processes: 0 if available, + //!< greater than 0 otherwise. +} GPUProperties; + +GPUInfoQueryStatus get_gpu_info(const char *bdf, uint32_t vendor_id, GPUProperties *out); + +#ifdef __cplusplus +} +#endif diff --git a/src/hwprobe/interops/linux/main.c b/src/hwprobe/interops/linux/main.c new file mode 100644 index 0000000..5d7b626 --- /dev/null +++ b/src/hwprobe/interops/linux/main.c @@ -0,0 +1,27 @@ +#include "gpu_info.h" +#include +#include +#include + +int main(int argc, char *argv[]) +{ + if (argc < 3) { + fprintf( + stderr, "Usage: %s \n e.g. %s 0000:01:00.0 0x1002\n", + argv[0], argv[0]); + return 1; + } + + uint32_t vendor_id = (uint32_t)strtoul(argv[2], NULL, 16); + GPUProperties g; + if (get_gpu_info(argv[1], vendor_id, &g) < 0) { + fprintf(stderr, "Failed to query GPU info for %s\n", argv[1]); + return 1; + } + + printf("GPU at %s:\n", argv[1]); + printf(" Name: %s\n", g.name[0] ? g.name : "(unknown)"); + printf(" VRAM Total: %lu MB\n", (unsigned long)g.vram_total_mb); + printf(" VRAM Used: %lu MB\n", (unsigned long)g.vram_used_mb); + return 0; +} diff --git a/src/hwprobe/interops/linux/src/gpu_info.cpp b/src/hwprobe/interops/linux/src/gpu_info.cpp new file mode 100644 index 0000000..fd63985 --- /dev/null +++ b/src/hwprobe/interops/linux/src/gpu_info.cpp @@ -0,0 +1,444 @@ +#include "../include/gpu_info.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include + +typedef struct VkGPU { + uint32_t vendor_id; + uint32_t device_id; + uint64_t vram_mb; + uint64_t used_mb; + char slot[64]; + char name[256]; +} VkGPU; + +inline GPUInfoQueryStatus operator|(GPUInfoQueryStatus a, GPUInfoQueryStatus b) { + return static_cast(static_cast(a) | static_cast(b)); +} + +inline GPUInfoQueryStatus operator|=(GPUInfoQueryStatus a, GPUInfoQueryStatus b) { + a = a | b; + return a; +} + +static uint64_t to_mb(uint64_t bytes) { return bytes / BYTES_PER_MB; } + +static PCIAddress parse_bdf_to_pci_addr(const char *bdf) +{ + PCIAddress pciAddr = {0}; + if (!bdf || strlen(bdf) == 0) { + return pciAddr; + } + + char *endptr = nullptr; + + uint16_t domain = strtoll(bdf, &endptr, 16u); + if (*endptr != ':') { + return pciAddr; + } + + uint8_t bus = strtoll(endptr + 1, &endptr, 16u); + if (*endptr != ':') { + return pciAddr; + } + + uint8_t device = strtoll(endptr + 1, &endptr, 16u); + if (*endptr != '.') { + return pciAddr; + } + + uint8_t function = strtoll(endptr + 1, &endptr, 16u); + + pciAddr.domain = domain; + pciAddr.bus = bus; + pciAddr.device = device; + pciAddr.function = function; + + return pciAddr; +} + +// +// DRM ioctl VRAM queries +// + +static void vram_amdgpu(int fd, GPUProperties *g) +{ + if (g == nullptr) { + return; + } + + drm_amdgpu_info req = {0}; + drm_amdgpu_memory_info mem; + + req.return_pointer = reinterpret_cast(&mem); + req.return_size = sizeof(mem); + req.query = AMDGPU_INFO_MEMORY; + + if (ioctl(fd, DRM_IOCTL_AMDGPU_INFO, &req) == 0) { + g->vram_total_mb = to_mb(mem.vram.total_heap_size); + g->vram_used_mb = to_mb(mem.vram.heap_usage); + } +} + +static void vram_radeon(int fd, GPUProperties *g) +{ + if (g == nullptr) { + return; + } + + drm_radeon_gem_info gem_info = {0}; + + if (ioctl(fd, DRM_IOCTL_RADEON_GEM_INFO, &gem_info) == 0) { + g->vram_total_mb = to_mb(gem_info.vram_size); + } + + drm_radeon_info info = {0}; + info.request = RADEON_INFO_VRAM_USAGE; + info.value = (uint64_t)(uintptr_t)(&g->vram_used_mb); + if (ioctl(fd, DRM_IOCTL_RADEON_INFO, &info) < 0) { + g->vram_used_mb = 0; + } +} + +static void vram_i915(int fd, GPUProperties *g) +{ + if (g == nullptr) { + return; + } + + drm_i915_query_item item = {0}; + item.query_id = DRM_I915_QUERY_MEMORY_REGIONS; + + drm_i915_query q = {0}; + q.num_items = 1; + q.items_ptr = reinterpret_cast(&item); + + if (ioctl(fd, DRM_IOCTL_I915_QUERY, &q) != 0 || item.length <= 0) { + return; + } + + uint8_t *buf = static_cast(calloc(1, item.length)); + if (!buf) { + return; + } + + item.data_ptr = reinterpret_cast(buf); + + if (ioctl(fd, DRM_IOCTL_I915_QUERY, &q) == 0) { + drm_i915_query_memory_regions *r = + reinterpret_cast(buf); + for (uint32_t i = 0; i < r->num_regions; i++) { + if (r->regions[i].region.memory_class == I915_MEMORY_CLASS_DEVICE) { + g->vram_total_mb = to_mb(r->regions[i].probed_size); + break; + } + } + } + + free(buf); +} + +static void vram_xe(int fd, GPUProperties *g) +{ + if (g == nullptr || fd < 0) { + return; + } + + struct drm_xe_device_query query = {}; + query.query = DRM_XE_DEVICE_QUERY_MEM_REGIONS; + + if (ioctl(fd, DRM_IOCTL_XE_DEVICE_QUERY, &query) != 0 || query.size == 0) { + g->vram_total_mb = 0; + g->vram_used_mb = 0; + return; + } + + std::vector buffer(query.size); + query.data = reinterpret_cast(buffer.data()); + + if (ioctl(fd, DRM_IOCTL_XE_DEVICE_QUERY, &query) != 0) { + g->vram_total_mb = 0; + g->vram_used_mb = 0; + return; + } + + const auto *regions = reinterpret_cast(buffer.data()); + + uint64_t total_vram = 0; + uint64_t used_vram = 0; + + for (uint32_t i = 0; i < regions->num_mem_regions; ++i) { + const auto &mem = regions->mem_regions[i]; + + // Filter out system memory (SYSMEM) to only aggregate discrete VRAM + if (mem.mem_class == DRM_XE_MEM_REGION_CLASS_VRAM) { + total_vram += mem.total_size; + used_vram += mem.used; + } + } + + g->vram_total_mb = to_mb(total_vram); + g->vram_used_mb = to_mb(used_vram); +} + +static void vram_nouveau(int fd, GPUProperties *g) +{ + if (g == nullptr) { + return; + } + + drm_nouveau_getparam p = {0}; + p.param = NOUVEAU_GETPARAM_FB_SIZE; + + if (ioctl(fd, DRM_IOCTL_NOUVEAU_GETPARAM, &p) == 0 && p.value > 0) { + g->vram_total_mb = to_mb(p.value); + } +} + +static int vulkan_query(VkGPU *out, const PCIAddress *pciAddr) +{ + if (pciAddr == NULL) { + return -1; + } + + // Request the get_physical_device_properties2 extension to read PCI bus + // info. + const char *instExts[] = { + VK_KHR_GET_PHYSICAL_DEVICE_PROPERTIES_2_EXTENSION_NAME}; + + VkApplicationInfo appInfo = {}; + appInfo.sType = VK_STRUCTURE_TYPE_APPLICATION_INFO; + appInfo.pApplicationName = "HWProbe"; + appInfo.applicationVersion = VK_MAKE_VERSION(1, 0, 0); +#ifdef VK_API_VERSION_1_2 + appInfo.apiVersion = VK_API_VERSION_1_2; +#else + appInfo.apiVersion = VK_MAKE_VERSION(1, 2, 0); +#endif + + VkInstanceCreateInfo instCreate = {}; + instCreate.sType = VK_STRUCTURE_TYPE_INSTANCE_CREATE_INFO; + instCreate.pApplicationInfo = &appInfo; + instCreate.enabledExtensionCount = 1; + instCreate.ppEnabledExtensionNames = instExts; + + VkInstance inst = VK_NULL_HANDLE; + VkResult r = vkCreateInstance(&instCreate, NULL, &inst); + if (r != VK_SUCCESS || inst == VK_NULL_HANDLE) { + return -1; + } + + uint32_t numberOfDevices = 0; + r = vkEnumeratePhysicalDevices(inst, &numberOfDevices, NULL); + if (r != VK_SUCCESS) { + vkDestroyInstance(inst, NULL); + return -1; + } + + if (numberOfDevices == 0) { + vkDestroyInstance(inst, NULL); + return 0; + } + + if (numberOfDevices > MAX_GPU_CARDS) { + numberOfDevices = MAX_GPU_CARDS; + } + + VkPhysicalDevice devs[MAX_GPU_CARDS]; + r = vkEnumeratePhysicalDevices(inst, &numberOfDevices, devs); + if (r != VK_SUCCESS) { + vkDestroyInstance(inst, NULL); + return -1; + } + + uint32_t cnt = 0; + + for (uint32_t i = 0; i < numberOfDevices && cnt < MAX_GPU_CARDS; i++) { + VkGPU *g = &out[cnt++]; + + VkPhysicalDevicePCIBusInfoPropertiesEXT pci = {}; + pci.sType = VK_STRUCTURE_TYPE_PHYSICAL_DEVICE_PCI_BUS_INFO_PROPERTIES_EXT; + + VkPhysicalDeviceProperties2 props2 = {}; + props2.sType = VK_STRUCTURE_TYPE_PHYSICAL_DEVICE_PROPERTIES_2; + props2.pNext = &pci; + vkGetPhysicalDeviceProperties2(devs[i], &props2); + + snprintf( + g->slot, sizeof(g->slot), "%04x:%02x:%02x.%x", pci.pciDomain, + pci.pciBus, pci.pciDevice, pci.pciFunction); + + if (pciAddr->domain != pci.pciDomain || pciAddr->bus != pci.pciBus || + pciAddr->device != pci.pciDevice || + pciAddr->function != pci.pciFunction) { + // Skip devices that don't match the PCI address. + continue; + } + + VkPhysicalDeviceMemoryBudgetPropertiesEXT budget = {}; + budget.sType = + VK_STRUCTURE_TYPE_PHYSICAL_DEVICE_MEMORY_BUDGET_PROPERTIES_EXT; + + VkPhysicalDeviceMemoryProperties2 memProps2 = {}; + memProps2.sType = VK_STRUCTURE_TYPE_PHYSICAL_DEVICE_MEMORY_PROPERTIES_2; + memProps2.pNext = &budget; + vkGetPhysicalDeviceMemoryProperties2(devs[i], &memProps2); + + uint64_t total = 0, used = 0; + uint32_t heapCnt = memProps2.memoryProperties.memoryHeapCount; + for (uint32_t h = 0; h < heapCnt; h++) { + if (memProps2.memoryProperties.memoryHeaps[h].flags & + VK_MEMORY_HEAP_DEVICE_LOCAL_BIT) { + total += memProps2.memoryProperties.memoryHeaps[h].size; + // total - budget ≈ system-wide VRAM usage + if (budget.heapBudget[h] > 0 && + budget.heapBudget[h] < + memProps2.memoryProperties.memoryHeaps[h].size) + used += memProps2.memoryProperties.memoryHeaps[h].size - + budget.heapBudget[h]; + } + } + + g->vendor_id = props2.properties.vendorID; + g->device_id = props2.properties.deviceID; + g->vram_mb = to_mb(total); + g->used_mb = to_mb(used); + + snprintf(g->name, sizeof(g->name), "%s", props2.properties.deviceName); + } + + vkDestroyInstance(inst, NULL); + return cnt; +} + +/** + * Get GPU information for a specific PCI device. + * + * @note This function also returns the current VRAM usage, even though it + * currently has no proper application in HWProbe. It shall stay here for future + * reference if we decide to query similar information on other platforms. + * + * @param bdf The PCI bus:device.function string (e.g., "0000:01:00.0"). + * @param vendor_id The PCI vendor ID of the GPU (e.g., 0x1002 for AMD, 0x8086 + * for Intel, 0x10DE for NVIDIA). + * @param out Pointer to a GPUProperties struct to receive the GPU information. + * @return 0 on success, -1 on failure. + */ +GPUInfoQueryStatus get_gpu_info(const char *bdf, uint32_t vendor_id, GPUProperties *out) +{ + if (!bdf || !out) { + return GPUInfoQueryStatus::FAILURE; + } + + GPUInfoQueryStatus status = GPUInfoQueryStatus::FAILURE; + + // Find the DRM card node under /sys/bus/pci/devices//drm/cardN. + char drm_dir_path[160]; + snprintf( + drm_dir_path, sizeof(drm_dir_path), "/sys/bus/pci/devices/%s/drm", bdf); + + char card_name[256] = {0}; + DIR *drm_dir = opendir(drm_dir_path); + + if (drm_dir) { + struct dirent *entry; + + while ((entry = readdir(drm_dir)) != NULL) { + if (strncmp(entry->d_name, "card", 4) == 0 && entry->d_name[4] >= '0' && + entry->d_name[4] <= '9') { + snprintf(card_name, sizeof(card_name), "%s", entry->d_name); + break; + } + } + closedir(drm_dir); + } + + GPUProperties g = {}; + + if (card_name[0]) { + char devpath[64]; + + snprintf(devpath, sizeof(devpath), "/dev/dri/%s", card_name); + + int fd = open(devpath, O_RDWR | O_CLOEXEC); + if (fd < 0) { + fd = open(devpath, O_RDONLY | O_CLOEXEC); + } + + if (fd >= 0) { + drmVersionPtr drm_version = drmGetVersion(fd); + + // TODO: Test all the DRM IOCTL methods; support from users + if (drm_version) { + if (strcmp(drm_version->name, "amdgpu") == 0) { + vram_amdgpu(fd, &g); + } + else if (strcmp(drm_version->name, "radeon") == 0) { + vram_radeon(fd, &g); + } + else if (strcmp(drm_version->name, "i915") == 0) { + vram_i915(fd, &g); + } + else if (strcmp(drm_version->name, "xe") == 0) { + vram_xe(fd, &g); + } + else if (strcmp(drm_version->name, "nouveau") == 0) { + vram_nouveau(fd, &g); + } + } + close(fd); + } + } + + // TODO: Remove 'g.name[0]' check when pci-ids parser is implemented! + // Vulkan fallback if DRM didn't give us complete info. + if (!g.vram_total_mb || !g.name[0]) { + PCIAddress pciAddr = parse_bdf_to_pci_addr(bdf); + VkGPU vk[MAX_GPU_CARDS]; + int vk_n = vulkan_query(vk, &pciAddr); + + if (vk_n > 0) { + for (int v = 0; v < vk_n; v++) { + if (strcmp(bdf, vk[v].slot) != 0) { + continue; + } + + if (!g.vram_total_mb) { + g.vram_total_mb = vk[v].vram_mb; + status = GPUInfoQueryStatus::VULKAN_VRAM_FALLBACK; + } + if (!g.vram_used_mb) { + g.vram_used_mb = vk[v].used_mb; + status = GPUInfoQueryStatus::VULKAN_VRAM_FALLBACK; + } + if (vk[v].name[0]) { + snprintf(g.name, sizeof(g.name), "%s", vk[v].name); + if (status == GPUInfoQueryStatus::VULKAN_VRAM_FALLBACK) { + status |= GPUInfoQueryStatus::VULKAN_NAME_FALLBACK; + } else { + status = GPUInfoQueryStatus::VULKAN_NAME_FALLBACK; + } + } + break; + } + } + } + + *out = g; + return status; +} diff --git a/src/hwprobe/models/display_models.py b/src/hwprobe/models/display_models.py index 6cf337d..3a4cc91 100644 --- a/src/hwprobe/models/display_models.py +++ b/src/hwprobe/models/display_models.py @@ -21,6 +21,7 @@ class ResolutionInfo(BaseModel): class DisplayModuleInfo(BaseModel): """Information for one Display is stored here""" + #: Brand name/model of the display name: Optional[str] = None #: Year it was manufactured / designed. @@ -29,9 +30,6 @@ class DisplayModuleInfo(BaseModel): #: ACPI path of the display device. acpi_path: Optional[str] = None - #: PCI path of the device - pci_path: Optional[str] = None - #: Parent GPU driving this display gpu_name: Optional[str] = None diff --git a/src/hwprobe/models/gpu_models.py b/src/hwprobe/models/gpu_models.py index 3ff85b8..c12a66a 100644 --- a/src/hwprobe/models/gpu_models.py +++ b/src/hwprobe/models/gpu_models.py @@ -43,6 +43,7 @@ class GPUInfo(BaseModel): #: ACPI device path, e.g. ``\\_SB.PC00.RP05.PXSX``. acpi_path: Optional[str] = None + #: PCI path from the firmware tree, e.g. ``PciRoot(0x0)/Pci(0x1C,0x5)/Pci(0x0,0x0)``. pci_path: Optional[str] = None diff --git a/src/hwprobe/models/info_models.py b/src/hwprobe/models/info_models.py index afcba07..a5b9aef 100644 --- a/src/hwprobe/models/info_models.py +++ b/src/hwprobe/models/info_models.py @@ -2,7 +2,10 @@ from pydantic import BaseModel +from hwprobe.models.audio_models import AudioInfo +from hwprobe.models.baseboard_models import BaseboardInfo from hwprobe.models.cpu_models import CPUInfo +from hwprobe.models.display_models import DisplayInfo from hwprobe.models.gpu_models import GraphicsInfo from hwprobe.models.memory_models import MemoryInfo from hwprobe.models.network_models import NetworkInfo @@ -53,3 +56,12 @@ def fetch_storage_info(self) -> StorageInfo: def fetch_network_info(self) -> NetworkInfo: """Fetches Network Information.""" + + def fetch_display_info(self) -> DisplayInfo: + """Fetches Display Information. Not available on every platform yet.""" + + def fetch_audio_info(self) -> AudioInfo: + """Fetches Audio Information. Not available on every platform yet.""" + + def fetch_baseboard_info(self) -> BaseboardInfo: + """Fetches Baseboard/Motherboard Information. Not available on every platform yet.""" diff --git a/src/hwprobe/util/nvidia.py b/src/hwprobe/util/nvidia.py deleted file mode 100644 index 6883a27..0000000 --- a/src/hwprobe/util/nvidia.py +++ /dev/null @@ -1,36 +0,0 @@ -import subprocess - - -def fetch_gpu_details_nvidia(device: str) -> tuple[str, int, int, int]: - """ - :param device: format: ::. - :return: GPU name, PCI Width, PCI Gen, Total VRAM in MB - """ - # Combine all queries into a single comma-separated string - # Fields: Name, PCIe Width, PCIe Gen, Memory Total - query_fields = "name,pcie.link.width.current,pcie.link.gen.current,memory.total" - - command = ["nvidia-smi", f"--id={device}", f"--query-gpu={query_fields}", "--format=csv,noheader,nounits"] - - # Run the command - result = subprocess.run(command, capture_output=True, text=True) - - # Check for execution errors - if result.returncode != 0: - raise RuntimeError(f"nvidia-smi failed: {result.stderr}") - - # Parse output (Expected: "Name, Width, Gen, Memory") - output = result.stdout.strip() - parts = output.split(",") - - # Validate we got exactly 4 fields back - if len(parts) != 4: - raise ValueError(f"Unexpected output format from nvidia-smi: {output}") - - # Parse and Type Convert - gpu_name = parts[0].strip() - pci_width = int(parts[1].strip()) # e.g., 16 - pci_gen = int(parts[2].strip()) # e.g., 3, 4, or 5 - vram_total = int(parts[3].strip()) # e.g., 16384 (MiB) - - return gpu_name, pci_width, pci_gen, vram_total diff --git a/tests/core/common/test_edid.py b/tests/core/common/test_edid.py index a3b99b0..2172b45 100644 --- a/tests/core/common/test_edid.py +++ b/tests/core/common/test_edid.py @@ -1,4 +1,4 @@ - +from typing import Optional from hwprobe.core.common.edid import parse_edid @@ -197,6 +197,82 @@ def test_analog_with_timing_gets_resolution(self): assert result.resolution.refresh_rate > 0 +def _build_cta_detailed_timing_descriptor( + width=1920, + height=1080, + h_blank=280, + v_blank=45, + pixel_clock_10khz=14850, +): + """Create a valid 18-byte CTA/EDID detailed timing descriptor.""" + descriptor = bytearray(18) + descriptor[0] = pixel_clock_10khz & 0xFF + descriptor[1] = (pixel_clock_10khz >> 8) & 0xFF + descriptor[2] = width & 0xFF + descriptor[3] = h_blank & 0xFF + descriptor[4] = ((width >> 8) & 0x0F) << 4 | ((h_blank >> 8) & 0x0F) + descriptor[5] = height & 0xFF + descriptor[6] = v_blank & 0xFF + descriptor[7] = ((height >> 8) & 0x0F) << 4 | ((v_blank >> 8) & 0x0F) + return bytes(descriptor) + + +def _build_cta_extension(dtd: bytes) -> bytes: + """Build a CTA-861 extension block with the DTD starting at byte offset 4.""" + ext = bytearray(128) + ext[0] = 0x02 + ext[1] = 0x03 + ext[2] = 0x04 + ext[4 : 4 + len(dtd)] = dtd + return bytes(ext) + + +def _build_displayid_type_i_timing_entry( + width=1920, + height=1080, + h_blank=280, + v_blank=45, + pixel_clock_10khz=14850, +): + """Build a 20-byte DisplayID Type-I timing entry using the actual-1 encoding.""" + entry = bytearray(20) + entry[0:3] = (pixel_clock_10khz - 1).to_bytes(3, byteorder="little") + entry[4:6] = (width - 1).to_bytes(2, byteorder="little") + entry[6:8] = (h_blank - 1).to_bytes(2, byteorder="little") + entry[8:10] = b"\x00\x00" + entry[10:12] = b"\x00\x00" + entry[12:14] = (height - 1).to_bytes(2, byteorder="little") + entry[14:16] = (v_blank - 1).to_bytes(2, byteorder="little") + entry[16:20] = b"\x00\x00\x00\x00" + return bytes(entry) + + +def _build_displayid_extension(payload: bytes, *, section_size: Optional[int] = None, truncated: bool = False) -> bytes: + """Build a DisplayID extension block with a Type-I timing payload. + + If truncated is True, the payload length is longer than the advertised section size + to ensure the parser must stop before overrunning the extension boundary. + """ + ext = bytearray(128) + ext[0] = 0x70 + ext[1] = 0x01 + ext[2] = section_size if section_size is not None else 5 + len(payload) + ext[3] = 0x00 + ext[4] = 0x00 + + if truncated: + ext[5] = 0x03 + ext[6] = 0x01 + ext[7] = 0x20 + else: + ext[5] = 0x03 + ext[6] = 0x01 + ext[7] = len(payload) + ext[8 : 8 + len(payload)] = payload + + return bytes(ext) + + class TestCommonFieldsAcrossVersions: def test_year_parsed(self): edid = _build_edid(year_offset=25) @@ -223,3 +299,47 @@ def test_name_and_serial_both_parsed(self): result = parse_edid(edid) assert result.name == "My Display" assert result.serial_number == "ABC123" + + +class TestEdidExtensionFixtures: + def test_parse_edid_reads_cta_dtd_from_extension(self): + dtd = _build_cta_detailed_timing_descriptor() + cta_ext = _build_cta_extension(dtd) + + edid = bytearray(128 * 2) + edid[0x7E] = 1 + edid[128 : 128 + 128] = cta_ext + + result = parse_edid(bytes(edid)) + + assert result.resolution.width == 1920 + assert result.resolution.height == 1080 + assert result.resolution.refresh_rate == 60.0 + + def test_parse_edid_reads_displayid_type_i_timing(self): + entry = _build_displayid_type_i_timing_entry() + displayid_ext = _build_displayid_extension(entry) + + edid = bytearray(128 * 2) + edid[0x7E] = 1 + edid[128 : 128 + 128] = displayid_ext + + result = parse_edid(bytes(edid)) + + assert result.resolution.width == 1920 + assert result.resolution.height == 1080 + assert result.resolution.refresh_rate == 60.0 + + def test_parse_edid_handles_truncated_displayid_payload(self): + entry = _build_displayid_type_i_timing_entry() + displayid_ext = _build_displayid_extension(entry, section_size=25, truncated=True) + + edid = bytearray(128 * 2) + edid[0x7E] = 1 + edid[128 : 128 + 128] = displayid_ext + + result = parse_edid(bytes(edid)) + + assert result.resolution.width is None + assert result.resolution.height is None + assert result.resolution.refresh_rate is None diff --git a/tests/core/linux/test_common.py b/tests/core/linux/test_common.py index 3410da6..9013d50 100644 --- a/tests/core/linux/test_common.py +++ b/tests/core/linux/test_common.py @@ -2,8 +2,7 @@ import pytest -from hwprobe.core.linux.common import _format_pci_component, pci_path_linux - +from hwprobe.core.linux.common import pci_path_linux class TestPciPathLinux: def test_single_device(self, monkeypatch): @@ -38,27 +37,23 @@ def test_non_zero_domain(self, monkeypatch): ) assert pci_path_linux("0001:00:00.0") == "PciRoot(0x1)/Pci(0x0,0x0)" + def test_bus_device(self, monkeypatch): + monkeypatch.setattr( + posixpath, + "realpath", + lambda _: "/sys/devices/platform/bus@0/14100000.pcie/pci0001:00/0001:00:00.0" + ) + assert pci_path_linux("0001:00:00.0") == "PciRoot(0x1)/Pci(0x0,0x0)" + def test_fallback_when_sysfs_has_no_pci(self, monkeypatch): monkeypatch.setattr( posixpath, "realpath", lambda _: "/sys/devices/platform/non-pci-device", ) - assert pci_path_linux("0000:03:00.0") == "PciRoot(0x0)/Pci(0x0,0x0)" + assert pci_path_linux("0000:03:00.0") == None @pytest.mark.parametrize("bad_slot", ["", "xyz", ":::"]) def test_invalid_device_slot_returns_none(self, bad_slot, monkeypatch): monkeypatch.setattr(posixpath, "realpath", lambda _: "") assert pci_path_linux(bad_slot) is None - - -class TestFormatPciComponent: - def test_standard_slot(self): - assert _format_pci_component("0000:00:02.0") == "0x2,0x0" - - def test_multifunction_slot(self): - assert _format_pci_component("0000:00:1f.3") == "0x1f,0x3" - - @pytest.mark.parametrize("bad_input", ["", "no-dot", None]) - def test_invalid_input_returns_none(self, bad_input): - assert _format_pci_component(bad_input) is None diff --git a/tests/core/linux/test_cpu.py b/tests/core/linux/test_cpu.py index 21093c7..aed394f 100644 --- a/tests/core/linux/test_cpu.py +++ b/tests/core/linux/test_cpu.py @@ -145,6 +145,8 @@ class TestX86Flags: def test_x86_flags_sse_variants(self): cpu_lines = "flags\t\t: sse sse2 sse3 ssse3 sse4_1 sse4_2\n" flags = _x86_flags(cpu_lines) + + assert flags is not None assert "SSE" in flags assert "SSE2" in flags assert "SSE3" in flags @@ -155,6 +157,7 @@ def test_x86_flags_sse_variants(self): def test_x86_flags_with_lm(self): cpu_lines = "flags\t\t: sse lm\n" flags = _x86_flags(cpu_lines) + assert flags is not None assert "LM" in flags def test_x86_flags_missing(self): diff --git a/tests/core/linux/test_display.py b/tests/core/linux/test_display.py index 3096a1d..ba4adc6 100644 --- a/tests/core/linux/test_display.py +++ b/tests/core/linux/test_display.py @@ -42,7 +42,7 @@ def _patch_exists(self, monkeypatch, paths): def test_returns_none_when_edid_missing(self, monkeypatch): self._patch_exists(monkeypatch, set()) - assert _fetch_individual_monitor_info(self.DEVICE_PATH) is None + assert _fetch_individual_monitor_info(self.DEVICE_PATH, []) is None def test_returns_none_when_edid_empty(self, monkeypatch): self._patch_exists(monkeypatch, {self.EDID_PATH}) @@ -51,65 +51,7 @@ def test_returns_none_when_edid_empty(self, monkeypatch): "open", lambda *a, **kw: mock_open(read_data=b"")(), ) - assert _fetch_individual_monitor_info(self.DEVICE_PATH) is None - - def test_pci_path_resolved_from_gpu_endpoint(self, monkeypatch): - self._patch_exists(monkeypatch, {self.EDID_PATH}) - monkeypatch.setattr( - builtins, - "open", - lambda *a, **kw: mock_open(read_data=b"\x01\x02")(), - ) - monkeypatch.setattr( - "hwprobe.core.linux.display.parse_edid", - lambda _: DisplayModuleInfo(name="Internal Display"), - ) - monkeypatch.setattr( - posixpath, - "realpath", - lambda _: "/sys/devices/pci0000:00/0000:00:02.0/0000:06:00.0/drm/card0", - ) - - pci_calls = [] - monkeypatch.setattr( - "hwprobe.core.linux.display.pci_path_linux", - lambda slot: (pci_calls.append(slot), "PciRoot(0x0)/Pci(0x6,0x0)")[-1], - ) - - monitor = _fetch_individual_monitor_info(self.DEVICE_PATH) - - assert monitor is not None - assert pci_calls == ["0000:06:00.0"] - assert monitor.pci_path == "PciRoot(0x0)/Pci(0x6,0x0)" - - def test_no_pci_path_for_non_pci_parent(self, monkeypatch): - self._patch_exists(monkeypatch, {self.EDID_PATH}) - monkeypatch.setattr( - builtins, - "open", - lambda *a, **kw: mock_open(read_data=b"\x01\x02")(), - ) - monkeypatch.setattr( - "hwprobe.core.linux.display.parse_edid", - lambda _: DisplayModuleInfo(name="Panel"), - ) - monkeypatch.setattr( - posixpath, - "realpath", - lambda _: "/sys/devices/platform/simple-framebuffer/drm/card0", - ) - - pci_calls = [] - monkeypatch.setattr( - "hwprobe.core.linux.display.pci_path_linux", - lambda slot: pci_calls.append(slot), - ) - - monitor = _fetch_individual_monitor_info(self.DEVICE_PATH) - - assert monitor is not None - assert pci_calls == [] - assert monitor.pci_path is None + assert _fetch_individual_monitor_info(self.DEVICE_PATH, []) is None def test_acpi_path_populated_when_firmware_node_exists(self, monkeypatch): self._patch_exists(monkeypatch, {self.EDID_PATH, self.ACPI_PATH}) @@ -138,7 +80,7 @@ def fake_open(path, *args, **kwargs): lambda slot: "PciRoot(0x0)/Pci(0x2,0x0)", ) - monitor = _fetch_individual_monitor_info(self.DEVICE_PATH) + monitor = _fetch_individual_monitor_info(self.DEVICE_PATH, []) assert monitor is not None assert monitor.acpi_path == r"\_SB.PCI0.GFX0.DD1F" @@ -151,20 +93,19 @@ def test_collects_monitors_from_drm(self, monkeypatch): os, "listdir", lambda path: { - "/sys/class/drm": ["card0", "renderD128", "version"], - "/sys/class/drm/card0": ["card0-eDP-1", "card0-HDMI-A-1", "device"], + "/sys/class/drm": ["card0", "card1"], + "/sys/class/drm/card0": ["card0-eDP-1"], + "/sys/class/drm/card1": ["card1-HDMI-A-1"], }.get(path, []), ) monkeypatch.setattr( "hwprobe.core.linux.display._fetch_individual_monitor_info", - lambda path: DisplayModuleInfo(name=posixpath.basename(path)), + lambda path, gpu_devices: DisplayModuleInfo(name=posixpath.basename(path)), ) - info = fetch_display_info() + info = fetch_display_info([]) assert len(info.modules) == 2 - names = {m.name for m in info.modules} - assert names == {"card0-eDP-1", "card0-HDMI-A-1"} def test_skips_monitors_returning_none(self, monkeypatch): monkeypatch.setattr(posixpath, "isdir", lambda p: p == "/sys/class/drm") @@ -178,17 +119,17 @@ def test_skips_monitors_returning_none(self, monkeypatch): ) monkeypatch.setattr( "hwprobe.core.linux.display._fetch_individual_monitor_info", - lambda path: None, + lambda path, gpu_devices: None, ) - info = fetch_display_info() + info = fetch_display_info([]) assert len(info.modules) == 0 def test_failed_when_drm_root_missing(self, monkeypatch): monkeypatch.setattr(posixpath, "isdir", lambda p: False) - info = fetch_display_info() + info = fetch_display_info([]) assert info.status.type == StatusType.FAILED assert any("/sys/class/drm" in m for m in info.status.messages) @@ -200,16 +141,14 @@ def test_partial_when_monitor_raises(self, monkeypatch): os, "listdir", lambda path: { - "/sys/class/drm": ["card0"], - "/sys/class/drm/card0": ["card0-eDP-1", "card0-HDMI-A-1"], + "/sys/class/drm": ["card0", "card1"], + "/sys/class/drm/card0": ["card0-DP-1"], + "/sys/class/drm/card1": ["card1-HDMI-A-1"], }.get(path, []), ) - call_count = {"n": 0} - - def _mock_fetch(path): - call_count["n"] += 1 - if call_count["n"] == 1: + def _mock_fetch(path, gpu_devices): + if path.endswith("card0-DP-1"): raise RuntimeError("EDID decode failed") return DisplayModuleInfo(name="Monitor B") @@ -218,10 +157,10 @@ def _mock_fetch(path): _mock_fetch, ) - info = fetch_display_info() + info = fetch_display_info([]) assert info.status.type == StatusType.PARTIAL - assert any("card0-eDP-1" in m for m in info.status.messages) + assert any("card0-DP-1" in m for m in info.status.messages) assert len(info.modules) == 1 assert info.modules[0].name == "Monitor B" @@ -275,7 +214,7 @@ def test_connector_overrides_edid_interface(self, monkeypatch): lambda _: "/sys/devices/platform/drm/card0", ) - monitor = _fetch_individual_monitor_info(device_path) + monitor = _fetch_individual_monitor_info(device_path, []) assert monitor is not None assert monitor.interface == "HDMI" diff --git a/tests/core/linux/test_graphics.py b/tests/core/linux/test_graphics.py index bf11b88..d7632e4 100644 --- a/tests/core/linux/test_graphics.py +++ b/tests/core/linux/test_graphics.py @@ -1,91 +1,16 @@ import builtins import os +import pytest import posixpath -import subprocess from unittest.mock import mock_open -from hwprobe.core.linux.graphics import ( - _check_gpu_class, - _pcie_gen, - _populate_amd_info, - _populate_lspci_info, - _populate_nvidia_info, - _vram_amd, - fetch_graphics_info, -) -from hwprobe.models.gpu_models import GPUInfo +from hwprobe.core.linux.common import _read_from_sysfs +from hwprobe.core.linux.graphics import _check_gpu_class, _pcie_gen, fetch_graphics_info +from hwprobe.interops.linux.bindings.gpu_info import GPUInfoQueryStatus, GPUProperties from hwprobe.models.status_models import StatusType - -class TestVramAmd: - """Tests for _vram_amd function.""" - - def test_vram_amd_success(self, monkeypatch): - device = "0000:03:00.0" - vram_path = f"/sys/bus/pci/devices/{device}/drm/card0/device/mem_info_vram_total" - - monkeypatch.setattr("glob.glob", lambda x: [vram_path]) - - def mock_open_func(file, *args, **kwargs): - if file == vram_path: - # 8GB in bytes - return mock_open(read_data=str(8 * 1024 * 1024 * 1024))() - raise FileNotFoundError(file) - - monkeypatch.setattr(builtins, "open", mock_open_func) - - vram_mb = _vram_amd(device) - assert vram_mb == 8192 - - def test_vram_amd_4gb(self, monkeypatch): - device = "0000:03:00.0" - vram_path = f"/sys/bus/pci/devices/{device}/drm/card0/device/mem_info_vram_total" - - monkeypatch.setattr("glob.glob", lambda x: [vram_path]) - - def mock_open_func(file, *args, **kwargs): - if file == vram_path: - # 4GB in bytes - return mock_open(read_data=str(4 * 1024 * 1024 * 1024))() - raise FileNotFoundError(file) - - monkeypatch.setattr(builtins, "open", mock_open_func) - - vram_mb = _vram_amd(device) - assert vram_mb == 4096 - - def test_vram_amd_no_file(self, monkeypatch): - device = "0000:03:00.0" - monkeypatch.setattr("glob.glob", lambda x: []) - - vram_mb = _vram_amd(device) - assert vram_mb is None - - def test_vram_amd_exception(self, monkeypatch): - def raise_error(x): - raise Exception("Glob failed") - - monkeypatch.setattr("glob.glob", raise_error) - - assert _vram_amd("0000:00:00.0") is None - - def test_vram_amd_read_error(self, monkeypatch): - device = "0000:03:00.0" - vram_path = f"/sys/bus/pci/devices/{device}/drm/card0/device/mem_info_vram_total" - - monkeypatch.setattr("glob.glob", lambda x: [vram_path]) - - def mock_open_func(file, *args, **kwargs): - raise OSError("Read error") - - monkeypatch.setattr(builtins, "open", mock_open_func) - - vram_mb = _vram_amd(device) - assert vram_mb is None - - class TestPcieGen: - """Tests for _pcie_gen function.""" + """Tests for _pcie_gen.""" def test_pcie_gen_success_gen4(self, monkeypatch): device = "0000:01:00.0" @@ -100,7 +25,10 @@ def mock_open_func(file, *args, **kwargs): monkeypatch.setattr(builtins, "open", mock_open_func) - gen = _pcie_gen(device) + speed = _read_from_sysfs(path) + assert speed is not None + + gen = _pcie_gen(speed) assert gen == 4 def test_pcie_gen_success_gen3(self, monkeypatch): @@ -116,7 +44,10 @@ def mock_open_func(file, *args, **kwargs): monkeypatch.setattr(builtins, "open", mock_open_func) - gen = _pcie_gen(device) + speed = _read_from_sysfs(path) + assert speed is not None + + gen = _pcie_gen(speed) assert gen == 3 def test_pcie_gen_success_gen2(self, monkeypatch): @@ -132,7 +63,10 @@ def mock_open_func(file, *args, **kwargs): monkeypatch.setattr(builtins, "open", mock_open_func) - gen = _pcie_gen(device) + speed = _read_from_sysfs(path) + assert speed is not None + + gen = _pcie_gen(speed) assert gen == 2 def test_pcie_gen_success_gen1(self, monkeypatch): @@ -148,7 +82,10 @@ def mock_open_func(file, *args, **kwargs): monkeypatch.setattr(builtins, "open", mock_open_func) - gen = _pcie_gen(device) + speed = _read_from_sysfs(path) + assert speed is not None + + gen = _pcie_gen(speed) assert gen == 1 def test_pcie_gen_success_gen5(self, monkeypatch): @@ -164,7 +101,10 @@ def mock_open_func(file, *args, **kwargs): monkeypatch.setattr(builtins, "open", mock_open_func) - gen = _pcie_gen(device) + speed = _read_from_sysfs(path) + assert speed is not None + + gen = _pcie_gen(speed) assert gen == 5 def test_pcie_gen_with_suffix(self, monkeypatch): @@ -180,7 +120,10 @@ def mock_open_func(file, *args, **kwargs): monkeypatch.setattr(builtins, "open", mock_open_func) - gen = _pcie_gen(device) + speed = _read_from_sysfs(path) + assert speed is not None + + gen = _pcie_gen(speed) assert gen == 3 def test_pcie_gen_unknown_speed(self, monkeypatch): @@ -196,14 +139,22 @@ def mock_open_func(file, *args, **kwargs): monkeypatch.setattr(builtins, "open", mock_open_func) - gen = _pcie_gen(device) + speed = _read_from_sysfs(path) + assert speed is not None + + gen = _pcie_gen(speed) assert gen is None def test_pcie_gen_file_not_found(self, monkeypatch): device = "0000:01:00.0" + path = f"/sys/bus/pci/devices/{device}/current_link_speed" + monkeypatch.setattr(posixpath, "exists", lambda x: False) - gen = _pcie_gen(device) + speed = _read_from_sysfs(path) + assert speed is None + + gen = _pcie_gen(speed) assert gen is None def test_pcie_gen_read_exception(self, monkeypatch): @@ -217,197 +168,34 @@ def mock_open_func(file, *args, **kwargs): monkeypatch.setattr(builtins, "open", mock_open_func) - gen = _pcie_gen(device) - assert gen is None + speed = _read_from_sysfs(path) + assert speed is None + gen = _pcie_gen(speed) + assert gen is None +@pytest.mark.parametrize("bdf", ["0000:01:00.0"]) class TestCheckGpuClass: - """Tests for _check_gpu_class function.""" - - def test_check_gpu_class_display_controller(self, monkeypatch): - device = "0000:01:00.0" - - def mock_open_func(file, *args, **kwargs): - if "class" in file: - return mock_open(read_data="0x030000")() - raise FileNotFoundError(file) - - monkeypatch.setattr(builtins, "open", mock_open_func) - - assert _check_gpu_class(device) is True - - def test_check_gpu_class_vga_controller(self, monkeypatch): - device = "0000:01:00.0" - - def mock_open_func(file, *args, **kwargs): - if "class" in file: - return mock_open(read_data="0x030200")() # 3D controller - raise FileNotFoundError(file) - - monkeypatch.setattr(builtins, "open", mock_open_func) - - assert _check_gpu_class(device) is True - - def test_check_gpu_class_network_controller(self, monkeypatch): - device = "0000:01:00.0" - - def mock_open_func(file, *args, **kwargs): - if "class" in file: - return mock_open(read_data="0x020000")() # Network controller - raise FileNotFoundError(file) - - monkeypatch.setattr(builtins, "open", mock_open_func) - - assert _check_gpu_class(device) is False - - def test_check_gpu_class_storage_controller(self, monkeypatch): - device = "0000:01:00.0" - - def mock_open_func(file, *args, **kwargs): - if "class" in file: - return mock_open(read_data="0x010000")() # Storage controller - raise FileNotFoundError(file) - - monkeypatch.setattr(builtins, "open", mock_open_func) - - assert _check_gpu_class(device) is False - - -class TestPopulateAmdInfo: - """Tests for _populate_amd_info function.""" - - def test_populate_amd_info_with_vram(self, monkeypatch): - device = "0000:03:00.0" - gpu = GPUInfo() - - vram_path = f"/sys/bus/pci/devices/{device}/drm/card0/device/mem_info_vram_total" - monkeypatch.setattr("glob.glob", lambda x: [vram_path]) - - def mock_open_func(file, *args, **kwargs): - if file == vram_path: - return mock_open(read_data=str(8 * 1024 * 1024 * 1024))() - raise FileNotFoundError(file) - - monkeypatch.setattr(builtins, "open", mock_open_func) - - gpu = _populate_amd_info(gpu, device) - - assert gpu.vram is not None - assert gpu.vram.capacity == 8192 - - def test_populate_amd_info_no_vram(self, monkeypatch): - device = "0000:03:00.0" - gpu = GPUInfo() - - monkeypatch.setattr("glob.glob", lambda x: []) - - gpu = _populate_amd_info(gpu, device) - - assert gpu.vram is None - - -class TestPopulateNvidiaInfo: - """Tests for _populate_nvidia_info function.""" - - def test_populate_nvidia_info_success(self, monkeypatch): - device = "0000:01:00.0" - gpu = GPUInfo() - - def mock_run(command, *args, **kwargs): - if command[0] == "nvidia-smi": - return subprocess.CompletedProcess(command, 0, stdout="GeForce RTX 3080, 16, 4, 10240\n") - return subprocess.CompletedProcess(command, 1) - - monkeypatch.setattr(subprocess, "run", mock_run) - - gpu = _populate_nvidia_info(gpu, device) - - assert gpu.name == "GeForce RTX 3080" - assert gpu.pcie_width == 16 - assert gpu.pcie_gen == 4 - assert gpu.vram.capacity == 10240 - - def test_populate_nvidia_info_failure(self, monkeypatch): - device = "0000:01:00.0" - gpu = GPUInfo() - - def mock_run(command, *args, **kwargs): - raise subprocess.CalledProcessError(1, command) - - monkeypatch.setattr(subprocess, "run", mock_run) - - try: - _populate_nvidia_info(gpu, device) - assert False, "Expected exception" - except: - pass # Expected - - -class TestPopulateLspciInfo: - """Tests for _populate_lspci_info function.""" - - def test_populate_lspci_info_full(self, monkeypatch): - device = "0000:01:00.0" - gpu = GPUInfo() - - def mock_run(command, *args, **kwargs): - if command[0] == "lspci": - output = ( - "Slot:\t01:00.0\n" - "Vendor:\tNVIDIA Corporation\n" - "Device:\tGeForce GTX 1080\n" - "SVendor:\tASUS\n" - "SDevice:\tROG STRIX GTX 1080\n" - ) - return subprocess.CompletedProcess(command, 0, stdout=output) - return subprocess.CompletedProcess(command, 1) - - monkeypatch.setattr(subprocess, "run", mock_run) - - gpu = _populate_lspci_info(gpu, device) - - assert gpu.manufacturer == "NVIDIA Corporation" - assert gpu.name == "GeForce GTX 1080" - assert gpu.subsystem_manufacturer == "ASUS" - assert gpu.subsystem_model == "ROG STRIX GTX 1080" - - def test_populate_lspci_info_minimal(self, monkeypatch): - device = "0000:01:00.0" - gpu = GPUInfo() - - def mock_run(command, *args, **kwargs): - if command[0] == "lspci": - output = "Vendor:\tIntel Corporation\nDevice:\tUHD Graphics 620\n" - return subprocess.CompletedProcess(command, 0, stdout=output) - return subprocess.CompletedProcess(command, 1) - - monkeypatch.setattr(subprocess, "run", mock_run) - - gpu = _populate_lspci_info(gpu, device) - - assert gpu.manufacturer == "Intel Corporation" - assert gpu.name == "UHD Graphics 620" - assert gpu.subsystem_manufacturer is None - assert gpu.subsystem_model is None - - def test_populate_lspci_info_failure(self, monkeypatch): - device = "0000:01:00.0" - gpu = GPUInfo() - - def mock_run(command, *args, **kwargs): - raise FileNotFoundError("lspci not found") - - monkeypatch.setattr(subprocess, "run", mock_run) - - try: - _populate_lspci_info(gpu, device) - assert False, "Expected exception" - except FileNotFoundError: - pass # Expected + """Tests for _check_gpu_class.""" + + @pytest.mark.parametrize( + ("device_class", "expected"), + [ + ("0x030000", True), + ("0x030200", True), + ("0x020000", False), + ], + ) + def test_check_gpu_class(self, bdf, monkeypatch, device_class, expected): + monkeypatch.setattr( + "hwprobe.core.linux.graphics._read_from_sysfs", + lambda *args: device_class, + ) + assert _check_gpu_class(bdf) is expected class TestFetchGraphicsInfo: - """Tests for fetch_graphics_info function.""" + """Tests for fetch_graphics_info.""" def test_fetch_graphics_info_root_not_found(self, monkeypatch): monkeypatch.setattr(posixpath, "exists", lambda x: False) @@ -426,12 +214,15 @@ def test_fetch_graphics_info_success_intel(self, monkeypatch): "class": "0x030000", "vendor": "0x8086", "device": "0x5917", - "current_link_width": "0", + "current_link_width": "16", "current_link_speed": "8.0 GT/s", + "max_link_width": "16", + "max_link_speed": "8.0 GT/s", "firmware_node/path": "\\_SB.PCI0.GFX0", } def custom_open(path, *args, **kwargs): + filename = posixpath.basename(path) filename = posixpath.basename(path) if filename == "path" and "firmware_node" in path: return mock_open(read_data=file_contents["firmware_node/path"])() @@ -441,29 +232,22 @@ def custom_open(path, *args, **kwargs): monkeypatch.setattr(builtins, "open", custom_open) monkeypatch.setattr("hwprobe.core.linux.graphics.pci_path_linux", lambda x: "PciRoot(0x0)/Pci(0x2,0x0)") - - def mock_run(command, *args, **kwargs): - if command[0] == "lspci": - output = "Vendor:\tIntel Corporation\nDevice:\tUHD Graphics 620\nSVendor:\tLenovo\nSDevice:\tThinkPad\n" - return subprocess.CompletedProcess(command, 0, stdout=output) - return subprocess.CompletedProcess(command, 1) - - monkeypatch.setattr(subprocess, "run", mock_run) + monkeypatch.setattr("hwprobe.core.linux.graphics.NATIVE_AVAILABLE", False) info = fetch_graphics_info() - assert info.status.type == StatusType.SUCCESS + assert info.status.type == StatusType.PARTIAL assert len(info.modules) == 1 - gpu = info.modules[0] assert gpu.vendor_id == "0x8086" assert gpu.device_id == "0x5917" assert gpu.acpi_path == "\\_SB.PCI0.GFX0" - assert gpu.manufacturer == "Intel Corporation" - assert gpu.name == "UHD Graphics 620" assert gpu.pcie_gen == 3 + assert gpu.pcie_width == 16 + # Uncomment this once PCI-IDs parser is implemented + # assert gpu.manufacturer == "Intel Corporation" - def test_fetch_graphics_info_nvidia(self, monkeypatch): + def test_fetch_graphics_info_native_nvidia(self, monkeypatch): monkeypatch.setattr(posixpath, "exists", lambda x: True) monkeypatch.setattr(os, "listdir", lambda x: ["0000:01:00.0"]) @@ -473,6 +257,8 @@ def test_fetch_graphics_info_nvidia(self, monkeypatch): "device": "0x1c03", "current_link_width": "16", "current_link_speed": "8.0 GT/s", + "max_link_width": "16", + "max_link_speed": "8.0 GT/s", "firmware_node/path": "\\_SB.PCI0.PEG0.PEGP", } @@ -486,26 +272,29 @@ def custom_open(path, *args, **kwargs): monkeypatch.setattr(builtins, "open", custom_open) monkeypatch.setattr("hwprobe.core.linux.graphics.pci_path_linux", lambda x: "PciRoot(0x0)/Pci(0x1,0x0)") - - def mock_run(command, *args, **kwargs): - if command[0] == "nvidia-smi": - return subprocess.CompletedProcess(command, 0, stdout="GeForce GTX 1060, 16, 3, 6144\n") - if command[0] == "lspci": - output = "Vendor:\tNVIDIA\nDevice:\tGeForce GTX 1060\n" - return subprocess.CompletedProcess(command, 0, stdout=output) - return subprocess.CompletedProcess(command, 1) - - monkeypatch.setattr(subprocess, "run", mock_run) + monkeypatch.setattr("hwprobe.core.linux.graphics.NATIVE_AVAILABLE", True) + monkeypatch.setattr( + "hwprobe.core.linux.graphics.native_gpu.get_gpu_info", + lambda *args, **kwargs: ( + GPUInfoQueryStatus.DRM_SUCCESS, + GPUProperties( + name="GeForce GTX 1060", + vram_total_mb=6144, + vram_used_mb=0, + ), + ), + ) info = fetch_graphics_info() assert len(info.modules) == 1 gpu = info.modules[0] assert gpu.vendor_id == "0x10de" + assert gpu.name == "GeForce GTX 1060" assert gpu.vram is not None assert gpu.vram.capacity == 6144 - def test_fetch_graphics_info_amd(self, monkeypatch): + def test_fetch_graphics_info_native_amd(self, monkeypatch): monkeypatch.setattr(posixpath, "exists", lambda x: True) monkeypatch.setattr(os, "listdir", lambda x: ["0000:03:00.0"]) @@ -515,6 +304,8 @@ def test_fetch_graphics_info_amd(self, monkeypatch): "device": "0x731f", "current_link_width": "16", "current_link_speed": "16.0 GT/s", + "max_link_width": "16", + "max_link_speed": "16.0 GT/s", "firmware_node/path": "\\_SB.PCI0.PEG0.PEGP", } @@ -524,29 +315,29 @@ def custom_open(path, *args, **kwargs): return mock_open(read_data=file_contents["firmware_node/path"])() if filename in file_contents: return mock_open(read_data=file_contents[filename])() - if filename == "mem_info_vram_total": - return mock_open(read_data=str(8 * 1024 * 1024 * 1024))() raise FileNotFoundError(path) monkeypatch.setattr(builtins, "open", custom_open) monkeypatch.setattr("hwprobe.core.linux.graphics.pci_path_linux", lambda x: "PciRoot(0x0)/Pci(0x3,0x0)") + monkeypatch.setattr("hwprobe.core.linux.graphics.NATIVE_AVAILABLE", True) monkeypatch.setattr( - "glob.glob", lambda x: ["/sys/bus/pci/devices/0000:03:00.0/drm/card0/device/mem_info_vram_total"] + "hwprobe.core.linux.graphics.native_gpu.get_gpu_info", + lambda *args, **kwargs: ( + GPUInfoQueryStatus.DRM_SUCCESS, + GPUProperties( + name="Radeon RX 5700 XT", + vram_total_mb=8192, + vram_used_mb=0, + ), + ) ) - def mock_run(command, *args, **kwargs): - if command[0] == "lspci": - output = "Vendor:\tAMD\nDevice:\tRadeon RX 5700 XT\n" - return subprocess.CompletedProcess(command, 0, stdout=output) - return subprocess.CompletedProcess(command, 1) - - monkeypatch.setattr(subprocess, "run", mock_run) - info = fetch_graphics_info() assert len(info.modules) == 1 gpu = info.modules[0] assert gpu.vendor_id == "0x1002" + assert gpu.name == "Radeon RX 5700 XT" assert gpu.vram is not None assert gpu.vram.capacity == 8192 assert gpu.pcie_gen == 4 @@ -555,14 +346,9 @@ def test_fetch_graphics_info_skip_non_display(self, monkeypatch): monkeypatch.setattr(posixpath, "exists", lambda x: True) monkeypatch.setattr(os, "listdir", lambda x: ["0000:04:00.0"]) - file_contents = { - "class": "0x020000", # Network Controller - } - def custom_open(path, *args, **kwargs): - filename = posixpath.basename(path) - if filename in file_contents: - return mock_open(read_data=file_contents[filename])() + if "class" in path: + return mock_open(read_data="0x020000")() raise FileNotFoundError(path) monkeypatch.setattr(builtins, "open", custom_open) @@ -570,13 +356,15 @@ def custom_open(path, *args, **kwargs): info = fetch_graphics_info() assert len(info.modules) == 0 - assert info.status.type == StatusType.SUCCESS + assert info.status.type == StatusType.FAILED def test_fetch_graphics_info_partial_failure(self, monkeypatch): + monkeypatch.setattr(posixpath, "exists", lambda x: True) monkeypatch.setattr(posixpath, "exists", lambda x: True) monkeypatch.setattr(os, "listdir", lambda x: ["0000:01:00.0"]) def custom_open(path, *args, **kwargs): + filename = posixpath.basename(path) filename = posixpath.basename(path) if filename == "class": return mock_open(read_data="0x030000")() @@ -584,19 +372,22 @@ def custom_open(path, *args, **kwargs): raise OSError("Permission denied") if filename == "device": return mock_open(read_data="0x1234")() - raise OSError("File not found") + if filename in {"current_link_speed", "max_link_speed"}: + return mock_open(read_data="8.0 GT/s")() + if filename in {"current_link_width", "max_link_width"}: + return mock_open(read_data="16")() + if filename == "path" and "firmware_node" in path: + return mock_open(read_data="\\_SB.PCI0.GFX0")() + raise FileNotFoundError(path) monkeypatch.setattr(builtins, "open", custom_open) - - def mock_run(*args, **kwargs): - raise FileNotFoundError("lspci not found") - - monkeypatch.setattr(subprocess, "run", mock_run) + monkeypatch.setattr("hwprobe.core.linux.graphics.pci_path_linux", lambda x: "PciRoot(0x0)/Pci(0x1,0x0)") info = fetch_graphics_info() assert info.status.type == StatusType.PARTIAL assert len(info.modules) == 1 + assert any("Could not read vendor ID" in msg for msg in info.status.messages) def test_fetch_graphics_info_acpi_path_failure(self, monkeypatch): monkeypatch.setattr(posixpath, "exists", lambda x: True) @@ -606,11 +397,15 @@ def test_fetch_graphics_info_acpi_path_failure(self, monkeypatch): "class": "0x030000", "vendor": "0x8086", "device": "0x5917", - "current_link_width": "0", + "current_link_width": "16", "current_link_speed": "8.0 GT/s", + "max_link_width": "16", + "max_link_speed": "8.0 GT/s", } + def custom_open(path, *args, **kwargs): + filename = posixpath.basename(path) filename = posixpath.basename(path) if filename == "path" and "firmware_node" in path: raise FileNotFoundError("No ACPI path") @@ -620,7 +415,6 @@ def custom_open(path, *args, **kwargs): monkeypatch.setattr(builtins, "open", custom_open) monkeypatch.setattr("hwprobe.core.linux.graphics.pci_path_linux", lambda x: "PciRoot(0x0)/Pci(0x2,0x0)") - monkeypatch.setattr(subprocess, "run", lambda *args, **kwargs: subprocess.CompletedProcess(args, 0, stdout="")) info = fetch_graphics_info() @@ -632,6 +426,7 @@ def custom_open(path, *args, **kwargs): assert any("ACPI path" in msg for msg in info.status.messages) def test_fetch_graphics_info_pci_path_failure(self, monkeypatch): + monkeypatch.setattr(posixpath, "exists", lambda x: True) monkeypatch.setattr(posixpath, "exists", lambda x: True) monkeypatch.setattr(os, "listdir", lambda x: ["0000:00:02.0"]) @@ -639,9 +434,10 @@ def test_fetch_graphics_info_pci_path_failure(self, monkeypatch): "class": "0x030000", "vendor": "0x8086", "device": "0x5917", - "current_link_width": "0", + "current_link_width": "16", "current_link_speed": "8.0 GT/s", - "firmware_node/path": "\\_SB.PCI0.GFX0", + "max_link_width": "16", + "max_link_speed": "8.0 GT/s", } def custom_open(path, *args, **kwargs): @@ -653,12 +449,7 @@ def custom_open(path, *args, **kwargs): raise FileNotFoundError(path) monkeypatch.setattr(builtins, "open", custom_open) - - def mock_pci_path(device): - raise Exception("PCI path failed") - - monkeypatch.setattr("hwprobe.core.linux.graphics.pci_path_linux", mock_pci_path) - monkeypatch.setattr(subprocess, "run", lambda *args, **kwargs: subprocess.CompletedProcess(args, 0, stdout="")) + monkeypatch.setattr("hwprobe.core.linux.graphics.pci_path_linux", lambda x: None) info = fetch_graphics_info() @@ -678,7 +469,8 @@ def test_fetch_graphics_info_nvidia_failure(self, monkeypatch): "device": "0x1c03", "current_link_width": "16", "current_link_speed": "8.0 GT/s", - "firmware_node/path": "\\_SB.PCI0.PEG0.PEGP", + "max_link_width": "16", + "max_link_speed": "8.0 GT/s", } def custom_open(path, *args, **kwargs): @@ -691,24 +483,29 @@ def custom_open(path, *args, **kwargs): monkeypatch.setattr(builtins, "open", custom_open) monkeypatch.setattr("hwprobe.core.linux.graphics.pci_path_linux", lambda x: "PciRoot(0x0)/Pci(0x1,0x0)") - - def mock_run(command, *args, **kwargs): - if command[0] == "nvidia-smi": - raise subprocess.CalledProcessError(1, command) - return subprocess.CompletedProcess(command, 0, stdout="") - - monkeypatch.setattr(subprocess, "run", mock_run) + monkeypatch.setattr("hwprobe.core.linux.graphics.NATIVE_AVAILABLE", True) + monkeypatch.setattr( + "hwprobe.core.linux.graphics.native_gpu.get_gpu_info", + lambda *args, **kwargs: ( + GPUInfoQueryStatus.DRM_SUCCESS, + GPUProperties( + name="GeForce GTX 1060", + vram_total_mb=6144, + vram_used_mb=0, + ), + ), + ) info = fetch_graphics_info() assert len(info.modules) == 1 gpu = info.modules[0] assert gpu.vendor_id == "0x10de" - assert gpu.vram is None - assert info.status.type == StatusType.PARTIAL - assert any("Could not get additional GPU info" in msg for msg in info.status.messages) + assert gpu.name == "GeForce GTX 1060" + assert gpu.vram is not None + assert gpu.vram.capacity == 6144 - def test_fetch_graphics_info_lspci_failure(self, monkeypatch): + def test_fetch_graphics_info_native_failure(self, monkeypatch): monkeypatch.setattr(posixpath, "exists", lambda x: True) monkeypatch.setattr(os, "listdir", lambda x: ["0000:00:02.0"]) @@ -716,28 +513,24 @@ def test_fetch_graphics_info_lspci_failure(self, monkeypatch): "class": "0x030000", "vendor": "0x8086", "device": "0x5917", - "current_link_width": "0", + "current_link_width": "16", + "max_link_width": "16", "current_link_speed": "8.0 GT/s", - "firmware_node/path": "\\_SB.PCI0.GFX0", + "max_link_speed": "8.0 GT/s", } def custom_open(path, *args, **kwargs): filename = posixpath.basename(path) if filename == "path" and "firmware_node" in path: - return mock_open(read_data=file_contents["firmware_node/path"])() + return mock_open(read_data="\\_SB.PCI0.GFX0")() if filename in file_contents: return mock_open(read_data=file_contents[filename])() raise FileNotFoundError(path) monkeypatch.setattr(builtins, "open", custom_open) monkeypatch.setattr("hwprobe.core.linux.graphics.pci_path_linux", lambda x: "PciRoot(0x0)/Pci(0x2,0x0)") - - def mock_run(command, *args, **kwargs): - if command[0] == "lspci": - raise FileNotFoundError("lspci not found") - return subprocess.CompletedProcess(command, 0, stdout="") - - monkeypatch.setattr(subprocess, "run", mock_run) + monkeypatch.setattr("hwprobe.core.linux.graphics.NATIVE_AVAILABLE", True) + monkeypatch.setattr("hwprobe.core.linux.graphics.native_gpu.get_gpu_info", lambda *args, **kwargs: None) info = fetch_graphics_info() @@ -745,10 +538,10 @@ def mock_run(command, *args, **kwargs): gpu = info.modules[0] assert gpu.vendor_id == "0x8086" assert info.status.type == StatusType.PARTIAL - assert any("LSPCI" in msg for msg in info.status.messages) + assert any("Native GPU info library could not fetch GPU name or VRAM" in msg for msg in info.status.messages) def test_fetch_graphics_info_pcie_gen_failure(self, monkeypatch): - monkeypatch.setattr(posixpath, "exists", lambda x: "/current_link_speed" not in x) + monkeypatch.setattr(posixpath, "exists", lambda x: "/current_link_speed" not in str(x)) monkeypatch.setattr(os, "listdir", lambda x: ["0000:00:02.0"]) file_contents = { @@ -762,22 +555,22 @@ def test_fetch_graphics_info_pcie_gen_failure(self, monkeypatch): def custom_open(path, *args, **kwargs): filename = posixpath.basename(path) if filename == "path" and "firmware_node" in path: - return mock_open(read_data=file_contents["firmware_node/path"])() + return mock_open(read_data="\\_SB.PCI0.GFX0")() if filename in file_contents: return mock_open(read_data=file_contents[filename])() raise FileNotFoundError(path) monkeypatch.setattr(builtins, "open", custom_open) monkeypatch.setattr("hwprobe.core.linux.graphics.pci_path_linux", lambda x: "PciRoot(0x0)/Pci(0x2,0x0)") - monkeypatch.setattr(subprocess, "run", lambda *args, **kwargs: subprocess.CompletedProcess(args, 0, stdout="")) info = fetch_graphics_info() assert len(info.modules) == 1 gpu = info.modules[0] + assert gpu.pcie_width is None assert gpu.pcie_gen is None assert info.status.type == StatusType.PARTIAL - assert any("PCI gen" in msg for msg in info.status.messages) + assert any("current link speed" in msg for msg in info.status.messages) def test_fetch_graphics_info_class_read_failure(self, monkeypatch): monkeypatch.setattr(posixpath, "exists", lambda x: True) @@ -792,7 +585,5 @@ def custom_open(path, *args, **kwargs): info = fetch_graphics_info() - # Device should be skipped due to class read failure assert len(info.modules) == 0 - assert info.status.type == StatusType.PARTIAL - assert any("Could not open file" in msg for msg in info.status.messages) + assert info.status.type == StatusType.FAILED