Coverage for trimesh/resolvers.py: 83%
221 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-07-31 23:55 +0000
1"""
2resolvers.py
3---------------
5Provides a common interface to load assets referenced by name
6like MTL files, texture images, etc. Assets can be from ZIP
7archives, web assets, or a local file path.
8"""
10import abc
11import itertools
12import os
13from pathlib import Path
14from typing import TypeAlias
16# URL parsing for remote resources via WebResolver
17from urllib.parse import urlparse
19from . import caching, util
20from .typed import HttpSessionLike, Mapping
23class Resolver(util.ABC):
24 """
25 The base class for resolvers.
26 """
28 @abc.abstractmethod
29 def __init__(self, *args, **kwargs):
30 raise NotImplementedError("Use a resolver subclass!")
32 @abc.abstractmethod
33 def get(self, key):
34 raise NotImplementedError()
36 @abc.abstractmethod
37 def write(self, name: str, data):
38 raise NotImplementedError("`write` not implemented!")
40 @abc.abstractmethod
41 def namespaced(self, namespace: str):
42 raise NotImplementedError("`namespaced` not implemented!")
44 @abc.abstractmethod
45 def keys(self):
46 raise NotImplementedError("`keys` not implemented!")
48 def __getitem__(self, key: str):
49 return self.get(key)
51 def __setitem__(self, key: str, value):
52 return self.write(key, value)
54 def __contains__(self, key: str) -> bool:
55 return key in self.keys()
58class FilePathResolver(Resolver):
59 """
60 Resolve files from a source path on the file system.
61 """
63 def __init__(self, source: str, allow_anywhere: bool = False):
64 """
65 Resolve files based on a source path.
67 Parameters
68 ------------
69 source : str
70 File path where mesh was loaded from
71 allow_anywhere : bool
72 If True allow assets to reference paths outside the
73 resolver root, i.e. `../textures/thing.png` — the
74 pre-5.0 behavior.
75 """
76 # remove everything other than absolute path
77 clean = os.path.expanduser(os.path.abspath(str(source)))
79 self.allow_anywhere = bool(allow_anywhere)
81 self.clean = clean
82 if os.path.isdir(clean):
83 # if we were passed a directory use it
84 self.parent = clean
85 else:
86 # otherwise get the parent directory we've been passed
87 split = os.path.split(clean)
88 self.parent = split[0]
90 # exit if directory doesn't exist
91 if not os.path.isdir(self.parent):
92 raise ValueError(f"path `{self.parent} `not a directory!")
94 self.file_path = source
95 self.file_name = os.path.basename(source)
97 def keys(self):
98 """
99 List all files available to be loaded.
101 Yields
102 -----------
103 name : str
104 Name of a file which can be accessed.
105 """
106 parent = self.parent
107 for path, _, names in os.walk(self.parent):
108 # strip any leading parent key
109 if path.startswith(parent):
110 path = path[len(parent) :]
111 # yield each name
112 for name in names:
113 yield os.path.join(path, name)
115 def namespaced(self, namespace: str) -> "FilePathResolver":
116 """
117 Return a resolver which changes the root of the
118 resolver by an added namespace.
120 Parameters
121 -------------
122 namespace : str
123 Probably a subdirectory
125 Returns
126 --------------
127 resolver : FilePathResolver
128 Resolver with root directory changed.
129 """
130 return FilePathResolver(
131 os.path.join(self.parent, namespace), allow_anywhere=self.allow_anywhere
132 )
134 def absolute(self, name: str) -> Path:
135 """
136 Resolve an asset name to an absolute path under the
137 resolver root.
139 Parameters
140 ------------
141 name : str
142 Name of an asset relative to the resolver root.
144 Returns
145 ------------
146 path : pathlib.Path
147 Absolute resolved path.
149 Raises
150 ------------
151 ValueError
152 If the path escapes the resolver root and
153 `allow_anywhere` was not set.
154 """
155 parent = Path(self.parent).resolve()
156 path = (parent / name.strip()).resolve()
157 if not self.allow_anywhere and not path.is_relative_to(parent):
158 raise ValueError(
159 f"'{name}' escapes resolver root '{parent}' — pass "
160 + "`FilePathResolver(path, allow_anywhere=True)` to allow it"
161 )
162 return path
164 def get(self, name: str):
165 """
166 Get an asset, restricted to the resolver root.
168 Parameters
169 -------------
170 name : str
171 Name of the asset. Must resolve inside the resolver root.
173 Returns
174 ------------
175 data : bytes
176 Loaded data from asset.
177 """
178 candidates = (
179 name.strip(),
180 name.strip().lstrip("/"),
181 os.path.split(name)[-1],
182 )
183 for candidate in candidates:
184 try:
185 path = self.absolute(candidate)
186 except ValueError:
187 continue
188 if path.exists():
189 with open(path, "rb") as f:
190 return f.read()
191 # if the requested name escaped the root this raises
192 # the actionable error instead of a plain not-found
193 self.absolute(name)
194 raise FileNotFoundError(name)
196 def write(self, name: str, data: str | bytes):
197 """
198 Write an asset to a file path, restricted to the resolver root.
200 Parameters
201 -----------
202 name : str
203 Name of the file to write. Must resolve inside the resolver root.
204 data : str or bytes
205 Data to write to the file.
206 """
207 with open(self.absolute(name), "wb") as f:
208 # handle encodings correctly for str/bytes
209 util.write_encoded(file_obj=f, stuff=data)
212class ZipResolver(Resolver):
213 """
214 Resolve files inside a ZIP archive.
215 """
217 def __init__(self, archive: dict | None = None, namespace: str | None = None):
218 """
219 Resolve files inside a ZIP archive as loaded by
220 trimesh.util.decompress
222 Parameters
223 -------------
224 archive : dict
225 Contains resources as file object
226 namespace : None or str
227 If passed will only show keys that start
228 with this value and this substring must be
229 removed for any get calls.
230 """
231 self.archive = archive
232 if isinstance(namespace, str):
233 self.namespace = namespace.strip().rstrip("/") + "/"
234 else:
235 self.namespace = None
237 def keys(self):
238 """
239 Get the available keys in the current archive.
241 Returns
242 -----------
243 keys : iterable
244 Keys in the current archive.
245 """
246 if self.namespace is not None:
247 namespace = self.namespace
248 length = len(namespace)
249 # only return keys that start with the namespace
250 # and strip off the namespace from the returned
251 # keys.
252 return [
253 k[length:]
254 for k in self.archive.keys()
255 if k.startswith(namespace) and len(k) > length
256 ]
257 return self.archive.keys()
259 def write(self, key: str, value) -> None:
260 """
261 Store a value in the current archive.
263 Parameters
264 -----------
265 key : hashable
266 Key to store data under.
267 value : str, bytes, file-like
268 Value to store.
269 """
270 if self.archive is None:
271 self.archive = {}
272 self.archive[key] = value
274 def get(self, name: str) -> bytes:
275 """
276 Get an asset from the ZIP archive.
278 Parameters
279 -------------
280 name : str
281 Name of the asset
283 Returns
284 -------------
285 data : bytes
286 Loaded data from asset
287 """
288 # not much we can do with None
289 if name is None:
290 return
291 # make sure name is a string
292 if hasattr(name, "decode"):
293 name = name.decode("utf-8")
294 # store reference to archive inside this function
295 archive = self.archive
296 # requested name not identical in
297 # storage so attempt to recover
298 if name not in archive:
299 # loop through unique results
300 for option in nearby_names(name, self.namespace):
301 if option in archive:
302 # cleaned option is in archive
303 # so store value and exit
304 name = option
305 break
307 # get the stored data
308 obj = archive[name]
309 # if the dict is storing data as bytes just return
310 if isinstance(obj, (bytes, str)):
311 return obj
312 # otherwise get it as a file object
313 # read file object from beginning
314 obj.seek(0)
315 # data is stored as a file object
316 data = obj.read()
317 obj.seek(0)
318 return data
320 def namespaced(self, namespace: str) -> "ZipResolver":
321 """
322 Return a "sub-resolver" with a root namespace.
324 Parameters
325 -------------
326 namespace : str
327 The root of the key to clip off, i.e. if
328 this resolver has key `a/b/c` you can get
329 'a/b/c' with resolver.namespaced('a/b').get('c')
331 Returns
332 -----------
333 resolver : Resolver
334 Namespaced resolver.
335 """
336 return ZipResolver(archive=self.archive, namespace=namespace)
338 def export(self) -> bytes:
339 """
340 Export the contents of the current archive as
341 a ZIP file.
343 Returns
344 ------------
345 compressed : bytes
346 Compressed data in ZIP format.
347 """
348 return util.compress(self.archive)
351class WebResolver(Resolver):
352 """
353 Resolve assets from a remote URL.
354 """
356 def __init__(
357 self,
358 url: str,
359 session: HttpSessionLike | None = None,
360 timeout: float = 30.0,
361 ):
362 """
363 Resolve assets from a base URL.
365 Parameters
366 --------------
367 url : str
368 Location where a mesh was stored or
369 directory where mesh was stored.
370 session : HttpSessionLike or None
371 Optional HTTP session used for fetches. Accepts
372 `httpx.Client` or `requests.Session`.
373 timeout : float
374 Per-request timeout in seconds.
375 """
376 if hasattr(url, "decode"):
377 url = url.decode("utf-8")
379 # parse string into namedtuple
380 parsed = urlparse(url)
381 # only http(s) is supported, reject `file://`, `gopher://`, etc.
382 if parsed.scheme not in ("http", "https"):
383 raise ValueError(f"scheme {parsed.scheme!r} not in ('http', 'https')")
385 if session is None:
386 # an explicit session will be required in a future release
387 import warnings
389 warnings.warn(
390 "`WebResolver` without a `session` is deprecated "
391 + "and will require one in a future release. "
392 + "pass an `httpx.Client` or `requests.Session`.",
393 category=DeprecationWarning,
394 stacklevel=2,
395 )
397 self.session = session
398 self.timeout = timeout
400 # per-library request kwargs: httpx and requests disagree on
401 # the redirect kwarg name — this is also where any future
402 # library-specific options should live
403 library = "httpx" if session is None else type(session).__module__.split(".")[0]
404 if library == "httpx":
405 # also the bare httpx module fallback when no session was passed
406 self.request_kwargs = {"follow_redirects": True, "timeout": timeout}
407 elif library == "requests":
408 self.request_kwargs = {"allow_redirects": True, "timeout": timeout}
409 elif library == "aiohttp":
410 # an aiohttp session can only be constructed inside a running
411 # event loop and its connector binds to that loop, so a
412 # synchronous fetch can never legally drive one
413 raise ValueError(
414 "`aiohttp` sessions are async-only and bound to their creation "
415 + "loop: pass an `httpx.Client` or `requests.Session` instead"
416 )
417 else:
418 # a duck-typed session gets `get(url)` with no assumed kwargs
419 self.request_kwargs = {}
421 # we want a base url
422 split = [i for i in parsed.path.split("/") if len(i) > 0]
424 # if the last item in the url path is a filename
425 # move up a "directory" for the base path
426 if len(split) == 0:
427 path = ""
428 elif "." in split[-1]:
429 # clip off last item
430 path = "/".join(split[:-1])
431 else:
432 # recombine into string ignoring any double slashes
433 path = "/".join(split)
435 # save the URL we were created with, i.e.
436 # `https://stuff.com/models/thing.glb`
437 self.url = url
438 # save the root url, i.e. `https://stuff.com/models`
439 self.base_url = (
440 "/".join(
441 i
442 for i in [parsed.scheme + ":/", parsed.netloc.strip("/"), path.strip("/")]
443 if len(i) > 0
444 )
445 + "/"
446 )
448 # our string handling should have never inserted double slashes
449 assert "//" not in self.base_url[len(parsed.scheme) + 3 :]
450 # we should always have ended with a single slash
451 assert self.base_url.endswith("/")
453 self.file_name = url.split("/")[-1]
455 def get(self, name: str) -> bytes:
456 """
457 Get a resource from the remote site.
459 Parameters
460 -------------
461 name : str
462 Asset name, i.e. 'quadknot.obj.mtl'
463 """
464 import httpx
466 # remove leading and trailing whitespace
467 name = name.strip()
469 # the caller's session or the bare httpx module, both expose `.get`
470 client = self.session or httpx
471 response = client.get(self.base_url + name, **self.request_kwargs)
473 if response.status_code >= 300:
474 # try to strip off filesystem crap
475 if name.startswith("./"):
476 name = name[2:]
477 response = client.get(self.base_url + name, **self.request_kwargs)
479 # now raise if we don't have
480 response.raise_for_status()
482 # return the bytes of the response
483 return response.content
485 def get_base(self) -> bytes:
486 """
487 Fetch the data at the full URL this resolver was
488 instantiated with, i.e. `https://stuff.com/hi.glb`
489 this will return the response.
491 Returns
492 --------
493 content
494 The value at `self.url`
495 """
496 import httpx
498 # just fetch the url we were created with
499 response = (self.session or httpx).get(self.url, **self.request_kwargs)
500 response.raise_for_status()
501 return response.content
503 def namespaced(self, namespace: str) -> "WebResolver":
504 """
505 Return a namespaced version of current resolver.
507 Parameters
508 -------------
509 namespace : str
510 URL fragment
512 Returns
513 -----------
514 resolver : WebResolver
515 With sub-url: `https://example.com/{namespace}`
516 """
517 # propagate session/timeout so the child keeps the same posture
518 return WebResolver(
519 url=self.base_url + namespace,
520 session=self.session,
521 timeout=self.timeout,
522 )
524 def write(self, key, value):
525 raise NotImplementedError("`WebResolver` is read-only!")
527 def keys(self):
528 raise NotImplementedError("`WebResolver` can't list keys")
531class GithubResolver(Resolver):
532 def __init__(
533 self,
534 repo: str,
535 branch: str | None = None,
536 commit: str | None = None,
537 save: str | None = None,
538 session: HttpSessionLike | None = None,
539 timeout: float = 30.0,
540 ):
541 """
542 Get files from a remote Github repository by
543 downloading a zip file with the entire branch
544 or a specific commit.
546 Parameters
547 -------------
548 repo
549 In the format of `owner/repo`.
550 branch
551 The remote branch you want to get files from.
552 commit
553 The full commit hash: pass either this OR branch.
554 save
555 A path if you want to save results locally.
556 session : HttpSessionLike or None
557 Optional HTTP session used for fetches. Accepts
558 `httpx.Client` or `requests.Session`.
559 timeout : float
560 Per-request timeout in seconds.
561 """
563 if commit is not None:
564 # just get the exact commit
565 self.url = f"https://github.com/{repo}/archive/{commit}.zip"
566 elif branch is not None:
567 # gets the latest commit on the specified branch.
568 self.url = f"https://github.com/{repo}/archive/refs/heads/{branch}.zip"
569 else:
570 raise ValueError("`commit` or `branch` must be passed!")
572 # reuse the session handling and deprecation warning
573 self.resolver = WebResolver(url=self.url, session=session, timeout=timeout)
575 if save is not None:
576 self.cache = caching.DiskCache(save)
577 else:
578 self.cache = None
580 def keys(self):
581 """
582 List the available files in the repository.
584 Returns
585 ----------
586 keys : iterable
587 Keys available to the resolved.
588 """
589 return self.zipped.keys()
591 def write(self, name, data):
592 raise NotImplementedError("`write` not implemented!")
594 @property
595 def zipped(self) -> ZipResolver:
596 """
597 - opened zip file
598 - locally saved zip file
599 - retrieve zip file and saved
600 """
602 if hasattr(self, "_zip"):
603 return self._zip
604 # download the archive or get from disc
605 raw = self.cache.get(self.url, self.resolver.get_base)
606 # create a zip resolver for the archive
607 # the root directory in the zip is the repo+commit so strip that off
608 # so the keys are usable, i.e. "models" instead of "trimesh-2232323/models"
609 self._zip = ZipResolver(
610 {
611 k.split("/", 1)[1]: v
612 for k, v in util.decompress(
613 util.wrap_as_stream(raw), file_type="zip"
614 ).items()
615 }
616 )
618 return self._zip
620 def get(self, key):
621 return self.zipped.get(key)
623 def namespaced(self, namespace):
624 """
625 Return a "sub-resolver" with a root namespace.
627 Parameters
628 -------------
629 namespace : str
630 The root of the key to clip off, i.e. if
631 this resolver has key `a/b/c` you can get
632 'a/b/c' with resolver.namespaced('a/b').get('c')
634 Returns
635 -----------
636 resolver : Resolver
637 Namespaced resolver.
638 """
639 return self.zipped.namespaced(namespace)
642def nearby_names(name, namespace=None):
643 """
644 Try to find nearby variants of a specified name.
646 Parameters
647 ------------
648 name : str
649 Initial name.
651 Yields
652 -----------
653 nearby : str
654 Name that is a lightly permutated version
655 of the initial name.
656 """
658 # the various operations that *might* result in a correct key
659 def trim(prefix, item):
660 if item.startswith(prefix):
661 return item[len(prefix) :]
662 return item
664 cleaners = [
665 lambda x: x,
666 lambda x: x.strip(),
667 lambda x: trim("./", x),
668 lambda x: trim(".\\", x),
669 lambda x: trim("\\", x),
670 lambda x: os.path.split(x)[-1],
671 lambda x: x.replace("%20", " "),
672 ]
674 if namespace is None:
675 namespace = ""
677 # make sure we don't return repeat values
678 hit = set()
679 for f in cleaners:
680 # try just one cleaning function
681 current = f(name)
682 if current in hit:
683 continue
684 hit.add(current)
685 yield namespace + current
687 for a, b in itertools.combinations(cleaners, 2):
688 # apply both clean functions
689 current = a(b(name))
690 if current in hit:
691 continue
692 hit.add(current)
693 yield namespace + current
695 # try applying in reverse order
696 current = b(a(name))
697 if current in hit:
698 continue
699 hit.add(current)
700 yield namespace + current
702 if ".." in name and namespace is not None:
703 # if someone specified relative paths give it one attempt
704 strip = namespace.strip("/").split("/")[: -name.count("..")]
705 strip.extend(name.split("..")[-1].strip("/").split("/"))
706 yield "/".join(strip)
709# most loaders can use a mapping in addition to a resolver
710ResolverLike: TypeAlias = Resolver | Mapping