diff --git a/CREDITS.md b/CREDITS.md index 57a077526b..b1928df105 100644 --- a/CREDITS.md +++ b/CREDITS.md @@ -2,7 +2,7 @@ ## Jetlink external inference -The comma-side Jetlink client and USB transport in `jetlink_repo` are by Zeph Leggett and the [Jetlink contributors](https://github.com/zoompilot/jetlink). They are vendored from revision `a863c3801da6fd925d14eb1f5b1e250f8efcc649` under the MIT license, retained in [jetlink_repo/LICENSE](jetlink_repo/LICENSE). StarPilot's adapter (`openpilot/starpilot/models/jetlink_adapter.py`) builds on Jetlink's openpilot layer and owns its param names, Always On Lateral rule, model runner reset, settings and alerts. Companion applications and inference servers remain distributed by Jetlink. +The comma-side Jetlink client and USB transport in `jetlink_repo` are by Zeph Leggett and the [Jetlink contributors](https://github.com/zoompilot/jetlink). They are vendored from revision `9cfde98eec1d7cc714d800504944cf1a3944513d` under the MIT license, retained in [jetlink_repo/LICENSE](jetlink_repo/LICENSE). StarPilot's adapter (`openpilot/starpilot/models/jetlink_adapter.py`) builds on Jetlink's openpilot layer and owns its param names, Always On Lateral rule, model runner reset, settings and alerts. Companion applications and inference servers remain distributed by Jetlink. ## MRR35 Hyundai angle and radar diff --git a/jetlink_repo/jetlink/__init__.py b/jetlink_repo/jetlink/__init__.py index 5458975a61..757eccbc2c 100644 --- a/jetlink_repo/jetlink/__init__.py +++ b/jetlink_repo/jetlink/__init__.py @@ -10,4 +10,4 @@ jetlink: run openpilot's large driving models on an attached Jetson. # so does the release check (macos/scripts/check-version.sh). The comma runs # jetlink from a checkout, not an install, so this is also all it has. The Swift # reads it from Pinned.swift: run JetlinkKit/Scripts/make_pins.py after a bump. -__version__ = '0.8.12' +__version__ = '0.9.0' diff --git a/jetlink_repo/jetlink/client.py b/jetlink_repo/jetlink/client.py index 22fe81ea62..35ea3f9363 100644 --- a/jetlink_repo/jetlink/client.py +++ b/jetlink_repo/jetlink/client.py @@ -228,8 +228,9 @@ class JetlinkClient: """Introduce this client. The server starts its session over on a hello, so this is also how a new owner of the gadget takes over one the server never saw end; see Session.greet in JetlinkServer. `pick` is the comma's - own big-model pick and its age, {ref, sha256, nbytes, name, age_s}; the - answer's `pick` is the newer of it and the server's (PickStore).""" + own big-model pick with its age and its date by the comma's clock, {ref, + sha256, nbytes, name, age_s, at}; the answer's `pick` is the newer of it + and the server's (PickStore).""" seq = self._next_seq() client = {'nonce': self.nonce, 'name': self.name} if pick is not None: diff --git a/jetlink_repo/jetlink/openpilot/__init__.py b/jetlink_repo/jetlink/openpilot/__init__.py index a99db2f8ce..f50a011277 100644 --- a/jetlink_repo/jetlink/openpilot/__init__.py +++ b/jetlink_repo/jetlink/openpilot/__init__.py @@ -40,7 +40,7 @@ from jetlink.openpilot.interface import MODES, STATES, Keys, ModelFace, Openpilo from jetlink.openpilot.parts import Parts, for_this_process from jetlink.openpilot.status import Status -API = 3 +API = 4 __all__ = ['API', 'MODES', 'STATES', 'Jetlink', 'Keys', 'ModelFace', 'Openpilot', 'OwnerConfig', 'Status', 'bind', 'conformance', 'SHUTDOWN_TIMEOUT', 'Flags', 'SwapEvents'] @@ -302,13 +302,34 @@ class Jetlink: self._shutdown_asked = time.monotonic() return True - def should_extend_catalog(self) -> bool: - """Should the model manager's big-model catalog carry the newer catalogs' - models? With no chestnut fitted. Hardware, not the link setting: the model - manager drops a pick its catalog does not list.""" + def lists_big_models(self) -> bool: + """Is the model manager's big-model list jetlink's (big_catalog) rather + than sunnypilot's? With no chestnut fitted. Hardware, not the link + setting: the model manager drops a pick its catalog does not list.""" return not self._parts.chestnut_fitted() + def big_catalog(self, selector: int) -> dict | None: + """The model index as the model manager's big-model catalog, every bundle + at `selector` (its REQUIRED_JSON_VERSION), fetched again + (Models.big_catalog). None when it cannot be had, offline say, and the + model manager keeps the list it has.""" + from jetlink.registry.index import RegistryError + try: + return self._parts.models.big_catalog(selector) + except RegistryError as e: + self._log.warning("jetlink: could not fetch the model index: %s", e) + return None + + def listed_ref(self, ref: str) -> str | None: + """The ref big_catalog() lists a pick's model under, for a pick a new list + no longer names: a recompile is listed once, under its newest ref. None + while the list is not jetlink's (lists_big_models).""" + return self._parts.models.listed_ref(ref) if self.lists_big_models() else None + + # API 3's, until the next minor release: the catalog is left as fetched. + # Its picks still drive, since the index knows every ref it ever saw + def should_extend_catalog(self) -> bool: + return False + def extend_catalog(self, catalog: dict) -> dict: - """The big-model catalog the model manager fetched, with the models newer - catalogs list folded in. Never raises.""" - return self._parts.models.big_catalog(catalog) + return catalog diff --git a/jetlink_repo/jetlink/openpilot/hooks.py b/jetlink_repo/jetlink/openpilot/hooks.py index 095f3a6922..7ffb5d22c6 100644 --- a/jetlink_repo/jetlink/openpilot/hooks.py +++ b/jetlink_repo/jetlink/openpilot/hooks.py @@ -11,9 +11,10 @@ manager, hardwared, the UI and the model manager call these on every device, link on or off, and modeld on every drive: whatever jetlink does wrong turns the link off and is logged, and never takes one of them down. Each hook has the answer it gets with the link off, and gives it on any failure. They are -not all False: attach() keeps modeld's model, extend_catalog() the catalog, -and shutdown_step() says yes to a power-off that is wanted, since a broken -jetlink must never keep a comma on to drain the car's battery. selfdrived's +not all False: attach() keeps modeld's model, big_catalog() is None so the +model manager keeps its own list, and shutdown_step() says yes to a power-off +that is wanted, since a broken jetlink must never keep a comma on to drain the +car's battery. selfdrived's swap events (swap_events()) answer no event at all. The fork's guard, which builds Hooks or else Off inside a function, and its @@ -39,7 +40,7 @@ from jetlink.openpilot import API, Flags, Jetlink, SwapEvents, bind # the APIs a fork's adapter may be written to: today's, and after a bump the # previous one too, for one minor release (jetlink/openpilot/__init__.py) -COMPATIBLE = frozenset({API}) +COMPATIBLE = frozenset({API, 3}) # the hooks that ask jetlink with link_on() saying off. status(): the UI shows # the big model's name from the model library. shutdown_step(): hardwared @@ -47,7 +48,7 @@ COMPATIBLE = frozenset({API}) # and its wait, for a link turned off while one is pending. The model # manager's: the catalog is hardware and the library, not the setting BINDS_WITH_LINK_OFF = frozenset({'status', 'shutdown_step', 'request_shutdown', 'shutdown_pending', 'model_state', - 'should_extend_catalog', 'extend_catalog'}) + 'lists_big_models', 'big_catalog', 'listed_ref', 'should_extend_catalog', 'extend_catalog'}) # what selfdrived is told when the swap events cannot run: none at all NO_EVENTS = Flags(False, False, False, False, False) @@ -156,6 +157,17 @@ class _Hooks: def model_state(self, ref: str) -> str | None: return self._call('model_state', None, ref) + def lists_big_models(self) -> bool: + return self._call('lists_big_models', False) + + def big_catalog(self, selector: int) -> dict | None: + """None with the link off or on any failure: the model manager keeps its list.""" + return self._call('big_catalog', None, selector) + + def listed_ref(self, ref: str) -> str | None: + return self._call('listed_ref', None, ref) + + # API 3's, until the next minor release def should_extend_catalog(self) -> bool: return self._call('should_extend_catalog', False) diff --git a/jetlink_repo/jetlink/openpilot/interface.py b/jetlink_repo/jetlink/openpilot/interface.py index f0fa5e135a..a2d5c2b7cd 100644 --- a/jetlink_repo/jetlink/openpilot/interface.py +++ b/jetlink_repo/jetlink/openpilot/interface.py @@ -45,7 +45,6 @@ class Keys: offroad: str # BOOL: is the car parked (manager's IsOffroad) progress: str # JSON {stage, frac, msg, drops}: provisioning and join progress spec: str # JSON: the built model's spec and whether its engine is built - pointers: str # JSON: catalog ref, or a hash-named pick's sha256 -> {oid, size} # sunnypilot's model manager: the big-model pick, JSON {ref, displayName}, # and the big-model catalog, JSON {bundles}. Without one both are None, and # jetlink runs its own default big model, or the one picked on the server. @@ -62,6 +61,8 @@ class Keys: # the server is written there, so the comma's own list shows it; without # the key it drives all the same select: str | None = None + # API 3's kept pointers, unread since the model index: until the next minor release + pointers: str | None = None @dataclass(frozen=True) @@ -97,7 +98,6 @@ class StatusSide(Protocol): """Every reader: manager, the UI, hardwared, the model manager.""" keys: Keys log: Any # logging.Logger-like (cloudlog): debug, info, warning, error, exception - catalog_selector: int # the model manager's REQUIRED_JSON_VERSION; 0 without a model manager def params_dir(self) -> Path: """The params store's directory, by params.cc's rule, worked out on every diff --git a/jetlink_repo/jetlink/openpilot/joining.py b/jetlink_repo/jetlink/openpilot/joining.py index 9c54254709..462275f5b5 100644 --- a/jetlink_repo/jetlink/openpilot/joining.py +++ b/jetlink_repo/jetlink/openpilot/joining.py @@ -116,7 +116,7 @@ LAGS = (BEHIND, BAD_OUTPUT) RECONNECTING = {LOST: ('reconnecting', 'lost'), BEHIND: ('fell behind, reconnecting', 'behind'), NOT_UP: ('load failed, reconnecting', 'load_failed'), BAD_OUTPUT: ('bad output, reconnecting', 'bad_output')} -# how long a model name the catalog had not resolved stays the sha's +# how long a model name the comma's index did not have stays the sha's NAME_RETRY = 30.0 @@ -694,8 +694,9 @@ class JoiningModelState: 'lags': self._lags} def _name_of(self, sha: str, names: dict) -> str: - """The catalog's name, kept once it is one. The sha's stand-in (a row not - resolved yet, or a lookup that failed) is asked again after NAME_RETRY.""" + """The index's name, kept once it is one. The sha's stand-in (a model the + comma's index does not list yet, or a lookup that failed) is asked again + after NAME_RETRY.""" name, retry_at = names.get(sha, (None, 0.0)) if time.monotonic() >= retry_at: try: diff --git a/jetlink_repo/jetlink/openpilot/link.py b/jetlink_repo/jetlink/openpilot/link.py index 7b3e6cbe51..8b1b0c193b 100644 --- a/jetlink_repo/jetlink/openpilot/link.py +++ b/jetlink_repo/jetlink/openpilot/link.py @@ -225,18 +225,16 @@ def connect_patiently(link: Link): # -- making the Jetson ready ------------------------------------------------------ def identity(parts, entry: dict) -> tuple[str, int]: - """The picked model's sha256 and byte count, from the catalog model's LFS - pointer, so the comma can name the model without holding or hashing the - ONNX. Looked up once per model ever and kept in a param: the one part of - provisioning that needs the internet. A model picked by its hash alone is - only sized.""" + """The picked model's sha256 and byte count, from the model index, so the + comma can name the model without holding or hashing the ONNX. A row has + them already where the comma's copy of the index knows the model; one it + does not fetches the index again, the one part of provisioning that may + need the internet.""" sha256, nbytes = entry.get('oid'), entry.get('size') if sha256 and nbytes: return sha256, int(nbytes) parts.progress.report('connect', 0.0, 'finding model', code='finding_model') - if entry.get('ref'): - return parts.models.resolve_pointer(entry['ref']) - return parts.models.resolve_size(sha256) + return parts.models.resolve(entry.get('ref'), sha256) # What has already been hashed in this process, keyed on the file as it was @@ -326,8 +324,7 @@ def stand_in(parts, hello: dict, pick: str, built) -> tuple[str, int] | None: if pick in have: return None if loaded: - # its size names it to the server, and only a catalog model resolved - # here, or the record, has one + # its size names it to the server: the record's, else the index's nbytes = built.nbytes if built is not None and built.sha256 == loaded else parts.models.size_for(loaded) if nbytes: return loaded, nbytes @@ -357,7 +354,8 @@ def open_link(parts, link: Link, should_stop=None): # booting; both resolve on their own client = connect_patiently(link) try: - hello = parts.models.heard(client.hello(timeout=10.0, pick=parts.models.hello_pick())) + sent = parts.models.hello_pick() + hello = parts.models.heard(client.hello(timeout=10.0, pick=sent), sent) parts.log.warning("jetlink: %s trt %s, engine %s, loaded %s", hello.get('device'), hello.get('trt_version'), hello.get('engine_state'), str(hello.get('loaded'))[:16]) diff --git a/jetlink_repo/jetlink/openpilot/models.py b/jetlink_repo/jetlink/openpilot/models.py index a95b4adace..82e8210d01 100644 --- a/jetlink_repo/jetlink/openpilot/models.py +++ b/jetlink_repo/jetlink/openpilot/models.py @@ -7,42 +7,48 @@ See the LICENSE file in the root directory for more details. Which large model runs, and where its bytes are. The large model is the model manager's big-model pick, the same slot a -chestnut runs from (Keys.big_model). Its bundles are tinygrad pkls for a GPU -the Jetson does not have, but each one names the comma commit it was compiled -from, and that commit's ONNX in comma's LFS is what the Jetson runs. The -catalog says what exists, the slot says which one, the pointer at the commit -says which bytes. A commit that ships a precompiled pkl instead names its -export, and jetlink's registry follows that to comma's model repo; see -jetlink.registry.lfs. +chestnut runs from (Keys.big_model). It names a comma commit, and what the +Jetson runs is the ONNX that commit was built from: the model index +(jetlink.registry.index) says which, for every big-model commit, and lists +what can be picked. The comma keeps the last index that parsed under +model_dir(), so it names, sizes and lists its models with no network; a pick +the copy does not know fetches it again (resolve). + +Without a chestnut fitted the model manager's big-model list is the index's +(big_catalog), so what the comma lists is what the Jetson runs. With one, the +list is sunnypilot's own, and jetlink stays out of it. Without a model manager (Keys.big_model and Keys.catalog None) there is no pick and no catalog, and the large model is the one picked on the server, else -jetlink's own default, named by its ref: its pointer resolves from the ref as -any pick's does. +jetlink's own default, named by its ref. A fork with a picker of its own but no catalog jetlink reads (Keys.big_model without Keys.catalog, StarPilot's) may name its pick by the ONNX's sha256 -instead of a ref, which is all its catalog knows of a model. Its size is looked -up from comma's LFS servers once (resolve_size). With no catalog no model -manager owns the slot either, so jetlink writes a newer pick from the server -into it itself, and the fork's picker shows what drives. +instead of a ref, which is all its catalog knows of a model; the index sizes +it. With no catalog no model manager owns the slot either, so jetlink writes a +newer pick from the server into it itself, and the fork's picker shows what +drives. """ from __future__ import annotations +import json import os import re import time from pathlib import Path -from jetlink.registry.catalog import DEFAULT_BIG_MODEL_NAME, DEFAULT_BIG_MODEL_REF, is_ref, is_sha256 +from jetlink.registry.index import (DEFAULT_BIG_MODEL_NAME, DEFAULT_BIG_MODEL_REF, ModelIndex, NetworkError, Pointer, RegistryError, + is_ref, is_sha256, parse_index) -POINTER_TIMEOUT = 10.0 -_REF = re.compile(r'[0-9a-f]{40}') # the build date a catalog name ends in, " (September 17, 2026)"; any other parenthesis stays _TRAILING_DATE = re.compile(r' \([A-Za-z]+ \d{1,2}, \d{4}\)$') -# the index and the slot are JSON params, and the UI names the active model -# every frame; the status line can lag a new pick by this long +# the index and the slot are a file and a JSON param, and the UI names the +# active model every frame; the status line can lag a new pick by this long INDEX_TTL = 2.0 +# the comma's copy of the model index, under model_dir() +INDEX_FILE = 'index.json' +# a pick the comma's copy does not know fetches the index again, at most this often +INDEX_MISS_RETRY = 60.0 # A big model's download over a car's connection: the waits between the # attempts of one provisioning run, one more attempt than waits. Each carries # on from the .part the last left, so a stall costs the wait, not the bytes @@ -50,6 +56,11 @@ INDEX_TTL = 2.0 DOWNLOAD_RETRY_DELAYS = (5.0, 15.0, 30.0, 60.0, 120.0) # server_pick() before this process has heard a hello _UNHEARD = object() +# heard() told nothing of the hello's own pick: an older caller +_UNTOLD = object() +# the slot as jetlink saw it last, under model_dir(): to tell a slot the user +# emptied (the default, chosen) from one never filled +SLOT_STATE = 'slot.json' def _dated(pick) -> dict | None: @@ -68,91 +79,99 @@ def _dated(pick) -> dict | None: class Models: def __init__(self, op): self.op = op - self._index_cache: tuple[float, list[dict], dict[str, dict]] | None = None + # (looked at, the file's (mtime, size), the index it held and its rows) + self._index_cache: tuple[float, tuple | None, ModelIndex | None, list[dict]] | None = None + self._refresh_after = 0.0 self._slot_cache: tuple[float, dict | None, float | None] | None = None + # SLOT_STATE as this process last read or wrote it + self._slot_state: dict | None = None # the pick of the last hello this process heard, and the owner's record # of the last hello any process heard self._heard: dict | None | object = _UNHEARD self._record_cache: tuple[float, dict | None] | None = None - def catalog(self) -> list[dict]: - """sunnypilot's big-model bundles as {name, ref}, newest first, as the model - manager's own picker lists them. From the cached JSON rather than the parsed - bundles: parsing builds capnp objects and writes chunk manifests, for two - fields. Never raises.""" - try: - key = self.op.keys.catalog - bundles = ((self.op.get(key) if key else None) or {}).get('bundles', []) - found = [b for b in bundles if _REF.fullmatch(str(b.get('ref'))) - and int(b.get('minimum_selector_version', 0)) == self.op.catalog_selector] - except Exception: - self.op.log.exception("jetlink: could not read the big-model catalog") - return [] - found.sort(key=lambda b: int(b.get('index', 0)), reverse=True) - return [{'name': str(b.get('display_name') or b['ref'][:10]), 'ref': b['ref']} for b in found] + # -- the model index -------------------------------------------------------- - def pointers(self) -> dict[str, dict]: - value = self.op.get(self.op.keys.pointers) - return value if isinstance(value, dict) else {} + def index_path(self) -> Path: + return self.model_dir() / INDEX_FILE - def fetch_pointer(self, ref: str) -> tuple[str, int]: - """The oid and size of the ONNX a comma commit names, in its tree or, for a - precompiled-pkl commit, in comma's model repo. The Jetson's registry does - the same lookup, so the two ends agree on every model's identity.""" - from jetlink.registry.lfs import fetch_pointer as registry_fetch_pointer - pointer = registry_fetch_pointer(ref, timeout=POINTER_TIMEOUT) - return pointer.oid, pointer.size - - def resolve_pointer(self, ref: str) -> tuple[str, int]: - """The oid and size behind a catalog model, fetched the first time and kept for good.""" - return self._resolve(ref, self.fetch_pointer) - - def fetch_size(self, sha256: str) -> tuple[str, int]: - """The ONNX with this hash and its byte count, from the LFS servers a - download asks.""" - from jetlink.registry.lfs import fetch_size as registry_fetch_size - pointer = registry_fetch_size(sha256, self.lfs_endpoints(), timeout=POINTER_TIMEOUT) - return pointer.oid, pointer.size - - def resolve_size(self, sha256: str) -> tuple[str, int]: - """A model picked by its hash, sized the first time and kept for good with - the catalog's pointers, under the hash.""" - return self._resolve(sha256, self.fetch_size) - - def _resolve(self, key: str, fetch) -> tuple[str, int]: - known = self.pointers() - if key in known: - return known[key]['oid'], int(known[key]['size']) - oid, size = fetch(key) - known[key] = {'oid': oid, 'size': size} - self._keep_pointers(known) - self.op.log.warning("jetlink: %s is %s, %d MB", key[:10], oid[:16], size >> 20) - return oid, size - - def _keep_pointers(self, known: dict[str, dict]) -> None: - # blocking: the next lookup reads this back, and a put still in flight would be lost under it - self.op.put(self.op.keys.pointers, known, block=True) - self._index_cache = None - - @staticmethod - def _row(name: str, ref: str | None, known: dict[str, dict], sha256: str | None = None) -> dict: - """A model as a catalog row, by its ref's pointer, else by its hash, sized - once resolve_size has looked it up.""" - p = known.get(ref) or known.get(sha256) or {} - return {'name': name, 'ref': ref, 'oid': p.get('oid') or sha256, 'size': int(p['size']) if p.get('size') else None} - - def _index(self) -> tuple[list[dict], dict[str, dict]]: - """The catalog's rows and the pointers behind them, read together once per INDEX_TTL.""" - now = time.monotonic() - if self._index_cache is None or now - self._index_cache[0] >= INDEX_TTL: - known = self.pointers() - self._index_cache = (now, [self._row(b['name'], b['ref'], known) for b in self.catalog()], known) - return self._index_cache[1], self._index_cache[2] + def index(self) -> ModelIndex | None: + """The comma's copy of the model index, None before the first fetch. + Looked at once per INDEX_TTL and parsed again only when the file + changed. Never raises.""" + return self._read_index()[0] def model_index(self) -> list[dict]: - """Every catalog model, {name, ref, oid, size}. oid and size are None until - the model has been selected and resolved. No network.""" - return self._index()[0] + """Every model the index lists, {name, ref, oid, size}, in its order. No network.""" + return self._read_index()[1] + + def _read_index(self) -> tuple[ModelIndex | None, list[dict]]: + now = time.monotonic() + cache = self._index_cache + if cache is not None and now - cache[0] < INDEX_TTL: + return cache[2], cache[3] + path = self.index_path() + try: + st = path.stat() + seen = (st.st_mtime_ns, st.st_size) + except OSError: + seen = None + if cache is not None and cache[1] == seen: + # the same file, parsed or not: a copy that does not parse is told of once + index = cache[2] + else: + index = None + if seen is not None: + try: + index = parse_index(json.loads(path.read_text())) + except (OSError, ValueError, RegistryError): + self.op.log.exception("jetlink: the comma's model index does not parse") + self._keep_index(now, seen, index) + return self._index_cache[2], self._index_cache[3] + + def _keep_index(self, now: float, seen: tuple | None, index: ModelIndex | None) -> None: + rows = self._index_cache[3] if self._index_cache is not None and self._index_cache[2] is index else \ + [{'name': m.name, 'ref': m.ref, 'oid': m.oid, 'size': m.size} for m in index.models] if index else [] + self._index_cache = (now, seen, index, rows) + + def refresh_index(self) -> ModelIndex: + """The model index, fetched and kept as the comma's copy, which is left + alone when nothing changed so no reader parses it again. Raises a + RegistryError for one that cannot be had or used, and keeps the copy.""" + from jetlink.comma import gadget + from jetlink.registry.index import fetch_index + data, index = fetch_index() + path = self.index_path() + if gadget.read_record(path) != data: + gadget.write_record(path, data) + st = path.stat() + self._keep_index(time.monotonic(), (st.st_mtime_ns, st.st_size), index) + return index + + def _row(self, name: str, ref: str | None, sha256: str | None = None) -> dict: + """A model as a row: its ONNX by its ref, from the index, else by its hash + and sized from the index. oid or size is None where the index does not know.""" + index = self.index() + pointer = index.pointer(ref) if index is not None and ref else None + if pointer is not None: + return {'name': name, 'ref': ref, 'oid': pointer.oid, 'size': pointer.size} + return {'name': name, 'ref': ref, 'oid': sha256, 'size': index.size(sha256) if index is not None and sha256 else None} + + def resolve(self, ref: str | None, sha256: str | None) -> tuple[str, int]: + """A pick's oid and size, from the index; one the comma's copy does not + know fetches it again, at most once per INDEX_MISS_RETRY. Raises a + RegistryError for a model the index does not know, or no index at all.""" + if (found := self._identity(ref, sha256)) is None and time.monotonic() >= self._refresh_after: + self._refresh_after = time.monotonic() + INDEX_MISS_RETRY + self.refresh_index() + found = self._identity(ref, sha256) + if found is None: + raise RegistryError(f"the model index does not know {(ref or sha256 or 'the model')[:16]}") + return found + + def _identity(self, ref: str | None, sha256: str | None) -> tuple[str, int] | None: + row = self._row('', ref, sha256) + return (row['oid'], int(row['size'])) if row['oid'] and row['size'] else None def selected_slot(self) -> dict | None: """The big-model slot's pick as {name, ref, sha256}, from the raw param @@ -175,18 +194,59 @@ class Models: at = os.stat(Path(self.op.params_dir()) / key).st_mtime except OSError: pass + self._note_slot(pick) self._slot_cache = (now, pick, at) return pick + def _note_slot(self, pick: dict | None) -> None: + """Keep SLOT_STATE in step with the slot: what is in it, and when it was + emptied. A pick taken out is the default chosen (the model manager's + Default, here or on sunnylink), a pick like any other, which the hello + names and a server's older pick must not undo. Read again only when the + slot no longer matches what this process knows, so per process it costs + a file read at a change. Never raises.""" + last = {'ref': pick['ref'], 'sha256': pick['sha256']} if pick else None + state = self._slot_state + if state is not None and state.get('last') == last: + return + path = self.model_dir() / SLOT_STATE + try: + state = json.loads(path.read_text()) + state = state if isinstance(state, dict) else {} + except (OSError, ValueError): + state = {} + if state and state.get('last') == last: + self._slot_state = state + return + if last is None and not state: + # empty with nothing known before it: never picked, as far as we know + self._slot_state = {'last': None, 'cleared_at': None} + return + state = {'last': last, 'cleared_at': time.time() if last is None else None} + self._slot_state = state + from jetlink.comma import gadget + try: + gadget.write_record(path, state) + except OSError: + self.op.log.exception("jetlink: could not keep the big-model slot's state") + + def default_chosen_at(self) -> float | None: + """When the slot was emptied by this clock, the default chosen; None while + it holds a pick, or was never filled.""" + if self.selected_slot() is not None or self._slot_state is None: + return None + at = self._slot_state.get('cleared_at') + return float(at) if isinstance(at, (int, float)) else None + def default_model(self) -> dict | None: - """What runs with no pick: jetlink's default big model, else the newest the - catalog lists. Not the chestnut's default, which is the model in the tree. - A fork with no catalog at all (Keys.catalog None) gets jetlink's default by - its ref; one whose catalog has not been fetched yet waits for it.""" - if self.op.keys.catalog is None: - return self._row(DEFAULT_BIG_MODEL_NAME, DEFAULT_BIG_MODEL_REF, self._index()[1]) + """What runs with no pick: the index's default, else the first it lists. + Not the chestnut's default, which is the model in the tree. Before the + comma has an index, jetlink's own by its ref, which a provisioning run + resolves by fetching one.""" + if (index := self.index()) is None: + return self._row(DEFAULT_BIG_MODEL_NAME, DEFAULT_BIG_MODEL_REF) models = self.model_index() - return next((m for m in models if m['ref'] == DEFAULT_BIG_MODEL_REF), models[0] if models else None) + return next((m for m in models if m['ref'] == index.default), models[0] if models else None) def default_model_name(self) -> str | None: """default_model()'s name without the catalog's trailing build date, the @@ -197,26 +257,40 @@ class Models: def selected_model(self) -> dict | None: """The model picked last, here or on the server: the slot's pick, listed - in the catalog or not (a ref is enough to find its pointer), unless the - server's last hello named a newer one. With neither, default_model().""" + or not (the index knows every ref it ever saw), unless the server's last + hello named a newer one. With neither, default_model().""" if (theirs := self._newer_server_pick()) is not None: return self._server_row(theirs) if (own := self.selected_slot()) is not None: return self._own_row(own) return self.default_model() + def _own(self) -> tuple[dict | None, float | None]: + """Our pick as a row, and when it was made by this clock: the slot's, or + the default when it was chosen by emptying the slot. (None, None) with + neither.""" + if (own := self.selected_slot()) is not None: + return self._own_row(own), self.picked_at() + if (at := self.default_chosen_at()) is not None and (default := self.default_model()) is not None: + return default, at + return None, None + def _own_row(self, own: dict) -> dict: - return self._row(own['name'], own['ref'], self._index()[1], own['sha256']) + return self._row(own['name'], own['ref'], own['sha256']) # -- picks on both ends ----------------------------------------------------- - # A pick is made on either end: in the comma's list, or on the server (its - # web page, an app's Use Model). Each hello carries the comma's own pick and - # how long ago it was made, and the server answers with the newer of that - # and its own (PickStore in JetlinkServer). The answer is kept with the rest - # of the hello (gadget.SERVER, through the owner), dated by this clock, and - # selected_model() takes whichever pick is newer; so a model picked on the - # server drives without being picked here too, and one picked here goes to - # the server at the next hello. + # A pick is made on either end: in the comma's list (the default included, + # chosen by emptying the slot), or on the server (its web page, an app's Use + # Model). Each hello carries the comma's own pick, when it was made by this + # clock and how long ago, and the server answers with the newer of that and + # its own (PickStore in JetlinkServer), which it settles by what the comma + # named before rather than by its clock where it can: a Jetson's clock + # starts again behind at every power-on. The answer is kept with the rest of + # the hello (gadget.SERVER, through the owner) along with the pick the hello + # carried, and selected_model() takes the server's when that is still our + # pick: the server chose knowing it. So a model picked on the server drives + # without being picked here too, and one picked here goes to the server at + # the next hello. def picked_at(self) -> float | None: """When the slot was last written, by this clock: when the pick was made. @@ -224,21 +298,24 @@ class Models: return self._slot_cache[2] if self.selected_slot() is not None and self._slot_cache else None def hello_pick(self) -> dict | None: - """Our own pick, for the hello: {ref, sha256, nbytes, name, age_s}, the - pointer where it is resolved already (no network here). None with nothing - picked here, or no date for it: the server's pick, if any, stands.""" - own, at = self.selected_slot(), self.picked_at() - if own is None or at is None: + """Our own pick, for the hello: {ref, sha256, nbytes, name, age_s, at}, + its oid and size where the index knows them (no network here), `at` when it + was made by this clock. None with nothing picked here, or no date for it: + the server's pick, if any, stands.""" + row, at = self._own() + if row is None or at is None or not (row['ref'] or row['oid']): return None - row = self._own_row(own) return {'ref': row['ref'], 'sha256': row['oid'], 'nbytes': row['size'], 'name': row['name'], - 'age_s': round(max(0.0, time.time() - at), 1)} + 'age_s': round(max(0.0, time.time() - at), 1), 'at': round(at, 3)} - def heard(self, hello: dict) -> dict: - """The server's hello with its `pick` dated by this clock (_dated), which - is how the owner keeps it (lending.SERVER_FIELDS) and server_pick() reads - it. This process goes by it at once, whatever the owner does with it.""" + def heard(self, hello: dict, asked=_UNTOLD) -> dict: + """The server's hello with its `pick` dated by this clock (_dated), and + `asked`, the hello_pick() the hello carried, kept with it: which is how + the owner keeps it (lending.SERVER_FIELDS) and server_pick() reads it. + This process goes by it at once, whatever the owner does with it.""" self._heard = _dated(hello.get('pick')) + if self._heard is not None and asked is not _UNTOLD: + self._heard['asked'] = {k: asked.get(k) for k in ('ref', 'sha256', 'at')} if isinstance(asked, dict) else None return {**hello, 'pick': self._heard} if self._heard is not None else hello def server_pick(self) -> dict | None: @@ -257,49 +334,52 @@ class Models: def _newer_server_pick(self) -> dict | None: """The server's last pick, when it is another model than ours and was - picked after it, or we have none.""" - theirs, own = self.server_pick(), self.selected_slot() - if theirs is None or (own is not None and (self._same(theirs, self._own_row(own)) or - theirs['at'] <= (self.picked_at() or 0.0))): + picked after it, or we have none. The server settled that when the hello + carried this pick of ours; a pick made here since that hello is newer. + An answer kept without the pick it was asked with (an older jetlink) goes + by its date.""" + theirs = self.server_pick() + if theirs is None: return None - return theirs + own, at = self._own() + if own is None: + return theirs + if self._same(theirs, own): + return None + if 'asked' in theirs: + asked = theirs['asked'] + fresh = isinstance(asked, dict) and self._same(asked, own) and isinstance(asked.get('at'), (int, float)) \ + and at is not None and abs(asked['at'] - at) < 1.0 + return theirs if fresh else None + return None if theirs['at'] <= (at or 0.0) else theirs @staticmethod def _same(pick: dict, row: dict) -> bool: """One model: by the catalog ref where both have one, since two refs can share one export, else by the hash. The server's PickStore.sameModel.""" - if pick['ref'] and row['ref']: + if pick.get('ref') and row['ref']: return pick['ref'] == row['ref'] - return bool(pick['sha256']) and pick['sha256'] == row['oid'] + return bool(pick.get('sha256')) and pick['sha256'] == row['oid'] def _server_row(self, pick: dict) -> dict: - """The server's pick as a catalog row: ours where we list it, with the - server's oid and size, which name the bytes it has.""" + """The server's pick as a row: named as we list it, with the server's oid + and size, which name the bytes it has.""" listed = next((m for m in self.model_index() if pick['ref'] and m['ref'] == pick['ref']), None) row = self._row(listed['name'] if listed else pick.get('name') or (pick['ref'] or pick['sha256'])[:16], - pick['ref'], self._index()[1], pick['sha256']) + pick['ref'], pick['sha256']) if pick['sha256'] and pick.get('nbytes'): row['oid'], row['size'] = pick['sha256'], int(pick['nbytes']) return row def adopt(self) -> None: - """Keep the server's pointer for its pick, by ref and by hash, so naming - the model here needs no lookup; and make a newer pick from the server - ours. With a catalog, ask the model manager for it (Keys.select), so the - comma's own list shows what drives: only a model the catalog lists, which - is all the model manager takes; any other drives all the same - (selected_model). Without one, no model manager owns the slot, so the - pick is written into it, whatever the model. Never raises.""" + """Make a newer pick from the server ours. With a catalog, ask the model + manager for it (Keys.select), so the comma's own list shows what drives: + only a model the catalog lists, which is all the model manager takes; any + other drives all the same (selected_model). Without one, no model manager + owns the slot, so the pick is written into it, whatever the model. Never + raises.""" try: - if (theirs := self.server_pick()) is None: - return - known = self.pointers() - if theirs['sha256'] and theirs.get('nbytes'): - pointer = {'oid': theirs['sha256'], 'size': int(theirs['nbytes'])} - missing = {k: pointer for k in (theirs['ref'], theirs['sha256']) if k and k not in known} - if missing: - self._keep_pointers({**known, **missing}) - if self._newer_server_pick() is None: + if (theirs := self.server_pick()) is None or self._newer_server_pick() is None: return model, keys = self._server_row(theirs), self.op.keys if keys.catalog is None: @@ -312,13 +392,20 @@ class Models: # a download the user asked for that the model manager has not taken # yet is left alone: the next hello asks again key = keys.select - if key and model['ref'] and self.op.get(key) is None and any(m['ref'] == model['ref'] for m in self.model_index()): + if key and model['ref'] and self.op.get(key) is None and model['ref'] in self._catalog_refs(): self.op.log.warning("jetlink: %s was picked on the server, picking it here too", model['name']) self.op.put(key, model['ref']) self._slot_cache = None except Exception: self.op.log.exception("jetlink: could not take the server's pick") + def _catalog_refs(self) -> set[str]: + """The refs the model manager's catalog lists: the index's without a + chestnut fitted (big_catalog), else sunnypilot's.""" + key = self.op.keys.catalog + bundles = ((self.op.get(key) if key else None) or {}).get('bundles') + return {b['ref'] for b in bundles if isinstance(b, dict) and is_ref(b.get('ref'))} if isinstance(bundles, list) else set() + def selected_model_name(self) -> str | None: selected = self.selected_model() return selected['name'] if selected else None @@ -372,8 +459,6 @@ class Models: the run for the next one. Each attempt asks the LFS server again: the address it hands out expires. `retrying(error, delay)` hears of each wait; a run that is stopped stops waiting.""" - from jetlink.registry.catalog import NetworkError - from jetlink.registry.lfs import Pointer model = self.selected_model() if model is None or not model['oid']: return None @@ -397,27 +482,23 @@ class Models: time.sleep(min(1.0, end - time.monotonic())) def _fetch_once(self, pointer, dest: Path, progress, should_stop) -> Path: - from jetlink.registry.catalog import NetworkError from jetlink.registry.lfs import lfs_download, lfs_hrefs for endpoint, href in lfs_hrefs(self.lfs_endpoints(), pointer): self.op.log.warning("jetlink: fetching the large model (%d MB) from %s", pointer.size >> 20, endpoint) return lfs_download(href, pointer, dest, progress=progress, should_stop=should_stop) raise NetworkError(f"no LFS server has {pointer.oid[:16]}") - def _row_for(self, oid: str) -> dict | None: - """The catalog model with this oid, among those we have resolved.""" - return next((m for m in self.model_index() if m['oid'] == oid), None) - def name_for(self, oid: str) -> str: - """The catalog's name for the model with this oid, or the oid's first 16 - characters when it lists none we have resolved.""" - row = self._row_for(oid) - return row['name'] if row is not None else oid[:16] + """The index's name for the model with this oid, or the oid's first 16 + characters when it lists none.""" + index = self.index() + listed = index.by_oid.get(oid) if index is not None else None + return listed.name if listed is not None else oid[:16] def size_for(self, oid: str) -> int | None: - """The byte count of the catalog model with this oid, once resolved here.""" - row = self._row_for(oid) - return row['size'] if row is not None else None + """The byte count of the ONNX with this oid, as the index knows it.""" + index = self.index() + return index.size(oid) if index is not None else None def has_file(self, model: dict) -> bool: """Is this model's ONNX on the comma, whole? Its size is the cheap check.""" @@ -430,54 +511,26 @@ class Models: # -- the model manager's catalog -------------------------------------------- - def big_catalog(self, catalog: dict) -> dict: - """The big-model catalog with every newer one sunnypilot has published folded in. - A Jetson runs the commit's ONNX, so a model sunnypilot only builds for its next - runtime is still one it can run; see jetlink.registry.catalog.fetch_catalogs. - Never raises: a probe that fails keeps what the last one found.""" - try: - from jetlink.registry.catalog import fetch_catalogs, merge_catalogs - fetched = fetch_catalogs() - self.keep_pins(fetched.get('pointers', {})) - merged = merge_catalogs([catalog, fetched], selector=self.op.catalog_selector) - merged.pop('pointers', None) # the pins are kept above, never in the model manager's catalog - added = len(merged.get('bundles', [])) - len(catalog.get('bundles', [])) - if added <= 0: - return catalog - self.op.log.warning("jetlink: %d model(s) only newer catalogs list", added) - return merged - except Exception: - self.op.log.exception("jetlink: could not check for newer catalogs") - return self._found_before(catalog) + def big_catalog(self, selector: int) -> dict: + """The index's models as a catalog in sunnypilot's shape, for the model + manager's big-model list when no chestnut is fitted: what the comma lists + is then what the Jetson runs, each with the index's fields, in its folder + and order. Only `selector` is the fork's (its model manager's + REQUIRED_JSON_VERSION, which it parses), and no artifacts: a chestnut + has nothing to download, and an accelerator runs the ONNX. The index is + fetched again; raises a RegistryError when it cannot be, so the model + manager keeps the list it has.""" + models = self.refresh_index().models + return {'bundles': [{'ref': m.ref, 'display_name': m.name, 'short_name': m.short_name or m.ref[:7], 'index': len(models) - i, + 'is_big': True, 'is_20hz': m.is_20hz, 'generation': m.generation, 'environment': m.environment, + 'runner': m.runner, 'minimum_selector_version': str(selector), 'build_time': m.build_time, + 'overrides': {'folder': m.folder}, 'models': []} for i, m in enumerate(models)]} - def keep_pins(self, pins: dict[str, dict]) -> None: - """Writes the list's pins over the kept pointers (see catalog.pinned_pointers).""" - known = self.pointers() - changed = {ref: pin for ref, pin in pins.items() if known.get(ref) != pin} - if not changed: - return - known.update(changed) - self._keep_pointers(known) - for ref, pin in changed.items(): - self.op.log.warning("jetlink: %s is pinned to %s", ref[:10], pin['oid'][:16]) + def listed_ref(self, ref: str) -> str | None: + """The ref the index lists a model under: its own, or the newest one with + the same ONNX (a recompile is listed once). None for a model it does not + list. No network.""" + index = self.index() + listed = index.listed(ref) if index is not None and is_ref(ref) else None + return listed.ref if listed is not None else None - def _found_before(self, catalog: dict) -> dict: - """The catalog with the models the last probe found folded in again, from the - model manager's cached copy: every entry there the fetched catalog does not - list. Not only merge_catalogs' entries with no artifacts: a newer catalog at - the same selector version comes through with its builds (v26 lists Cinque - Terre V3 that way). Dropped with a probe that failed, a pick only a newer - catalog lists would be reset at the manager's next start and the owner would - provision the default in its place; a refresh on a flaky network is enough. - A model sunnypilot withdrew stays listed until a probe works.""" - try: - listed = {b.get('ref') for b in catalog.get('bundles', [])} - key = self.op.keys.catalog - cached = ((self.op.get(key) if key else None) or {}).get('bundles', []) - kept = [b for b in cached if isinstance(b, dict) and b.get('ref') not in listed] - except Exception: - return catalog - if not kept: - return catalog - self.op.log.warning("jetlink: keeping %d model(s) the last probe found", len(kept)) - return {**catalog, 'bundles': [*catalog.get('bundles', []), *kept]} diff --git a/jetlink_repo/jetlink/openpilot/provision.py b/jetlink_repo/jetlink/openpilot/provision.py index 0c04d908e9..1e95fd2f18 100644 --- a/jetlink_repo/jetlink/openpilot/provision.py +++ b/jetlink_repo/jetlink/openpilot/provision.py @@ -191,7 +191,8 @@ class ProvisioningRun: what it costs, never speaks the protocol to learn it, and keeps the server's pick and whether it downloads for the comma (Models.heard).""" models = self.parts.models - hello = models.heard(self.client.hello(timeout=10.0, pick=models.hello_pick())) + sent = models.hello_pick() + hello = models.heard(self.client.hello(timeout=10.0, pick=sent), sent) if self.loan is not None: self.loan.note_server(hello) return hello diff --git a/jetlink_repo/jetlink/openpilot/status.py b/jetlink_repo/jetlink/openpilot/status.py index c47ee99cb9..f7350a0bd5 100644 --- a/jetlink_repo/jetlink/openpilot/status.py +++ b/jetlink_repo/jetlink/openpilot/status.py @@ -433,11 +433,11 @@ def read(parts, mode: str) -> Status: def model_state(parts, ref: str) -> str | None: - """One catalog model, for the picker's list: 'ready' when the Jetson has + """One listed model, for the picker's list: 'ready' when the Jetson has built it (the spec record), 'downloaded' when the comma has its file, else - None. Records and one stat; a model never resolved has neither.""" + None. Records and one stat.""" model = next((m for m in parts.models.model_index() if m['ref'] == ref), None) - if model is None or not model.get('oid'): + if model is None: return None if parts.spec.engine_ready_for(model['oid']): return 'ready' diff --git a/jetlink_repo/jetlink/openpilot/stock.py b/jetlink_repo/jetlink/openpilot/stock.py index 776fdfb6e8..84b6a1b0c9 100644 --- a/jetlink_repo/jetlink/openpilot/stock.py +++ b/jetlink_repo/jetlink/openpilot/stock.py @@ -16,8 +16,8 @@ capnp, swaglog loads zmq, and model_face() imports modeld itself, so it is for modeld alone, where that module is already loaded. Data are class attributes, behaviour small methods, each a fork may override; -one override point per value: a fork sets KEYS and CATALOG_SELECTOR, and the -interface's keys and catalog_selector read them. Both are jetlink's contract: +one override point per value: a fork sets KEYS, and the interface's keys +reads it. It is jetlink's contract: renaming or removing one, or changing what one means, is a breaking change and bumps API. scripts/upstream (the upstream-paths CI job) checks the paths and literals here against commaai/openpilot master weekly. @@ -57,9 +57,7 @@ class Stock: # the canonical names, which a fork declares in params_keys.h. No model # manager upstream: no big-model slot or catalog, so jetlink runs its default KEYS = Keys(link='JetlinkLink', offroad='IsOffroad', progress='JetlinkProgress', spec='JetlinkSpec', - pointers='JetlinkModelPointers', charge_phone='JetlinkChargePhone') - # the model manager's catalog version; none upstream - CATALOG_SELECTOR = 0 + charge_phone='JetlinkChargePhone') # comma's chestnut, running and in its ROM (common.hardware.usb's # CHESTNUT_USB_IDS + CHESTNUT_ROM_USB_IDS, which CI holds this to): written # out, as the owner cannot import that package @@ -88,10 +86,6 @@ class Stock: def keys(self) -> Keys: return self.KEYS - @property - def catalog_selector(self) -> int: - return self.CATALOG_SELECTOR - @cached_property def log(self): from openpilot.common.swaglog import cloudlog diff --git a/jetlink_repo/jetlink/openpilot/testing.py b/jetlink_repo/jetlink/openpilot/testing.py index 77add7c4f8..99ec572171 100644 --- a/jetlink_repo/jetlink/openpilot/testing.py +++ b/jetlink_repo/jetlink/openpilot/testing.py @@ -192,7 +192,8 @@ _MODEL = object() HOOK_CALLS = {'should_run': [(False, None, None)], 'status': [()], 'reason': [()], 'prepare': [(False,), (True,)], 'attach': [(_MODEL, object(), 1928, 1208)], 'handed_over': [(_MODEL,)], 'state': [(_MODEL,)], 'shutdown_step': [(False,), (True, 'test')], 'request_shutdown': [('test',)], 'shutdown_pending': [()], - 'model_state': [('f' * 40,)], 'should_extend_catalog': [()], 'extend_catalog': [({'bundles': []},)], + 'model_state': [('f' * 40,)], 'lists_big_models': [()], 'big_catalog': [(21,)], 'listed_ref': [('f' * 40,)], + 'should_extend_catalog': [()], 'extend_catalog': [({'bundles': []},)], 'runtime': [()], 'swap_events': [(0.01,)]} diff --git a/jetlink_repo/jetlink/registry/__init__.py b/jetlink_repo/jetlink/registry/__init__.py index cc8e725d68..9cd0f99c70 100644 --- a/jetlink_repo/jetlink/registry/__init__.py +++ b/jetlink_repo/jetlink/registry/__init__.py @@ -4,8 +4,8 @@ Copyright (c) 2026-, Zeph Leggett. This file is part of jetlink and is licensed under the MIT License. See the LICENSE file in the root directory for more details. -Which large models exist and how to get their bytes: sunnypilot's catalog -(catalog) and a model's ONNX by its git-lfs pointer (lfs). +Which large models exist and how to get their bytes: the model index, which +names every big model's ONNX (index), and an ONNX by its oid (lfs). The comma imports both, so this package stays stdlib only and imports nothing itself. The server's side of it is JetlinkRegistry in Swift, held to these diff --git a/jetlink_repo/jetlink/registry/catalog.py b/jetlink_repo/jetlink/registry/catalog.py deleted file mode 100644 index c8b6dfe495..0000000000 --- a/jetlink_repo/jetlink/registry/catalog.py +++ /dev/null @@ -1,211 +0,0 @@ -""" -Copyright (c) 2026-, Zeph Leggett. - -This file is part of jetlink and is licensed under the MIT License. -See the LICENSE file in the root directory for more details. - -sunnypilot's big-model catalog: which large models exist and which commit each one is. - -The catalog is the same JSON the comma's model manager caches for its chestnut -slot, so a model picked here is a model the comma can ask for. A bundle's -artifacts are tinygrad pkls for a GPU we do not have; the only field that -matters is `ref`, the comma openpilot commit the bundle was compiled from, -because that commit's ONNX is what a jetlink server runs. - -Stdlib only, and every network call takes an `opener` so tests stay offline. -""" -from __future__ import annotations - -import json -import re -import urllib.error -import urllib.request -from dataclasses import dataclass - - -CATALOG_URL_TEMPLATE = 'https://raw.githubusercontent.com/sunnypilot/sunnypilot-models/refs/heads/gh-pages/docs/driving_models_chestnut_v{version}.json' -# v26 is v25 plus Cinque Terre V3, at the same selector version. v27 onwards -# are selector 20 and a newer tinygrad, for sunnypilot's next sync. -CATALOG_VERSION = 26 -CATALOG_URL = CATALOG_URL_TEMPLATE.format(version=CATALOG_VERSION) -# sunnypilot publishes a new catalog version for new models or a new runtime, -# and keeps the old ones. The versions after the pinned one are probed up to -# the first that is not there, so a model published later is listed without a -# release of this package. See merge_catalogs. -PROBE_LIMIT = 10 -# zoompilot's own big models: ones sunnypilot does not list (yet), such as -# previews from comma's open pull requests, in the catalog's shape. Merged -# after sunnypilot's catalogs, so their entry for the same commit wins. An -# edit on main reaches every comma within the hour, with no release. -EXTRA_CATALOG_URL = 'https://raw.githubusercontent.com/zoompilot/jetlink/refs/heads/main/catalog/extra_big_models.json' -# The selector version the fork requires (REQUIRED_JSON_VERSION on the comma). -# It is a string in the JSON; bundles at any other version describe fields we -# would misread. -REQUIRED_SELECTOR_VERSION = 19 -# Cinque Terre V3 Model (September 17, 2026) -DEFAULT_BIG_MODEL_REF = 'bf3e3631b3f91d92a1020a5e0dd4298b93ff4244' -# its name as sunnypilot's catalog lists it, without the build date: what a -# comma with no catalog to name it from shows (jetlink.openpilot, on a fork -# without sunnypilot's model manager) -DEFAULT_BIG_MODEL_NAME = 'Cinque Terre V3 Model' -CATALOG_TIMEOUT = 10.0 - -_REF = re.compile(r'[0-9a-f]{40}') -_SHA256 = re.compile(r'[0-9a-f]{64}') - - -class RegistryError(Exception): - """Anything the registry refuses to do.""" - - -class NetworkError(RegistryError): - """A server could not be reached or did not answer sensibly.""" - - -class VerifyError(RegistryError): - """Bytes arrived, but not the bytes that were asked for.""" - - -class NotFound(NetworkError): - """The server answered that there is nothing at the URL.""" - - -@dataclass(frozen=True) -class CatalogModel: - name: str - short_name: str - ref: str - build_time: str - index: int - - -def is_ref(value: str) -> bool: - """A comma openpilot commit: 40 lowercase hex characters.""" - return isinstance(value, str) and _REF.fullmatch(value) is not None - - -def is_sha256(value: str) -> bool: - """A model identity: 64 lowercase hex characters, which is also the LFS oid.""" - return isinstance(value, str) and _SHA256.fullmatch(value) is not None - - -def parse_catalog(data: dict) -> list[CatalogModel]: - """The big-model bundles, newest first. No IO, and no bundle can raise. - - A malformed entry in a catalog served to every comma must cost that entry and - nothing else, so each bundle is parsed in its own try. - """ - found: dict[str, CatalogModel] = {} - for bundle in _bundles(data): - try: - ref = bundle.get('ref') - if not is_ref(ref) or ref in found: - continue - if _selector(bundle) != REQUIRED_SELECTOR_VERSION or not bundle.get('is_big'): - continue - found[ref] = CatalogModel(name=str(bundle.get('display_name') or ref[:10]), short_name=str(bundle.get('short_name') or ''), - ref=ref, build_time=str(bundle.get('build_time') or ''), index=int(bundle.get('index', 0))) - except (AttributeError, TypeError, ValueError): - continue - return sorted(found.values(), key=lambda m: m.index, reverse=True) - - -def http_get(url: str, timeout: float, opener=None, limit: int = -1) -> bytes: - """The body at `url`, at most `limit` bytes. A 404 is NotFound, and every - other failure a NetworkError, so a caller only catches what it treats apart.""" - opener = opener or urllib.request.urlopen - try: - with opener(url, timeout=timeout) as response: - return response.read(limit) - except urllib.error.HTTPError as e: - raise (NotFound if e.code == 404 else NetworkError)(f"could not fetch {url}: {e}") from e - except (OSError, ValueError) as e: - raise NetworkError(f"could not fetch {url}: {e}") from e - - -def http_json(url: str, timeout: float, opener=None, kind: type = dict): - """http_get, parsed, and of the type asked for.""" - body = http_get(url, timeout, opener) - try: - data = json.loads(body.decode()) - except ValueError as e: - raise NetworkError(f"{url} did not serve JSON: {e}") from e - if not isinstance(data, kind): - raise NetworkError(f"{url} did not serve a JSON {kind.__name__}") - return data - - -def fetch_catalog(url: str = CATALOG_URL, timeout: float = CATALOG_TIMEOUT, opener=None) -> dict: - """The catalog JSON. Every failure, transport or content, is a NetworkError.""" - return http_json(url, timeout, opener) - - -def fetch_catalogs(opener=None) -> dict: - """The pinned catalog merged with every one sunnypilot has published since, - whose commits the server runs too, and then zoompilot's extra models. The - probe stops at the first version that is not there, and a missing extra list - adds nothing. Any other failure raises: a short list would drop a model - found only past it from whatever the caller caches, so the caller keeps its - last complete one instead.""" - found = [fetch_catalog(CATALOG_URL, opener=opener)] - for v in range(CATALOG_VERSION + 1, CATALOG_VERSION + 1 + PROBE_LIMIT): - try: - found.append(fetch_catalog(CATALOG_URL_TEMPLATE.format(version=v), opener=opener)) - except NotFound: - break - try: - extra = fetch_catalog(EXTRA_CATALOG_URL, opener=opener) - except NotFound: - extra = {} - merged, pins = merge_catalogs([*found, extra]), pinned_pointers(extra) - return {**merged, 'pointers': pins} if pins else merged - - -def pinned_pointers(data: dict) -> dict[str, dict]: - """The extra list's well-formed pins, {ref: {'oid', 'size'}}: the ONNX jetlink - runs for that commit whatever a lookup said. See catalog/README.md.""" - pins = data.get('pointers') - if not isinstance(pins, dict): - return {} - return {ref: {'oid': p['oid'], 'size': p['size']} for ref, p in pins.items() - if is_ref(ref) and isinstance(p, dict) and is_sha256(p.get('oid')) - and type(p.get('size')) is int and p['size'] > 0} - - -def merge_catalogs(catalogs: list[dict], selector: int = REQUIRED_SELECTOR_VERSION) -> dict: - """One catalog in the first one's shape, listing every big model of them all. - - A model some catalog lists at `selector` comes through as published, so a - chestnut can still fetch its build. One listed only at another selector - version, which is where sunnypilot puts every model once it moves runtimes, - comes from the newest catalog that has it, retyped to `selector` and with no - artifacts: an accelerator runs the commit's ONNX and needs nothing else from - the entry, and a chestnut has nothing to download. - """ - if not catalogs: - return {} - kept: dict[str, dict] = {} - others: dict[str, dict] = {} - for data in catalogs: - for bundle in _bundles(data): - ref = bundle.get('ref') - if not is_ref(ref): - continue - if _selector(bundle) == selector: - kept.setdefault(ref, bundle) - elif bundle.get('is_big'): - others[ref] = bundle # the newest catalog's entry wins - extra = [{**b, 'minimum_selector_version': str(selector), 'models': []} for ref, b in others.items() if ref not in kept] - return {**catalogs[0], 'bundles': [*kept.values(), *extra]} - - -def _bundles(data: dict) -> list[dict]: - bundles = data.get('bundles') if isinstance(data, dict) else None - return [b for b in bundles if isinstance(b, dict)] if isinstance(bundles, list) else [] - - -def _selector(bundle: dict) -> int | None: - try: - return int(bundle.get('minimum_selector_version', 0)) - except (TypeError, ValueError): - return None diff --git a/jetlink_repo/jetlink/registry/index.py b/jetlink_repo/jetlink/registry/index.py new file mode 100644 index 0000000000..890c96e3d9 --- /dev/null +++ b/jetlink_repo/jetlink/registry/index.py @@ -0,0 +1,193 @@ +""" +Copyright (c) 2026-, Zeph Leggett. + +This file is part of jetlink and is licensed under the MIT License. +See the LICENSE file in the root directory for more details. + +Which big models a Jetson can run, and which ONNX each one is: the model +index, zoompilot/jetlink-model-index. + +A big model is a comma openpilot commit (its ref), and what a Jetson runs is +the ONNX that commit was built from. Finding that ONNX takes reading comma's +trees, its precompiled pkls' weights and its model repo, so the index repo does +it once an hour for every commit any catalog, comma master or comma pull +request has named, and publishes the answers: + + default the model a Jetson runs with nothing picked + refs every big-model commit ever seen -> {oid, size}: an old pick still resolves + onnx every ONNX comma has published -> {size}: a pick by hash alone is sized + models what to list, one entry per ONNX, in display order, with its folder + and the catalog fields a list in sunnypilot's shape needs + +The comma and the server each keep the last index that parsed, so both work +offline with what they had, and neither looks anything up itself. + +Stdlib only, and every network call takes an `opener` so tests stay offline. +""" +from __future__ import annotations + +import json +import re +import urllib.request +from dataclasses import dataclass, field + +INDEX_URL = 'https://raw.githubusercontent.com/zoompilot/jetlink-model-index/main/index/v1.json' +# the index's own version: another one describes fields we would misread +SCHEMA = 1 +INDEX_TIMEOUT = 10.0 +# a listed model with no folder of its own +OTHER_FOLDER = 'Other Models' +# the runtime fields of a listed model the index has none for (an index older +# than them): what sunnypilot's big models are built for +RUNTIME = {'generation': '12', 'environment': 'development', 'runner': 'tinygrad'} +# what runs with nothing picked before the comma or server has an index, which +# names its own (ModelIndex.default): Cinque Terre V3 Model (September 17, 2026) +DEFAULT_BIG_MODEL_REF = 'bf3e3631b3f91d92a1020a5e0dd4298b93ff4244' +# its name without the build date: what a comma with no index to name it from +# shows (jetlink.openpilot, on a fork without sunnypilot's model manager) +DEFAULT_BIG_MODEL_NAME = 'Cinque Terre V3 Model' + +_REF = re.compile(r'[0-9a-f]{40}') +_SHA256 = re.compile(r'[0-9a-f]{64}') + + +class RegistryError(Exception): + """Anything the registry refuses to do.""" + + +class NetworkError(RegistryError): + """A server could not be reached or did not answer sensibly.""" + + +class VerifyError(RegistryError): + """Bytes arrived, but not the bytes that were asked for.""" + + +class IndexInvalid(RegistryError): + """An index that cannot be used: another schema, or nothing listed.""" + + +def is_ref(value) -> bool: + """A comma openpilot commit: 40 lowercase hex characters.""" + return isinstance(value, str) and _REF.fullmatch(value) is not None + + +def is_sha256(value) -> bool: + """A model identity: 64 lowercase hex characters, which is also the LFS oid.""" + return isinstance(value, str) and _SHA256.fullmatch(value) is not None + + +@dataclass(frozen=True) +class Pointer: + oid: str # the ONNX SHA-256 + size: int + + +@dataclass(frozen=True) +class IndexModel: + ref: str + oid: str + size: int + name: str # as sunnypilot's catalog names it, build date and all + short_name: str # '' when the index has none + folder: str + build_time: str # ISO-8601, or '' + pr: int | None # the comma pull request a model listed only as one is + is_20hz: bool + generation: str # the runtime it was built for, as sunnypilot's catalog says + environment: str + runner: str + + +@dataclass(frozen=True) +class ModelIndex: + models: tuple[IndexModel, ...] # in display order + refs: dict[str, Pointer] # every ref the index resolves, the listed ones included + sizes: dict[str, int] # every ONNX the index knows, by oid + default: str | None = None # the listed ref that runs with nothing picked + # the listed models by their ONNX: one entry per ONNX, so one each + by_oid: dict[str, IndexModel] = field(init=False, repr=False, compare=False) + + def __post_init__(self): + object.__setattr__(self, 'by_oid', {m.oid: m for m in reversed(self.models)}) + + def pointer(self, ref: str) -> Pointer | None: + return self.refs.get(ref) + + def size(self, oid: str) -> int | None: + return self.sizes.get(oid) + + def listed(self, ref: str) -> IndexModel | None: + """The listed model for a ref's ONNX: its own entry, or for a recompile of a + listed model, the newest ref's (each ONNX is listed once).""" + pointer = self.refs.get(ref) + return self.by_oid.get(pointer.oid) if pointer is not None else None + + +def _pointer(value) -> Pointer | None: + if not isinstance(value, dict) or not is_sha256(value.get('oid')): + return None + size = _size(value.get('size')) + return Pointer(value['oid'], size) if size else None + + +def _size(value) -> int | None: + # bool is an int; a size is not + return value if isinstance(value, int) and not isinstance(value, bool) and value > 0 else None + + +def _text(value) -> str: + return value if isinstance(value, str) else '' + + +def parse_index(data) -> ModelIndex: + """The index, or IndexInvalid naming why it cannot be used. No IO. An entry + that is malformed costs that entry and nothing else; an index of another + schema, or one listing nothing, is refused whole, so a caller keeps the + last one that parsed rather than an empty list.""" + if not isinstance(data, dict): + raise IndexInvalid('the index is not a JSON object') + if data.get('schema') != SCHEMA: + raise IndexInvalid(f"the index is schema {data.get('schema')!r}, not {SCHEMA}") + refs = {ref: p for ref, value in _items(data.get('refs')) if is_ref(ref) and (p := _pointer(value))} + sizes = {oid: s for oid, value in _items(data.get('onnx')) if is_sha256(oid) and isinstance(value, dict) and (s := _size(value.get('size')))} + models: dict[str, IndexModel] = {} + for entry in data.get('models') if isinstance(data.get('models'), list) else []: + if not isinstance(entry, dict) or not is_ref(entry.get('ref')) or entry['ref'] in models or (p := _pointer(entry)) is None: + continue + ref = entry['ref'] + models[ref] = IndexModel(ref=ref, oid=p.oid, size=p.size, name=_text(entry.get('name')) or ref[:10], + short_name=_text(entry.get('short_name')), folder=_text(entry.get('folder')) or OTHER_FOLDER, + build_time=_text(entry.get('build_time')), pr=_size(entry.get('pr')), + is_20hz=entry['is_20hz'] if isinstance(entry.get('is_20hz'), bool) else True, + **{k: _text(entry.get(k)) or v for k, v in RUNTIME.items()}) + if not models: + raise IndexInvalid('the index lists no model') + # a listed model is what is shown and run, so its own entry names its ONNX + for m in models.values(): + refs[m.ref] = Pointer(m.oid, m.size) + for p in refs.values(): + sizes.setdefault(p.oid, p.size) + default = data.get('default') + return ModelIndex(models=tuple(models.values()), refs=refs, sizes=sizes, default=default if default in models else None) + + +def _items(value) -> list[tuple]: + return list(value.items()) if isinstance(value, dict) else [] + + +def fetch_index(url: str = INDEX_URL, timeout: float = INDEX_TIMEOUT, opener=None) -> tuple[dict, ModelIndex]: + """The index as served, and parsed: the raw JSON is what a caller keeps. + A transport failure is a NetworkError, an index that cannot be used + IndexInvalid; both are RegistryErrors, and either way nothing is kept.""" + opener = opener or urllib.request.urlopen + try: + with opener(url, timeout=timeout) as response: + body = response.read() + except (OSError, ValueError) as e: # HTTPError is an OSError + raise NetworkError(f"could not fetch {url}: {e}") from e + try: + data = json.loads(body.decode()) + except ValueError as e: + raise NetworkError(f"{url} did not serve JSON: {e}") from e + return data, parse_index(data) diff --git a/jetlink_repo/jetlink/registry/lfs.py b/jetlink_repo/jetlink/registry/lfs.py index f474445e21..a608787172 100644 --- a/jetlink_repo/jetlink/registry/lfs.py +++ b/jetlink_repo/jetlink/registry/lfs.py @@ -4,27 +4,14 @@ Copyright (c) 2026-, Zeph Leggett. This file is part of jetlink and is licensed under the MIT License. See the LICENSE file in the root directory for more details. -Getting a large model's ONNX by its git-lfs oid. +Getting a large model's ONNX by its oid, the one the model index names +(jetlink.registry.index). -comma overwrites one file per model, so the commit is the only name a big -model's ONNX has. GitHub's raw host serves the LFS pointer for any commit it -holds, merged or not, and that pointer carries the oid and size the comma will -ask a jetlink server for. The bytes themselves are on comma's LFS servers, -which is GitLab and not GitHub: each is asked in turn because which one has an -object varies with the model's age. - -Since openpilot #38930 a commit can ship a precompiled tinygrad -pkl instead, with no ONNX in the tree at all; Cinque Terre V3 is one. Its -subject names the export ("Use f78ed37d for the precompiled eGPU driving -model") and the ONNX is in comma's HuggingFace model repo, in the folder that -id starts. sunnypilot's model builds find it the same way. That repo speaks -the LFS batch protocol too, so it is one more endpoint to ask. - -Not every such commit names its export. ResAction's pull request (#39037) -adds the ONNX in one commit and replaces it with the pkl in the -next, subject "compiled", and a squash merge's subject only names the pull -request. The pointer is still in a diff: the one the commit's own patch -deletes, or the last one its pull request's patch carries. +The bytes are on comma's LFS servers. Since openpilot #39206 comma's own LFS +client stores objects in a HuggingFace bucket, read by oid; the older LFS +servers, and comma's model repo for the exports behind a precompiled pkl, +speak the LFS batch protocol, and which one has an object varies with the +model's age, so each is asked in turn. The zoompilot fork downloads through this module too, asking its own .lfsconfig endpoint first. No openpilot imports. @@ -34,49 +21,29 @@ from __future__ import annotations import hashlib import json import logging -import re import shutil -import urllib.parse import urllib.request from collections.abc import Callable -from dataclasses import dataclass from pathlib import Path -from jetlink.registry.catalog import NetworkError, NotFound, RegistryError, VerifyError, http_get, http_json, is_sha256 +from jetlink.registry.index import NetworkError, Pointer, RegistryError, VerifyError + +__all__ = ['BUCKET', 'LFS_ENDPOINTS', 'Pointer', 'lfs_download', 'lfs_hrefs', 'lfs_resolve'] log = logging.getLogger('jetlink.registry') -BIG_ONNX = 'big_driving_supercombo.onnx' -POINTER_URL = 'https://raw.githubusercontent.com/commaai/openpilot/{ref}/openpilot/selfdrive/modeld/models/' + BIG_ONNX -# a commit's subject without the API, whose anonymous limit a car behind CGNAT shares -COMMIT_PATCH_URL = 'https://github.com/commaai/openpilot/commit/{ref}.patch' -# every commit of a pull request, one patch after another, for a squash merge's subject -PULL_PATCH_URL = 'https://github.com/commaai/openpilot/pull/{number}.patch' -# how much of a patch is read for the pointers in its diffs; a model commit's is 3 KB -PATCH_MAX = 256 << 10 -# the diff header of the big ONNX, wherever the tree keeps selfdrive -_ONNX_DIFF = re.compile(r'^diff --git a/\S*selfdrive/modeld/models/' + re.escape(BIG_ONNX) + r' ') -# a squash merge's subject ends with its pull request -_PULL = re.compile(r'\(#([0-9]+)\)$') +# where comma's LFS client keeps objects since openpilot #39206. A HuggingFace +# bucket is not an LFS server: an object is read by its oid under it (lfs_resolve) +BUCKET = 'https://huggingface.co/buckets/commaai/openpilot-lfs' DRIVING_MODELS_REPO = 'commaai/openpilot_driving_models' -DRIVING_MODELS_TREE_URL = f'https://huggingface.co/api/models/{DRIVING_MODELS_REPO}/tree/main' LFS_ENDPOINTS = ( + BUCKET, 'https://gitlab.com/commaai/openpilot-lfs.git/info/lfs', # every object, older and PR-branch models included - 'https://huggingface.co/commaai/openpilot-lfs.git/info/lfs', # where comma is moving them; the current ones + 'https://huggingface.co/commaai/openpilot-lfs.git/info/lfs', # the current ones f'https://huggingface.co/{DRIVING_MODELS_REPO}.git/info/lfs', # the exports behind a precompiled pkl ) -# an export folder is a uuid; the subject names its first eight -_EXPORT_ID = re.compile(r'\b[0-9a-f]{8}\b') -# how comma's model commits name an export: the whole uuid, or its first eight -# and the checkpoint ('8d736683/12864'). A bare eight could be any hex word -_NAMED_EXPORT = r'\b{}[-/]' LFS_MEDIA_TYPE = 'application/vnd.git-lfs+json' -POINTER_TIMEOUT = 10.0 CONNECT_TIMEOUT = 30.0 -# A pointer is 134 bytes. Anything larger is the ONNX itself, served by a host -# that resolved the LFS filter for us, and reading a gigabyte to find that out -# is not on. -POINTER_MAX = 4096 CHUNK = 4 << 20 FREE_SLACK = 64 << 20 @@ -84,190 +51,30 @@ ProgressFn = Callable[[float], None] StopFn = Callable[[], bool] -@dataclass(frozen=True) -class Pointer: - oid: str # the ONNX SHA-256 - size: int - - -def parse_pointer_text(text: str) -> Pointer | None: - """The oid and size in a git-lfs pointer's text, or None if it is not one.""" - if not isinstance(text, str) or len(text.encode('utf-8', 'replace')) > POINTER_MAX: - return None - oid = None - size = None - for line in text.splitlines(): - key, _, value = line.partition(' ') - if key == 'oid': - oid = value.removeprefix('sha256:').strip() - elif key == 'size': - try: - size = int(value) - except ValueError: - return None - if not is_sha256(oid) or size is None or size <= 0: - return None - return Pointer(oid, size) - - -def fetch_pointer(ref: str, timeout: float = POINTER_TIMEOUT, opener=None) -> Pointer: - """The oid and size of the ONNX at a comma commit: in its tree, or for a - commit that ships a precompiled pkl instead, the ONNX it was built from.""" +def _bucket_href(bucket: str, pointer: Pointer, timeout: float, opener) -> str | None: + """A bucket's address for the object, or None when it does not have it or + cannot be reached: its first byte is asked for, as cheap as a HEAD.""" + url = f"{bucket}/resolve/sha256/{pointer.oid}" try: - text = http_get(POINTER_URL.format(ref=ref), timeout, opener, POINTER_MAX).decode('utf-8', 'replace') - except NotFound: - return fetch_export_pointer(ref, timeout=timeout, opener=opener) - pointer = parse_pointer_text(text) - if pointer is None: - raise RegistryError(f"{ref[:10]} did not serve an lfs pointer") - return pointer - - -def patch_subjects(text: str) -> list[str]: - """Every commit's subject in a patch, in order: one for a commit's, one per - commit for a pull request's.""" - lines = text.splitlines() - subjects = [] - for i, line in enumerate(lines): - if line.startswith('Subject:'): - subject = [line.removeprefix('Subject:').strip()] - # a long subject is folded onto indented lines; the headers end at a blank one - for cont in lines[i + 1:]: - if not cont[:1].isspace() or not cont.strip(): - break - subject.append(cont.strip()) - subjects.append(re.sub(r'^\[PATCH[^\]]*\]\s*', '', ' '.join(subject))) - return subjects - - -def diff_pointer(text: str) -> Pointer | None: - """The big ONNX's pointer in a patch's diffs, the last one wins: the pointer a - diff gives the file, or for a diff that deletes it, the pointer it had. A - commit that swaps the ONNX for a pkl deletes the ONNX the pkl was built from.""" - found = None - section: list[str] | None = None - - def close(): - nonlocal found - if section is not None: - added = parse_pointer_text('\n'.join(x[1:] for x in section if x.startswith('+') and not x.startswith('+++'))) - removed = parse_pointer_text('\n'.join(x[1:] for x in section if x.startswith('-') and not x.startswith('---'))) - found = added or removed or found - - for line in text.splitlines(): - # each diff, and each commit of a pull request's patch, ends the one before - if line.startswith(('diff --git ', 'From ')): - close() - section = [] if _ONNX_DIFF.match(line) else None - elif section is not None: - section.append(line) - close() - return found - - -def commit_subject(ref: str, timeout: float = POINTER_TIMEOUT, opener=None) -> str: - """A comma commit's subject line, from the head of its patch.""" - url = COMMIT_PATCH_URL.format(ref=ref) - subjects = patch_subjects(http_get(url, timeout, opener, POINTER_MAX).decode('utf-8', 'replace')) - if not subjects: - raise RegistryError(f"{url} has no subject line") - return subjects[0] - - -def _tree(path: str, timeout: float, opener) -> list[dict]: - url = DRIVING_MODELS_TREE_URL + (f"/{urllib.parse.quote(path)}?recursive=true" if path else '') - return [e for e in http_json(url, timeout, opener, list) if isinstance(e, dict) and isinstance(e.get('path'), str)] - - -def fetch_export_pointer(ref: str, timeout: float = POINTER_TIMEOUT, opener=None) -> Pointer: - """The big ONNX a precompiled-pkl commit was built from. In turn: the export - its subject names, in comma's model repo; the pointer its own diff deletes; - and for a squash merge, the last pointer its pull request carried, or the - export one of that pull request's commits names. - - The newest export named wins: a pull request's commits are in order, and - Mountain Dew's (#39047) named three before the one it shipped. One that is - not in the model repo yet stops the search rather than falling back to an - older one, a different model.""" - url = COMMIT_PATCH_URL.format(ref=ref) - patch = http_get(url, timeout, opener, PATCH_MAX).decode('utf-8', 'replace') - subjects = patch_subjects(patch) - if not subjects: - raise RegistryError(f"{url} has no subject line") - subject = subjects[0] - ids = _newest_first(subject) - pointer, missing = _export_pointer(ref, ids, subject, timeout, opener) - if pointer is not None: - return pointer - tried = [f"its subject {_not_uploaded(missing)}" if missing else f"no folder in {DRIVING_MODELS_REPO} for {', '.join(ids)}" if ids - else 'its subject names no export'] - if (pointer := diff_pointer(patch)) is not None: - log.info("%s deletes %s %s", ref[:10], BIG_ONNX, pointer.oid[:16]) - return pointer - tried.append(f"its diff has no {BIG_ONNX}") - if (pull := _PULL.search(subject)) is not None: - number = pull.group(1) - pull_patch = http_get(PULL_PATCH_URL.format(number=number), timeout, opener, PATCH_MAX).decode('utf-8', 'replace') - if (pointer := diff_pointer(pull_patch)) is not None: - log.info("%s merged #%s, whose last %s is %s", ref[:10], number, BIG_ONNX, pointer.oid[:16]) - return pointer - if missing is None: # an export the subject names and the repo lacks stops the search here too - pull_subjects = ' '.join(patch_subjects(pull_patch)) - pull_ids = [i for i in _newest_first(pull_subjects) if i not in ids] - pointer, pull_missing = _export_pointer(ref, pull_ids, pull_subjects, timeout, opener) - if pointer is not None: - return pointer - tried.append(f"pull request #{number} {_not_uploaded(pull_missing)}" if pull_missing else f"nor does pull request #{number}") - raise RegistryError(f"{ref[:10]} has no {BIG_ONNX}: {'; '.join(tried)} ({subject!r})") - - -def _newest_first(text: str) -> list[str]: - """The eight-hex ids `text` names, the last mention first.""" - return list(dict.fromkeys(reversed(_EXPORT_ID.findall(text)))) - - -def _not_uploaded(export: str) -> str: - return f"names export {export}, which is not in {DRIVING_MODELS_REPO} yet" - - -def _export_pointer(ref: str, ids: list[str], subject: str, timeout: float, opener) -> tuple[Pointer | None, str | None]: - """The ONNX in the first export folder one of `ids` starts, and None; or - None and the first id `subject` names as an export that has no folder, where - the search stops; or None and None when no id has a folder.""" - if not ids: - return None, None - folders = [e['path'] for e in _tree('', timeout, opener) if e.get('type') == 'directory'] - for export in ids: - matches = [f for f in folders if f.startswith(export)] - if not matches: - if re.search(_NAMED_EXPORT.format(export), subject): - return None, export - continue - if len(matches) > 1: - raise RegistryError(f"{ref[:10]}: {export} starts {len(matches)} folders in {DRIVING_MODELS_REPO}") - files = [e for e in _tree(matches[0], timeout, opener) - if e.get('type') == 'file' and e['path'].rsplit('/', 1)[-1] == BIG_ONNX] - if len(files) > 1: - # a subject that names the checkpoint, as '1a421175-.../12864' does, picks one - files = [f for f in files if f['path'].rsplit('/', 1)[0] in subject] or files - if len(files) != 1: - raise RegistryError(f"{ref[:10]}: {len(files)} copies of {BIG_ONNX} under {matches[0]}") - lfs = files[0].get('lfs') or {} - oid, size = lfs.get('oid'), lfs.get('size') - if not is_sha256(oid) or not isinstance(size, int) or size <= 0: - raise RegistryError(f"{files[0]['path']} is not an lfs object") - log.info("%s names export %s: %s", ref[:10], export, files[0]['path']) - return Pointer(oid, size), None - return None, None + with opener(urllib.request.Request(url, headers={'Range': 'bytes=0-0'}), timeout=timeout) as response: + response.read(1) + except (OSError, ValueError) as e: + log.warning("%s has no %s (%s)", bucket, pointer.oid[:16], e) + return None + return url def lfs_resolve(endpoint: str, pointer: Pointer, timeout: float = CONNECT_TIMEOUT, opener=None) -> str | None: """Ask one LFS server for a download href, or None if it does not have it. + A HuggingFace bucket, comma's (BUCKET) or one a checkout's .lfsconfig + names, is asked for the object itself. A server that is down is not different from a server that lacks the object: either way the caller moves to the next one, so nothing raises here. """ opener = opener or urllib.request.urlopen + if '/buckets/' in endpoint: + return _bucket_href(endpoint, pointer, timeout, opener) body = json.dumps({ 'operation': 'download', 'transfers': ['basic'], @@ -302,28 +109,6 @@ def lfs_hrefs(endpoints, pointer: Pointer, timeout: float = CONNECT_TIMEOUT, ope yield endpoint, href -def fetch_size(oid: str, endpoints=LFS_ENDPOINTS, timeout: float = POINTER_TIMEOUT, opener=None) -> Pointer: - """The pointer for an ONNX known by its oid alone, as a fork whose catalog - names a model by its hash knows it (StarPilot's). comma's LFS servers hand - out a download address for any oid they hold, whatever size the batch asks - for, and the first byte of that download says how many there are. No - server with it is NotFound; one that would not say, a NetworkError.""" - opener = opener or urllib.request.urlopen - for endpoint, href in lfs_hrefs(endpoints, Pointer(oid, 1), timeout=timeout, opener=opener): - request = urllib.request.Request(href, headers={'Range': 'bytes=0-0'}) - try: - with opener(request, timeout=timeout) as response: - # a server that ignores the range answers 200 with the whole length - total = (response.headers.get('Content-Range') or '').rpartition('/')[2] \ - if response.status == 206 else response.headers.get('Content-Length') - except (OSError, ValueError) as e: - raise NetworkError(f"could not size {oid[:16]} at {endpoint}: {e}") from e - if not str(total).isdigit() or int(total) <= 0: - raise NetworkError(f"{endpoint} did not say how big {oid[:16]} is") - return Pointer(oid, int(total)) - raise NotFound(f"no LFS server has {oid[:16]}") - - def lfs_download(href: str, pointer: Pointer, dest: Path, progress: ProgressFn | None = None, should_stop: StopFn | None = None, opener=None) -> Path: """Stream to a .part file, hashing as we go, and only then take the name. diff --git a/openpilot/starpilot/models/JETLINK.md b/openpilot/starpilot/models/JETLINK.md index 1b6a55c60d..db1d5a2b1f 100644 --- a/openpilot/starpilot/models/JETLINK.md +++ b/openpilot/starpilot/models/JETLINK.md @@ -28,14 +28,14 @@ release still needs a successful compatible handshake and device validation. ### Jetson and Linux PC -On a supported NVIDIA host, review and run the [upstream installer](https://github.com/zoompilot/jetlink/blob/v0.8.12/install.sh). +On a supported NVIDIA host, review and run the [upstream installer](https://github.com/zoompilot/jetlink/blob/v0.9.0/install.sh). The upstream Jetson guide specifies an Orin Nano Super 8 GB and JetPack 7.2.1; the Linux guide specifies an RTX 20-series or newer GPU with driver 580 or newer. Keep the computer powered, cool, and awake. Connect its USB-A port to the comma's USB-C port. Use `jetlink status` to check the host and `jetlink logs` to inspect problems. Host power, sleep, and driver setup details remain in the credited -[Jetson](https://github.com/zoompilot/jetlink/blob/v0.8.12/docs/jetson.md) and -[Linux](https://github.com/zoompilot/jetlink/blob/v0.8.12/docs/linux-pc.md) references; +[Jetson](https://github.com/zoompilot/jetlink/blob/v0.9.0/docs/jetson.md) and +[Linux](https://github.com/zoompilot/jetlink/blob/v0.9.0/docs/linux-pc.md) references; their comma-side ZoomPilot instructions are replaced by **Set up StarPilot** above. ### Mac @@ -45,7 +45,7 @@ Use an Apple silicon Mac with macOS 15 or newer. Get the Jetlink DMG from in Applications, and start it. Keep it awake and connect by USB 3. Use **USB computer** in StarPilot's Galaxy, rather than following the comma-side ZoomPilot installation instructions in the credited -[Mac host reference](https://github.com/zoompilot/jetlink/blob/v0.8.12/docs/macos-app.md). +[Mac host reference](https://github.com/zoompilot/jetlink/blob/v0.9.0/docs/macos-app.md). ### iPhone and iPad @@ -54,7 +54,7 @@ It requires a USB-C device supporting USB 3 and iOS/iPadOS 26.1 or newer. Choose **iPhone / iPad** in StarPilot, allow the host app's Local Network access, connect by USB 3, and keep the app open and the device unlocked. Check sustained thermal performance with the app's benchmark; that benchmark excludes cable latency. -The [credited Apple host reference](https://github.com/zoompilot/jetlink/blob/v0.8.12/docs/iphone-app.md) +The [credited Apple host reference](https://github.com/zoompilot/jetlink/blob/v0.9.0/docs/iphone-app.md) contains hardware details. StarPilot supplies the comma-side instructions above. ### Android @@ -64,7 +64,7 @@ upstream Jetlink Android APK from [its releases](https://github.com/zoompilot/je Use a recent USB 3-capable Android 12+ phone, a powered USB 3 hub, and separate comma power. Choose **USB computer** in StarPilot. Follow the host's USB permission prompt and benchmark sustained performance before use. -[Credited Android host reference](https://github.com/zoompilot/jetlink/blob/v0.8.12/docs/android-app.md). +[Credited Android host reference](https://github.com/zoompilot/jetlink/blob/v0.9.0/docs/android-app.md). ## Use and recovery @@ -89,7 +89,7 @@ and defaults off. ## Attribution and license This StarPilot guide adapts the host setup structure from -[ZoomPilot / Jetlink at v0.8.12](https://github.com/zoompilot/jetlink/tree/v0.8.12/docs). +[ZoomPilot / Jetlink at v0.9.0](https://github.com/zoompilot/jetlink/tree/v0.9.0/docs). Jetlink is copyright (c) 2026 Zeph Leggett and licensed under the [MIT license](../../../jetlink_repo/LICENSE), whose full text is preserved. ZoomPilot source links, project names, authorship, and history are attribution; diff --git a/upstream-sync.json b/upstream-sync.json index c59e3619d9..ac5acde7e8 100644 --- a/upstream-sync.json +++ b/upstream-sync.json @@ -65,8 +65,8 @@ { "path": "jetlink_repo", "url": "https://github.com/zoompilot/jetlink.git", - "commit": "a863c3801da6fd925d14eb1f5b1e250f8efcc649", - "tree": "70cabdd689e2f1d95f31c95ede0c7990b41e550c", + "commit": "9cfde98eec1d7cc714d800504944cf1a3944513d", + "tree": "dbcb53149dbc8ae5df9d8236deeb216f2493a5f9", "include": [ "jetlink", "LICENSE",