arozantsev commited on
Commit
62765f3
·
verified ·
1 Parent(s): 8bc8124

Deploy simready-assets run 20261020-a3cb7f2-captions (GPU variant) (part 2)

Browse files
Files changed (3) hide show
  1. pipeline/config.py +5 -0
  2. pipeline/export.py +27 -0
  3. pipeline/profiles.py +59 -10
pipeline/config.py CHANGED
@@ -419,6 +419,11 @@ class Config:
419
  # a text query's judge reads each hit's path (below its dataset or Isaac content root) and vision
420
  # caption, and an image + words query asks whether the candidate matches both (usdsearch !279).
421
  "USDSEARCH_VISION_VALIDATION_IMAGE_MAX_PX": "448",
 
 
 
 
 
422
  # Each stored view is fitted and encoded for its check once per container, not once per check (most
423
  # of a check's API CPU): docker/hf-space/check_image_cache.py, GPU image only (the CPU image has no
424
  # hook: the variable does nothing there). Off: delete the Space Variable and restart; a deploy sets it
 
419
  # a text query's judge reads each hit's path (below its dataset or Isaac content root) and vision
420
  # caption, and an image + words query asks whether the candidate matches both (usdsearch !279).
421
  "USDSEARCH_VISION_VALIDATION_IMAGE_MAX_PX": "448",
422
+ # Views and query images go to the engine as JPEG q90 (transparency on white), not PNG: less API CPU per
423
+ # check. Needs wheels with usdsearch !303 (older ones ignore it). Part of the verdict cache identity and of
424
+ # check_image_cache.py's folder (it keys on the whole validation config), so the warm-up prepares JPEGs.
425
+ # Off: delete the Space Variable and restart; a deploy sets it again while this line is here.
426
+ "USDSEARCH_VISION_VALIDATION_IMAGE_FORMAT": "jpeg",
427
  # Each stored view is fitted and encoded for its check once per container, not once per check (most
428
  # of a check's API CPU): docker/hf-space/check_image_cache.py, GPU image only (the CPU image has no
429
  # hook: the variable does nothing there). Off: delete the Space Variable and restart; a deploy sets it
pipeline/export.py CHANGED
@@ -275,8 +275,35 @@ def export_neo4j(cfg: Config) -> Dict:
275
  }
276
 
277
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
278
  def run(cfg: Config) -> int:
279
  manifest = load_manifest()
 
 
 
 
280
  ARTIFACTS.mkdir(parents=True, exist_ok=True)
281
 
282
  print("Exporting OpenSearch ...")
 
275
  }
276
 
277
 
278
+ def package_sizes_problem(manifest: Dict, path: Path) -> str:
279
+ """Why `path` is not this run's package_sizes.json ("" when it is): its (repo, revision) groups must be exactly
280
+ the manifest's hf_repos. On 2026-10-20 a test overwrote .build/package_sizes.json with its fixture
281
+ (nvidia/repo-a@main) and export shipped it with run 20261020-a3cb7f2-captions; this is the check that run
282
+ lacked. Only a storage_api run has one."""
283
+ if manifest.get("storage_backend") != "storage_api":
284
+ return ""
285
+ if not path.exists():
286
+ return f"{path.name} is missing: re-run `enumerate`"
287
+ try:
288
+ groups = json.loads(path.read_text()).get("groups") or []
289
+ except (OSError, ValueError, AttributeError) as e:
290
+ return f"{path.name} is unreadable ({e}): re-run `enumerate`"
291
+ have = {(g.get("repo_id"), g.get("revision")) for g in groups if isinstance(g, dict)}
292
+ want = {(repo_id, revision) for _host, _type, repo_id, revision in manifest.get("hf_repos") or []}
293
+ if have != want:
294
+ extra, missing = sorted(have - want), sorted(want - have)
295
+ return (f"{path.name} is not this run's: {len(extra)} (repo, revision) group(s) the run doesn't list "
296
+ f"{extra[:3]}, {len(missing)} of its own missing {missing[:3]}. Re-run `enumerate` with this "
297
+ f"run's RUN_ID, or restore the file from the run's S3 prefix.")
298
+ return ""
299
+
300
+
301
  def run(cfg: Config) -> int:
302
  manifest = load_manifest()
303
+ problem = package_sizes_problem(manifest, BUILD_DIR / "package_sizes.json")
304
+ if problem:
305
+ print(f"export: {problem}")
306
+ return 1
307
  ARTIFACTS.mkdir(parents=True, exist_ok=True)
308
 
309
  print("Exporting OpenSearch ...")
