Merge pull request #118 from dlstreamer/v1.0

v1.0

.gitignore

@@ -8,3 +8,7 @@ docker/Dockerfile.env
 docker/final.env
 models
 samples/edgex_bridge/edgex/**/*
+samples/kubernetes/values.yaml
+samples/kubernetes/charts/
+samples/kubernetes/Chart.lock
+samples/nginx/cert/*

README.md

@@ -36,7 +36,6 @@ The sample microservice includes five categories of media analytics pipelines. C
 | **[object_classification](pipelines/gstreamer/object_classification)** | As object_detection adding meta-data such as object subtype and color
 | **[object_tracking](pipelines/gstreamer/object_tracking)** | As object_classification adding tracking identifier to meta-data
 | **[audio_detection](pipelines/gstreamer/audio_detection)** | Analyze audio streams for events such as breaking glass or barking dogs.
-| [Preview] **[action_recognition](pipelines/gstreamer/action_recognition/general/README.md)**  | Classifies general purpose actions in input video such as tying a bow tie or shaking hands.
 
 # Getting Started
 
@@ -123,7 +122,6 @@ In new shell run the following command:
  ```text
  - object_classification/vehicle_attributes
  - audio_detection/environment
- - action_recognition/general
  - object_tracking/object_line_crossing
  - object_tracking/person_vehicle_bike
  - object_detection/object_zone_count
@@ -221,7 +219,7 @@ Starting pipeline object_detection/person_vehicle_bike, instance = 8ad2c85af4bd4
 ```
 
 ```bash
-./client/pipeline_client.sh status object_detection/person_vehicle_bike 8ad2c85a-f4bd473e8a693aff562be316
+./client/pipeline_client.sh status object_detection/person_vehicle_bike 8ad2c85af4bd473e8a693aff562be316
 ```
 
 ```text
@@ -259,7 +257,7 @@ The error state covers a number of outcomes such as the request could not be sat
 
 ```text
 <snip>
-Starting pipeline object_detection/person_vehicle_bike, instance = 2bb2d219-310a4ee881faf258fbcc4355
+Starting pipeline object_detection/person_vehicle_bike, instance = 2bb2d219310a4ee881faf258fbcc4355
 ```
 
 Note that the Pipeline Server does not report an error at this stage as it goes into `QUEUED` state before it realizes that the source is not providing media.
@@ -278,7 +276,7 @@ ERROR (0fps)
 
 ## Change Pipeline and Source Media
 
-With pipeline_client it is easy to customize service requests. Here will use a vehicle classification pipeline `object_classification/vehicle_attributes` with the Iot Devkit video `car-detection.mp4`. Note how pipeline_client now displays classification metadata including type and color of vehicle.
+With pipeline_client it is easy to customize service requests. Here will use a vehicle classification pipeline `object_classification/vehicle_attributes` with the IoT Devkit video `car-detection.mp4`. Note how pipeline_client now displays classification metadata including type and color of vehicle.
 
 ```bash
 ./client/pipeline_client.sh run object_classification/vehicle_attributes https://github.com/intel-iot-devkit/sample-videos/blob/master/car-detection.mp4?raw=true

client/arguments.py

@@ -1,35 +1,13 @@
 '''
-* Copyright (C) 2019-2020 Intel Corporation.
+* Copyright (C) 2019 Intel Corporation.
 *
-* SPDX-License-Identifier: MIT License
-*
-*****
-*
-* MIT License
-*
-* Copyright (c) Microsoft Corporation.
-*
-* Permission is hereby granted, free of charge, to any person obtaining a copy
-* of this software and associated documentation files (the "Software"), to deal
-* in the Software without restriction, including without limitation the rights
-* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
-* copies of the Software, and to permit persons to whom the Software is
-* furnished to do so, subject to the following conditions:
-*
-* The above copyright notice and this permission notice shall be included in all
-* copies or substantial portions of the Software.
-*
-* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
-* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
-* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
-* SOFTWARE
+* SPDX-License-Identifier: BSD-3-Clause
 '''
 import sys
 import json
 import argparse
+import os
+from urllib.parse import urlparse
 import pipeline_client
 
 
@@ -124,4 +102,7 @@ def parse_args(program_name="Pipeline Client"):
     if args.subparsers in ['start', 'run'] and not args.uri and not args.request_file:
         parser.error("at least one of uri or --request-file is required")
 
+    if urlparse(args.server_address).scheme == "https" and not os.environ["ENV_CERT"]:
+        parser.error("ENV_CERT environment must be set if you are using HTTPS")
+
     return args

client/pipeline_client.py

@@ -5,16 +5,21 @@
 * SPDX-License-Identifier: BSD-3-Clause
 '''
 
-from urllib.parse import urljoin
+from urllib.parse import urljoin, urlparse
 import json
 import time
 import os
 import sys
+
 from html.parser import HTMLParser
 import requests
 import results_watcher
+# Used to workaround warning shown by Self-signed certificate
+import urllib3
 from server.pipeline import Pipeline
 
+urllib3.disable_warnings(urllib3.exceptions.SecurityWarning)
+
 RESPONSE_SUCCESS = 200
 TIMEOUT = 30
 SLEEP_FOR_STATUS = 0.5
@@ -156,7 +161,10 @@ def wait(args):
 def status(args):
     pipeline_status = get_pipeline_status(args.server_address, args.instance, args.show_request)
     if pipeline_status is not None and "state" in pipeline_status:
-        print("{} ({}fps)".format(pipeline_status["state"], round(pipeline_status["avg_fps"])))
+        if pipeline_status["state"] == "ERROR":
+            print("{} ({})".format(pipeline_status["state"], pipeline_status["message"]))
+        else:
+            print("{} ({}fps)".format(pipeline_status["state"], round(pipeline_status["avg_fps"])))
     else:
         print("Unable to fetch status")
 
@@ -171,14 +179,14 @@ def list_instances(args):
     statuses = get(url, args.show_request)
     for status in statuses:
         url = urljoin(args.server_address, "pipelines/{}".format(status["id"]))
-        response = requests.get(url, timeout=TIMEOUT)
-        request_status = json.loads(response.text)
-        response.close()
+        time.sleep(SLEEP_FOR_STATUS)
+        request_status = get(url, args.show_request)
         pipeline = request_status["request"]["pipeline"]
         print("{}: {}/{}".format(status["id"], pipeline["name"], pipeline["version"]))
         print("state: {}".format(status["state"]))
         print("fps: {:.2f}".format(status["avg_fps"]))
-        print("source: {}".format(json.dumps(request_status["request"]["source"], indent=4)))
+        if request_status["request"].get("source") is not None:
+            print("source: {}".format(json.dumps(request_status["request"]["source"], indent=4)))
         if request_status["request"].get("destination") is not None:
             print("destination: {}".format(json.dumps(request_status["request"]["destination"], indent=4)))
         if request_status["request"].get("parameters") is not None:
@@ -266,25 +274,25 @@ def wait_for_pipeline_running(server_address,
     status = {"state" : "QUEUED"}
     timeout_count = 0
     while status and not Pipeline.State[status["state"]] == Pipeline.State.RUNNING:
+        time.sleep(SLEEP_FOR_STATUS)
         status = get_pipeline_status(server_address, instance_id)
         if not status or Pipeline.State[status["state"]].stopped():
             break
-        time.sleep(SLEEP_FOR_STATUS)
         timeout_count += 1
         if timeout_count * SLEEP_FOR_STATUS >= timeout_sec:
             print("Timed out waiting for RUNNING status")
             break
     if not status or status["state"] == "ERROR":
-        raise ValueError("Error in pipeline, please check pipeline-server log messages")
+        raise ValueError(status["message"])
     return Pipeline.State[status["state"]] == Pipeline.State.RUNNING
 
 def wait_for_pipeline_completion(server_address, instance_id):
     status = {"state" : "RUNNING"}
     while status and not Pipeline.State[status["state"]].stopped():
-        status = get_pipeline_status(server_address, instance_id)
         time.sleep(SLEEP_FOR_STATUS)
+        status = get_pipeline_status(server_address, instance_id)
     if status and status["state"] == "ERROR":
-        raise ValueError("Error in pipeline, please check pipeline-server log messages")
+        raise ValueError(status["message"])
 
     return status
 
@@ -318,7 +326,7 @@ def wait_for_all_pipeline_completions(server_address, instance_ids, status_only=
             stopped = Pipeline.State[status["state"]].stopped()
             status_list.append(status)
     if status and status["state"] == "ERROR":
-        raise ValueError("Error in pipeline, please check pipeline-server log messages")
+        raise ValueError(status["message"])
     return status_list
 
 def get_pipeline_status(server_address, instance_id, show_request=False):
@@ -337,25 +345,34 @@ def _list(server_address, list_name, show_request=False):
         return
     print_list(response)
 
+def https_request(url):
+    return urlparse(url).scheme == "https"
+
 def post(url, body, show_request=False):
     try:
         if show_request:
             print('POST {}\nBody:{}'.format(url, body))
             sys.exit(0)
-        launch_response = requests.post(url, json=body, timeout=TIMEOUT)
+        if https_request(url):
+            launch_response = requests.post(url, json=body, timeout=TIMEOUT, verify=os.environ["ENV_CERT"])
+        else:
+            launch_response = requests.post(url, json=body, timeout=TIMEOUT)
         if launch_response.status_code == RESPONSE_SUCCESS:
             instance_id = json.loads(launch_response.text)
             return instance_id
     except requests.exceptions.ConnectionError as error:
         raise ConnectionError(SERVER_CONNECTION_FAILURE_MESSAGE) from error
-    raise RuntimeError(html_to_text(launch_response.text))
+    raise RuntimeError("{} - {}".format(launch_response.status_code, html_to_text(launch_response.text)))
 
 def get(url, show_request=False):
     try:
         if show_request:
             print('GET {}'.format(url))
             sys.exit(0)
-        status_response = requests.get(url, timeout=TIMEOUT)
+        if https_request(url):
+            status_response = requests.get(url, timeout=TIMEOUT, verify=os.environ["ENV_CERT"])
+        else:
+            status_response = requests.get(url, timeout=TIMEOUT)
         if status_response.status_code == RESPONSE_SUCCESS:
             return json.loads(status_response.text)
         print("Got unsuccessful status code: {}".format(status_response.status_code))
@@ -369,7 +386,10 @@ def delete(url, show_request=False):
         if show_request:
             print('DELETE {}'.format(url))
             sys.exit(0)
-        stop_response = requests.delete(url, timeout=TIMEOUT)
+        if https_request(url):
+            stop_response = requests.delete(url, timeout=TIMEOUT, verify=os.environ["ENV_CERT"])
+        else:
+            stop_response = requests.delete(url, timeout=TIMEOUT)
         if stop_response.status_code != RESPONSE_SUCCESS:
             print(html_to_text(stop_response.text))
         return stop_response.status_code

client/pipeline_client.sh

@@ -9,8 +9,46 @@ VOLUME_MOUNT="-v /tmp:/tmp "
 IMAGE="dlstreamer-pipeline-server-gstreamer"
 PIPELINE_SERVER_ROOT=/home/pipeline-server
 ENTRYPOINT="python3"
-ENTRYPOINT_ARGS="$PIPELINE_SERVER_ROOT/client $@"
 LOCAL_CLIENT_DIR=$(dirname $(readlink -f "$0"))
 ROOT_DIR=$(dirname $LOCAL_CLIENT_DIR)
+ARGS=
+ENV_CERT=
+MQTT_CLUSTER_BROKER=
 
-"$ROOT_DIR/docker/run.sh" $INTERACTIVE --name \"\" --network host --image  $IMAGE $VOLUME_MOUNT --entrypoint $ENTRYPOINT --entrypoint-args "$ENTRYPOINT_ARGS"
+error() {
+    printf '%s\n' "$1" >&2
+    exit 1
+}
+
+while [[ "$#" -ge 0 ]]; do
+    case $1  in
+            --server-cert)
+                if [ "$2" ]; then
+                    VOLUME_MOUNT="$VOLUME_MOUNT -v $2:/etc/ssl/certs/server.crt "
+                    ENV_CERT=/etc/ssl/certs/server.crt
+                    shift
+                else
+                    error 'ERROR: "--server-cert" requires an argument.'
+                fi
+                ;;
+            --mqtt-cluster-broker)
+                if [ "$2" ]; then
+                    MQTT_CLUSTER_BROKER=$2
+                    shift
+                else
+                    error 'ERROR: "--mqtt-cluster-broker" requires an argument.'
+                fi
+                ;;
+            *)
+                ARGS="${ARGS} ${1}"
+                ;;
+        esac
+    if [[ "$#" -eq 0 ]];
+    then
+        break
+    fi
+    shift
+done
+ENTRYPOINT_ARGS="$PIPELINE_SERVER_ROOT/client $ARGS"
+
+"$ROOT_DIR/docker/run.sh" $INTERACTIVE --name \"\" --network host --image  $IMAGE $VOLUME_MOUNT -e "ENV_CERT=${ENV_CERT}" -e "MQTT_CLUSTER_BROKER=${MQTT_CLUSTER_BROKER}" -e "REQUESTS_CA_BUNDLE=${ENV_CERT}" --entrypoint $ENTRYPOINT --entrypoint-args "$ENTRYPOINT_ARGS"

client/results_watcher.py

@@ -7,6 +7,7 @@
 
 import json
 import time
+import os
 import socket
 from threading import Thread, Event
 from abc import ABC, abstractmethod
@@ -140,7 +141,11 @@ class MqttWatcher(ResultsWatcher):
     def __init__(self, destination):
         super().__init__()
         self._client = mqtt.Client("Intel(R) DL Streamer Results Watcher", userdata=destination)
-        broker_address = destination["host"].split(':')
+        if os.environ["MQTT_CLUSTER_BROKER"]:
+            mqtt_host = os.environ["MQTT_CLUSTER_BROKER"]
+            broker_address = mqtt_host.split(':')
+        else:
+            broker_address = destination["host"].split(':')
         self._host = broker_address[0]
         if len(broker_address) == 2:
             self._port = int(broker_address[1])

docker/build.sh

@@ -10,7 +10,7 @@ DOCKERFILE_DIR=$(dirname "$(readlink -f "$0")")
 SOURCE_DIR=$(dirname "$DOCKERFILE_DIR")
 
 BASE_IMAGE_FFMPEG="openvisualcloud/xeone3-ubuntu1804-analytics-ffmpeg:20.10"
-BASE_IMAGE_GSTREAMER="intel/dlstreamer:2022.1.0-ubuntu20"
+BASE_IMAGE_GSTREAMER="intel/dlstreamer:2022.2.0-ubuntu20-gpu815"
 
 BASE_IMAGE=${BASE_IMAGE:-""}
 BASE_BUILD_CONTEXT=
@@ -36,7 +36,7 @@ BASE_BUILD_OPTIONS="--network=host "
 
 SUPPORTED_IMAGES=($BASE_IMAGE_GSTREAMER $BASE_IMAGE_FFMPEG)
 DEFAULT_OMZ_IMAGE_GSTREAMER="intel/dlstreamer"
-DEFAULT_OMZ_VERSION_GSTREAMER="2022.1.0-ubuntu20-devel"
+DEFAULT_OMZ_VERSION_GSTREAMER="2022.2.0-ubuntu20-gpu815-devel"
 DEFAULT_OMZ_IMAGE_FFMPEG="openvino/ubuntu18_data_dev"
 DEFAULT_OMZ_VERSION_FFMPEG="2021.2"
 FORCE_MODEL_DOWNLOAD=