diff --git a/.gitignore b/.gitignore index 74676532260..49db3eddb54 100644 --- a/.gitignore +++ b/.gitignore @@ -18,6 +18,7 @@ cmake-ios-out/ cmake-out* cmake-out-android/ backends/webgpu/third-party/ +backends/arm/test/models/neural_graphics_data/ build-android/ build-x86/ build-direct/ diff --git a/backends/arm/scripts/generate_neural_graphics_test_data.py b/backends/arm/scripts/generate_neural_graphics_test_data.py index 67bfa2a8754..e69223c4832 100644 --- a/backends/arm/scripts/generate_neural_graphics_test_data.py +++ b/backends/arm/scripts/generate_neural_graphics_test_data.py @@ -2,7 +2,7 @@ # # This source code is licensed under the BSD-style license found in the # LICENSE file in the root directory of this source tree. -"""Generate NSS autoencoder calibration and verification data.""" +"""Generate NSS and NFRU autoencoder calibration and verification data.""" from __future__ import annotations @@ -15,16 +15,28 @@ import torch import torch.nn.functional as F from executorch.backends.arm.scripts.neural_graphics_test_data import ( + _NFRU_INPUT_CHANNELS, + _NSS_INPUT_CHANNELS, + download_safetensors_prefix, + NFRU_CALIBRATION_SAMPLES, + nfru_input_shape, + NFRU_INPUT_SPATIAL_SIZE, + nfru_test_calibration_path, + nfru_test_data_root, + nfru_test_verification_path, + NFRU_VERIFICATION_SAMPLES, + NSS_CALIBRATION_SAMPLES, + NSS_CALIBRATION_SPATIAL_SIZE, nss_input_shape, nss_test_calibration_path, nss_test_data_root, nss_test_verification_path, + NSS_VERIFICATION_SAMPLES, ) from safetensors.torch import save_file os.environ.setdefault("HF_HUB_DISABLE_XET", "1") -from huggingface_hub import snapshot_download from ng_model_gym.core.config.config_model import ( # type: ignore[import-not-found,import-untyped] ConfigModel, ) @@ -34,24 +46,56 @@ tonemap_forward, ToneMapperMode, ) +from ng_model_gym.usecases.nfru.data.dataset import ( # type: ignore[import-not-found,import-untyped] + NFRUDataset, +) +from ng_model_gym.usecases.nfru.model.nfru_v1 import ( # type: ignore[import-not-found,import-untyped] + NFRUv1, +) from ng_model_gym.usecases.nss.data.dataset import ( # type: ignore[import-not-found,import-untyped] NSSDataset, ) _DATASET_REPO_ID = "Arm/neural-graphics-dataset" -_CALIBRATION_SOURCE_ALLOW_PATTERNS = [ - "train/**/*.safetensors", - "nss/train/**/*.safetensors", -] -_EVALUATION_SOURCE_ALLOW_PATTERNS = [ - "test/test_full_resolution_sample.safetensors", - "nss/test/test_full_resolution_sample.safetensors", -] +_NSS_DATASET_REVISION = "main" +_NSS_CAPTURE_FRAMES = 99 +_NSS_TRAIN_CAPTURES = ( + *range(80, 90), + *range(91, 107), + *range(108, 119), +) +_NSS_SOURCE_TENSORS = { + "colour_linear", + "depth", + "exposure", + "ground_truth_linear", + "motion_lr", + "render_size", +} +_NSS_EVALUATION_SOURCE_FRAMES = {"nss/test/test_full_resolution_sample.safetensors": 1} EPS = 1e-7 NSS_V1_SPATIAL_MULTIPLE = 8 +_NFRU_DATASET_REVISION = "main" +_NFRU_CALIBRATION_SOURCE = "nfru/train/0002.safetensors" +_NFRU_EVALUATION_SOURCE = "nfru/test/0000.safetensors" +_NFRU_SOURCE_TENSORS = { + "DepthParams", + "FarPlane", + "FovY", + "NearPlane", + "ViewProj", + "depth", + "exposure", + "infinite_zFar", + "mv_{}_f30_m1", + "rgb_linear", + "sy_{}_f30_m1", + "sy_{}_f30_p1", +} + def _luminance(rgb: torch.Tensor) -> torch.Tensor: weights = torch.tensor( @@ -266,34 +310,52 @@ def _raw_dataset_root(snapshot_path: Path) -> Path: return snapshot_path -def _env_int(name: str, default: int) -> int: - value = os.environ.get(name) - if value is None: - return default - parsed = int(value) - if parsed <= 0: - raise ValueError(f"{name} must be positive.") - return parsed +def _has_samples(path: Path, num_samples: int) -> bool: + if path.is_file(): + files = [path] + elif path.is_dir(): + files = list(path.glob("*.safetensors")) + else: + return False + return ( + bool(files) and sum(nss_input_shape(file)[0] for file in files) == num_samples + ) + +def _calibration_source_frames(num_samples: int) -> dict[str, int]: + max_samples = len(_NSS_TRAIN_CAPTURES) * _NSS_CAPTURE_FRAMES + if num_samples > max_samples: + raise ValueError( + f"Requested {num_samples} calibration samples, but only " + f"{max_samples} are available." + ) -def _has_safetensors(path: Path) -> bool: - return path.is_file() or (path.is_dir() and any(path.glob("*.safetensors"))) + source_frames = {} + remaining = num_samples + for capture in _NSS_TRAIN_CAPTURES: + if remaining == 0: + break + frames = min(remaining, _NSS_CAPTURE_FRAMES) + source_frames[f"nss/train/bistro/0002/{capture:04d}/0002.safetensors"] = frames + remaining -= frames + return source_frames def _download_raw_sources( - allow_patterns: list[str], *, force_download: bool = False + source_frames: dict[str, int], *, force_download: bool = False ) -> Path: - snapshot = Path( - snapshot_download( + for source, num_frames in source_frames.items(): + download_safetensors_prefix( repo_id=_DATASET_REPO_ID, - repo_type="dataset", - revision="5039ce015d7c877980fad44f87893fc5ac0927e2", - allow_patterns=allow_patterns, - local_dir=_raw_source_path(), + filename=source, + revision=_NSS_DATASET_REVISION, + destination=_raw_source_path() / source, + num_frames=num_frames, + tensor_names=_NSS_SOURCE_TENSORS, force_download=force_download, + preserve_source_layout=True, ) - ) - return _raw_dataset_root(snapshot) + return _raw_dataset_root(_raw_source_path()) def _delete_raw_split(raw_root: Path, split: str, keep_env: str) -> None: @@ -315,14 +377,14 @@ def _has_test_calibration_samples( if not path.is_dir() or len(list(path.glob("*.safetensors"))) != num_samples: return False return all( - nss_input_shape(file_path)[1:] == (12, *spatial_size) + nss_input_shape(file_path) == (1, _NSS_INPUT_CHANNELS, *spatial_size) for file_path in path.glob("*.safetensors") ) def ensure_generated_test_calibration_dataset( - num_samples: int = 3663, - spatial_size: tuple[int, int] = (128, 128), + num_samples: int = NSS_CALIBRATION_SAMPLES, + spatial_size: tuple[int, int] = NSS_CALIBRATION_SPATIAL_SIZE, force_download: bool = False, ) -> Path: """Generate evenly distributed, test-ready NSS calibration samples.""" @@ -335,7 +397,7 @@ def ensure_generated_test_calibration_dataset( return calibration_path raw_root = _download_raw_sources( - _CALIBRATION_SOURCE_ALLOW_PATTERNS, force_download=force_download + _calibration_source_frames(num_samples), force_download=force_download ) dataset = _model_gym_dataset(raw_root / "train") total_samples = len(dataset) @@ -347,14 +409,7 @@ def ensure_generated_test_calibration_dataset( _remove_stale_shards(calibration_path) calibration_path.mkdir(parents=True, exist_ok=True) - sample_indices = ( - [0] - if num_samples == 1 - else [ - index * (total_samples - 1) // (num_samples - 1) - for index in range(num_samples) - ] - ) + sample_indices = range(num_samples) for output_index, sample_index in enumerate(sample_indices): tensor = _autoencoder_sample(dataset, sample_index) if tensor.shape[-2:] != spatial_size: @@ -381,25 +436,25 @@ def ensure_generated_verification_dataset( """Generate the held-out NSS verification input without calibration data.""" verification_path = nss_test_verification_path() - if _has_safetensors(verification_path): + if _has_samples(verification_path, NSS_VERIFICATION_SAMPLES): return verification_path raw_root = _download_raw_sources( - _EVALUATION_SOURCE_ALLOW_PATTERNS, force_download=force_download + _NSS_EVALUATION_SOURCE_FRAMES, force_download=force_download ) _generate_verification_dataset( raw_root / "test", verification_path, - _env_int("NSS_GENERATED_EVALUATION_SAMPLES", 1), - _env_int("NSS_GENERATED_EVALUATION_SHARD_SIZE", 10), + NSS_VERIFICATION_SAMPLES, + NSS_VERIFICATION_SAMPLES, ) _delete_raw_split(raw_root, "test", "NSS_KEEP_RAW_EVALUATION_DATA") return verification_path def ensure_generated_test_datasets( - calibration_samples: int = 3663, - spatial_size: tuple[int, int] = (128, 128), + calibration_samples: int = NSS_CALIBRATION_SAMPLES, + spatial_size: tuple[int, int] = NSS_CALIBRATION_SPATIAL_SIZE, ) -> tuple[Path, Path]: """Generate the NSS artifacts consumed directly by ``test_nss.py``.""" @@ -412,8 +467,8 @@ def ensure_generated_test_datasets( def generate_test_datasets_from_scratch( - calibration_samples: int = 3663, - spatial_size: tuple[int, int] = (128, 128), + calibration_samples: int = NSS_CALIBRATION_SAMPLES, + spatial_size: tuple[int, int] = NSS_CALIBRATION_SPATIAL_SIZE, ) -> tuple[Path, Path]: """Download and regenerate the NSS artifacts consumed by ``test_nss.py``.""" @@ -433,3 +488,251 @@ def generate_test_datasets_from_scratch( verification_path = ensure_generated_verification_dataset(force_download=True) _delete_raw_sources() return calibration_path, verification_path + + +class _NFRUAutoencoderInputCaptured(Exception): + pass + + +def _nfru_model_gym_config(src: Path) -> ConfigModel: + config_path = files("ng_model_gym.usecases.nfru.configs").joinpath( + "nfru_template.json" + ) + config = json.loads(config_path.read_text(encoding="utf-8")) + for split in ("train", "validation", "test"): + config["dataset"]["path"][split] = str(src) + config["dataset"].update(gt_augmentation=False, align_data=True) + config["model"]["processing_backend"] = "torch" + return ConfigModel.model_validate(config) + + +def _nfru_model_gym_dataset(src: Path, loader_mode: DataLoaderMode) -> NFRUDataset: + return NFRUDataset(_nfru_model_gym_config(src), loader_mode, DatasetType.SAFETENSOR) + + +def _nfru_preprocessing_model(src: Path) -> NFRUv1: + model = NFRUv1(_nfru_model_gym_config(src)).eval() + model.on_evaluation_start() + return model + + +def _capture_nfru_autoencoder_input( + model: NFRUv1, inputs: dict[str, torch.Tensor], random_seed: int +) -> torch.Tensor: + captured: torch.Tensor | None = None + + def capture(_module, args): + nonlocal captured + captured = args[0].detach().cpu() + raise _NFRUAutoencoderInputCaptured + + hook = model.get_neural_network().register_forward_pre_hook(capture) + try: + try: + with torch.random.fork_rng(devices=[]), torch.no_grad(): + torch.manual_seed(random_seed) + model({name: value.unsqueeze(0) for name, value in inputs.items()}) + except _NFRUAutoencoderInputCaptured: + pass + finally: + hook.remove() + + if captured is None: + raise RuntimeError("Failed to capture the NFRU autoencoder input.") + return captured + + +def _nfru_autoencoder_sample( + dataset: NFRUDataset, model: NFRUv1, index: int +) -> tuple[Path, torch.Tensor]: + inputs, _ = dataset[index] + source = dataset.frame_indexes[index][0] + return source, _capture_nfru_autoencoder_input(model, inputs, index) + + +def _nfru_metadata(src: Path, tensor: torch.Tensor, sample: int) -> dict[str, str]: + return { + "format": "nfru_v1_autoencoder_calibration", + "source": str(src), + "sample": str(sample), + "samples": str(tensor.shape[0]), + "shape": json.dumps(list(tensor.shape)), + "preprocess": "nfru_v1_torch_preprocess", + "channels": json.dumps( + [ + "rgb_m1_mv.r", + "rgb_m1_mv.g", + "rgb_m1_mv.b", + "rgb_p1_mv.r", + "rgb_p1_mv.g", + "rgb_p1_mv.b", + "rgb_m1_flow.r", + "rgb_m1_flow.g", + "rgb_m1_flow.b", + "rgb_p1_flow.r", + "rgb_p1_flow.g", + "rgb_p1_flow.b", + "depth_m1", + "depth_p1", + "disocclusion_m1", + "disocclusion_p1", + ] + ), + } + + +def _write_nfru_tensor(src: Path, dst: Path, tensor: torch.Tensor, sample: int) -> None: + dst.parent.mkdir(parents=True, exist_ok=True) + save_file( + {"input": tensor.contiguous()}, + dst, + metadata=_nfru_metadata(src, tensor, sample), + ) + + +def _nfru_raw_source_path() -> Path: + return nfru_test_data_root() / "source" + + +def _download_nfru_raw_source( + source: str, num_frames: int, *, force_download: bool = False +) -> Path: + download_safetensors_prefix( + repo_id=_DATASET_REPO_ID, + filename=source, + revision=_NFRU_DATASET_REVISION, + destination=_nfru_raw_source_path() / source, + num_frames=num_frames, + tensor_names=_NFRU_SOURCE_TENSORS, + force_download=force_download, + preserve_source_layout=True, + ) + return _nfru_raw_source_path() / "nfru" + + +def _has_nfru_samples(path: Path, num_samples: int) -> bool: + if path.is_file(): + files = [path] + elif path.is_dir(): + files = list(path.glob("*.safetensors")) + else: + return False + return ( + bool(files) and sum(nfru_input_shape(file)[0] for file in files) == num_samples + ) + + +def _has_nfru_calibration_samples(path: Path, num_samples: int) -> bool: + if not path.is_dir() or len(list(path.glob("*.safetensors"))) != num_samples: + return False + return all( + nfru_input_shape(file_path) + == (1, _NFRU_INPUT_CHANNELS, *NFRU_INPUT_SPATIAL_SIZE) + for file_path in path.glob("*.safetensors") + ) + + +def ensure_generated_nfru_calibration_dataset( + num_samples: int = NFRU_CALIBRATION_SAMPLES, + force_download: bool = False, +) -> Path: + """Generate evenly distributed NFRU training inputs for calibration.""" + + if num_samples <= 0: + raise ValueError("num_samples must be positive.") + + calibration_path = nfru_test_calibration_path(num_samples) + if _has_nfru_calibration_samples(calibration_path, num_samples): + return calibration_path + + calibration_frames = 2 * num_samples + 3 + raw_root = _download_nfru_raw_source( + _NFRU_CALIBRATION_SOURCE, + calibration_frames, + force_download=force_download, + ) + dataset = _nfru_model_gym_dataset(raw_root / "train", DataLoaderMode.TRAIN) + if num_samples > len(dataset): + raise ValueError( + f"Requested {num_samples} calibration samples, but only found " + f"{len(dataset)}." + ) + + _remove_stale_shards(calibration_path) + calibration_path.mkdir(parents=True, exist_ok=True) + model = _nfru_preprocessing_model(raw_root / "train") + sample_indices = range(num_samples) + for output_index, sample_index in enumerate(sample_indices): + source, tensor = _nfru_autoencoder_sample(dataset, model, sample_index) + _write_nfru_tensor( + source, + calibration_path / f"{output_index:04d}.safetensors", + tensor, + output_index, + ) + + return calibration_path + + +def ensure_generated_nfru_verification_dataset( + force_download: bool = False, +) -> Path: + """Generate a held-out deployment-resolution NFRU autoencoder input.""" + + verification_path = nfru_test_verification_path() + if _has_nfru_samples(verification_path, NFRU_VERIFICATION_SAMPLES): + return verification_path + + raw_root = _download_nfru_raw_source( + _NFRU_EVALUATION_SOURCE, + 2 * NFRU_VERIFICATION_SAMPLES + 3, + force_download=force_download, + ) + dataset = _nfru_model_gym_dataset(raw_root / "test", DataLoaderMode.TEST) + model = _nfru_preprocessing_model(raw_root / "test") + source, tensor = _nfru_autoencoder_sample(dataset, model, 0) + _remove_stale_shards(verification_path) + _write_nfru_tensor(source, verification_path / "0000.safetensors", tensor, 0) + return verification_path + + +def _delete_nfru_raw_sources() -> None: + if not os.environ.get("NFRU_KEEP_RAW_DATA"): + shutil.rmtree(_nfru_raw_source_path(), ignore_errors=True) + + +def ensure_generated_nfru_test_datasets( + calibration_samples: int = NFRU_CALIBRATION_SAMPLES, +) -> tuple[Path, Path]: + """Generate the NFRU artifacts consumed directly by ``test_nfru.py``.""" + + calibration_path = ensure_generated_nfru_calibration_dataset(calibration_samples) + verification_path = ensure_generated_nfru_verification_dataset() + _delete_nfru_raw_sources() + return calibration_path, verification_path + + +def generate_nfru_test_datasets_from_scratch( + calibration_samples: int = NFRU_CALIBRATION_SAMPLES, +) -> tuple[Path, Path]: + """Download and regenerate the NFRU artifacts consumed by + ``test_nfru.py``. + """ + + paths = ( + nfru_test_calibration_path(calibration_samples), + nfru_test_verification_path(), + _nfru_raw_source_path(), + ) + for path in paths: + if path.is_dir(): + shutil.rmtree(path) + elif path.exists(): + path.unlink() + + calibration_path = ensure_generated_nfru_calibration_dataset( + calibration_samples, force_download=True + ) + verification_path = ensure_generated_nfru_verification_dataset(force_download=True) + _delete_nfru_raw_sources() + return calibration_path, verification_path diff --git a/backends/arm/scripts/install_models_for_test.sh b/backends/arm/scripts/install_models_for_test.sh index c7239ee2760..2e6a7caca8f 100644 --- a/backends/arm/scripts/install_models_for_test.sh +++ b/backends/arm/scripts/install_models_for_test.sh @@ -22,10 +22,10 @@ pip install . --no-deps cd .. rm -rf neural-graphics-model-gym -# Prepare the fixed NSS artifacts before pytest. The calibration data is -# generated from raw 128x128 training frames; evaluation retains the -# deployment-resolution input. +# Prepare fixed neural-graphics artifacts before pytest. NFRU calibration and +# evaluation retain the native autoencoder input resolution. python3 -c ' -from executorch.backends.arm.scripts.generate_neural_graphics_test_data import generate_test_datasets_from_scratch +from executorch.backends.arm.scripts.generate_neural_graphics_test_data import generate_nfru_test_datasets_from_scratch, generate_test_datasets_from_scratch generate_test_datasets_from_scratch() +generate_nfru_test_datasets_from_scratch() ' diff --git a/backends/arm/scripts/neural_graphics_test_data.py b/backends/arm/scripts/neural_graphics_test_data.py index e404394dca2..96835bb20c7 100644 --- a/backends/arm/scripts/neural_graphics_test_data.py +++ b/backends/arm/scripts/neural_graphics_test_data.py @@ -2,55 +2,101 @@ # # This source code is licensed under the BSD-style license found in the # LICENSE file in the root directory of this source tree. -"""Load generated NSS autoencoder calibration and verification data.""" +"""Fetch and load neural-graphics calibration and verification data.""" from __future__ import annotations +import json import os +from concurrent.futures import ThreadPoolExecutor +from io import BytesIO from pathlib import Path -from typing import Iterator +from typing import BinaryIO, Callable, Collection, Iterator import torch +from huggingface_hub import hf_hub_url +from huggingface_hub.utils import build_hf_headers, get_session from safetensors import safe_open NSS_DATASET_REVISION = "main" -_TEST_CALIBRATION_DIR = "calibration/nss_v1_autoencoder_cpu_calibration" -_VERIFICATION_DIR = "evaluation/nss_v1_autoencoder_cpu_evaluation" +NSS_CALIBRATION_SAMPLES = 1024 +NSS_VERIFICATION_SAMPLES = 1 +NSS_CALIBRATION_SPATIAL_SIZE = (128, 128) +NFRU_DATASET_REVISION = "main" +NFRU_CALIBRATION_SAMPLES = 32 +NFRU_VERIFICATION_SAMPLES = 1 +NFRU_INPUT_SPATIAL_SIZE = (270, 480) +_NSS_INPUT_CHANNELS = 12 +_NFRU_INPUT_CHANNELS = 16 +_NSS_CALIBRATION_DIR = "calibration/nss_v1_autoencoder_cpu_calibration" +_NSS_VERIFICATION_DIR = "evaluation/nss_v1_autoencoder_cpu_evaluation" +_NFRU_CALIBRATION_DIR = "calibration/nfru_v1_autoencoder_cpu_calibration" +_NFRU_VERIFICATION_DIR = "evaluation/nfru_v1_autoencoder_cpu_evaluation" + +_InputShape = tuple[int, int, int, int] +_InputShapeFn = Callable[[Path], _InputShape] +_ModelInputs = tuple[torch.Tensor] + + +def neural_graphics_test_data_root() -> Path: + if "NEURAL_GRAPHICS_GENERATED_DATASET_ROOT" in os.environ: + return Path(os.environ["NEURAL_GRAPHICS_GENERATED_DATASET_ROOT"]) + return ( + Path(__file__).resolve().parents[1] / "test" / "models" / "neural_graphics_data" + ) def nss_test_data_root() -> Path: if "NSS_GENERATED_DATASET_ROOT" in os.environ: return Path(os.environ["NSS_GENERATED_DATASET_ROOT"]) - else: - return ( - Path(__file__).resolve().parents[1] - / "test" - / "models" - / "nss_data" - / NSS_DATASET_REVISION - ) + return neural_graphics_test_data_root() / "nss" / NSS_DATASET_REVISION def nss_test_calibration_path( - num_samples: int = 3663, spatial_size: tuple[int, int] = (128, 128) + num_samples: int = NSS_CALIBRATION_SAMPLES, + spatial_size: tuple[int, int] = NSS_CALIBRATION_SPATIAL_SIZE, ) -> Path: """Return the preprocessed calibration dataset path used by NSS tests.""" height, width = spatial_size return nss_test_data_root() / ( - f"{_TEST_CALIBRATION_DIR}_{num_samples}_{height}x{width}" + f"{_NSS_CALIBRATION_DIR}_{num_samples}_{height}x{width}" ) def nss_test_verification_path() -> Path: """Return the preprocessed verification dataset path used by NSS tests.""" - return nss_test_data_root() / _VERIFICATION_DIR + return nss_test_data_root() / _NSS_VERIFICATION_DIR + + +def nfru_test_data_root() -> Path: + if "NFRU_GENERATED_DATASET_ROOT" in os.environ: + return Path(os.environ["NFRU_GENERATED_DATASET_ROOT"]) + return neural_graphics_test_data_root() / "nfru" / NFRU_DATASET_REVISION + + +def nfru_test_calibration_path( + num_samples: int = NFRU_CALIBRATION_SAMPLES, +) -> Path: + """Return the preprocessed calibration dataset path used by NFRU tests.""" + + height, width = NFRU_INPUT_SPATIAL_SIZE + return nfru_test_data_root() / ( + f"{_NFRU_CALIBRATION_DIR}_{num_samples}_{height}x{width}" + ) + +def nfru_test_verification_path() -> Path: + """Return the preprocessed verification dataset path used by NFRU tests.""" -def nss_input_shape(path: Path) -> tuple[int, int, int, int]: - """Return the shape of the ``input`` tensor in a safetensors file.""" + return nfru_test_data_root() / _NFRU_VERIFICATION_DIR + + +def _validate_input_shape( + path: Path, expected_channels: int, model_name: str +) -> _InputShape: with safe_open(path, framework="pt", device="cpu") as handle: keys = set(handle.keys()) @@ -61,16 +107,36 @@ def nss_input_shape(path: Path) -> tuple[int, int, int, int]: if len(shape) != 4: raise ValueError(f"Expected NCHW `input`, got shape {shape} in {path}.") - if shape[1] != 12: - raise ValueError(f"Expected 12 NSS input channels, got shape {shape}.") + if shape[1] != expected_channels: + raise ValueError( + f"Expected {expected_channels} {model_name} input channels, got shape " + f"{shape}." + ) return shape # type: ignore[return-value] -def _load_input_slice(path: Path, start: int, stop: int) -> torch.Tensor: +def nss_input_shape(path: Path) -> _InputShape: + """Return the shape of an NSS ``input`` tensor.""" + + return _validate_input_shape(path, _NSS_INPUT_CHANNELS, "NSS") + + +def nfru_input_shape(path: Path) -> _InputShape: + """Return the shape of an NFRU ``input`` tensor.""" + + return _validate_input_shape(path, _NFRU_INPUT_CHANNELS, "NFRU") + + +def _load_input_slice( + path: Path, + start: int, + stop: int, + input_shape_fn: _InputShapeFn, +) -> torch.Tensor: if not path.is_file(): raise FileNotFoundError(path) - shape = nss_input_shape(path) + shape = input_shape_fn(path) if start < 0 or stop <= start or stop > shape[0]: raise ValueError(f"Invalid slice [{start}:{stop}] for shape {shape}.") @@ -92,27 +158,38 @@ def _safetensor_files(path: Path) -> list[Path]: raise FileNotFoundError(f"No safetensors found at {path}") -def _load_sample(path: Path, start: int = 0) -> tuple[torch.Tensor]: - if start < 0: - raise ValueError("start must be non-negative.") +def _load_input_sample( + path: Path, + sample_index: int, + input_shape_fn: _InputShapeFn, +) -> _ModelInputs: + if sample_index < 0: + raise ValueError("sample_index must be non-negative.") skipped = 0 for file_path in _safetensor_files(path): - file_samples = nss_input_shape(file_path)[0] - if start < skipped + file_samples: - local_index = start - skipped - return (_load_input_slice(file_path, local_index, local_index + 1),) + file_samples = input_shape_fn(file_path)[0] + if sample_index < skipped + file_samples: + local_index = sample_index - skipped + return ( + _load_input_slice( + file_path, local_index, local_index + 1, input_shape_fn + ), + ) skipped += file_samples - raise ValueError(f"Sample {start} is outside the {skipped} samples in {path}.") + raise ValueError( + f"Sample {sample_index} is outside the {skipped} samples in {path}." + ) -def iter_calibration_samples( +def _iter_calibration_samples( path: Path, *, - num_samples: int = 8, -) -> Iterator[tuple[torch.Tensor]]: - """Stream NSS calibration samples without retaining them in memory.""" + num_samples: int, + input_shape_fn: _InputShapeFn, +) -> Iterator[_ModelInputs]: + """Stream calibration samples without retaining them in memory.""" if num_samples <= 0: raise ValueError("num_samples must be positive.") @@ -124,15 +201,290 @@ def iter_calibration_samples( ) for file_path in files[:num_samples]: - yield (_load_input_slice(file_path, 0, 1),) + yield (_load_input_slice(file_path, 0, 1, input_shape_fn),) -def load_verification_inputs( +def load_nss_verification_inputs( path: Path | None = None, *, - start: int = 0, -) -> tuple[torch.Tensor]: - """Load one held-out verification sample in ``example_inputs`` format.""" + sample_index: int = 0, +) -> _ModelInputs: + """Load one held-out NSS verification sample in ``example_inputs`` + format. + """ path = nss_test_verification_path() if path is None else path - return _load_sample(path, start) + return _load_input_sample(path, sample_index, nss_input_shape) + + +def load_nfru_verification_inputs( + path: Path | None = None, + *, + sample_index: int = 0, +) -> _ModelInputs: + """Load one held-out NFRU verification sample in ``example_inputs`` + format. + """ + + path = nfru_test_verification_path() if path is None else path + return _load_input_sample(path, sample_index, nfru_input_shape) + + +def _require_calibration_path(path: Path, model_name: str) -> Path: + if not path.exists(): + raise RuntimeError( + f"{model_name} calibration data is prepared by " + "backends/arm/scripts/install_models_for_test.sh." + ) + return path + + +def iter_nss_calibration_samples( + path: Path, + *, + num_samples: int = 8, +) -> Iterator[_ModelInputs]: + """Stream NSS calibration samples without retaining them in memory.""" + + return _iter_calibration_samples( + path, num_samples=num_samples, input_shape_fn=nss_input_shape + ) + + +def iter_nfru_calibration_samples( + path: Path, + *, + num_samples: int = 8, +) -> Iterator[_ModelInputs]: + """Stream NFRU calibration samples without retaining them in memory.""" + + return _iter_calibration_samples( + path, num_samples=num_samples, input_shape_fn=nfru_input_shape + ) + + +def iter_nss_test_calibration_samples( + num_samples: int = NSS_CALIBRATION_SAMPLES, +) -> Iterator[_ModelInputs]: + """Stream the generated NSS calibration dataset used by model tests.""" + + path = _require_calibration_path(nss_test_calibration_path(), "NSS") + return iter_nss_calibration_samples(path, num_samples=num_samples) + + +def iter_nfru_test_calibration_samples( + num_samples: int = NFRU_CALIBRATION_SAMPLES, +) -> Iterator[_ModelInputs]: + """Stream the generated NFRU calibration dataset used by model tests.""" + + path = _require_calibration_path(nfru_test_calibration_path(), "NFRU") + return iter_nfru_calibration_samples(path, num_samples=num_samples) + + +_HEADER_LENGTH_BYTES = 8 +_COPY_CHUNK_SIZE = 1024 * 1024 + + +def _download_range(url: str, start: int, stop: int, destination: BinaryIO) -> None: + if stop <= start: + return + + headers = build_hf_headers() + headers.update( + { + "Accept-Encoding": "identity", + "Range": f"bytes={start}-{stop - 1}", + } + ) + with get_session().stream( + "GET", url, headers=headers, follow_redirects=True, timeout=None + ) as response: + if response.status_code != 206: + raise RuntimeError( + f"Expected an HTTP range response for {url}, got " + f"status {response.status_code}." + ) + written = 0 + for chunk in response.iter_bytes(_COPY_CHUNK_SIZE): + destination.write(chunk) + written += len(chunk) + + expected = stop - start + if written != expected: + raise RuntimeError(f"Expected {expected} bytes from {url}, received {written}.") + + +def _read_range(url: str, start: int, stop: int) -> bytes: + destination = BytesIO() + _download_range(url, start, stop, destination) + return destination.getvalue() + + +def _local_prefix_matches( + path: Path, + num_frames: int, + tensor_names: Collection[str], + preserve_source_layout: bool, +) -> bool: + try: + with path.open("rb") as source: + header_length = int.from_bytes(source.read(_HEADER_LENGTH_BYTES), "little") + header = json.loads(source.read(header_length)) + metadata = header.get("__metadata__", {}) + expected_frames = ( + metadata.get("RangePrefixFrames") + if preserve_source_layout + else metadata.get("Length") + ) + return int(expected_frames or -1) == num_frames and set(tensor_names).issubset( + header + ) + except (OSError, ValueError): + return False + + +def _remote_header(url: str) -> tuple[dict, int]: + header_length = int.from_bytes(_read_range(url, 0, _HEADER_LENGTH_BYTES), "little") + if header_length <= 0: + raise ValueError(f"Invalid safetensor header length {header_length} in {url}.") + header = json.loads( + _read_range( + url, + _HEADER_LENGTH_BYTES, + _HEADER_LENGTH_BYTES + header_length, + ) + ) + return header, _HEADER_LENGTH_BYTES + header_length + + +def _prefix_plan( + header: dict, + source_data_offset: int, + num_frames: int, + tensor_names: Collection[str], + preserve_source_layout: bool = False, +) -> tuple[bytes, list[tuple[int, int, int]], int]: + metadata = dict(header.get("__metadata__", {})) + source_frames = int(metadata["Length"]) + if num_frames <= 0 or num_frames > source_frames: + raise ValueError( + f"Requested {num_frames} frames from a {source_frames}-frame safetensor." + ) + + missing = set(tensor_names).difference(header) + if missing: + raise KeyError(f"Missing tensors: {sorted(missing)}") + + output_metadata = dict(metadata) + if preserve_source_layout: + output_metadata["RangePrefixFrames"] = str(num_frames) + else: + output_metadata["Length"] = str(num_frames) + output_header: dict = {"__metadata__": output_metadata} + source_ranges: list[tuple[int, int, int]] = [] + output_offset = 0 + for name, descriptor in header.items(): + if name == "__metadata__" or name not in tensor_names: + continue + + shape = list(descriptor["shape"]) + if not shape or shape[0] != source_frames: + raise ValueError( + f"Tensor {name} has shape {shape}; expected a leading frame " + f"dimension of {source_frames}." + ) + + source_start, source_stop = descriptor["data_offsets"] + source_bytes = source_stop - source_start + if source_bytes % source_frames: + raise ValueError( + f"Tensor {name} byte size {source_bytes} is not divisible by " + f"its {source_frames} frames." + ) + + downloaded_bytes = source_bytes // source_frames * num_frames + output_bytes = source_bytes if preserve_source_layout else downloaded_bytes + if not preserve_source_layout: + shape[0] = num_frames + output_header[name] = { + "dtype": descriptor["dtype"], + "shape": shape, + "data_offsets": [output_offset, output_offset + output_bytes], + } + source_ranges.append( + ( + source_data_offset + source_start, + source_data_offset + source_start + downloaded_bytes, + output_offset, + ) + ) + output_offset += output_bytes + + encoded_header = json.dumps(output_header, separators=(",", ":")).encode("utf-8") + encoded_header += b" " * (-len(encoded_header) % 8) + return ( + len(encoded_header).to_bytes(_HEADER_LENGTH_BYTES, "little") + encoded_header, + source_ranges, + output_offset, + ) + + +def download_safetensors_prefix( + *, + repo_id: str, + filename: str, + revision: str, + destination: Path, + num_frames: int, + tensor_names: Collection[str], + force_download: bool = False, + preserve_source_layout: bool = False, +) -> Path: + """Download selected tensors for the first ``num_frames`` frames.""" + + if ( + destination.is_file() + and not force_download + and _local_prefix_matches( + destination, num_frames, tensor_names, preserve_source_layout + ) + ): + return destination + + url = hf_hub_url( + repo_id=repo_id, + filename=filename, + repo_type="dataset", + revision=revision, + ) + header, source_data_offset = _remote_header(url) + output_header, source_ranges, output_size = _prefix_plan( + header, + source_data_offset, + num_frames, + tensor_names, + preserve_source_layout, + ) + if not source_ranges: + raise ValueError("At least one tensor must be selected.") + + destination.parent.mkdir(parents=True, exist_ok=True) + partial = destination.with_name(f"{destination.name}.partial") + try: + with partial.open("wb") as output: + output.write(output_header) + output.truncate(len(output_header) + output_size) + + def download_range(source_range: tuple[int, int, int]) -> None: + start, stop, output_offset = source_range + with partial.open("r+b") as output: + output.seek(len(output_header) + output_offset) + _download_range(url, start, stop, output) + + with ThreadPoolExecutor(max_workers=min(4, len(source_ranges))) as executor: + list(executor.map(download_range, source_ranges)) + os.replace(partial, destination) + finally: + partial.unlink(missing_ok=True) + + return destination diff --git a/backends/arm/test/models/model_test_utils.py b/backends/arm/test/models/model_test_utils.py index c49db594a46..4007cc90669 100644 --- a/backends/arm/test/models/model_test_utils.py +++ b/backends/arm/test/models/model_test_utils.py @@ -3,11 +3,39 @@ # This source code is licensed under the BSD-style license found in the # LICENSE file in the root directory of this source tree. +import os from typing import Any +import pytest import torch +REAL_AND_RANDOM_DATA = { + "real_data": True, + "random_data": False, +} + +PTQ_AND_QAT_DATA = { + "ptq": False, + "qat": True, +} + + +def skip_if_frozen_release(model_name: str): + refs = ( + os.environ.get("GITHUB_REF", ""), + os.environ.get("GITHUB_REF_NAME", ""), + os.environ.get("GITHUB_BASE_REF", ""), + ) + is_frozen_release = any( + ref.removeprefix("refs/heads/").startswith("release/") for ref in refs + ) + return pytest.mark.skipif( + is_frozen_release, + reason=f"{model_name} tests depend on resources fetched from main.", + ) + + def to_bfloat16( model: torch.nn.Module, inputs: tuple[Any, ...] ) -> tuple[torch.nn.Module, tuple[Any, ...]]: diff --git a/backends/arm/test/models/test_nfru.py b/backends/arm/test/models/test_nfru.py index 6122770da05..e00302fe971 100644 --- a/backends/arm/test/models/test_nfru.py +++ b/backends/arm/test/models/test_nfru.py @@ -8,11 +8,19 @@ os.environ.setdefault("HF_HUB_DISABLE_XET", "1") -import pytest - import torch +from executorch.backends.arm.scripts.neural_graphics_test_data import ( + _NFRU_INPUT_CHANNELS, + iter_nfru_test_calibration_samples, + load_nfru_verification_inputs, +) from executorch.backends.arm.test import common +from executorch.backends.arm.test.models.model_test_utils import ( + PTQ_AND_QAT_DATA, + REAL_AND_RANDOM_DATA, + skip_if_frozen_release, +) from executorch.backends.arm.test.tester.test_pipeline import ( TosaPipelineFP, TosaPipelineINT, @@ -25,20 +33,7 @@ input_t = Tuple[torch.Tensor] # Input x -_RELEASE_REFS = ( - os.environ.get("GITHUB_REF", ""), - os.environ.get("GITHUB_REF_NAME", ""), - os.environ.get("GITHUB_BASE_REF", ""), -) -_IS_FROZEN_RELEASE = any( - ref.removeprefix("refs/heads/").startswith("release/") for ref in _RELEASE_REFS -) -pytestmark = pytest.mark.skipif( - _IS_FROZEN_RELEASE, - reason="NFRU tests depend on resources fetched from main.", -) - -_NFRU_QAT_INPUTS = (torch.ones((1, 16, 64, 64)),) +pytestmark = skip_if_frozen_release("NFRU") def nfru() -> NFRUAutoEncoder: @@ -65,48 +60,86 @@ def nfru() -> NFRUAutoEncoder: return model -def example_inputs(memory_format: torch.memory_format = torch.channels_last): - return (torch.randn((1, 16, 270, 480)).to(memory_format=memory_format),) +def example_inputs(): + return load_nfru_verification_inputs() + + +def random_inputs(memory_format: torch.memory_format = torch.channels_last): + return ( + torch.randn((1, _NFRU_INPUT_CHANNELS, 270, 480)).to( + memory_format=memory_format + ), + ) + + +input_test_data = REAL_AND_RANDOM_DATA +is_qat_test_data = PTQ_AND_QAT_DATA -def test_nfru_tosa_FP(): +def _set_nfru_calibration_samples(pipeline): + return pipeline.set_quantization_calibration(iter_nfru_test_calibration_samples()) + + +@common.parametrize("use_real_data", input_test_data) +def test_nfru_tosa_FP(use_real_data): pipeline = TosaPipelineFP[input_t]( nfru().eval(), - example_inputs(), + example_inputs() if use_real_data else random_inputs(), aten_op=[], exir_op=[], ) pipeline.run() -def test_nfru_tosa_INT(): +@common.parametrize("is_qat", is_qat_test_data) +@common.parametrize("use_real_data", input_test_data) +def test_nfru_tosa_INT(use_real_data, is_qat): + pipeline_kwargs = ( + { + "per_channel_quantization": False, + "use_to_edge_transform_and_lower": True, + "frobenius_threshold": None, + "cosine_threshold": None, + } + if is_qat + else {} + ) pipeline = TosaPipelineINT[input_t]( nfru().eval(), - example_inputs(), + example_inputs() if use_real_data else random_inputs(), aten_op=[], exir_op=[], atol=0.2, + qtol=2 if use_real_data else 1, + is_qat=is_qat, + **pipeline_kwargs, ) + if use_real_data: + _set_nfru_calibration_samples(pipeline) pipeline.run() -def test_nfru_tosa_INT_a16w8(): +@common.parametrize("use_real_data", input_test_data) +def test_nfru_tosa_INT_a16w8(use_real_data): pipeline = TosaPipelineINT[input_t]( nfru().eval(), - example_inputs(), + example_inputs() if use_real_data else random_inputs(), aten_op=[], exir_op=[], tosa_extensions=["int16"], atol=0.1, ) + if use_real_data: + _set_nfru_calibration_samples(pipeline) pipeline.run() @common.SkipIfNoModelConverter -def test_nfru_vgf_no_quant(): +@common.parametrize("use_real_data", input_test_data) +def test_nfru_vgf_no_quant(use_real_data): pipeline = VgfPipeline[input_t]( nfru().eval(), - example_inputs(), + example_inputs() if use_real_data else random_inputs(), aten_op=[], exir_op=[], tosa_version="TOSA-1.0+FP", @@ -116,24 +149,42 @@ def test_nfru_vgf_no_quant(): @common.SkipIfNoModelConverter -def test_nfru_vgf_quant(): +@common.parametrize("is_qat", is_qat_test_data) +@common.parametrize("use_real_data", input_test_data) +def test_nfru_vgf_quant(use_real_data, is_qat): + pipeline_kwargs = ( + { + "run_on_vulkan_runtime": True, + "per_channel_quantization": False, + "use_to_edge_transform_and_lower": True, + } + if is_qat + else {} + ) pipeline = VgfPipeline[input_t]( nfru().eval(), - example_inputs(), + example_inputs() if use_real_data else random_inputs(), aten_op=[], exir_op=[], tosa_version="TOSA-1.0+INT", symmetric_io_quantization=True, + quantize=True, + is_qat=is_qat, atol=0.2, + qtol=2 if use_real_data else 1, + **pipeline_kwargs, ) + if use_real_data: + _set_nfru_calibration_samples(pipeline) pipeline.run() @common.SkipIfNoModelConverter -def test_nfru_vgf_quant_a16w8(): +@common.parametrize("use_real_data", input_test_data) +def test_nfru_vgf_quant_a16w8(use_real_data): pipeline = VgfPipeline[input_t]( nfru().eval(), - example_inputs(), + example_inputs() if use_real_data else random_inputs(), aten_op=[], exir_op=[], tosa_version="TOSA-1.0+INT", @@ -141,36 +192,6 @@ def test_nfru_vgf_quant_a16w8(): symmetric_io_quantization=True, atol=0.2, ) - pipeline.run() - - -def test_nfru_qat_tosa_INT() -> None: - pipeline = TosaPipelineINT( - nfru(), - _NFRU_QAT_INPUTS, - aten_op=[], - exir_op=[], - per_channel_quantization=False, - use_to_edge_transform_and_lower=True, - is_qat=True, - frobenius_threshold=None, - cosine_threshold=None, - ) - pipeline.run() - - -@common.SkipIfNoModelConverter -def test_nfru_qat_vgf_INT() -> None: - pipeline = VgfPipeline( - nfru(), - _NFRU_QAT_INPUTS, - aten_op=[], - exir_op=[], - run_on_vulkan_runtime=True, - quantize=True, - per_channel_quantization=False, - use_to_edge_transform_and_lower=True, - is_qat=True, - tosa_version="TOSA-1.0+INT", - ) + if use_real_data: + _set_nfru_calibration_samples(pipeline) pipeline.run() diff --git a/backends/arm/test/models/test_nss.py b/backends/arm/test/models/test_nss.py index 3010a98848a..8a5a5f0d5a6 100644 --- a/backends/arm/test/models/test_nss.py +++ b/backends/arm/test/models/test_nss.py @@ -3,19 +3,22 @@ # This source code is licensed under the BSD-style license found in the # LICENSE file in the root directory of this source tree. -import os -from pathlib import Path from typing import Tuple import pytest import torch from executorch.backends.arm.scripts.neural_graphics_test_data import ( - iter_calibration_samples, - load_verification_inputs, - nss_test_calibration_path, + _NSS_INPUT_CHANNELS, + iter_nss_test_calibration_samples, + load_nss_verification_inputs, ) from executorch.backends.arm.test import common +from executorch.backends.arm.test.models.model_test_utils import ( + PTQ_AND_QAT_DATA, + REAL_AND_RANDOM_DATA, + skip_if_frozen_release, +) from executorch.backends.arm.test.tester.test_pipeline import ( EthosU55PipelineINT, EthosU85PipelineINT, @@ -33,18 +36,7 @@ input_t = Tuple[torch.Tensor] # Input x -_RELEASE_REFS = ( - os.environ.get("GITHUB_REF", ""), - os.environ.get("GITHUB_REF_NAME", ""), - os.environ.get("GITHUB_BASE_REF", ""), -) -_IS_FROZEN_RELEASE = any( - ref.removeprefix("refs/heads/").startswith("release/") for ref in _RELEASE_REFS -) -pytestmark = pytest.mark.skipif( - _IS_FROZEN_RELEASE, - reason="NSS tests depend on resources fetched from main.", -) +pytestmark = skip_if_frozen_release("NSS") _NSS_HEIGHT = 8 * Dim("_nss_height", min=16, max=68) _NSS_WIDTH = 8 * Dim("_nss_width", min=16, max=120) @@ -80,36 +72,22 @@ def nss() -> AutoEncoderV1: def example_inputs(): - return load_verification_inputs() + return load_nss_verification_inputs() def random_inputs(): - return (torch.rand((1, 12, 544, 960)),) + return (torch.rand((1, _NSS_INPUT_CHANNELS, 544, 960)),) -input_test_data = { - "real_data": True, - "random_data": False, -} - - -def _nss_calibration_path() -> Path: - path = nss_test_calibration_path() - if not path.exists(): - raise RuntimeError( - "NSS calibration data is prepared by " - "backends/arm/scripts/install_models_for_test.sh." - ) - return path +input_test_data = REAL_AND_RANDOM_DATA +is_qat_test_data = PTQ_AND_QAT_DATA def _set_nss_calibration_samples(pipeline): - quantize_stage = pipeline._stages[pipeline.find_pos("quantize")].args[0] - quantize_stage.dynamic_shapes = _NSS_QUANTIZATION_DYNAMIC_SHAPES - quantize_stage.calibration_samples = iter_calibration_samples( - _nss_calibration_path(), num_samples=3663 + return pipeline.set_quantization_calibration( + iter_nss_test_calibration_samples(), + dynamic_shapes=_NSS_QUANTIZATION_DYNAMIC_SHAPES, ) - return pipeline @common.parametrize("use_real_data", input_test_data) @@ -126,17 +104,27 @@ def test_nss_tosa_FP(use_real_data): pipeline.run() +@common.parametrize("is_qat", is_qat_test_data) @common.parametrize("use_real_data", input_test_data) -def test_nss_tosa_INT(use_real_data): - pipeline_kwargs = ( - {"frobenius_threshold": 0.32, "qtol": 12} if use_real_data else {"qtol": 7} - ) +def test_nss_tosa_INT(use_real_data, is_qat): + if is_qat: + pipeline_kwargs = { + # Frobenius norm & cosine theshold check disabled for QAT as smoke test has innacurate results and only checks flow functionality. + "frobenius_threshold": None, + "cosine_threshold": None, + "qtol": 12 if use_real_data else 8, + } + else: + pipeline_kwargs = ( + {"frobenius_threshold": 0.32, "qtol": 12} if use_real_data else {"qtol": 7} + ) pipeline = TosaPipelineINT[input_t]( nss().eval(), example_inputs() if use_real_data else random_inputs(), aten_op=[], exir_op=[], use_to_edge_transform_and_lower=True, + is_qat=is_qat, **pipeline_kwargs, ) if use_real_data: @@ -194,8 +182,9 @@ def test_nss_vgf_FP(use_real_data): @common.SkipIfNoModelConverter +@common.parametrize("is_qat", is_qat_test_data) @common.parametrize("use_real_data", input_test_data) -def test_nss_vgf_INT(use_real_data): +def test_nss_vgf_INT(use_real_data, is_qat): pipeline = VgfPipeline[input_t]( nss().eval(), example_inputs() if use_real_data else random_inputs(), @@ -205,45 +194,11 @@ def test_nss_vgf_INT(use_real_data): use_to_edge_transform_and_lower=True, run_on_vulkan_runtime=True, quantize=True, + is_qat=is_qat, # Override tosa version to test INT-only path tosa_version="TOSA-1.0+INT", - qtol=12 if use_real_data else 7, + qtol=12 if use_real_data else (8 if is_qat else 7), ) if use_real_data: _set_nss_calibration_samples(pipeline) pipeline.run() - - -def test_nss_qat_tosa_INT() -> None: - pipeline = TosaPipelineINT[input_t]( - nss().eval(), - example_inputs(), - aten_op=[], - exir_op=[], - use_to_edge_transform_and_lower=True, - is_qat=True, - frobenius_threshold=None, - cosine_threshold=None, - qtol=12, - ) - _set_nss_calibration_samples(pipeline) - pipeline.run() - - -@common.SkipIfNoModelConverter -def test_nss_qat_vgf_INT() -> None: - pipeline = VgfPipeline[input_t]( - nss().eval(), - example_inputs(), - aten_op=[], - exir_op=[], - symmetric_io_quantization=True, - use_to_edge_transform_and_lower=True, - run_on_vulkan_runtime=True, - quantize=True, - is_qat=True, - tosa_version="TOSA-1.0+INT", - qtol=12, - ) - _set_nss_calibration_samples(pipeline) - pipeline.run()