pipeline/profiles.py CHANGED
@@ -18,6 +18,12 @@ writes it; a top-level `profile` / `profile_version` (the shape simready-central
18
  that still fails after PROFILES_TRIES attempts fails the stage; nothing is written then. The record goes to
19
  .build/profiles.json.
20
 
 
 
 
 
 
 
21
  python -m pipeline.run_pipeline profiles --env-file ... [--dry-run]
22
  """
23
 
@@ -25,9 +31,11 @@ from __future__ import annotations
25
 
26
  import json
27
  import os
 
28
  import time
29
  from collections import Counter
30
  from dataclasses import dataclass
 
31
  from typing import Dict, Iterable, List, Optional, Tuple
32
  from urllib.parse import quote
33
 
@@ -38,6 +46,8 @@ from .config import BUILD_DIR, Config
38
  PROPERTY = "simready_metadata_validation__profile"
39
  VERSION_PROPERTY = "simready_metadata_validation__profile_version"
40
  REPORT = BUILD_DIR / "profiles.json"
 
 
41
 
42
 
43
  @dataclass(frozen=True)
@@ -52,6 +62,10 @@ class PackageRef:
52
  def owner(self) -> str:
53
  return self.repo_id.split("/")[0]
54
 
 
 
 
 
55
  def sidecar_url(self) -> str:
56
  rel = f"{self.package}/.metadata/{self.package}_validation.json"
57
  return f"https://{self.host}/datasets/{self.repo_id}/resolve/{self.revision}/{quote(rel)}"
@@ -157,6 +171,24 @@ def fetch_sidecar(http, url: str, tries: int, backoff_s: float) -> Tuple[str, ob
157
  raise SidecarError(f"{url}: {last}")
158
 
159
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
160
  def _headers() -> Dict[str, str]:
161
  # Public datasets read anonymously; $HF_TOKEN (environment only, as everywhere in this pipeline) when set.
162
  token = os.environ.get("HF_TOKEN")
@@ -182,6 +214,11 @@ def run(cfg: Config, dry_run: bool = False) -> int:
182
  updates: List[Tuple[str, List[Dict]]] = []
183
  expected: Dict[str, Tuple[List[str], List[str]]] = {}
184
  read = 0
 
 
 
 
 
185
  with httpx.Client(timeout=60.0, follow_redirects=True) as index_http, \
186
  httpx.Client(timeout=60.0, follow_redirects=True, headers=_headers()) as hf_http:
187
  for hit in _scroll(index_http, base, index):
@@ -196,14 +233,23 @@ def run(cfg: Config, dry_run: bool = False) -> int:
196
  usd = prop_values(props, PROPERTY)
197
  new_props = props
198
  if needs_sidecar(ref, props, vendors):
199
- if read:
200
- time.sleep(interval)
201
- read += 1
202
- try:
203
- state, doc = fetch_sidecar(hf_http, ref.sidecar_url(), tries, backoff_s=2.0)
204
- except SidecarError as e:
205
- errors.append(str(e))
206
- continue
 
 
 
 
 
 
 
 
 
207
  if state == "missing":
208
  outcomes["no-sidecar"] += 1
209
  else:
@@ -225,8 +271,10 @@ def run(cfg: Config, dry_run: bool = False) -> int:
225
  updates.append((hit["_id"], new_props))
226
  expected[hit["_id"]] = (final, prop_values(new_props, VERSION_PROPERTY))
227
 
228
- print(f"{sum(sources.values())} assets in {index}; {read} sidecar(s) read; outcomes: {dict(outcomes)}; "
229
- f"{len(updates)} doc(s) to update")
 
 
230
  print(f"{PROPERTY}: {json.dumps(dict(values), sort_keys=True)}")
231
  for d in disagreements:
232
  print(f"sidecar overrides the USD: {d['url']}: {d['usd']} -> {d['sidecar']}")
@@ -235,6 +283,7 @@ def run(cfg: Config, dry_run: bool = False) -> int:
235
  REPORT.parent.mkdir(parents=True, exist_ok=True)
236
  REPORT.write_text(json.dumps({
237
  "property": PROPERTY, "index": index, "assets": sum(sources.values()), "sidecars_read": read,
 
238
  "outcomes": dict(outcomes), "values": dict(sorted(values.items())), "disagreements": disagreements,
239
  "without": without, "errors": errors, "updated": len(updates), "dry_run": dry_run,
240
  }, indent=1) + "\n")
 
18
  that still fails after PROFILES_TRIES attempts fails the stage; nothing is written then. The record goes to
19
  .build/profiles.json.
20
 
21
+ A sidecar at a pinned revision (a 40-hex commit) never changes, so every one read there is kept in
22
+ .build/profiles-sidecars.json (the 404s too) and later runs read only packages pinned since: a run that re-pins
23
+ nothing reads none, where reading all ~1,060 partner sidecars at PROFILES_READ_INTERVAL_S took ~9 minutes. A sidecar
24
+ at a branch (`main`) is read every time. Within a run each sidecar is read once, however many of its package's
25
+ assets the index holds. PROFILES_CACHE=off reads everything again (and still refreshes the file).
26
+
27
  python -m pipeline.run_pipeline profiles --env-file ... [--dry-run]
28
  """
29
 
 
31
 
32
  import json
33
  import os
34
+ import re
35
  import time
36
  from collections import Counter
37
  from dataclasses import dataclass
38
+ from pathlib import Path
39
  from typing import Dict, Iterable, List, Optional, Tuple
40
  from urllib.parse import quote
41
 
 
46
  PROPERTY = "simready_metadata_validation__profile"
47
  VERSION_PROPERTY = "simready_metadata_validation__profile_version"
48
  REPORT = BUILD_DIR / "profiles.json"
49
+ CACHE = BUILD_DIR / "profiles-sidecars.json"
50
+ PINNED_REVISION = re.compile(r"^[0-9a-f]{40}$")
51
 
52
 
53
  @dataclass(frozen=True)
 
62
  def owner(self) -> str:
63
  return self.repo_id.split("/")[0]
64
 
65
+ @property
66
+ def pinned(self) -> bool:
67
+ return bool(PINNED_REVISION.match(self.revision))
68
+
69
  def sidecar_url(self) -> str:
70
  rel = f"{self.package}/.metadata/{self.package}_validation.json"
71
  return f"https://{self.host}/datasets/{self.repo_id}/resolve/{self.revision}/{quote(rel)}"
 
171
  raise SidecarError(f"{url}: {last}")
172
 
173
 
174
+ def load_cache(path: Path) -> Dict[str, Tuple[str, object]]:
175
+ """sidecar URL -> (state, parsed sidecar) as fetch_sidecar returned it; {} when there is no usable file."""
176
+ try:
177
+ raw = json.loads(path.read_text())
178
+ except (OSError, ValueError):
179
+ return {}
180
+ entries = raw.get("sidecars") if isinstance(raw, dict) else None
181
+ return {url: (v[0], v[1]) for url, v in (entries or {}).items()
182
+ if isinstance(v, list) and len(v) == 2 and v[0] in ("ok", "missing", "malformed")}
183
+
184
+
185
+ def save_cache(path: Path, cache: Dict[str, Tuple[str, object]]) -> None:
186
+ path.parent.mkdir(parents=True, exist_ok=True)
187
+ tmp = path.with_suffix(".tmp")
188
+ tmp.write_text(json.dumps({"sidecars": {url: list(v) for url, v in sorted(cache.items())}}) + "\n")
189
+ tmp.replace(path)
190
+
191
+
192
  def _headers() -> Dict[str, str]:
193
  # Public datasets read anonymously; $HF_TOKEN (environment only, as everywhere in this pipeline) when set.
194
  token = os.environ.get("HF_TOKEN")
 
214
  updates: List[Tuple[str, List[Dict]]] = []
215
  expected: Dict[str, Tuple[List[str], List[str]]] = {}
216
  read = 0
217
+ reused = 0
218
+ use_cache = os.environ.get("PROFILES_CACHE", "on").strip().lower() not in ("off", "0", "false", "no")
219
+ cache = load_cache(CACHE) if use_cache else {}
220
+ pinned = len(cache)
221
+ seen: Dict[str, Tuple[str, object]] = {} # this run's reads, branch revisions included
222
  with httpx.Client(timeout=60.0, follow_redirects=True) as index_http, \
223
  httpx.Client(timeout=60.0, follow_redirects=True, headers=_headers()) as hf_http:
224
  for hit in _scroll(index_http, base, index):
 
233
  usd = prop_values(props, PROPERTY)
234
  new_props = props
235
  if needs_sidecar(ref, props, vendors):
236
+ sidecar_url = ref.sidecar_url()
237
+ known = seen.get(sidecar_url) or cache.get(sidecar_url)
238
+ if known:
239
+ reused += 1
240
+ state, doc = known
241
+ else:
242
+ if read:
243
+ time.sleep(interval)
244
+ read += 1
245
+ try:
246
+ state, doc = fetch_sidecar(hf_http, sidecar_url, tries, backoff_s=2.0)
247
+ except SidecarError as e:
248
+ errors.append(str(e))
249
+ continue
250
+ seen[sidecar_url] = (state, doc)
251
+ if ref.pinned:
252
+ cache[sidecar_url] = (state, doc)
253
  if state == "missing":
254
  outcomes["no-sidecar"] += 1
255
  else:
 
271
  updates.append((hit["_id"], new_props))
272
  expected[hit["_id"]] = (final, prop_values(new_props, VERSION_PROPERTY))
273
 
274
+ if len(cache) > pinned:
275
+ save_cache(CACHE, cache) # before any error below: a retry reads only what failed
276
+ print(f"{sum(sources.values())} assets in {index}; {read} sidecar(s) read, {reused} already known "
277
+ f"({CACHE.name}: {len(cache)} pinned); outcomes: {dict(outcomes)}; {len(updates)} doc(s) to update")
278
  print(f"{PROPERTY}: {json.dumps(dict(values), sort_keys=True)}")
279
  for d in disagreements:
280
  print(f"sidecar overrides the USD: {d['url']}: {d['usd']} -> {d['sidecar']}")
 
283
  REPORT.parent.mkdir(parents=True, exist_ok=True)
284
  REPORT.write_text(json.dumps({
285
  "property": PROPERTY, "index": index, "assets": sum(sources.values()), "sidecars_read": read,
286
+ "sidecars_reused": reused,
287
  "outcomes": dict(outcomes), "values": dict(sorted(values.items())), "disagreements": disagreements,
288
  "without": without, "errors": errors, "updated": len(updates), "dry_run": dry_run,
289
  }, indent=1) + "\n")