Using gvapython in VA Serving Pipelines about pipeline-server HOT 6 CLOSED

'''
* Copyright (C) 2019-2020 Intel Corporation.
*
* SPDX-License-Identifier: BSD-3-Clause
'''
import os
import json

class FrameInfo:

    def __init__(self):
        self._output_path = "/tmp"
        self._initialized = False
        self._source = "example"

    def write_caps(self, frame):
        path = "{}/caps.json".format(self._output_path)
        try:
            os.makedirs(self._output_path,exist_ok=True)
            os.remove(path)
        except Exception as e:
            pass
        
        with open(path,"w") as file:
            value = {'caps':str(frame.video_info().to_caps()),
                     'source':self._source}
            file.write(json.dumps(value))
            file.flush()

    @staticmethod
    def read_caps(input_path):
        path = "{}/caps.json".format(input_path)
        caps = None
        with open(path,"r") as file:
            caps = json.load(file)
            
        return caps
        
    
    def process_frame(self,frame):
        if (not self._initialized):
            self.write_caps(frame)
            self._initialized = True
        return True

{
  "name": "main_settings",
  "version": 2,
  "type": "GStreamer",
  "template": [
    "urisourcebin buffer-size=4096 name=source ! decodebin ! videoconvert ! video/x-raw,format=BGR ! queue ! ",
    "gvadetect model-instance-id=inf0 model={models[person-detection-retail-0013][1][network]} model-proc={models[person-detection-retail-0013][1][proc]} name=detection ! queue ! ",
    "gvapython module=\"/home/openvino/gst/main/py_modules/test_area.py\" class=FrameInfo name=main ! ",
    "appsink name=appsink"
  ],
  "description": "persons detection w.o. face recognition",
  "parameters": {
    "type": "object",
    "properties": {
      "kwarg": {
          "element": "main",
          "type": "string",
          "default": "{\"config\":\"/home/openvino/gst/main/model_proc/filter_detections.json\",\"cam_name\":\"CAM_00\"}"
      }
    }
  }
}

def __init__(self, height=1, width=1, zonemap=[]):
    self._output_path = "/home/openvino/area"
    self._initialized = False
    self._source = "example"
    self.k = {'h': height, 'w': width}

def write_caps(self, frame):
    path = "{}/caps2.json".format(self._output_path)
    try:
        os.makedirs(self._output_path, exist_ok=True)
        os.remove(path)
    except Exception as e:
        pass

    with open(path, "w") as file:
        file.write(json.dumps(self.k))
        file.flush()
    with frame.data() as mat:
        cv2.imwrite(os.path.join(self._output_path, 'im.jpg'), img=mat)

@staticmethod
def read_caps(input_path):
    path = "{}/caps2.json".format(input_path)
    caps = None
    with open(path, "r") as file:
        caps = json.load(file)

    return caps

def process_frame(self, frame):
    if (not self._initialized):
        self.write_caps(frame)
        self._initialized = True
    return True`

{
	"type": "GStreamer",
	"template": ["videotestsrc ! gvapython name=main module=temp.py class=Test ! fakesink"],
	"description": "Test Pipeline",
	"parameters": {
	    "type": "object",
	    "properties": {
		"main":{
		    "element": {
			"name":"main",
			"format":"json",
			"property":"kwarg"
		    },
		    "type":"object",
		    "properties": {
			"height": {
			    "type":"integer",
			    "minimum":1,
			    "maximum":200
			},
			"width": {
			    "type":"integer",
			    "minimum":1,
			    "maximum":100
			}
		    },
		    "default": {
			"height":100,
			"width":100
		    }
		}
	    }
	}
}

class Test:
    def __init__(self,height,width):
        print(height,width)

    def process_frame(self,frame):
        pass

'''
* Copyright (C) 2019 Intel Corporation.
*
* SPDX-License-Identifier: BSD-3-Clause
'''

import os
import json

from vaserving.vaserving import VAServing


request = {
    "source": {
        "uri": "file:///example",
        "type": "uri"
    },
    "destination": {
        "type":"file",
        "path":"/dev/stdout"
    }
}


if __name__=="__main__":

    
    VAServing.start({'log_level': 'DEBUG',
                     "ignore_init_errors":True})


    pipeline = VAServing.pipeline("test", 1)

    parameters = {}
    # uncomment to change default
    # parameters = {"main":{"height":50,"width":6}}
    
    pipeline.start(request,
                   parameters=parameters)
    
    
    VAServing.stop()