def run()

in components/artifacts/aws.sagemaker.edgeManagerPythonClient/0.1.0/edge_manager_python_client.py [0:0]

• 54 lines of code
• 6 McCabe index (conditional complexity)

def run():
    with grpc.insecure_channel('unix:///tmp/sagemaker_edge_agent_example.sock') as channel:

        edge_manager_client = agent_pb2_grpc.AgentStub(channel)

        try:
            response = edge_manager_client.LoadModel(
                LoadModelRequest(url=model_url, name=model_name))
        except Exception as e:
            print(e)
            print('Model already loaded!')

        response = edge_manager_client.ListModels(ListModelsRequest())

        response = edge_manager_client.DescribeModel(
            DescribeModelRequest(name=model_name))

        # Mean and Std deviation of the RGB colors (collected from Imagenet dataset)
        mean = [123.68, 116.779, 103.939]
        std = [58.393, 57.12, 57.375]

        img = cv2.imread(image_url)
        frame = resize_short_within(img, short=SIZE, max_size=SIZE * 2)
        nn_input_size = SIZE
        nn_input = cv2.resize(frame, (nn_input_size, int(nn_input_size / 4 * 3)))
        nn_input = cv2.copyMakeBorder(nn_input, int(nn_input_size / 8), int(nn_input_size / 8),
                                      0, 0, cv2.BORDER_CONSTANT, value=(0, 0, 0))
        copy_frame = nn_input[:]
        nn_input = nn_input.astype('float32')
        nn_input = nn_input.reshape((nn_input_size * nn_input_size, 3))
        scaled_frame = np.transpose(nn_input)
        scaled_frame[0, :] = scaled_frame[0, :] - mean[0]
        scaled_frame[0, :] = scaled_frame[0, :] / std[0]
        scaled_frame[1, :] = scaled_frame[1, :] - mean[1]
        scaled_frame[1, :] = scaled_frame[1, :] / std[1]
        scaled_frame[2, :] = scaled_frame[2, :] - mean[2]
        scaled_frame[2, :] = scaled_frame[2, :] / std[2]

        request = PredictRequest(name=model_name, tensors=[Tensor(tensor_metadata=TensorMetadata(
            name=bytes(tensor_name, 'utf-8'), data_type=5, shape=tensor_shape), byte_data=scaled_frame.tobytes())])

        response = edge_manager_client.Predict(request)

        # read output tensors
        i = 0
        detections = []

        for t in response.tensors:
            print("Flattened RAW Output Tensor : " + str(i + 1))
            i += 1
            deserialized_bytes = np.frombuffer(t.byte_data, dtype=np.float32)
            detections.append(np.asarray(deserialized_bytes))

        print(detections)
        # convert the bounding boxes
        new_list = []
        for index, item in enumerate(detections[2]):
            if index % 4 == 0:
                new_list.append(detections[2][index - 4:index])
        detections[2] = new_list[1:]

        # get objects, scores, bboxes
        objects = detections[0]
        scores = detections[1]
        bounding_boxes = new_list[1:]

        print(objects)
        print(scores)
        print(bounding_boxes)

        response = edge_manager_client.UnLoadModel(
            UnLoadModelRequest(name=model_name))