Input for the ZMQ Connector type

Internal multi-part message (Topic, meta data, binary)

Since v1.4 the AI Inference Server can send pipeline predictions/outputs to ZMQ with textual metadata. For that an instance of the following Connector should be created via the UI/API: ZMQ Connector:

The ZMQ Connector allows high-performance data sending over ZMQ sockets using the Publisher/Subscriber communication pattern. AI Inference Server works as a Publisher and connects to the socket; the external component should behave as a Subscriber and bind to the socket. AI Inference Server sends multi-part messages: Topic, meta data, binary data. It may send multiple topics in the same message.

The topic is specified at variable mapping; the meta data is arbitrary text/string.

Connector with this payload type can only be used as a data destination.

The external component is an Edge Box app that runs on the same box. The host and port must be specified in the connector configuration. They are used when publishing the output payloads.

How to create an Edge Box app is out of scope of this User Manual.

Example script that assumes that the variable thumbnail in the Pipeline output type has been set to Object in the Pipeline configuration (by the AI SDK):

# ZMQ Output with metadata

def create_image_output(image: Image):

    output_image = {

        "metadata": json.dumps( {

            "resolutionWidth": image.width,

            "resolutionHeight": image.height,

        } ),

        "bytes": image.tobytes()

    }

    return output_image

...

def process_input (data: dict):

    vision_payload = data['vision_payload']

    pil_image = get_image_from_vision_payload(vision_payload)

    if pil_image is None:

        return None

    prediction, probability = predict_from_image(pil_image)

    return {

        "prediction": str(prediction),

        "probability": metric_output(probability),

        "thumbnail": create_image_output(pil_image)

        }

Internal multi-part message (Topic, binary data)

This payload type enables general payload processing both when receiving inputs and when sending predictions/outputs. The model should convert from binary type to the type for which it is prepared (type cast).

For example, JSON payloads can be sent and the Python script can be easily converted from the "Python type bytes()" to the "String" type:

The ZMQ Connector with internal multi-part message (Topic, binary) allows high-performance data sending/receiving over ZMQ sockets using the Publisher/Subscriber communication pattern.

During data receiving AI Inference Server works as a Subscriber and connects to the socket. The external component should behave as a Publisher and bind to the socket. AI Inference Server receives multi-part messages: Topic, binary data.

During data sending AI Inference Server works as a Publisher and connects to the socket. The external component should behave as a Subscriber and bind to the socket.

AI Inference Server sends multi-part messages: Topic, binary data.

The topic is specified at variable mapping both for pipeline input and output.

Example script for receiving and sending binary pipeline input and output, assuming the pipeline input variable name is set to "rawBinaryData" and the output is set to "rawBinaryPrediction". Both variable types should be set to Binary in the AI SDK.

In addition, the following examples show how to use an originally JSON input sent via ZMQ as a binary file:

import json

from log_module import LogModule

logger = LogModule()

def process_input(data: dict):

    logger.info("Preprocessing step received raw binary data")

    input_json_str = data[ "rawBinaryData" ].decode("utf-8")

    input = json.loads(input_json_str)

    ...

    prediction = {"prediction" : 42 }

    prediction_json_str = json.dumps(prediction)

    output = {"rawBinaryPrediction": bytes(prediction_json_str, 'utf-8')}

    return output

Socket binding in a ZMQ data transfer-based app

Example:

import sys

import zmq

port = "5556"

if len(sys.argv) > 1:

    port = sys.argv[1]

    int(port)

# Socket to talk to server

context = zmq.Context()

socket = contect.socket(zmq.SUB)

print("Collecting updates from weather server...")

socket.bind("tcp://*:%S" % port)

Vision payload

ZMQ Connector allows high performance data receiving based on the standardized Image Connector payload format of Siemens Industrial AI.

AI Inference Server connects to all camera applications that implement the Image Connector application interface and uses the "ImageSet" payload type for the data communication. Connectors with this payload type can only be used as a data source.

The payload is exactly the same as for the high data throughput of IE Vision Connector. For a detailed description, see section Data acquisition from the IE Vision Connector.