Added Python interface for Arm Ethos-U NPU driver library.
Python `ethosu_driver` could be built as part of Arm Ethos-U Linux
driver library CMake flow.
See driver_library/python/README.md for more details.
Change-Id: I177a890add5c13df9a839f4f43621f972afe5ab1
Signed-off-by: Kshitij Sisodia <kshitij.sisodia@arm.com>
diff --git a/driver_library/python/test/conftest.py b/driver_library/python/test/conftest.py
new file mode 100644
index 0000000..8eef1f8
--- /dev/null
+++ b/driver_library/python/test/conftest.py
@@ -0,0 +1,34 @@
+#
+# SPDX-FileCopyrightText: Copyright 2021-2022 Arm Limited and/or its affiliates <open-source-office@arm.com>
+# SPDX-License-Identifier: Apache-2.0
+#
+import os
+import pytest
+
+
+@pytest.fixture(scope="module")
+def data_folder_per_test(request):
+ """
+ This fixture returns path to folder with test resources (one per test module)
+ """
+
+ basedir, script = request.fspath.dirname, request.fspath.basename
+ return str(os.path.join(basedir, "testdata", os.path.splitext(script)[0]))
+
+
+@pytest.fixture(scope="module")
+def shared_data_folder(request):
+ """
+ This fixture returns path to folder with shared test resources among all tests
+ """
+
+ return str(os.path.join(request.fspath.dirname, "testdata", "shared"))
+
+
+@pytest.fixture(scope="function")
+def tmpdir(tmpdir):
+ """
+ This fixture returns path to temp folder. Fixture was added for py35 compatibility
+ """
+
+ return str(tmpdir)
diff --git a/driver_library/python/test/test_capabilities.py b/driver_library/python/test/test_capabilities.py
new file mode 100644
index 0000000..ffb201c
--- /dev/null
+++ b/driver_library/python/test/test_capabilities.py
@@ -0,0 +1,73 @@
+#
+# SPDX-FileCopyrightText: Copyright 2021-2022 Arm Limited and/or its affiliates <open-source-office@arm.com>
+# SPDX-License-Identifier: Apache-2.0
+#
+from ethosu_driver._generated.driver import SemanticVersion
+from ethosu_driver._generated.driver import HardwareId
+from ethosu_driver._generated.driver import HardwareConfiguration
+from ethosu_driver._generated.driver import Capabilities
+
+
+def check_semantic_version(ma, mi, pa, sv):
+ assert ma == sv.major
+ assert mi == sv.minor
+ assert pa == sv.patch
+
+
+def test_semantic_version():
+ sv = SemanticVersion(1, 2, 3)
+ assert '{ major=1, minor=2, patch=3 }' == sv.__str__()
+ check_semantic_version(1, 2, 3, sv)
+
+
+def test_hardware_id():
+ version = SemanticVersion(1, 2, 3)
+ product = SemanticVersion(4, 5, 6)
+ architecture = SemanticVersion(7, 8, 9)
+ hw_id = HardwareId(1, version, product, architecture)
+
+ assert 1 == hw_id.versionStatus
+
+ check_semantic_version(1, 2, 3, hw_id.version)
+ check_semantic_version(4, 5, 6, hw_id.product)
+ check_semantic_version(7, 8, 9, hw_id.architecture)
+
+ assert '{versionStatus=1, version={ major=1, minor=2, patch=3 }, product={ major=4, minor=5, patch=6 }, ' \
+ 'architecture={ major=7, minor=8, patch=9 }}' == hw_id.__str__()
+
+
+def test_hw_configuration():
+ hw_cfg = HardwareConfiguration(128, 1, True)
+
+ assert 1 == hw_cfg.cmdStreamVersion
+ assert 128 == hw_cfg.macsPerClockCycle
+ assert hw_cfg.customDma
+
+ assert "{macsPerClockCycle=128, cmdStreamVersion=1, customDma=True}" == hw_cfg.__str__()
+
+
+def test_capabilities():
+ version = SemanticVersion(100, 200, 300)
+ product = SemanticVersion(400, 500, 600)
+ architecture = SemanticVersion(700, 800, 900)
+ hw_id = HardwareId(1, version, product, architecture)
+ hw_cfg = HardwareConfiguration(256, 1000, False)
+ driver_v = SemanticVersion(10, 20, 30)
+
+ cap = Capabilities(hw_id, hw_cfg, driver_v)
+
+ check_semantic_version(10, 20, 30, cap.driver)
+
+ check_semantic_version(100, 200, 300, cap.hwId.version)
+ check_semantic_version(400, 500, 600, cap.hwId.product)
+ check_semantic_version(700, 800, 900, cap.hwId.architecture)
+
+ assert 1000 == cap.hwCfg.cmdStreamVersion
+ assert 256 == cap.hwCfg.macsPerClockCycle
+ assert not cap.hwCfg.customDma
+
+ assert '{hwId={versionStatus=1, version={ major=100, minor=200, patch=300 }, ' \
+ 'product={ major=400, minor=500, patch=600 }, ' \
+ 'architecture={ major=700, minor=800, patch=900 }}, ' \
+ 'hwCfg={macsPerClockCycle=256, cmdStreamVersion=1000, customDma=False}, ' \
+ 'driver={ major=10, minor=20, patch=30 }}' == cap.__str__()
diff --git a/driver_library/python/test/test_driver.py b/driver_library/python/test/test_driver.py
new file mode 100644
index 0000000..5496aed
--- /dev/null
+++ b/driver_library/python/test/test_driver.py
@@ -0,0 +1,179 @@
+#
+# SPDX-FileCopyrightText: Copyright 2021-2022 Arm Limited and/or its affiliates <open-source-office@arm.com>
+# SPDX-License-Identifier: Apache-2.0
+#
+import pytest
+import os
+import ethosu_driver as driver
+from ethosu_driver.inference_runner import read_npy_file_to_buf
+
+
+@pytest.fixture()
+def device(device_name):
+ device = driver.Device("/dev/{}".format(device_name))
+ yield device
+
+
+@pytest.fixture()
+def network_buffer(device, model_name, shared_data_folder):
+ network_file = os.path.join(shared_data_folder, model_name)
+ network_buffer = driver.Buffer(device, network_file)
+ yield network_buffer
+
+
+@pytest.mark.parametrize('device_name', ['ethosu0'])
+def test_check_device_swig_ownership(device):
+ # Check to see that SWIG has ownership for parser. This instructs SWIG to take
+ # ownership of the return value. This allows the value to be automatically
+ # garbage-collected when it is no longer in use
+ assert device.thisown
+
+
+@pytest.mark.parametrize('device_name', ['ethosu0'])
+def test_device_ping(device):
+ device.ping()
+
+
+@pytest.mark.parametrize('device_name', ['blabla'])
+def test_device_wrong_name(device_name):
+ with pytest.raises(RuntimeError) as err:
+ driver.Device("/dev/{}".format(device_name))
+ # Only check for part of the exception since the exception returns
+ # absolute path which will change on different machines.
+ assert 'Failed to open device' in str(err.value)
+
+
+@pytest.mark.parametrize('device_name', ['ethosu0'])
+def test_driver_network_filenotfound_exception(device, shared_data_folder):
+
+ network_file = os.path.join(shared_data_folder, "some_unknown_model.tflite")
+
+ with pytest.raises(RuntimeError) as err:
+ network_buffer = driver.Buffer(device, network_file)
+
+ # Only check for part of the exception since the exception returns
+ # absolute path which will change on different machines.
+ assert 'Failed to open file:' in str(err.value)
+
+
+@pytest.mark.parametrize('device_name', ['ethosu0'])
+@pytest.mark.parametrize('model_name', ['model.tflite'])
+def test_check_buffer_swig_ownership(network_buffer):
+ # Check to see that SWIG has ownership for parser. This instructs SWIG to take
+ # ownership of the return value. This allows the value to be automatically
+ # garbage-collected when it is no longer in use
+ assert network_buffer.thisown
+
+
+@pytest.mark.parametrize('device_name', ['ethosu0'])
+@pytest.mark.parametrize('model_name', ['model.tflite'])
+def test_check_buffer_capacity(network_buffer):
+ assert network_buffer.capacity() > 0
+
+
+@pytest.mark.parametrize('device_name', ['ethosu0'])
+@pytest.mark.parametrize('model_name', ['model.tflite'])
+def test_check_buffer_size(network_buffer):
+ assert network_buffer.size() > 0
+
+
+@pytest.mark.parametrize('device_name', ['ethosu0'])
+@pytest.mark.parametrize('model_name', ['model.tflite'])
+def test_check_buffer_clear(network_buffer):
+ network_buffer.clear()
+ assert network_buffer.size() == 0
+
+
+@pytest.mark.parametrize('device_name', ['ethosu0'])
+@pytest.mark.parametrize('model_name', ['model.tflite'])
+def test_check_buffer_resize(network_buffer):
+ offset = 1
+ new_size = network_buffer.capacity() - offset
+ network_buffer.resize(new_size, offset)
+ assert network_buffer.size() == new_size
+
+
+@pytest.mark.parametrize('device_name', ['ethosu0'])
+@pytest.mark.parametrize('model_name', ['model.tflite'])
+def test_check_buffer_getFd(network_buffer):
+ assert network_buffer.getFd() >= 0
+
+
+@pytest.mark.parametrize('device_name', ['ethosu0'])
+@pytest.mark.parametrize('model_name', ['model.tflite'])
+def test_check_network_ifm_size(device, network_buffer):
+ network = driver.Network(device, network_buffer)
+ assert network.getIfmSize() > 0
+ assert network_buffer.thisown
+
+
+@pytest.mark.parametrize('device_name', [('ethosu0')])
+def test_check_network_buffer_none(device):
+
+ with pytest.raises(RuntimeError) as err:
+ driver.Network(device, None)
+
+ # Only check for part of the exception since the exception returns
+ # absolute path which will change on different machines.
+ assert 'Failed to create the network' in str(err.value)
+
+
+@pytest.mark.parametrize('device_name', ['ethosu0'])
+@pytest.mark.parametrize('model_name', ['model.tflite'])
+def test_check_network_ofm_size(device, network_buffer):
+ network = driver.Network(device, network_buffer)
+ assert network.getOfmSize() > 0
+
+
+def test_getMaxPmuEventCounters():
+ assert driver.Inference.getMaxPmuEventCounters() > 0
+
+
+@pytest.fixture()
+def inf(device_name, model_name, input_files, timeout, shared_data_folder):
+ # Prepate full path of model and inputs
+ full_path_model_file = os.path.join(shared_data_folder, model_name)
+ full_path_input_files = []
+ for input_file in input_files:
+ full_path_input_files.append(os.path.join(shared_data_folder, input_file))
+
+ ifms_data = []
+ for ifm_file in full_path_input_files:
+ ifms_data.append(read_npy_file_to_buf(ifm_file))
+
+ device = driver.open_device(device_name)
+ device.ping()
+ network = driver.load_model(device, full_path_model_file)
+ ofms = driver.allocate_buffers(device, network.getOfmDims())
+ ifms = driver.allocate_buffers(device, network.getIfmDims())
+
+ # ofm_buffers = runner.run(ifms_data,timeout, ethos_pmu_counters)
+ driver.populate_buffers(ifms_data, ifms)
+ ethos_pmu_counters = [1]
+ enable_cycle_counter = True
+ inf_inst = driver.Inference(network, ifms, ofms, ethos_pmu_counters, enable_cycle_counter)
+ inf_inst.wait(int(timeout))
+
+ yield inf_inst
+
+
+@pytest.mark.parametrize('device_name, model_name, timeout, input_files',
+ [('ethosu0', 'model.tflite', 5000000000, ['model_ifm.npy'])])
+def test_inf_get_cycle_counter(inf):
+ total_cycles = inf.getCycleCounter()
+ assert total_cycles >= 0
+
+
+@pytest.mark.parametrize('device_name, model_name, timeout, input_files',
+ [('ethosu0', 'model.tflite', 5000000000, ['model_ifm.npy'])])
+def test_inf_get_pmu_counters(inf):
+ inf_pmu_counter = inf.getPmuCounters()
+ assert len(inf_pmu_counter) > 0
+
+
+@pytest.mark.parametrize('device_name', ['ethosu0'])
+def test_capabilities(device):
+ cap = device.capabilities()
+ assert cap.hwId
+ assert cap.hwCfg
+ assert cap.driver
diff --git a/driver_library/python/test/test_driver_utilities.py b/driver_library/python/test/test_driver_utilities.py
new file mode 100644
index 0000000..fc8e921
--- /dev/null
+++ b/driver_library/python/test/test_driver_utilities.py
@@ -0,0 +1,77 @@
+#
+# SPDX-FileCopyrightText: Copyright 2021-2022 Arm Limited and/or its affiliates <open-source-office@arm.com>
+# SPDX-License-Identifier: Apache-2.0
+#
+import pytest
+import os
+import ethosu_driver as driver
+from ethosu_driver.inference_runner import read_npy_file_to_buf
+
+
+@pytest.fixture()
+def device(device_name):
+ device = driver.open_device(device_name)
+ yield device
+
+
+@pytest.fixture()
+def network(device, model_name, shared_data_folder):
+ network_file = os.path.join(shared_data_folder, model_name)
+ network = driver.load_model(device, network_file)
+ yield network
+
+
+@pytest.mark.parametrize('device_name', ['blabla'])
+def test_open_device_wrong_name(device_name):
+ with pytest.raises(RuntimeError) as err:
+ device = driver.open_device(device_name)
+ # Only check for part of the exception since the exception returns
+ # absolute path which will change on different machines.
+ assert 'Failed to open device' in str(err.value)
+
+
+@pytest.mark.parametrize('device_name', ['ethosu0'])
+def test_network_filenotfound_exception(device, shared_data_folder):
+
+ network_file = os.path.join(shared_data_folder, "some_unknown_model.tflite")
+
+ with pytest.raises(RuntimeError) as err:
+ driver.load_model(device, network_file)
+
+ # Only check for part of the exception since the exception returns
+ # absolute path which will change on different machines.
+ assert 'Failed to open file:' in str(err.value)
+
+
+@pytest.mark.parametrize('device_name', ['ethosu0'])
+@pytest.mark.parametrize('model_name', ['model.tflite'])
+def test_check_network_ifm_size(network):
+ assert network.getIfmSize() > 0
+
+
+@pytest.mark.parametrize('device_name', ['ethosu0'])
+def test_allocate_buffers(device):
+ buffers = driver.allocate_buffers(device, [128, 256])
+ assert len(buffers) == 2
+ assert buffers[0].size() == 0
+ assert buffers[0].capacity() == 128
+ assert buffers[1].size() == 0
+ assert buffers[1].capacity() == 256
+
+
+@pytest.mark.parametrize('device_name', ['ethosu0'])
+@pytest.mark.parametrize('model_name', ['model.tflite'])
+@pytest.mark.parametrize('ifms_file_list', [['model_ifm.npy']])
+def test_set_ifm_buffers(device, network, ifms_file_list, shared_data_folder):
+ full_path_input_files = []
+ for input_file in ifms_file_list:
+ full_path_input_files.append(os.path.join(shared_data_folder, input_file))
+
+ ifms_data = []
+ for ifm_file in full_path_input_files:
+ ifms_data.append(read_npy_file_to_buf(ifm_file))
+
+ ifms = driver.allocate_buffers(device, network.getIfmDims())
+ driver.populate_buffers(ifms_data, ifms)
+ assert len(ifms) > 0
+
diff --git a/driver_library/python/test/test_inference.py b/driver_library/python/test/test_inference.py
new file mode 100644
index 0000000..bfb4068
--- /dev/null
+++ b/driver_library/python/test/test_inference.py
@@ -0,0 +1,50 @@
+#
+# SPDX-FileCopyrightText: Copyright 2021-2022 Arm Limited and/or its affiliates <open-source-office@arm.com>
+# SPDX-License-Identifier: Apache-2.0
+#
+import pytest
+import os
+import ethosu_driver as driver
+from ethosu_driver.inference_runner import read_npy_file_to_buf
+
+
+def run_inference_test(runner, timeout, input_files, golden_outputs, shared_data_folder):
+
+ full_path_input_files = []
+ for input_file in input_files:
+ full_path_input_files.append(os.path.join(shared_data_folder, input_file))
+
+ ifms_data = []
+ for ifm_file in full_path_input_files:
+ ifms_data.append(read_npy_file_to_buf(ifm_file))
+
+ ofm_buffers = runner.run(ifms_data, timeout)
+
+ for index, buffer_out in enumerate(ofm_buffers):
+ golden_output = read_npy_file_to_buf(os.path.join(shared_data_folder, golden_outputs[index]))
+ assert buffer_out.data().nbytes == golden_output.nbytes
+ for index, golden_value in enumerate(golden_output):
+ assert golden_value == buffer_out.data()[index]
+
+
+@pytest.mark.parametrize('device_name, model_name, timeout, input_files, golden_outputs',
+ [('ethosu0', 'model.tflite', 5000000000, ['model_ifm.npy'], ['model_ofm.npy'])])
+def test_inference(device_name, model_name, input_files, timeout, golden_outputs, shared_data_folder):
+ # Prepate full path of model and inputs
+ full_path_model_file = os.path.join(shared_data_folder, model_name)
+
+ runner = driver.InferenceRunner(device_name, full_path_model_file)
+ run_inference_test(runner, timeout, input_files, golden_outputs, shared_data_folder)
+
+
+@pytest.mark.parametrize('device_name, model_name, timeout, input_files, golden_outputs',
+ [('ethosu0', 'model.tflite', 5000000000,
+ [['model_ifm.npy'], ['model_ifm.npy']],
+ [['model_ofm.npy'], ['model_ofm.npy']])])
+def test_inference_loop(device_name, model_name, input_files, timeout, golden_outputs, shared_data_folder):
+ # Prepare full path of model and inputs
+ full_path_model_file = os.path.join(shared_data_folder, model_name)
+
+ runner = driver.InferenceRunner(device_name, full_path_model_file)
+ for input_file, golden_output in zip(input_files, golden_outputs):
+ run_inference_test(runner, timeout, input_file, golden_output, shared_data_folder)
diff --git a/driver_library/python/test/test_shadow_classes.py b/driver_library/python/test/test_shadow_classes.py
new file mode 100644
index 0000000..055c10b
--- /dev/null
+++ b/driver_library/python/test/test_shadow_classes.py
@@ -0,0 +1,20 @@
+#
+# SPDX-FileCopyrightText: Copyright 2021-2022 Arm Limited and/or its affiliates <open-source-office@arm.com>
+# SPDX-License-Identifier: Apache-2.0
+#
+import inspect
+import pytest
+import ethosu_driver._generated.driver as driver_shadow
+
+
+def get_classes():
+ ignored_class_names = ('_SwigNonDynamicMeta', '_object', '_swig_property')
+ return list(filter(lambda x: x[0] not in ignored_class_names,
+ inspect.getmembers(driver_shadow, inspect.isclass)))
+
+
+@pytest.mark.parametrize("class_instance", get_classes(), ids=lambda x: 'class={}'.format(x[0]))
+class TestOwnership:
+
+ def test_destructors_exist_per_class(self, class_instance):
+ assert getattr(class_instance[1], '__swig_destroy__', None)
diff --git a/driver_library/python/test/testdata/download.py b/driver_library/python/test/testdata/download.py
new file mode 100644
index 0000000..18aa9af
--- /dev/null
+++ b/driver_library/python/test/testdata/download.py
@@ -0,0 +1,46 @@
+#
+# SPDX-FileCopyrightText: Copyright 2021-2022 Arm Limited and/or its affiliates <open-source-office@arm.com>
+# SPDX-License-Identifier: Apache-2.0
+#
+import os
+from pathlib import Path
+from typing import List
+from urllib.request import urlopen
+"""
+Downloads resources for tests from Arm public model zoo.
+Run this script before executing tests.
+"""
+
+
+PMZ_URL = 'https://github.com/ARM-software/ML-zoo/raw/9f506fe52b39df545f0e6c5ff9223f671bc5ae00/models'
+test_resources = [
+ {'model': '{}/visual_wake_words/micronet_vww2/tflite_int8/vww2_50_50_INT8.tflite'.format(PMZ_URL),
+ 'ifm': '{}/visual_wake_words/micronet_vww2/tflite_int8/testing_input/input/0.npy'.format(PMZ_URL),
+ 'ofm': '{}/visual_wake_words/micronet_vww2/tflite_int8/testing_output/Identity/0.npy'.format(PMZ_URL)}
+]
+
+
+def download(path: str, url: str):
+ with urlopen(url) as response, open(path, 'wb') as file:
+ print("Downloading {} ...".format(url))
+ file.write(response.read())
+ file.seek(0)
+ print("Finished downloading {}.".format(url))
+
+
+def download_test_resources(test_res_entries: List[dict], where_to: str):
+ os.makedirs(where_to, exist_ok=True)
+
+ for resources in test_res_entries:
+ download(os.path.join(where_to, 'model.tflite'), resources['model'])
+ download(os.path.join(where_to, 'model_ifm.npy'), resources['ifm'])
+ download(os.path.join(where_to, 'model_ofm.npy'), resources['ofm'])
+
+
+def main():
+ current_dir = str(Path(__file__).parent.absolute())
+ download_test_resources(test_resources, os.path.join(current_dir, 'shared'))
+
+
+if __name__ == '__main__':
+ main()