inference-engine/ie_bridges/python/tests/test_ExecutableNetwork.py

   1 import numpy as np
   2 import os
   3 import pytest
   4 import warnings
   5
   6 from openvino.inference_engine import ie_api as ie
   7 from conftest import model_path, image_path
   8
   9
  10 is_myriad = os.environ.get("TEST_DEVICE") == "MYRIAD"
  11 path_to_image = image_path()
  12 test_net_xml, test_net_bin = model_path(is_myriad)
  13
  14
  15 def read_image():
  16     import cv2
  17     n, c, h, w = (1, 3, 32, 32)
  18     image = cv2.imread(path_to_image)
  19     if image is None:
  20         raise FileNotFoundError("Input image not found")
  21
  22     image = cv2.resize(image, (h, w)) / 255
  23     image = image.transpose((2, 0, 1))
  24     image = image.reshape((n, c, h, w))
  25     return image
  26
  27
  28 def test_infer(device):
  29     ie_core = ie.IECore()
  30     net = ie_core.read_network(model=test_net_xml, weights=test_net_bin)
  31     exec_net = ie_core.load_network(net, device)
  32     img = read_image()
  33     res = exec_net.infer({'data': img})
  34     assert np.argmax(res['fc_out'][0]) == 2
  35     del exec_net
  36     del ie_core
  37
  38
  39 def test_infer_net_from_buffer(device):
  40     ie_core = ie.IECore()
  41     with open(test_net_bin, 'rb') as f:
  42         bin = f.read()
  43     with open(test_net_xml, 'rb') as f:
  44         xml = f.read()
  45     net = ie_core.read_network(model=xml, weights=bin, init_from_buffer=True)
  46     net2 = ie_core.read_network(model=test_net_xml, weights=test_net_bin)
  47     exec_net = ie_core.load_network(net, device)
  48     exec_net2 = ie_core.load_network(net2, device)
  49     img = read_image()
  50     res = exec_net.infer({'data': img})
  51     res2 = exec_net2.infer({'data': img})
  52     del ie_core
  53     del exec_net
  54     del exec_net2
  55     assert np.allclose(res['fc_out'], res2['fc_out'], atol=1E-4, rtol=1E-4)
  56
  57
  58 def test_infer_wrong_input_name(device):
  59     ie_core = ie.IECore()
  60     net = ie_core.read_network(model=test_net_xml, weights=test_net_bin)
  61     exec_net = ie_core.load_network(net, device)
  62     img = read_image()
  63     with pytest.raises(AssertionError) as e:
  64         exec_net.infer({'_data_': img})
  65     assert "No input with name _data_ found in network" in str(e.value)
  66     del exec_net
  67     del ie_core
  68
  69
  70 def test_input_info(device):
  71     ie_core = ie.IECore()
  72     net = ie_core.read_network(model=test_net_xml, weights=test_net_bin)
  73     exec_net = ie_core.load_network(net, device, num_requests=5)
  74     assert isinstance(exec_net.input_info['data'], ie.InputInfoCPtr)
  75     assert exec_net.input_info['data'].name == "data"
  76     assert exec_net.input_info['data'].precision == "FP32"
  77     assert isinstance(exec_net.input_info['data'].input_data, ie.DataPtr)
  78     del exec_net
  79     del ie_core
  80
  81
  82 def test_inputs_deprecated(device):
  83     ie_core = ie.IECore()
  84     net = ie_core.read_network(model=test_net_xml, weights=test_net_bin)
  85     exec_net = ie_core.load_network(net, device, num_requests=5)
  86     with warnings.catch_warnings(record=True) as w:
  87         assert len(exec_net.inputs) == 1
  88         assert "data" in exec_net.inputs
  89         assert isinstance(exec_net.inputs['data'], ie.DataPtr)
  90     assert len(w) == 3
  91     for i in range (len(w)):
  92         assert "'inputs' property of ExecutableNetwork class is deprecated. " \
  93             "To access DataPtrs user need to use 'input_data' property " \
  94             "of InputInfoCPtr objects which " \
  95             "can be accessed by 'input_info' property." in str(w[i].message)
  96     del exec_net
  97     del ie_core
  98
  99
 100 def test_outputs(device):
 101     ie_core = ie.IECore()
 102     net = ie_core.read_network(model=test_net_xml, weights=test_net_bin)
 103     exec_net = ie_core.load_network(net, device, num_requests=5)
 104     assert len(exec_net.outputs) == 1
 105     assert "fc_out" in exec_net.outputs
 106     assert isinstance(exec_net.outputs['fc_out'], ie.CDataPtr)
 107     del exec_net
 108     del ie_core
 109
 110
 111 def test_access_requests(device):
 112     ie_core = ie.IECore()
 113     net = ie_core.read_network(model=test_net_xml, weights=test_net_bin)
 114     exec_net = ie_core.load_network(net, device, num_requests=5)
 115     assert len(exec_net.requests) == 5
 116     assert isinstance(exec_net.requests[0], ie.InferRequest)
 117     del exec_net
 118     del ie_core
 119
 120
 121 def test_async_infer_one_req(device):
 122     ie_core = ie.IECore()
 123     net = ie_core.read_network(model=test_net_xml, weights=test_net_bin)
 124     exec_net = ie_core.load_network(net, device, num_requests=1)
 125     img = read_image()
 126     request_handler = exec_net.start_async(request_id=0, inputs={'data': img})
 127     request_handler.wait()
 128     res = request_handler.output_blobs['fc_out'].buffer
 129     assert np.argmax(res) == 2
 130     del exec_net
 131     del ie_core
 132
 133
 134 def test_async_infer_many_req(device):
 135     ie_core = ie.IECore()
 136     net = ie_core.read_network(model=test_net_xml, weights=test_net_bin)
 137     exec_net = ie_core.load_network(net, device, num_requests=5)
 138     img = read_image()
 139     for id in range(5):
 140         request_handler = exec_net.start_async(request_id=id, inputs={'data': img})
 141         request_handler.wait()
 142         res = request_handler.output_blobs['fc_out'].buffer
 143         assert np.argmax(res) == 2
 144     del exec_net
 145     del ie_core
 146
 147
 148 def test_async_infer_many_req_get_idle(device):
 149     ie_core = ie.IECore()
 150     net = ie_core.read_network(model=test_net_xml, weights=test_net_bin)
 151     num_requests = 5
 152     exec_net = ie_core.load_network(net, device, num_requests=num_requests)
 153     img = read_image()
 154     check_id = set()
 155     for id in range(2*num_requests):
 156         request_id = exec_net.get_idle_request_id()
 157         if request_id == -1:
 158             status = exec_net.wait(num_requests=1, timeout=ie.WaitMode.RESULT_READY)
 159             assert(status == ie.StatusCode.OK)
 160         request_id = exec_net.get_idle_request_id()
 161         assert(request_id >= 0)
 162         request_handler = exec_net.start_async(request_id=request_id, inputs={'data': img})
 163         check_id.add(request_id)
 164     status = exec_net.wait(timeout=ie.WaitMode.RESULT_READY)
 165     assert status == ie.StatusCode.OK
 166     for id in range(num_requests):
 167         if id in check_id:
 168             assert np.argmax(exec_net.requests[id].output_blobs['fc_out'].buffer) == 2
 169     del exec_net
 170     del ie_core
 171
 172
 173 def test_wait_before_start(device):
 174   ie_core = ie.IECore()
 175   net = ie_core.read_network(model=test_net_xml, weights=test_net_bin)
 176   num_requests = 5
 177   exec_net = ie_core.load_network(net, device, num_requests=num_requests)
 178   img = read_image()
 179   requests = exec_net.requests
 180   for id in range(num_requests):
 181       status = requests[id].wait()
 182       assert status == ie.StatusCode.INFER_NOT_STARTED
 183       request_handler = exec_net.start_async(request_id=id, inputs={'data': img})
 184       status = requests[id].wait()
 185       assert status == ie.StatusCode.OK
 186       assert np.argmax(request_handler.output_blobs['fc_out'].buffer) == 2
 187   del exec_net
 188   del ie_core
 189
 190
 191 def test_wrong_request_id(device):
 192     ie_core = ie.IECore()
 193     net = ie_core.read_network(model=test_net_xml, weights=test_net_bin)
 194     exec_net = ie_core.load_network(net, device, num_requests=1)
 195     img = read_image()
 196     with pytest.raises(ValueError) as e:
 197         exec_net.start_async(request_id=20, inputs={'data': img})
 198     assert "Incorrect request_id specified!" in str(e.value)
 199     del exec_net
 200     del ie_core
 201
 202
 203 def test_wrong_num_requests(device):
 204     with pytest.raises(ValueError) as e:
 205         ie_core = ie.IECore()
 206         net = ie_core.read_network(model=test_net_xml, weights=test_net_bin)
 207         ie_core.load_network(net, device, num_requests=-1)
 208         assert "Incorrect number of requests specified: -1. Expected positive integer number or zero for auto detection" \
 209            in str(e.value)
 210         del ie_core
 211
 212 def test_wrong_num_requests_core(device):
 213     with pytest.raises(ValueError) as e:
 214         ie_core = ie.IECore()
 215         net = ie_core.read_network(model=test_net_xml, weights=test_net_bin)
 216         exec_net = ie_core.load_network(net, device, num_requests=-1)
 217         assert "Incorrect number of requests specified: -1. Expected positive integer number or zero for auto detection" \
 218            in str(e.value)
 219         del ie_core
 220
 221 def test_plugin_accessible_after_deletion(device):
 222     ie_core = ie.IECore()
 223     net = ie_core.read_network(model=test_net_xml, weights=test_net_bin)
 224     exec_net = ie_core.load_network(net, device)
 225     img = read_image()
 226     res = exec_net.infer({'data': img})
 227     assert np.argmax(res['fc_out'][0]) == 2
 228     del exec_net
 229     del ie_core
 230
 231
 232 def test_exec_graph(device):
 233     ie_core = ie.IECore()
 234     net = ie_core.read_network(model=test_net_xml, weights=test_net_bin)
 235     exec_net = ie_core.load_network(net, device)
 236     img = read_image()
 237     res = exec_net.infer({'data': img})
 238     exec_graph = exec_net.get_exec_graph_info()
 239     exec_graph_file = 'exec_graph.xml'
 240     exec_graph.serialize(exec_graph_file)
 241     assert os.path.exists(exec_graph_file)
 242     os.remove(exec_graph_file)
 243     del exec_net
 244     del exec_graph
 245     del ie_core
 246
 247
 248 @pytest.mark.skipif(os.environ.get("TEST_DEVICE", "CPU") != "MYRIAD", reason="Device specific test. "
 249                                                                              "Only MYRIAD plugin implements network export")
 250 def test_export_import():
 251     ie_core = ie.IECore()
 252     net = ie_core.read_network(model=test_net_xml, weights=test_net_bin)
 253     exec_net = ie_core.load_network(net, "MYRIAD")
 254     exported_net_file = 'exported_model.bin'
 255     exec_net.export(exported_net_file)
 256     assert os.path.exists(exported_net_file)
 257     exec_net = ie_core.import_network(exported_net_file, "MYRIAD")
 258     os.remove(exported_net_file)
 259     img = read_image()
 260     res = exec_net.infer({'data': img})
 261     assert np.argmax(res['fc_out'][0]) == 3
 262     del exec_net
 263     del ie_core
 264
 265
 266 def test_multi_out_data(device):
 267     # Regression test CVS-23965
 268     # Check that CDataPtr for all output layers not copied  between outputs map items
 269     ie_core = ie.IECore()
 270     net = ie_core.read_network(model=test_net_xml, weights=test_net_bin)
 271     net.add_outputs(['28/Reshape'])
 272     exec_net = ie_core.load_network(net, device)
 273     assert "fc_out" in exec_net.outputs and "28/Reshape" in exec_net.outputs
 274     assert isinstance(exec_net.outputs["fc_out"], ie.CDataPtr)
 275     assert isinstance(exec_net.outputs["28/Reshape"], ie.CDataPtr)
 276     assert exec_net.outputs["fc_out"].name == "fc_out" and exec_net.outputs["fc_out"].shape == [1, 10]
 277     assert exec_net.outputs["28/Reshape"].name == "28/Reshape" and exec_net.outputs["28/Reshape"].shape == [1, 5184]
 278     del ie_core
 279     pass
 280
 281
 282 def test_get_metric(device):
 283     ie_core = ie.IECore()
 284     net = ie_core.read_network(model=test_net_xml, weights=test_net_bin)
 285     exec_net = ie_core.load_network(net, "CPU")
 286     network_name = exec_net.get_metric("NETWORK_NAME")
 287     assert network_name == "test_model"
 288
 289
 290 @pytest.mark.skipif(os.environ.get("TEST_DEVICE", "CPU") != "CPU", reason="Device independent test")
 291 def test_get_config(device):
 292     ie_core = ie.IECore()
 293     net = ie_core.read_network(model=test_net_xml, weights=test_net_bin)
 294     exec_net = ie_core.load_network(net, device)
 295     config = exec_net.get_config("PERF_COUNT")
 296     assert config == "NO"