From 21c34c8f4daddffb6a4053d988ee1f1cd7e77ab6 Mon Sep 17 00:00:00 2001 From: Zeph Leggett Date: Thu, 29 Oct 2026 15:45:41 -0500 Subject: [PATCH] Jetlink v0.8.12 --- CREDITS.md | 2 +- jetlink_repo/jetlink/__init__.py | 2 +- jetlink_repo/jetlink/client.py | 26 ++- jetlink_repo/jetlink/comma/lending.py | 2 +- jetlink_repo/jetlink/openpilot/interface.py | 12 +- jetlink_repo/jetlink/openpilot/link.py | 36 +++- jetlink_repo/jetlink/openpilot/models.py | 224 +++++++++++++++++--- jetlink_repo/jetlink/openpilot/provision.py | 85 +++++--- jetlink_repo/jetlink/openpilot/status.py | 16 +- jetlink_repo/jetlink/registry/lfs.py | 30 +++ openpilot/starpilot/models/JETLINK.md | 14 +- upstream-sync.json | 4 +- 12 files changed, 357 insertions(+), 96 deletions(-) diff --git a/CREDITS.md b/CREDITS.md index e909915393..63ab91f273 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 `bf4dc940e98f5d3ed9b7163adb021439010d634f` 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 `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. ## MRR35 Hyundai angle and radar diff --git a/jetlink_repo/jetlink/__init__.py b/jetlink_repo/jetlink/__init__.py index cd1741435a..5458975a61 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.9' +__version__ = '0.8.12' diff --git a/jetlink_repo/jetlink/client.py b/jetlink_repo/jetlink/client.py index 09b6d9617b..22fe81ea62 100644 --- a/jetlink_repo/jetlink/client.py +++ b/jetlink_repo/jetlink/client.py @@ -224,12 +224,16 @@ class JetlinkClient: # -- handshake ------------------------------------------------------------ - def hello(self, timeout: float = 5.0) -> dict: + def hello(self, timeout: float = 5.0, pick: dict | None = None) -> dict: """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.""" + 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).""" seq = self._next_seq() client = {'nonce': self.nonce, 'name': self.name} + if pick is not None: + client['pick'] = pick try: link = self.t.link_info() except Exception: @@ -301,7 +305,8 @@ class JetlinkClient: frame_skip: int = DEFAULT_FRAME_SKIP, progress: ProgressFn | None = None, build_timeout: float = 900.0, - should_stop: StopFn | None = None) -> ModelSpec: + should_stop: StopFn | None = None, + fetch: bool = False, ref: str | None = None) -> ModelSpec: """Make the server ready to run this model, uploading and building if needed. Blocks until ready or the build fails, and returns the spec the server @@ -309,8 +314,12 @@ class JetlinkClient: parser rejects an export (tinygrad and the org.tinygrad domain) still runs. `onnx_path` None means the caller cannot upload (modeld) and a missing - engine is EngineMissing. `progress(stage, frac, msg)` has stage in - upload/patch/parse/build/load. `should_stop` is polled during long waits. + engine is EngineMissing. `fetch` asks a server that lacks the model to + download it itself (state downloading, stage download) rather than answer + need_upload; one that cannot, or is too old to, answers need_upload as + before. `ref` is the catalog ref the model is known by. + `progress(stage, frac, msg)` has stage in + download/upload/patch/parse/build/load. `should_stop` is polled during long waits. """ self.progress_cb = progress self._should_stop = should_stop or (lambda: False) @@ -318,7 +327,12 @@ class JetlinkClient: self._engine_state = None seq = self._next_seq() - self.t.send_json(P.Msg.ENGINE_REQ, seq, {'sha256': sha256, 'nbytes': nbytes, 'frame_skip': frame_skip}) + request = {'sha256': sha256, 'nbytes': nbytes, 'frame_skip': frame_skip} + if fetch: + request['fetch'] = True + if ref: + request['ref'] = ref + self.t.send_json(P.Msg.ENGINE_REQ, seq, request) resp = json.loads(bytes(self._expect(P.Msg.ENGINE_RESP, seq, 60.0).payload)) self._engine_state = resp log.info("server engine state: %s (%s)", resp['state'], resp.get('detail', '')) diff --git a/jetlink_repo/jetlink/comma/lending.py b/jetlink_repo/jetlink/comma/lending.py index 3102e7b8a5..1b7f5dce81 100644 --- a/jetlink_repo/jetlink/comma/lending.py +++ b/jetlink_repo/jetlink/comma/lending.py @@ -55,7 +55,7 @@ POLL = 0.5 # what a borrower passes on of the server's hello (Loan.note_server): the owner # never speaks the protocol, and sleep_after is how it knows whether letting # go of the gadget lets the Jetson sleep. The rest is for its status record -SERVER_FIELDS = ('protocol', 'device', 'backend', 'runtime_version', 'trt_version', 'sleep_after') +SERVER_FIELDS = ('protocol', 'device', 'backend', 'runtime_version', 'trt_version', 'sleep_after', 'pick', 'fetches') def _send(conn: socket.socket, msg: dict) -> None: diff --git a/jetlink_repo/jetlink/openpilot/interface.py b/jetlink_repo/jetlink/openpilot/interface.py index 7c917d9f5b..f0fa5e135a 100644 --- a/jetlink_repo/jetlink/openpilot/interface.py +++ b/jetlink_repo/jetlink/openpilot/interface.py @@ -45,15 +45,23 @@ 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 -> {oid, size} + 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 + # jetlink runs its own default big model, or the one picked on the server. + # A fork with a picker of its own and no catalog (StarPilot) declares only + # big_model, and may name its pick by the ONNX's hash, {sha256, + # displayName}; a newer pick made on the server is written into it big_model: str | None = None catalog: str | None = None # BOOL: an iPhone on a direct cable is asked to charge from the comma # (comma.port). Off when unset, and without the key charge_phone: str | None = None + # STRING: where the model manager takes a pick by its ref and stores it in + # big_model (sunnypilot's ModelManager_DownloadRef). A newer pick made on + # 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 @dataclass(frozen=True) diff --git a/jetlink_repo/jetlink/openpilot/link.py b/jetlink_repo/jetlink/openpilot/link.py index d8ec1b7b48..7b3e6cbe51 100644 --- a/jetlink_repo/jetlink/openpilot/link.py +++ b/jetlink_repo/jetlink/openpilot/link.py @@ -43,6 +43,8 @@ PRESENT_TIMEOUT = 5.0 # hardware; the ceiling is only there so a server that has stopped answering # does not hold the caller for the rest of the day. BUILD_TIMEOUT = 1800.0 +# and on top of that, a server downloading the model itself: 1.75 GB at 1 MB/s +FETCH_TIMEOUT = 1800.0 class ModelMissing(RuntimeError): @@ -226,12 +228,15 @@ 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.""" + provisioning that needs the internet. A model picked by its hash alone is + only sized.""" 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') - return parts.models.resolve_pointer(entry['ref']) + if entry.get('ref'): + return parts.models.resolve_pointer(entry['ref']) + return parts.models.resolve_size(sha256) # What has already been hashed in this process, keyed on the file as it was @@ -278,21 +283,27 @@ def verified_upload(log, model_path: Path | None, sha256: str, nbytes: int) -> P def ensure(parts, client, sha256: str, nbytes: int, model_path: Path | None, *, - progress=None, should_stop=None, build_timeout: float = BUILD_TIMEOUT): + progress=None, should_stop=None, build_timeout: float = BUILD_TIMEOUT, ref: str | None = None, + fetch: bool = True): """Make the server ready for this model and remember what it answered. Asks without the file first: the server answers from the sha alone when it - has the model, which is every parked poll and every join. EngineMissing means - the Jetson has neither the plan nor the bytes, and neither has the caller.""" + has the model, which is every parked poll and every join. Without the file + here either, the server is asked to download the model itself, which only + the Jetson needs. EngineMissing means the Jetson has neither the plan nor + the bytes and could not get them, and neither has the caller. `fetch` False + never asks for the download: a stand-in is a model the server has. `ref`, + the catalog's name for the model, goes with a download.""" from jetlink.client import EngineMissing - ask = functools.partial(client.ensure_engine, sha256, nbytes, progress=progress, - build_timeout=build_timeout, should_stop=should_stop) + fetch = fetch and model_path is None + ask = functools.partial(client.ensure_engine, sha256, nbytes, progress=progress, should_stop=should_stop) try: - spec = ask(onnx_path=None) + spec = ask(onnx_path=None, fetch=fetch, ref=ref if fetch else None, + build_timeout=build_timeout + (FETCH_TIMEOUT if fetch else 0.0)) except EngineMissing: upload = verified_upload(parts.log, model_path, sha256, nbytes) if upload is None: raise - spec = ask(onnx_path=upload) + spec = ask(onnx_path=upload, build_timeout=build_timeout) parts.spec.store(spec) return spec @@ -346,11 +357,14 @@ def open_link(parts, link: Link, should_stop=None): # booting; both resolve on their own client = connect_patiently(link) try: - hello = client.hello(timeout=10.0) + hello = parts.models.heard(client.hello(timeout=10.0, pick=parts.models.hello_pick())) 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]) link.note_server(hello) + # a newer pick on the server drives, and is ours from now on + parts.models.adopt() + selected = parts.models.selected_model() or selected sha256, nbytes = identity(parts, selected) path = parts.models.shipped_model_path() built = parts.spec.ready_spec() @@ -370,7 +384,7 @@ def open_link(parts, link: Link, should_stop=None): # server that restarted reloads from the plan cache, 13 to 25 s; one # that has never seen this model builds it, 102 to 294 s spec = ensure(parts, client, sha256, nbytes, path, progress=parts.progress.report_with_eta, - should_stop=should_stop) + should_stop=should_stop, ref=selected.get('ref'), fetch=standin is None) client.deadline = INFERENCE_TIMEOUT return client, spec except EngineMissing as e: diff --git a/jetlink_repo/jetlink/openpilot/models.py b/jetlink_repo/jetlink/openpilot/models.py index 336a720fb8..a95b4adace 100644 --- a/jetlink_repo/jetlink/openpilot/models.py +++ b/jetlink_repo/jetlink/openpilot/models.py @@ -16,16 +16,25 @@ export, and jetlink's registry follows that to comma's model repo; see jetlink.registry.lfs. Without a model manager (Keys.big_model and Keys.catalog None) there is no -pick and no catalog, and the large model is jetlink's own default, named by its -ref: its pointer resolves from the ref as any pick's does. +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. + +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. """ from __future__ import annotations +import os import re import time from pathlib import Path -from jetlink.registry.catalog import DEFAULT_BIG_MODEL_NAME, DEFAULT_BIG_MODEL_REF +from jetlink.registry.catalog import DEFAULT_BIG_MODEL_NAME, DEFAULT_BIG_MODEL_REF, is_ref, is_sha256 POINTER_TIMEOUT = 10.0 _REF = re.compile(r'[0-9a-f]{40}') @@ -39,13 +48,32 @@ INDEX_TTL = 2.0 # on from the .part the last left, so a stall costs the wait, not the bytes # (lfs.lfs_download) DOWNLOAD_RETRY_DELAYS = (5.0, 15.0, 30.0, 60.0, 120.0) +# server_pick() before this process has heard a hello +_UNHEARD = object() + + +def _dated(pick) -> dict | None: + """A server's pick as its hello carries it, dated by this clock: + {ref, sha256, nbytes, name, at}. None for none, or one naming no model.""" + if not isinstance(pick, dict) or not isinstance(pick.get('age_s'), (int, float)): + return None + ref = pick['ref'] if is_ref(pick.get('ref')) else None + sha256 = pick['sha256'] if is_sha256(pick.get('sha256')) else None + if ref is None and sha256 is None: + return None + return {'ref': ref, 'sha256': sha256, 'nbytes': pick.get('nbytes'), 'name': pick.get('name'), + 'at': time.time() - float(pick['age_s'])} class Models: def __init__(self, op): self.op = op self._index_cache: tuple[float, list[dict], dict[str, dict]] | None = None - self._slot_cache: tuple[float, dict | None] | None = None + self._slot_cache: tuple[float, dict | None, float | None] | 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 @@ -77,13 +105,28 @@ class Models: 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 ref in known: - return known[ref]['oid'], int(known[ref]['size']) - oid, size = self.fetch_pointer(ref) - known[ref] = {'oid': oid, 'size': size} + 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", ref[:10], oid[:16], size >> 20) + 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: @@ -92,9 +135,11 @@ class Models: self._index_cache = None @staticmethod - def _row(name: str, ref: str, known: dict[str, dict]) -> dict: - p = known.get(ref) or {} - return {'name': name, 'ref': ref, 'oid': p.get('oid'), 'size': int(p['size']) if p.get('size') else None} + 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.""" @@ -110,18 +155,27 @@ class Models: return self._index()[0] def selected_slot(self) -> dict | None: - """The big-model slot's pick as {name, ref}, from the raw param rather than a - parsed bundle, and memoised: the UI asks every frame.""" + """The big-model slot's pick as {name, ref, sha256}, from the raw param + rather than a parsed bundle, and memoised with when it was made + (picked_at): the UI asks every frame. It names its model by the catalog's + ref, by the ONNX's sha256 (a fork whose catalog knows only that), or by + both; the one it lacks is None.""" now = time.monotonic() if self._slot_cache is not None and now - self._slot_cache[0] < INDEX_TTL: return self._slot_cache[1] key = self.op.keys.big_model slot = self.op.get(key) if key else None - ref = slot.get('ref') if isinstance(slot, dict) else None - pick = None - if isinstance(ref, str) and ref: - pick = {'name': str(slot.get('displayName') or ref[:10]), 'ref': ref} - self._slot_cache = (now, pick) + slot = slot if isinstance(slot, dict) else {} + ref = slot['ref'] if isinstance(slot.get('ref'), str) and slot['ref'] else None + sha256 = slot['sha256'] if is_sha256(slot.get('sha256')) else None + pick, at = None, None + if ref or sha256: + pick = {'name': str(slot.get('displayName') or (ref or sha256)[:10]), 'ref': ref, 'sha256': sha256} + try: + at = os.stat(Path(self.op.params_dir()) / key).st_mtime + except OSError: + pass + self._slot_cache = (now, pick, at) return pick def default_model(self) -> dict | None: @@ -142,12 +196,129 @@ class Models: return _TRAILING_DATE.sub('', model['name']) if model else None def selected_model(self) -> dict | None: - """The slot's pick, listed in the catalog or not: a ref is enough to find its - pointer. With no pick, default_model().""" - if (pick := self.selected_slot()) is not None: - return self._row(pick['name'], pick['ref'], self._index()[1]) + """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().""" + 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_row(self, own: dict) -> dict: + return self._row(own['name'], own['ref'], self._index()[1], 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. + + def picked_at(self) -> float | None: + """When the slot was last written, by this clock: when the pick was made. + None with no pick of our own, or no date for it.""" + 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: + 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)} + + 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.""" + self._heard = _dated(hello.get('pick')) + return {**hello, 'pick': self._heard} if self._heard is not None else hello + + def server_pick(self) -> dict | None: + """The pick the server last answered a hello with, {ref, sha256, nbytes, + name, at}, or None: a server too old to say, or one with no pick. What + this process heard last, else the owner's record of the last hello: the + UI says no hello, and reads it every frame.""" + if self._heard is not _UNHEARD: + return self._heard + now = time.monotonic() + if self._record_cache is None or now - self._record_cache[0] >= INDEX_TTL: + from jetlink.comma import gadget + pick = (gadget.read_record(gadget.SERVER) or {}).get('pick') + self._record_cache = (now, pick if isinstance(pick, dict) and isinstance(pick.get('at'), (int, float)) else None) + return self._record_cache[1] + + 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))): + return None + return 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']: + return pick['ref'] == row['ref'] + return bool(pick['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.""" + 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']) + 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.""" + 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: + return + model, keys = self._server_row(theirs), self.op.keys + if keys.catalog is None: + if keys.big_model: + self.op.log.warning("jetlink: %s was picked on the server, picking it here too", model['name']) + slot = {k: v for k, v in (('displayName', model['name']), ('ref', model['ref']), ('sha256', model['oid'])) if v} + self.op.put(keys.big_model, slot, block=True) + self._slot_cache = None + return + # 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()): + 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 selected_model_name(self) -> str | None: selected = self.selected_model() return selected['name'] if selected else None @@ -227,11 +398,8 @@ class Models: 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_resolve - for endpoint in self.lfs_endpoints(): - href = lfs_resolve(endpoint, pointer) - if href is None: - continue + 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]}") diff --git a/jetlink_repo/jetlink/openpilot/provision.py b/jetlink_repo/jetlink/openpilot/provision.py index 5d4836ad6a..0c04d908e9 100644 --- a/jetlink_repo/jetlink/openpilot/provision.py +++ b/jetlink_repo/jetlink/openpilot/provision.py @@ -126,10 +126,8 @@ class ProvisioningRun: return path def needs_download(self) -> bool: - """Does the comma have to fetch the picked model before a Jetson can build - it? Only while its engine is not known to be built and the file is not here. - Asked without the Jetson: a download needs only the internet, and a car - whose Jetson goes off with the ignition has no Jetson while parked.""" + """Could the Jetson need the picked model from the comma? Only while its + engine is not known to be built and the file is not here.""" parts = self.parts entry = parts.models.selected_model() if entry is None: @@ -140,12 +138,17 @@ class ProvisioningRun: return parts.models.shipped_model_path() is None def provision(self) -> bool: - """Make the Jetson ready for the selected model; a host is attached. run() - has fetched the file already unless the spec record said the model was - built; a server that lost it since asks for the bytes, and they are - fetched then.""" + """Make the Jetson ready for the picked model; a host is attached. The + hello settles which pick that is, the comma's or a newer one made on the + server. A server without the model downloads it itself; only one that + cannot (no internet, or too old) gets the bytes from the comma, fetched + here then.""" from jetlink.client import EngineMissing parts = self.parts + parts.progress.report('connect', 0.0, 'connecting', code='connecting') + hello = self.hello() + self.log.warning("jetlink: server %s trt %s", hello.get('device'), hello.get('trt_version')) + parts.models.adopt() entry = parts.models.selected_model() if entry is None: # no catalog yet; not an error @@ -154,27 +157,24 @@ class ProvisioningRun: return False sha256, nbytes = link.identity(parts, entry) - # only needed if the server turns out not to have this model; None is a - # legitimate state here, see EngineMissing below + # only needed if the server turns out not to have this model and cannot + # get it; None is a legitimate state here, see EngineMissing below model_path = parts.models.shipped_model_path() self.log.warning("jetlink: provisioning %s (%d MB, sha %s)", entry.get('name', sha256[:16]), nbytes >> 20, sha256[:16]) - parts.progress.report('connect', 0.0, 'connecting', code='connecting') def ensure(path): return link.ensure(parts, self.client, sha256, nbytes, path, progress=parts.progress.report_with_eta, - should_stop=lambda: self.stop) + should_stop=lambda: self.stop, ref=entry.get('ref')) - hello = self.hello() - self.log.warning("jetlink: server %s trt %s", hello.get('device'), hello.get('trt_version')) try: spec = ensure(model_path) except EngineMissing: - # the server has nothing to build from, and neither have we: fetch the - # model and hand it over in this run. The owner lends the link for the - # whole run; leaving after the download would put the build a - # WORKER_BACKOFF later + # the server has nothing to build from and could not download it, and + # neither have we: fetch the model and hand it over in this run. The + # owner lends the link for the whole run; leaving after the download + # would put the build a WORKER_BACKOFF later if model_path is not None or (model_path := self.fetch_model()) is None: raise # a download of minutes can outlast the session; a hello starts a new one @@ -186,16 +186,35 @@ class ProvisioningRun: return True def hello(self) -> dict: - """Say hello, and pass the answer on to the owner, which needs its - sleep_after to decide whether letting go of the gadget is worth what it - costs and never speaks the protocol to learn it.""" - hello = self.client.hello(timeout=10.0) + """Say hello with our pick, and pass the answer on to the owner, which + needs its sleep_after to decide whether letting go of the gadget is worth + 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())) if self.loan is not None: self.loan.note_server(hello) return hello # -- the parked car ------------------------------------------------------- + def without_jetson(self) -> None: + """No Jetson came to the bind: a car whose Jetson is switched with the + ignition has none while parked. Only the Jetson needs the model, and one + that can download it gets it itself when it is on; the comma fetches it + now only for a Jetson that cannot (no internet when it was last here, or + too old to say), so the build can start the moment it is.""" + if not self.has_work(): + return + if self.needs_download(): + if server_fetches(): + self.parts.progress.report('waiting', 0.0, 'the jetson downloads it when it is on', code='waiting_for_host') + return + if self.fetch_model() is None: + return + self.parts.progress.report('waiting', 0.0, 'downloaded, builds when the jetson is on', + code='downloaded_waiting_host') + def has_work(self) -> bool: """Is there a reason to wake the Jetson? Only things the link can fix count.""" spec = self.parts.spec.load() @@ -259,25 +278,21 @@ class ProvisioningRun: self.log.warning("jetlink: no warp built for this camera, nothing to provision for") return True - if not self.has_work(): + # Nothing to build, unless a model was picked on the server since: a + # Jetson awake on the bus is asked, which costs a hello and no wake. Not + # a phone's cable, which is there only once the app dials in + if not self.has_work() and not (gadget.host_attached() and gadget.link_kind(self.parts.settings.mode()) == 'usb'): self.log.warning("jetlink: nothing to provision") return True finished = False try: - # the bytes first while no host is on the cable: a parked car whose - # Jetson is switched with the ignition never has one to wait for. A host - # that is there is asked first, since one with the model built needs - # nothing from the comma; provision() fetches it if the host has not - if self.needs_download() and not gadget.host_attached() and self.fetch_model() is None: - return False if not self.open_link(): return False if not self.wait_for_jetson(): self.log.warning("jetlink: no jetson within %.0f s, leaving it for the next run", WAKE_TIMEOUT) - if self.parts.models.shipped_model_path() is not None: - self.parts.progress.report('waiting', 0.0, 'downloaded, builds when the jetson is on', - code='downloaded_waiting_host') + self.close_link() + self.without_jetson() return False finished = self.provision() except Exception: @@ -288,6 +303,12 @@ class ProvisioningRun: return finished +def server_fetches() -> bool: + """Did the server the comma last spoke to say it downloads a model itself + (its hello's `fetches`, as the owner keeps it)? No for one too old to say.""" + return (gadget.read_record(gadget.SERVER) or {}).get('fetches') is True + + def main(argv: list[str] | None = None) -> None: """One provisioning round over the fork's adapter, as the owner starts it. Exits 0 when nothing is left to do, and 1 when the owner should try again.""" diff --git a/jetlink_repo/jetlink/openpilot/status.py b/jetlink_repo/jetlink/openpilot/status.py index 284854c754..c47ee99cb9 100644 --- a/jetlink_repo/jetlink/openpilot/status.py +++ b/jetlink_repo/jetlink/openpilot/status.py @@ -31,14 +31,17 @@ PROGRESS_MIN_INTERVAL = 0.25 # what a progress record's `code` may be, for a fork that words the line # itself; `msg` stays the English one. `args`, where a code has them: -# downloading and upload {got_mb, total_mb}, download_stalled {retry_s}, -# reconnecting {why: lost, behind, load_failed or bad_output}, build {eta_s}. -# From upload on they are stage names, and the server's stages are its own: +# downloading, download and upload {got_mb, total_mb}, download_stalled +# {retry_s}, reconnecting {why: lost, behind, load_failed or bad_output}, +# build {eta_s}. downloading is the comma fetching the model, download the +# server fetching it itself. From download on they are stage names, and the +# server's stages are its own: # a server can add one, so a fork falls back to `msg` for a code it does not # know. A server message with no stage has no code PROGRESS_CODES = ('waiting_for_host', 'finding_model', 'model_missing', 'ready_waiting_window', 'reconnecting', 'connecting', 'downloading', 'download_stalled', 'download_failed', 'downloaded_waiting_host', - 'engine_ready', 'failed', 'upload', 'patch', 'parse', 'convert', 'compile', 'build', 'load', 'warm') + 'engine_ready', 'failed', 'download', 'upload', 'patch', 'parse', 'convert', 'compile', 'build', + 'load', 'warm') def estimated_build_seconds(size: int) -> int: @@ -104,12 +107,15 @@ class Progress: size. Only the build: the upload reports MB of MB and a connect has nothing to predict.""" args = None - size = self._size() if self._size is not None and stage in ('build', 'upload') else None + size = self._size() if self._size is not None and stage in ('build', 'upload', 'download') else None if size and stage == 'build': left = estimated_build_seconds(size) * max(0.0, 1.0 - frac) msg, args = _eta(left), {'eta_s': round(left)} elif size: args = {'got_mb': int(frac * size) >> 20, 'total_mb': size >> 20} + if stage == 'download' and args: + # the server downloading the model itself: not the comma's own 'downloading' + msg = f"jetson downloading {args['got_mb']} of {args['total_mb']} MB" self.report(stage, frac, msg, code=stage or None, args=args) diff --git a/jetlink_repo/jetlink/registry/lfs.py b/jetlink_repo/jetlink/registry/lfs.py index c2439a8df7..f474445e21 100644 --- a/jetlink_repo/jetlink/registry/lfs.py +++ b/jetlink_repo/jetlink/registry/lfs.py @@ -294,6 +294,36 @@ def lfs_resolve(endpoint: str, pointer: Pointer, timeout: float = CONNECT_TIMEOU return None +def lfs_hrefs(endpoints, pointer: Pointer, timeout: float = CONNECT_TIMEOUT, opener=None): + """Each of `endpoints` that has the object, in order, with its download href.""" + for endpoint in endpoints: + href = lfs_resolve(endpoint, pointer, timeout=timeout, opener=opener) + if href is not None: + 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 ecbd89e2b3..1b6a55c60d 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/bf4dc94/install.sh). +On a supported NVIDIA host, review and run the [upstream installer](https://github.com/zoompilot/jetlink/blob/v0.8.12/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/bf4dc94/docs/jetson.md) and -[Linux](https://github.com/zoompilot/jetlink/blob/bf4dc94/docs/linux-pc.md) references; +[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; 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/bf4dc94/docs/macos-app.md). +[Mac host reference](https://github.com/zoompilot/jetlink/blob/v0.8.12/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/bf4dc94/docs/iphone-app.md) +The [credited Apple host reference](https://github.com/zoompilot/jetlink/blob/v0.8.12/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/bf4dc94/docs/android-app.md). +[Credited Android host reference](https://github.com/zoompilot/jetlink/blob/v0.8.12/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 bf4dc94](https://github.com/zoompilot/jetlink/tree/bf4dc94/docs). +[ZoomPilot / Jetlink at v0.8.12](https://github.com/zoompilot/jetlink/tree/v0.8.12/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 b8150a7037..6dde723c3c 100644 --- a/upstream-sync.json +++ b/upstream-sync.json @@ -65,8 +65,8 @@ { "path": "jetlink_repo", "url": "https://github.com/zoompilot/jetlink.git", - "commit": "bf4dc940e98f5d3ed9b7163adb021439010d634f", - "tree": "81237da0aa4423d8b47c4e18c4b02763afbc1917", + "commit": "a863c3801da6fd925d14eb1f5b1e250f8efcc649", + "tree": "70cabdd689e2f1d95f31c95ede0c7990b41e550c", "include": [ "jetlink", "LICENSE",