mirror of
https://github.com/firestar5683/StarPilot.git
synced 2026-10-28 00:53:45 +08:00
Jetlink runs modeld's driving warp
jetlink runs the driving_warp pickles modeld already ships, so its own warp build, the four jetlink_warp pickles and their QCOM import and validation go.
This commit is contained in:
Zeph Leggett
@@ -18,7 +18,6 @@ if arch not in ('comma_arm64', 'Darwin') and ctypes.util.find_library('LLVM') is
|
||||
|
||||
lenv = env.Clone()
|
||||
lenv.PrependENVPath('PYTHONPATH', Dir('#tinygrad_repo').abspath)
|
||||
lenv.PrependENVPath('PYTHONPATH', Dir('#jetlink_repo').abspath)
|
||||
|
||||
tinygrad_root = env.Dir("#").abspath
|
||||
tinygrad_files = ["#"+x for x in glob.glob(env.Dir("#tinygrad_repo").relpath + "/**", recursive=True, root_dir=tinygrad_root)
|
||||
@@ -84,28 +83,6 @@ for cam_w, cam_h in camera_configs:
|
||||
if laptop_qcom_import:
|
||||
lenv.AlwaysBuild(node)
|
||||
|
||||
# External inference uses a paired current-frame warp, independently of the
|
||||
# historical ordinary driving capture. Keep its QCOM package separate.
|
||||
from tools.laptop_device_build.jetlink_artifacts import CAMERAS, MODEL_SIZES, import_artifact as import_jetlink_artifact
|
||||
jetlink_package = Path(Dir('#.cache/jetlink-qcom-artifact-package').abspath)
|
||||
jetlink_sources = [File('#' + path) for path in ('openpilot/starpilot/models/jetlink_warp.py',
|
||||
'jetlink_repo/jetlink/openpilot/warp.py', 'jetlink_repo/jetlink/openpilot/interface.py')]
|
||||
for cam_w, cam_h in CAMERAS:
|
||||
for model_w, model_h in MODEL_SIZES:
|
||||
jetlink_path = File(f'models/jetlink_warp_{cam_w}x{cam_h}_{model_w}x{model_h}.pkl').abspath
|
||||
cmd = (f'{tg_flags} {mac_brew_string} {taskset}python3 -m jetlink.openpilot.warp '
|
||||
f'--adapter openpilot.starpilot.models.jetlink_warp --camera {cam_w}x{cam_h} '
|
||||
f'--model {model_w}x{model_h} --output {jetlink_path}')
|
||||
if laptop_qcom_import:
|
||||
def import_jetlink(target, source, env):
|
||||
import_jetlink_artifact(jetlink_package, Path(Dir('#').abspath), Path(str(target[0])))
|
||||
action = Action(import_jetlink, ' [JETLINK QCOM IMPORT] $TARGET')
|
||||
else:
|
||||
action = Action(cmd, ' [JETLINK WARP] $TARGET')
|
||||
node = lenv.Command(jetlink_path, tinygrad_files + warp_deps + jetlink_sources + [Value(cmd)], action)
|
||||
if laptop_qcom_import:
|
||||
lenv.AlwaysBuild(node)
|
||||
|
||||
dm_w, dm_h = DM_INPUT_SIZE
|
||||
for cam_w, cam_h in camera_configs:
|
||||
dm_pkl_path = File(f"models/dm_warp_{cam_w}x{cam_h}_tinygrad.pkl").abspath
|
||||
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
@@ -1,24 +0,0 @@
|
||||
"""Current-frame NV12 warp supplied to Jetlink's build API."""
|
||||
|
||||
CAMERA_SIZES = ((1928, 1208), (1344, 760))
|
||||
MODEL_SIZES = ((512, 256), (1024, 512))
|
||||
|
||||
|
||||
def make_warp(cam_w, cam_h, model_w, model_h):
|
||||
from examples.openpilot.compile_warp import NV12Frame, make_frame_prepare
|
||||
from tinygrad import Tensor
|
||||
from openpilot.system.camerad.cameras.nv12_info import get_nv12_info
|
||||
|
||||
if (cam_w, cam_h) not in CAMERA_SIZES or (model_w, model_h) not in MODEL_SIZES:
|
||||
raise ValueError("unsupported Jetlink warp geometry")
|
||||
layout = get_nv12_info(cam_w, cam_h)
|
||||
prepare = make_frame_prepare(NV12Frame(cam_w, cam_h, *layout), model_w, model_h)
|
||||
|
||||
def warp(tfm, big_tfm, frame, big_frame):
|
||||
return Tensor.stack(prepare(frame, tfm), prepare(big_frame, big_tfm))
|
||||
return warp, layout[3]
|
||||
|
||||
|
||||
def adapter():
|
||||
from types import SimpleNamespace
|
||||
return SimpleNamespace(make_warp=make_warp)
|
||||
@@ -14,8 +14,6 @@ EXPECTED_TARGETS = {
|
||||
"dmonitoring_model_tinygrad.pkl",
|
||||
"driving_warp_1344x760_tinygrad.pkl", "driving_warp_1928x1208_tinygrad.pkl",
|
||||
"dm_warp_1344x760_tinygrad.pkl", "dm_warp_1928x1208_tinygrad.pkl",
|
||||
"jetlink_warp_1344x760_512x256.pkl", "jetlink_warp_1344x760_1024x512.pkl",
|
||||
"jetlink_warp_1928x1208_512x256.pkl", "jetlink_warp_1928x1208_1024x512.pkl",
|
||||
}
|
||||
|
||||
|
||||
@@ -80,7 +78,7 @@ def test_supported_backend_retains_models_and_camera_warp_graph(graph, monkeypat
|
||||
assert library.call_count == (1 if arch == "aarch64" else 0)
|
||||
present.assert_not_called()
|
||||
calls = env.Command.call_args_list
|
||||
assert len(calls) == 9
|
||||
assert len(calls) == 5
|
||||
targets = [Path(call.args[0]).name for call in calls]
|
||||
assert set(targets) == EXPECTED_TARGETS
|
||||
assert targets[0] == "dmonitoring_model_tinygrad.pkl"
|
||||
@@ -108,7 +106,7 @@ def test_build_does_not_probe_chestnut_or_regenerate_shipped_big_artifacts(graph
|
||||
evaluate("Darwin")
|
||||
library.assert_not_called()
|
||||
present.assert_not_called()
|
||||
assert env.Command.call_count == 9
|
||||
assert env.Command.call_count == 5
|
||||
assert {Path(call.args[0]).name for call in env.Command.call_args_list} == EXPECTED_TARGETS
|
||||
assert not any(Path(call.args[0]).name.startswith("big_driving_") for call in env.Command.call_args_list)
|
||||
env.SideEffect.assert_not_called()
|
||||
|
||||
@@ -432,16 +432,6 @@ class ArtifactCheckTest(unittest.TestCase):
|
||||
chunks.parent.mkdir(parents=True)
|
||||
(self.root / "openpilot/__init__.py").touch()
|
||||
shutil.copy2(SOURCE.parent / "openpilot/common/file_chunker.py", chunks)
|
||||
for name in ("tools/laptop_device_build/jetlink_artifacts.py", "tools/laptop_device_build/jetlink_pickle.py",
|
||||
"openpilot/system/camerad/cameras/nv12_info.py", "tinygrad_repo/tinygrad/uop/__init__.py"):
|
||||
target = self.root / name
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
shutil.copy2(SOURCE.parent / name, target)
|
||||
from tools.laptop_device_build.jetlink_artifacts import MODEL_DIR, warp_names
|
||||
for name in warp_names():
|
||||
target = self.root / MODEL_DIR / name
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
shutil.copy2(SOURCE.parent / MODEL_DIR / name, target)
|
||||
shipped = b"verified RDF artifact fixture"
|
||||
self.write("openpilot/selfdrive/modeld/models/rdf43_driving_tinygrad.pkl", shipped)
|
||||
self.write("openpilot/starpilot/models/catalog.py",
|
||||
|
||||
@@ -1,110 +0,0 @@
|
||||
"""Separate, source-bound QCOM warp package for external Jetlink inference."""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import json
|
||||
from pathlib import Path
|
||||
import shutil
|
||||
import stat
|
||||
|
||||
from tools.laptop_device_build.jetlink_pickle import require_jetlink_qcom_pickle
|
||||
|
||||
CAMERAS = ((1344, 760), (1928, 1208))
|
||||
MODEL_SIZES = ((512, 256), (1024, 512))
|
||||
MODEL_DIR = Path('openpilot/selfdrive/modeld/models')
|
||||
|
||||
|
||||
def warp_names() -> tuple[str, ...]:
|
||||
return tuple(f'jetlink_warp_{cw}x{ch}_{mw}x{mh}.pkl' for cw, ch in CAMERAS for mw, mh in MODEL_SIZES)
|
||||
|
||||
|
||||
def sha(path: Path) -> str:
|
||||
return hashlib.sha256(path.read_bytes()).hexdigest()
|
||||
|
||||
|
||||
def sources(root: Path) -> dict[str, str]:
|
||||
files = [root / 'openpilot/starpilot/models/jetlink_warp.py',
|
||||
root / 'jetlink_repo/jetlink/openpilot/warp.py',
|
||||
root / 'jetlink_repo/jetlink/openpilot/interface.py',
|
||||
root / 'openpilot/common/transformations/camera.py',
|
||||
root / 'openpilot/common/transformations/model.py',
|
||||
root / 'openpilot/system/camerad/cameras/nv12_info.py']
|
||||
for folder in ('tinygrad_repo/tinygrad', 'tinygrad_repo/extra', 'tinygrad_repo/examples/openpilot'):
|
||||
files.extend(p for p in (root / folder).rglob('*') if p.is_file() and p.suffix in ('.py', '.c', '.h', '.cl') and '__pycache__' not in p.parts)
|
||||
if not (root / 'jetlink_repo/jetlink/openpilot/warp.py').is_file():
|
||||
raise ValueError('Pinned Jetlink warp sources are required')
|
||||
return {p.relative_to(root).as_posix(): sha(p) for p in sorted(files)}
|
||||
|
||||
|
||||
def create(root: Path, package: Path) -> None:
|
||||
package.mkdir(parents=True, exist_ok=True)
|
||||
rows = {}
|
||||
for name in warp_names():
|
||||
path = root / MODEL_DIR / name
|
||||
require_jetlink_qcom_pickle(path)
|
||||
rows[name] = {'sha256': sha(path), 'size': path.stat().st_size}
|
||||
shutil.copyfile(path, package / name)
|
||||
(package / 'manifest.json').write_text(json.dumps({'version': 1, 'sources': sources(root), 'artifacts': rows}, indent=2) + '\n')
|
||||
|
||||
|
||||
|
||||
RUNTIME_HELPER = 'tinygrad_repo/examples/openpilot/helpers.py'
|
||||
RUNTIME_HELPER_BEFORE = '7849829700c575339de7d95ffa869507ad2d65143af4c23a4411ad4d95ba3ae2'
|
||||
RUNTIME_HELPER_AFTER = '6494f23757a1058d8c90db8434004e8f703bb61d55903ac4500c04a7d511c8a3'
|
||||
RUNTIME_PROFILE = 'unused-empty-arena-loader-v1'
|
||||
|
||||
|
||||
def source_signature(source: dict[str, str]) -> str:
|
||||
return hashlib.sha256(json.dumps(source, sort_keys=True, separators=(',', ':')).encode()).hexdigest()
|
||||
|
||||
|
||||
def _compatible_sources(package: Path, manifest: dict, current: dict[str, str]) -> bool:
|
||||
saved = manifest.get('sources')
|
||||
if saved == current:
|
||||
return True
|
||||
if not isinstance(saved, dict) or saved.get(RUNTIME_HELPER) != RUNTIME_HELPER_BEFORE or current.get(RUNTIME_HELPER) != RUNTIME_HELPER_AFTER:
|
||||
return False
|
||||
if {**current, RUNTIME_HELPER: RUNTIME_HELPER_BEFORE} != saved:
|
||||
return False
|
||||
sidecar = package / 'runtime-compatibility.json'
|
||||
try:
|
||||
info = sidecar.lstat()
|
||||
if not stat.S_ISREG(info.st_mode) or not 0 < info.st_size <= 4096:
|
||||
return False
|
||||
expected = {'version': 1, 'profile': RUNTIME_PROFILE,
|
||||
'build_manifest_sha256': sha(package / 'manifest.json'),
|
||||
'compatible_source_sha256': source_signature(current)}
|
||||
extension = json.loads(sidecar.read_text())
|
||||
return type(extension) is dict and type(extension.get('version')) is int and extension == expected
|
||||
except (OSError, ValueError):
|
||||
return False
|
||||
|
||||
|
||||
def import_artifact(package: Path, root: Path, target: Path) -> None:
|
||||
manifest = json.loads((package / 'manifest.json').read_text())
|
||||
if manifest.get('version') != 1 or not _compatible_sources(package, manifest, sources(root)):
|
||||
raise ValueError('Jetlink QCOM warp source closure differs; compile a current package on supported QCOM hardware')
|
||||
if set(manifest.get('artifacts', {})) != set(warp_names()):
|
||||
raise ValueError('Jetlink QCOM warp package has an incomplete geometry set')
|
||||
if target.parent.resolve() != (root / MODEL_DIR).resolve() or target.name not in warp_names():
|
||||
raise ValueError('Unexpected Jetlink warp target')
|
||||
row = manifest['artifacts'][target.name]
|
||||
artifact = package / target.name
|
||||
if row != {'sha256': sha(artifact), 'size': artifact.stat().st_size}:
|
||||
raise ValueError('Jetlink QCOM warp artifact digest or size differs')
|
||||
require_jetlink_qcom_pickle(artifact)
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
shutil.copyfile(artifact, target)
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument('--root', type=Path, default=Path.cwd())
|
||||
parser.add_argument('--package', type=Path, required=True)
|
||||
args = parser.parse_args()
|
||||
create(args.root.resolve(), args.package.resolve())
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
@@ -1,158 +0,0 @@
|
||||
"""Inspect bare Jetlink captures with inert pickle globals; never run their code."""
|
||||
import ast
|
||||
import io
|
||||
import pickle
|
||||
from pathlib import Path
|
||||
import re
|
||||
|
||||
GLOBALS = {
|
||||
'tinygrad.engine.jit._TinyJit', 'tinygrad.engine.jit.CapturedJit', 'tinygrad.tensor.Tensor',
|
||||
'tinygrad.uop.ops.UOp', 'tinygrad.uop.Ops', 'tinygrad.uop.ops.ParamArg', 'tinygrad.dtype.DType',
|
||||
'tinygrad.dtype.AddrSpace', 'tinygrad.device.Buffer', 'tinygrad.device.BufferSpec', 'builtins.bytearray',
|
||||
'tinygrad.uop.ops.CallInfo', 'tinygrad.uop.ops.AxisType', 'tinygrad.uop.ops.KernelInfo',
|
||||
'tinygrad.renderer.Estimates', 'tinygrad.uop.ops.ProgramInfo', 'tinygrad.helpers.Target',
|
||||
'tinygrad.runtime.support.hcq2.HCQInfo',
|
||||
}
|
||||
|
||||
|
||||
class Inert:
|
||||
def __new__(cls, *args):
|
||||
value = object.__new__(cls)
|
||||
value.args = args
|
||||
return value
|
||||
|
||||
def __init__(self, *args):
|
||||
pass
|
||||
|
||||
def __setstate__(self, state):
|
||||
self.state = state
|
||||
|
||||
|
||||
class Inspector(pickle.Unpickler):
|
||||
def find_class(self, module, name):
|
||||
tag = f'{module}.{name}'
|
||||
if tag not in GLOBALS:
|
||||
raise ValueError(f'Unexpected Jetlink pickle global: {tag}')
|
||||
return type(name, (Inert,), {'tag': tag})
|
||||
|
||||
def persistent_load(self, pid):
|
||||
raise ValueError('Persistent pickle references are not Jetlink captures')
|
||||
|
||||
|
||||
def tagged(value, tag):
|
||||
return isinstance(value, Inert) and value.tag == tag
|
||||
|
||||
|
||||
def walk(root):
|
||||
stack, seen = [root], set()
|
||||
while stack:
|
||||
value = stack.pop()
|
||||
if id(value) in seen:
|
||||
continue
|
||||
seen.add(id(value))
|
||||
if len(seen) > 200_000:
|
||||
raise ValueError('Jetlink capture exceeds inspection object bound')
|
||||
yield value
|
||||
if isinstance(value, Inert):
|
||||
stack.extend((value.args, getattr(value, 'state', None)))
|
||||
elif isinstance(value, (tuple, list)):
|
||||
stack.extend(value)
|
||||
elif isinstance(value, dict):
|
||||
stack.extend(value.values())
|
||||
|
||||
|
||||
def op_codes():
|
||||
path = Path(__file__).resolve().parents[2] / 'tinygrad_repo/tinygrad/uop/__init__.py'
|
||||
body = next(node.body for node in ast.parse(path.read_text()).body if isinstance(node, ast.ClassDef) and node.name == 'Ops')
|
||||
names = [node.targets[0].id for node in body if isinstance(node, ast.Assign)]
|
||||
return {name: index + 1 for index, name in enumerate(names)}
|
||||
|
||||
|
||||
def shape(view, codes):
|
||||
if not tagged(view, 'tinygrad.uop.ops.UOp') or len(view.args) != 5:
|
||||
raise ValueError('Missing Jetlink input/output shape UOp')
|
||||
op, sources = view.args[:2]
|
||||
if not tagged(op, 'tinygrad.uop.Ops') or op.args != (codes['RESHAPE'],):
|
||||
raise ValueError('Jetlink input/output is not explicitly shaped')
|
||||
dimensions = sources[1]
|
||||
if dimensions.args[0].args != (codes['STACK'],):
|
||||
raise ValueError('Missing Jetlink shape dimensions')
|
||||
values = []
|
||||
for dimension in dimensions.args[1]:
|
||||
if dimension.args[0].args != (codes['CONST'],) or type(dimension.args[2]) is not int:
|
||||
raise ValueError('Jetlink shape is not fixed')
|
||||
values.append(dimension.args[2])
|
||||
return tuple(values)
|
||||
|
||||
|
||||
def dtype(value, name):
|
||||
return tagged(value, 'tinygrad.dtype.DType') and len(value.args) == 4 and value.args[2:] == name
|
||||
|
||||
|
||||
def require_jetlink_qcom_pickle(path: Path) -> None:
|
||||
from openpilot.system.camerad.cameras.nv12_info import get_nv12_info
|
||||
match = re.fullmatch(r'jetlink_warp_(1344x760|1928x1208)_(512x256|1024x512)\.pkl', path.name)
|
||||
if not match:
|
||||
raise ValueError('Unsupported Jetlink warp geometry name')
|
||||
if not 0 < path.stat().st_size <= 100_000_000:
|
||||
raise ValueError('Jetlink pickle exceeds inspection byte bound')
|
||||
stream = io.BytesIO(path.read_bytes())
|
||||
try:
|
||||
root = Inspector(stream).load()
|
||||
if stream.read(1):
|
||||
raise ValueError('Trailing bytes after Jetlink capture')
|
||||
if not tagged(root, 'tinygrad.engine.jit._TinyJit') or len(root.args) != 2 or root.args[0] is not None:
|
||||
raise ValueError('Jetlink warp is not a bare captured TinyJit')
|
||||
capture = root.args[1]
|
||||
if not tagged(capture, 'tinygrad.engine.jit.CapturedJit') or len(capture.args) != 4:
|
||||
raise ValueError('Missing Jetlink CapturedJit')
|
||||
ret, linear, names, inputs = capture.args
|
||||
if names != ['big_frame', 'big_tfm', 'frame', 'tfm'] or len(inputs) != 4:
|
||||
raise ValueError('Jetlink warp input names/count differ')
|
||||
codes = op_codes()
|
||||
for index, (view, variables, kind, device) in enumerate(inputs):
|
||||
if variables or device != ('QCOM' if index in (0, 2) else 'NPY'):
|
||||
raise ValueError('Jetlink warp captured input backend differs')
|
||||
if not dtype(kind, ('unsigned char', 'B') if index in (0, 2) else ('float', 'f')):
|
||||
raise ValueError('Jetlink warp captured input dtype differs')
|
||||
if index in (1, 3) and shape(view, codes) != (3, 3):
|
||||
raise ValueError('Jetlink transform input shape differs')
|
||||
if not tagged(ret, 'tinygrad.tensor.Tensor') or not tagged(linear, 'tinygrad.uop.ops.UOp'):
|
||||
raise ValueError('Missing Jetlink tensor return or executable graph')
|
||||
mw, mh = map(int, match[2].split('x'))
|
||||
if shape(ret.state[1]['uop'], codes) != (2, 6, mh // 2, mw // 2):
|
||||
raise ValueError('Jetlink output shape differs')
|
||||
output = ret.state[1]['uop']
|
||||
for _ in range(100):
|
||||
parameter = output.args[2]
|
||||
if tagged(parameter, 'tinygrad.uop.ops.ParamArg'):
|
||||
state = parameter.state
|
||||
if state['device'] != 'QCOM' or state['size'] != 2 * 6 * (mh // 2) * (mw // 2) or not dtype(state['dtype'], ('unsigned char', 'B')):
|
||||
raise ValueError('Jetlink output buffer signature differs')
|
||||
break
|
||||
output = output.args[1][0]
|
||||
else:
|
||||
raise ValueError('Missing Jetlink output buffer')
|
||||
frame_size = get_nv12_info(*map(int, match[1].split('x')))[3]
|
||||
slots, compiled = set(), set()
|
||||
for node in walk(root):
|
||||
if tagged(node, 'tinygrad.uop.ops.UOp'):
|
||||
operation = node.args[0].args[0]
|
||||
if operation == codes['PROGRAM']:
|
||||
compiled.add('program')
|
||||
if operation == codes['BINARY'] and isinstance(node.args[2], bytes) and node.args[2]:
|
||||
compiled.add('binary')
|
||||
if isinstance(node, str) and (node.startswith(('METAL', 'AMD', 'USB+AMD'))):
|
||||
raise ValueError('Foreign Jetlink captured backend')
|
||||
if tagged(node, 'tinygrad.uop.ops.ParamArg'):
|
||||
state = node.state
|
||||
if state['device'] == 'QCOM' and state['slot'] in (0, 2):
|
||||
if state['size'] != frame_size or not dtype(state['dtype'], ('unsigned char', 'B')) or state['buffer'] is not None:
|
||||
raise ValueError('Jetlink frame input signature differs')
|
||||
slots.add(state['slot'])
|
||||
if slots != {0, 2}:
|
||||
raise ValueError('Missing paired QCOM frame inputs')
|
||||
if compiled != {'program', 'binary'}:
|
||||
raise ValueError('Missing compiled Jetlink execution graph')
|
||||
except (AttributeError, IndexError, KeyError, TypeError, pickle.UnpicklingError, EOFError) as error:
|
||||
raise ValueError('Malformed Jetlink capture') from error
|
||||
@@ -1,148 +0,0 @@
|
||||
"""Separate Jetlink artifact admission rejects stale or modified captures."""
|
||||
import json
|
||||
from pathlib import Path
|
||||
import tempfile
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
from tools.laptop_device_build import jetlink_artifacts as artifacts
|
||||
|
||||
|
||||
class TestJetlinkArtifacts(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.temp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.temp.cleanup)
|
||||
self.root = Path(self.temp.name) / 'root'
|
||||
self.package = Path(self.temp.name) / 'package'
|
||||
self.root.mkdir()
|
||||
self.package.mkdir()
|
||||
self.source = {'jetlink_repo/jetlink/openpilot/warp.py': 'known-source'}
|
||||
self.rows = {}
|
||||
for name in artifacts.warp_names():
|
||||
p = self.package / name
|
||||
p.write_bytes(('independently-captured:' + name).encode())
|
||||
self.rows[name] = {'sha256': artifacts.sha(p), 'size': p.stat().st_size}
|
||||
self.manifest = {'version': 1, 'sources': self.source, 'artifacts': self.rows}
|
||||
self.write_manifest()
|
||||
self.target = self.root / artifacts.MODEL_DIR / artifacts.warp_names()[0]
|
||||
|
||||
def write_manifest(self):
|
||||
(self.package / 'manifest.json').write_text(json.dumps(self.manifest))
|
||||
|
||||
def admit(self):
|
||||
# Backend capture validation belongs to the actual QCOM build. This test
|
||||
# checks the independent source/digest/geometry guards around that boundary.
|
||||
with patch.object(artifacts, 'sources', return_value=self.source), patch.object(artifacts, 'require_jetlink_qcom_pickle') as capture:
|
||||
artifacts.import_artifact(self.package, self.root, self.target)
|
||||
return capture
|
||||
|
||||
def test_verified_capture_is_copied_only_to_declared_geometry(self):
|
||||
capture = self.admit()
|
||||
self.assertEqual(self.target.read_bytes(), (self.package / self.target.name).read_bytes())
|
||||
capture.assert_called_once_with(self.package / self.target.name)
|
||||
self.target = self.root / 'other' / self.target.name
|
||||
with self.assertRaisesRegex(ValueError, 'Unexpected'):
|
||||
self.admit()
|
||||
|
||||
def test_changed_bytes_or_source_never_replaces_target(self):
|
||||
self.admit()
|
||||
previous = self.target.read_bytes()
|
||||
(self.package / self.target.name).write_bytes(b'changed')
|
||||
with self.assertRaisesRegex(ValueError, 'digest or size'):
|
||||
self.admit()
|
||||
self.assertEqual(self.target.read_bytes(), previous)
|
||||
self.manifest['sources'] = {'jetlink_repo/jetlink/openpilot/warp.py': 'old-source'}
|
||||
self.write_manifest()
|
||||
with self.assertRaisesRegex(ValueError, 'source closure'):
|
||||
self.admit()
|
||||
self.assertEqual(self.target.read_bytes(), previous)
|
||||
|
||||
def test_exact_unused_loader_extension_keeps_build_manifest_and_captures(self):
|
||||
self.manifest['sources'] = {**self.source, artifacts.RUNTIME_HELPER: artifacts.RUNTIME_HELPER_BEFORE}
|
||||
self.write_manifest()
|
||||
self.source = {**self.source, artifacts.RUNTIME_HELPER: artifacts.RUNTIME_HELPER_AFTER}
|
||||
sidecar = self.package / 'runtime-compatibility.json'
|
||||
extension = {'version': 1, 'profile': artifacts.RUNTIME_PROFILE,
|
||||
'build_manifest_sha256': artifacts.sha(self.package / 'manifest.json'),
|
||||
'compatible_source_sha256': artifacts.source_signature(self.source)}
|
||||
before = (self.package / 'manifest.json').read_bytes(), dict(self.rows)
|
||||
with self.assertRaisesRegex(ValueError, 'source closure'):
|
||||
self.admit()
|
||||
sidecar.write_text(json.dumps(extension))
|
||||
self.admit()
|
||||
self.assertEqual(before, ((self.package / 'manifest.json').read_bytes(), self.rows))
|
||||
for field in ('profile', 'build_manifest_sha256', 'compatible_source_sha256'):
|
||||
with self.subTest(field=field):
|
||||
sidecar.write_text(json.dumps({**extension, field: 'wrong'}))
|
||||
with self.assertRaisesRegex(ValueError, 'source closure'):
|
||||
self.admit()
|
||||
sidecar.write_text(json.dumps(extension))
|
||||
self.source['jetlink_repo/jetlink/openpilot/warp.py'] = 'changed-compiler'
|
||||
with self.assertRaisesRegex(ValueError, 'source closure'):
|
||||
self.admit()
|
||||
self.source['jetlink_repo/jetlink/openpilot/warp.py'] = 'known-source'
|
||||
self.source[artifacts.RUNTIME_HELPER] = 'unreviewed-helper'
|
||||
with self.assertRaisesRegex(ValueError, 'source closure'):
|
||||
self.admit()
|
||||
self.source[artifacts.RUNTIME_HELPER] = artifacts.RUNTIME_HELPER_AFTER
|
||||
sidecar.unlink()
|
||||
external = self.package / 'external.json'
|
||||
external.write_text(json.dumps(extension))
|
||||
sidecar.symlink_to(external)
|
||||
with self.assertRaisesRegex(ValueError, 'source closure'):
|
||||
self.admit()
|
||||
|
||||
def test_incomplete_or_extra_geometry_is_rejected(self):
|
||||
for change in ('remove', 'extra'):
|
||||
with self.subTest(change=change):
|
||||
rows = dict(self.rows)
|
||||
if change == 'remove':
|
||||
rows.pop(artifacts.warp_names()[-1])
|
||||
else:
|
||||
rows['jetlink_warp_unsupported.pkl'] = {}
|
||||
self.manifest['artifacts'] = rows
|
||||
self.write_manifest()
|
||||
with self.assertRaisesRegex(ValueError, 'geometry set'):
|
||||
self.admit()
|
||||
|
||||
|
||||
class TestBareJetlinkCapture(unittest.TestCase):
|
||||
def test_shipped_qcom_captures_have_exact_input_and_output_signatures(self):
|
||||
from tools.laptop_device_build.jetlink_pickle import require_jetlink_qcom_pickle
|
||||
root = Path(__file__).resolve().parents[3]
|
||||
for name in artifacts.warp_names():
|
||||
with self.subTest(name=name):
|
||||
require_jetlink_qcom_pickle(root / artifacts.MODEL_DIR / name)
|
||||
|
||||
def test_foreign_backend_dtype_and_wrong_geometry_are_rejected(self):
|
||||
from tools.laptop_device_build.jetlink_pickle import require_jetlink_qcom_pickle
|
||||
root = Path(__file__).resolve().parents[3]
|
||||
original = (root / artifacts.MODEL_DIR / 'jetlink_warp_1344x760_512x256.pkl').read_bytes()
|
||||
cases = (
|
||||
('jetlink_warp_1344x760_512x256.pkl', original.replace(b'QCOM', b'AMD!', 1)),
|
||||
('jetlink_warp_1344x760_512x256.pkl', original.replace(b'unsigned char', b'unsigned long', 1)),
|
||||
('jetlink_warp_1344x760_1024x512.pkl', original),
|
||||
('jetlink_warp_1928x1208_512x256.pkl', original),
|
||||
)
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
for name, data in cases:
|
||||
with self.subTest(name=name, changed=data != original):
|
||||
path = Path(directory) / name
|
||||
path.write_bytes(data)
|
||||
with self.assertRaises(ValueError):
|
||||
require_jetlink_qcom_pickle(path)
|
||||
|
||||
def test_bare_capture_does_not_relax_ordinary_wrapped_validator(self):
|
||||
from tools.laptop_device_build.validate_artifacts import require_qcom_pickle
|
||||
root = Path(__file__).resolve().parents[3]
|
||||
with self.assertRaises(ValueError):
|
||||
require_qcom_pickle(root / artifacts.MODEL_DIR / artifacts.warp_names()[0])
|
||||
|
||||
def test_unexpected_pickle_globals_are_rejected_without_execution(self):
|
||||
import pickle
|
||||
from tools.laptop_device_build.jetlink_pickle import require_jetlink_qcom_pickle
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
path = Path(directory) / artifacts.warp_names()[0]
|
||||
path.write_bytes(pickle.dumps(eval))
|
||||
with self.assertRaisesRegex(ValueError, 'Unexpected Jetlink pickle global'):
|
||||
require_jetlink_qcom_pickle(path)
|
||||
@@ -206,10 +206,6 @@ def validate(root: Path) -> None:
|
||||
raise ValueError(f"camera warp set differs from current camera configs: {sorted(actual_warps)}")
|
||||
for name in (*MODELS, *WARPS):
|
||||
require_qcom_pickle(model_dir / name)
|
||||
from tools.laptop_device_build.jetlink_artifacts import warp_names
|
||||
from tools.laptop_device_build.jetlink_pickle import require_jetlink_qcom_pickle
|
||||
for name in warp_names():
|
||||
require_jetlink_qcom_pickle(model_dir / name)
|
||||
shipped = materialize_file_chunked(model_dir / f"{DEFAULT_SMALL}_driving_tinygrad.pkl", DEFAULT_SMALL_SHA256)
|
||||
if shipped.stat().st_size != DEFAULT_SMALL_SIZE:
|
||||
raise ValueError("Shipped default model size differs from the published artifact")
|
||||
|
||||
Reference in New Issue
Block a user