(
model,
test_dir,
test_name,
calibration_dataset_dir,
testing_dataset_dir,
input_spec,
dlg_model_verifier,
npu_results_dir,
mocker,
use_qat: bool = False,
train_fn: Callable[[torch.fx.GraphModule], None] | None = None,
use_new_flow_neutron_c: bool = False,
)
| 62 | |
| 63 | |
| 64 | def _run_delegated_executorch_program( |
| 65 | model, |
| 66 | test_dir, |
| 67 | test_name, |
| 68 | calibration_dataset_dir, |
| 69 | testing_dataset_dir, |
| 70 | input_spec, |
| 71 | dlg_model_verifier, |
| 72 | npu_results_dir, |
| 73 | mocker, |
| 74 | use_qat: bool = False, |
| 75 | train_fn: Callable[[torch.fx.GraphModule], None] | None = None, |
| 76 | use_new_flow_neutron_c: bool = False, |
| 77 | ) -> ExportedProgram: |
| 78 | if len(input_spec) == 1: |
| 79 | # Single input, use --dataset |
| 80 | dataset_cli = "--dataset" |
| 81 | dataset_or_inputs = testing_dataset_dir |
| 82 | else: |
| 83 | # Multiple input, use --inputs with subdirectories |
| 84 | dataset_cli = "--inputs" |
| 85 | dataset_or_inputs = ",".join( |
| 86 | sorted( |
| 87 | [ |
| 88 | os.path.join(testing_dataset_dir, d) |
| 89 | for d in os.listdir(testing_dataset_dir) |
| 90 | ] |
| 91 | ) |
| 92 | ) |
| 93 | |
| 94 | # Run nxp_executor_runner with program delegated to NPU |
| 95 | delegated_model_path = os.path.abspath( |
| 96 | os.path.join(test_dir, f"{test_name}_delegated.pte") |
| 97 | ) |
| 98 | |
| 99 | delegated_cmd = f"{NEUTRON_TEST_PATH} --model {delegated_model_path} {dataset_cli} {dataset_or_inputs} \ |
| 100 | --output {npu_results_dir} --firmware {NSYS_FIRMWARE_PATH} --nsys {NSYS_PATH} --nsys_config {NSYS_CONFIG_PATH}" |
| 101 | try: |
| 102 | if mocker: |
| 103 | method = getattr(NeutronPartitioner, "partition") # noqa B009 |
| 104 | |
| 105 | def wrapper(*args, **kwargs): |
| 106 | result = method(*args, **kwargs) |
| 107 | visualize_with_clusters( |
| 108 | result.tagged_exported_program, |
| 109 | os.path.join(test_dir, test_name + "_partitioned.json"), |
| 110 | False, |
| 111 | ) |
| 112 | return result |
| 113 | |
| 114 | wrapped = functools.update_wrapper(wrapper, method) |
| 115 | mocker.patch.object( |
| 116 | NeutronPartitioner, "partition", side_effect=wrapped, autospec=True |
| 117 | ) |
| 118 | delegated_program = to_quantized_executorch_program( |
| 119 | model, |
| 120 | input_spec, |
| 121 | dataset_dir=calibration_dataset_dir, |
no test coverage detected