Coverage for cuda/pathfinder/_dynamic_libs/search_platform.py: 95.62%
137 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-03 02:41 +0000
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-03 02:41 +0000
1# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
2# SPDX-License-Identifier: Apache-2.0
4"""Platform abstraction for filesystem search steps.
6The goal is to keep :mod:`search_steps` platform-agnostic: it should not branch
7on OS flags like ``IS_WINDOWS``. Instead, it calls through the single
8``PLATFORM`` instance exported here.
9"""
11from __future__ import annotations
13import os
14from collections.abc import Sequence
15from dataclasses import dataclass
16from pathlib import PurePath
17from typing import Protocol, cast
19from cuda.pathfinder._dynamic_libs.lib_descriptor import LibDescriptor
20from cuda.pathfinder._dynamic_libs.supported_nvidia_libs import is_suppressed_dll_file
21from cuda.pathfinder._utils.find_sub_dirs import find_sub_dirs_all_sitepackages
22from cuda.pathfinder._utils.platform_aware import IS_WINDOWS
23from cuda.pathfinder._utils.windows_arch import windows_pe_matches_arch, windows_python_arch
26def _no_such_file_in_sub_dirs(
27 sub_dirs: Sequence[str], file_description: str, error_messages: list[str], attachments: list[str]
28) -> None:
29 error_messages.append(f"No such file: {file_description}") 1agFijklref
30 for sub_dir in find_sub_dirs_all_sitepackages(sub_dirs): 1agFijklref
31 attachments.append(f' listdir("{sub_dir}"):') 1ijklref
32 for node in sorted(os.listdir(sub_dir)): 1ijklref
33 attachments.append(f" {node}") 1ijklef
36def _find_descriptor_so_under_dir(dirpath: str, desc: LibDescriptor) -> str | None:
37 for soname in desc.linux_sonames: 1ambcnoST2UVBCDE3GHIijklrJWXYZ01
38 path = os.path.join(dirpath, soname) 1ambcnoST2UVBCDE3GHIijklrJWXYZ01
39 if os.path.isfile(path): 1ambcnoST2UVBCDE3GHIijklrJWXYZ01
40 return path 1ambcnoST2UV3GHIJWXYZ01
41 return None 1BCDEijklr
44def _find_so_in_rel_dirs(
45 rel_dirs: tuple[str, ...],
46 desc: LibDescriptor,
47 file_description: str,
48 error_messages: list[str],
49 attachments: list[str],
50) -> str | None:
51 sub_dirs_searched: list[tuple[str, ...]] = [] 1agGHIijklrJ
52 for rel_dir in rel_dirs: 1agGHIijklrJ
53 sub_dir = PurePath(rel_dir).parts 1agGHIijklrJ
54 for abs_dir in find_sub_dirs_all_sitepackages(sub_dir): 1agGHIijklrJ
55 so_path = _find_descriptor_so_under_dir(abs_dir, desc) 1GHIijklrJ
56 if so_path is not None: 1GHIijklrJ
57 return so_path 1GHIJ
58 sub_dirs_searched.append(sub_dir) 1agijklr
59 for sub_dir in sub_dirs_searched: 1agijklr
60 _no_such_file_in_sub_dirs(sub_dir, file_description, error_messages, attachments) 1agijklr
61 return None 1agijklr
64def _find_descriptor_dll_under_dir(
65 dirpath: str,
66 desc: LibDescriptor,
67 target_arch: str | None = None,
68) -> str | None:
69 def candidate_is_usable(path: str) -> bool: 1ambcnoOwKLPQpMNqtusve456789fhyzAdx
70 if not os.path.isfile(path): 1ambcnoOwKLPQpMNqtusve456789fhyzAdx
71 return False 1ambcnopMNqtuse56789fhd
72 if is_suppressed_dll_file(os.path.basename(path)): 1ambcnoOwKLPQpMNqtusv4yzAdx
73 return False
74 return target_arch is None or windows_pe_matches_arch(path, target_arch) 1ambcnoOwKLPQpMNqtusv4yzAdx
76 for dll_basename in desc.windows_dlls: 1ambcnoOwKLPQpMNqtusve456789fhyzAdx
77 path = os.path.join(dirpath, dll_basename) 1ambcnoOwKLPQpMNqtusve456789fhyzAdx
78 if candidate_is_usable(path): 1ambcnoOwKLPQpMNqtusve456789fhyzAdx
79 return path 1ambcnoOwKLPQMNtusv4yzAdx
80 return None 1pqe56789fhd
83def _find_dll_in_rel_dirs(
84 rel_dirs: tuple[str, ...],
85 desc: LibDescriptor,
86 target_arch: str,
87 lib_searched_for: str,
88 error_messages: list[str],
89 attachments: list[str],
90) -> str | None:
91 sub_dirs_searched: list[tuple[str, ...]] = [] 1agtFusvef
92 checked_arch = target_arch if desc.requires_windows_binary_arch_check else None 1agtFusvef
93 for rel_dir in rel_dirs: 1agtFusvef
94 sub_dir = PurePath(rel_dir).parts 1agtFusvef
95 for abs_dir in find_sub_dirs_all_sitepackages(sub_dir): 1agtFusvef
96 dll_name = _find_descriptor_dll_under_dir(abs_dir, desc, checked_arch) 1tusvef
97 if dll_name is not None: 1tusvef
98 return dll_name 1tusv
99 sub_dirs_searched.append(sub_dir) 1agFsef
100 for sub_dir in sub_dirs_searched: 1agFef
101 _no_such_file_in_sub_dirs(sub_dir, lib_searched_for, error_messages, attachments) 1agFef
102 return None 1agFef
105class SearchPlatform(Protocol):
106 def lib_searched_for(self, desc: LibDescriptor) -> str: ...
108 def site_packages_rel_dirs(self, desc: LibDescriptor) -> tuple[str, ...]: ...
110 def conda_anchor_point(self, conda_prefix: str) -> str: ...
112 def anchor_rel_dirs(self, desc: LibDescriptor) -> tuple[str, ...]: ...
114 def install_root_env_vars(self, desc: LibDescriptor) -> tuple[str, ...]: ...
116 def install_root_env_rel_dirs(self, desc: LibDescriptor) -> tuple[str, ...]: ...
118 def program_files_root_globs(self, desc: LibDescriptor) -> tuple[str, ...]: ...
120 def find_in_site_packages(
121 self,
122 rel_dirs: tuple[str, ...],
123 desc: LibDescriptor,
124 error_messages: list[str],
125 attachments: list[str],
126 ) -> str | None: ...
128 def find_in_lib_dir(
129 self,
130 lib_dir: str,
131 desc: LibDescriptor,
132 error_messages: list[str],
133 attachments: list[str],
134 ) -> str | None: ...
137@dataclass(frozen=True, slots=True)
138class LinuxSearchPlatform:
139 def lib_searched_for(self, desc: LibDescriptor) -> str:
140 return " or ".join(desc.linux_sonames) 1aRg)BCDEGHIijklrJ*+,
142 def site_packages_rel_dirs(self, desc: LibDescriptor) -> tuple[str, ...]:
143 return cast(tuple[str, ...], desc.site_packages_linux) 1agGHIijklrJ-
145 def conda_anchor_point(self, conda_prefix: str) -> str:
146 return conda_prefix 1STUVBCDE
148 def anchor_rel_dirs(self, desc: LibDescriptor) -> tuple[str, ...]:
149 return cast(tuple[str, ...], desc.anchor_rel_dirs_linux) 1amb%cno'ST./2UVBCDE3
151 def install_root_env_vars(self, desc: LibDescriptor) -> tuple[str, ...]:
152 return cast(tuple[str, ...], desc.install_root_env_vars_linux) 1abcRWXYZ01
154 def install_root_env_rel_dirs(self, desc: LibDescriptor) -> tuple[str, ...]:
155 return cast(tuple[str, ...], desc.install_root_env_rel_dirs_linux) 1abcRWXYZ01
157 def program_files_root_globs(self, _desc: LibDescriptor) -> tuple[str, ...]:
158 return () 1bR
160 def find_in_site_packages(
161 self,
162 rel_dirs: tuple[str, ...],
163 desc: LibDescriptor,
164 error_messages: list[str],
165 attachments: list[str],
166 ) -> str | None:
167 return _find_so_in_rel_dirs(rel_dirs, desc, self.lib_searched_for(desc), error_messages, attachments) 1agGHIijklrJ
169 def find_in_lib_dir(
170 self,
171 lib_dir: str,
172 desc: LibDescriptor,
173 error_messages: list[str],
174 attachments: list[str],
175 ) -> str | None:
176 so_path = _find_descriptor_so_under_dir(lib_dir, desc) 1ambcnoST2UVBCDE3WXYZ01
177 if so_path is not None: 1ambcnoST2UVBCDE3WXYZ01
178 return so_path 1ambcnoST2UV3WXYZ01
179 error_messages.append(f"No such file: {self.lib_searched_for(desc)}") 1BCDE
180 attachments.append(f' listdir("{lib_dir}"):') 1BCDE
181 if not os.path.isdir(lib_dir): 1BCDE
182 attachments.append(" DIRECTORY DOES NOT EXIST")
183 else:
184 for node in sorted(os.listdir(lib_dir)): 1BCDE
185 attachments.append(f" {node}") 1BCDE
186 return None 1BCDE
189@dataclass(frozen=True, slots=True)
190class WindowsSearchPlatform:
191 target_arch: str
193 def lib_searched_for(self, desc: LibDescriptor) -> str:
194 return f"known {desc.name} DLL" 1aRgpqtFusvef:hd
196 def site_packages_rel_dirs(self, desc: LibDescriptor) -> tuple[str, ...]:
197 return cast(tuple[str, ...], desc.site_packages_windows.for_arch(self.target_arch)) 1agwtFusvef!
199 def conda_anchor_point(self, conda_prefix: str) -> str:
200 return os.path.join(conda_prefix, "Library") 1KL!
202 def anchor_rel_dirs(self, desc: LibDescriptor) -> tuple[str, ...]:
203 return cast(tuple[str, ...], desc.anchor_rel_dirs_windows.for_arch(self.target_arch)) 1amb%cno'O;=?@KLPQpMNq!dx
205 def install_root_env_vars(self, desc: LibDescriptor) -> tuple[str, ...]:
206 if self.target_arch not in desc.supported_windows_arch: 1abcRw#$hyzA
207 return ()
208 return cast(tuple[str, ...], desc.install_root_env_vars_windows) 1abcRw#$hyzA
210 def install_root_env_rel_dirs(self, desc: LibDescriptor) -> tuple[str, ...]:
211 if self.target_arch not in desc.supported_windows_arch: 1abcRw#$hyzA
212 return ()
213 return cast(tuple[str, ...], desc.install_root_env_rel_dirs_windows.for_arch(self.target_arch)) 1abcRw#$hyzA
215 def program_files_root_globs(self, desc: LibDescriptor) -> tuple[str, ...]:
216 program_files = os.environ.get("PROGRAMW6432") or os.environ.get("PROGRAMFILES") 1bR!dx
217 if not program_files: 1bR!dx
218 return ()
219 rel_globs = desc.program_files_root_globs_windows.for_arch(self.target_arch) 1bR!dx
220 return tuple(os.path.join(program_files, rel_glob) for rel_glob in rel_globs) 1bR!dx
222 def find_in_site_packages(
223 self,
224 rel_dirs: tuple[str, ...],
225 desc: LibDescriptor,
226 error_messages: list[str],
227 attachments: list[str],
228 ) -> str | None:
229 return _find_dll_in_rel_dirs( 1agtFusvef
230 rel_dirs,
231 desc,
232 self.target_arch,
233 self.lib_searched_for(desc),
234 error_messages,
235 attachments,
236 )
238 def find_in_lib_dir(
239 self,
240 lib_dir: str,
241 desc: LibDescriptor,
242 error_messages: list[str],
243 attachments: list[str],
244 ) -> str | None:
245 target_arch = self.target_arch if desc.requires_windows_binary_arch_check else None 1ambcnoOwKLPQpMNqhyzAdx
246 dll_name = _find_descriptor_dll_under_dir(lib_dir, desc, target_arch) 1ambcnoOwKLPQpMNqhyzAdx
247 if dll_name is not None: 1ambcnoOwKLPQpMNqhyzAdx
248 return dll_name 1ambcnoOwKLPQMNyzAdx
249 lib_searched_for = self.lib_searched_for(desc) 1pqhd
250 if target_arch is None: 1pqhd
251 error_messages.append(f"No such file: {lib_searched_for}") 1hd
252 else:
253 error_messages.append(f"No {target_arch}-compatible PE file: {lib_searched_for}") 1pq
254 attachments.append(f' listdir("{lib_dir}"):') 1pqhd
255 if not os.path.isdir(lib_dir): 1pqhd
256 attachments.append(" DIRECTORY DOES NOT EXIST")
257 else:
258 for node in sorted(os.listdir(lib_dir)): 1pqhd
259 attachments.append(f" {node}") 1pqhd
260 return None 1pqhd
263def _platform_for_current_system() -> SearchPlatform:
264 if IS_WINDOWS: 1a(
265 return WindowsSearchPlatform(target_arch=windows_python_arch())
266 return LinuxSearchPlatform() 1a(
269PLATFORM = _platform_for_current_system()