(url:str, name:pathlib.Path|str|None=None, subdir:str|None=None, gunzip:bool=False, allow_caching=not getenv("DISABLE_HTTP_CACHE"),
headers:dict[str, str]={}, sha256:str|None=None)
| 448 | return pathlib.Path(cache_dir) / "downloads" |
| 449 | |
| 450 | def fetch(url:str, name:pathlib.Path|str|None=None, subdir:str|None=None, gunzip:bool=False, allow_caching=not getenv("DISABLE_HTTP_CACHE"), |
| 451 | headers:dict[str, str]={}, sha256:str|None=None) -> pathlib.Path: |
| 452 | import urllib.request |
| 453 | if url.startswith(("/", ".")): return pathlib.Path(url) |
| 454 | if name is not None and (isinstance(name, pathlib.Path) or '/' in name): fp = pathlib.Path(name) |
| 455 | else: |
| 456 | hh = "_"+hashlib.md5(("\n".join(f"{k.strip()}:{v.strip()}" for k,v in sorted(headers.items()))).encode("utf-8")).hexdigest() if headers else "" |
| 457 | fp = _ensure_downloads_dir() / (subdir or "") / ((name or hashlib.md5(url.encode('utf-8')).hexdigest()) + hh + (".gunzip" if gunzip else "")) |
| 458 | if not fp.is_file() or not allow_caching or (sha256 and hashlib.sha256(fp.read_bytes()).hexdigest() != sha256): |
| 459 | (_dir := fp.parent).mkdir(parents=True, exist_ok=True) |
| 460 | with urllib.request.urlopen(urllib.request.Request(url, headers={"User-Agent": "tinygrad 0.13.0", **headers}), timeout=10) as r: |
| 461 | assert r.status in {200, 206}, r.status |
| 462 | length = int(r.headers.get('content-length', 0)) if not gunzip else None |
| 463 | readfile = gzip.GzipFile(fileobj=r) if gunzip else r |
| 464 | progress_bar:tqdm = tqdm(total=length, unit='B', unit_scale=True, desc=f"{url}") |
| 465 | h = hashlib.sha256() if sha256 else None |
| 466 | with tempfile.NamedTemporaryFile(dir=_dir, delete=False) as f: |
| 467 | while chunk := readfile.read(16384): |
| 468 | if h: h.update(chunk) |
| 469 | progress_bar.update(f.write(chunk)) |
| 470 | f.close() |
| 471 | if h and (actual_sha256:=h.hexdigest()) != sha256: raise RuntimeError(f"fetch sha mismatch, expected {sha256} but got {actual_sha256}") |
| 472 | pathlib.Path(f.name).rename(fp) |
| 473 | progress_bar.update(close=True) |
| 474 | if length and (file_size:=os.stat(fp).st_size) < length: raise RuntimeError(f"fetch size incomplete, {file_size} < {length}") |
| 475 | return fp |
| 476 | |
| 477 | def fetch_fw(path:str, name:str, sha256:str) -> bytes: |
| 478 | if sys.version_info >= (3,14) and (p:=pathlib.Path(f"/lib/firmware/{path}/{name}.zst")).is_file(): |
searching dependent graphs…