|
- from __future__ import absolute_import
- from __future__ import division
- from __future__ import print_function
-
- import os
- from src.utils import logging
- from src.datasets.build import build_dataset
- from src.utils.parser import load_config, parse_args
- from src.config.defaults import assert_and_infer_cfg
- from hiai.nn_tensor_lib import NNTensor
- from hiai.nntensor_list import NNTensorList
- from model_service.hiai_model_service import HiaiBaseService
- from src.utils.meters import AVAMeter
- import numpy as np
- """AIPP example
- aipp_op {
- aipp_mode: static
- input_format : RGB888_U8
-
- mean_chn_0 : 123
- mean_chn_1 : 117
- mean_chn_2 : 104
- }
- """
- class DemoService(HiaiBaseService):
-
- def _preprocess(self, data):
- args = parse_args()
- cfg = load_config(args)
- cfg = assert_and_infer_cfg(cfg)
- # setup logger
- logger = logging.get_logger(__name__)
- logging.setup_logging()
- logger.info(cfg)
- # build dataset
- dataset = build_dataset(cfg, "test")
- iterator = dataset.create_tuple_iterator(output_numpy=True)
- preprocessed_data = {}
- slowpath_list = []
- fastpath_list = []
- boxes_list = []
- for cur_iter, (slowpath, fastpath, boxes, labels, ori_boxes, metadata, mask) in enumerate(iterator):
- slowpath_list.append(slowpath)
- fastpath_list.append(fastpath)
- boxes_list.append(boxes)
- preprocessed_data["slowpath"] = slowpath_list
- preprocessed_data["fastpath"] = fastpath_list
- preprocessed_data["boxes"] = boxes_list
- self.cfg = cfg
- self.mask = mask.asnumpy().astype(bool)
- test_meter = AVAMeter(dataset.get_dataset_size(), cfg, mode="test")
- test_meter.iter_tic()
- self.test_meter = test_meter
- return preprocessed_data
-
- def _inference(self, data, image_info=None):
- for k, v in data.items():
- preds = self.model.proc(v["slowpath"], v["fastpath"], v["boxes"])
- preds = preds.asnumpy().reshape(self.mask.shape + (self.cfg.MODEL.NUM_CLASSES,))
- mask = self.mask
- preds = preds[mask]
- padded_idx = np.tile(np.arange(mask.shape[0]).reshape(
- (-1, 1, 1)), (1, mask.shape[1], 1))
- ori_boxes = np.concatenate(
- (padded_idx, ori_boxes.asnumpy()), axis=2)[mask]
- metadata = metadata.asnumpy()[mask]
- # update stats
- test_meter = self.test_meter
- test_meter.update_stats(preds, ori_boxes, metadata)
- test_meter.log_iter_stats(None, k)
- test_meter.iter_tic()
- return test_meter.finalize_metrics()
-
- def _postprocess(self, data):
- outputs = {}
- outputs["result"] = data
- return outputs
-
|