8. Typhoon HIL RPC API

RPC (Remote Procedure Call) is an interprocess communication technique that allows two independent processes to communicate. RPC is most commonly used to create distributed client/server programs where:

  • client is a process (program or task) that requests the service provided by another program, and

  • server is a process (program or task) that provides the service (responds to the requests from a client).

Typhoon HIL’s RPC API allows you to write custom APIs in any language you want as long as it is supported by ZMQ [1] library.

8.1. Message format

Messages exchanged between the client and the server side are compatible with the JSON-RPC protocol [2].

Request message contains following members:

  • api - a string specifying the version of the Typhoon HIL RPC API.

  • jsonrpc - a string specifying the version of the JSON-RPC protocol. MUST be exactly “2.0”.

  • method - a string containing the name of the method to be invoked.

  • params - a list or dictionary of parameter values to be used during the invocation of the method. This member is optional.

  • id - an identifier established by the client. It can be either a string or an integer.

Response message contains following members:

  • jsonrpc - a string specifying the version of the JSON-RPC protocol. MUST be exactly “2.0”.

  • result - this member exists only if method is invoked and finished successfully.

  • error - this member exists only if error occurred during the method invocation or execution.

  • warnings - this member exists only if warnings occurred during the method execution.

  • id - an identifier that must be the same as the value of the id member in the request.

For more detailed information about message formats, check JSON-RPC specification [3].

8.2. Usage

Typhoon HIL RPC API can be used for following APIs:

For the full list of available methods, check the documentation of before-mentioned APIs. Each API is using different port number. The port numbers can be obtained by subscribing to the announcement port. Announcement port number is picked from the range defined in settings.conf file, which is located at the API version-specific folder at %APPDATA%\typhoon-api (i.e. for API version 1.7.0, that folder is %APPDATA%\typhoon-api\1.7.0)

8.3. Examples

Following examples illustrate how you can use Typhoon HIL RPC API.

8.3.1. Example 1

This example shows how you can obtain port numbers of Typhoon APIs.

import zmq
SERVICE_REGISTRY_HEADER = "typhoon-service-registry"


def discover(start_port=50000, end_port=50100, req_retries=30, timeout=1000):
    """Discovers port numbers for each of Typhoon APIs, in a given range:
    (start_port, end_port).

    Note: Check if the default param values match values in your settings.conf.

    Args:
        start_port (int): defines start of the port discovery range
        end_port (int): defines end of the port discovery range
        req_retries (int): defines number of times function will try to reach
            Typhoon API server
        timeout (int): defines timeout between each request retry, in
            milliseconds.

    Returns:
        dict
    """
    context = zmq.Context()
    socket = context.socket(zmq.SUB)

    offset = end_port - start_port + 1

    for i in range(offset):
        port = start_port + i
        socket.connect(f"tcp://localhost:{port}")

    socket.setsockopt_string(zmq.SUBSCRIBE, "")
    socket.setsockopt(zmq.LINGER, 0)

    poller = zmq.Poller()
    poller.register(socket, zmq.POLLIN)

    while req_retries != 0:

        socks = dict(poller.poll(timeout))
        if socks.get(socket) == zmq.POLLIN:
            response = socket.recv_json()

            if not response:
                break

            result = response.get("result")

            # Continue discovery if a received message is not from Typhoon
            # HIL Control Center.
            header = result[0]
            if header != SERVICE_REGISTRY_HEADER:
                continue

            return result
        else:
            req_retries -= 1
            if req_retries == 0:
                break

    raise RuntimeError("Server not found.")

8.3.2. Example 2

This example shows how load method can be invoked.

import zmq

context = zmq.Context()
req_socket = context.socket(zmq.REQ)
# Connect with the server
# In this example we assume that the server listens on the port 51357.
# Example 1 shows how to always get correct port number.
req_socket.connect("tcp://localhost:51357")

# Request message
message = {
    "api": "1.0",
    "jsonrpc": "2.0",
    "method": "load",
    "params": {"filename": "abs_path_to_the_model"},
    "id": 1
}

req_socket.send_json(message)
response = req_socket.recv_json()

8.3.3. Example 3

This example shows how compile method can be invoked.

import zmq

context = zmq.Context()
req_socket = context.socket(zmq.REQ)
# Connect with the server
# In this example we assume that the server listens on the port 51357.
# Example 1 shows how to always get correct port number.
req_socket.connect("tcp://localhost:51357")

# Request message
message = {
    "api": "1.0",
    "jsonrpc": "2.0",
    "method": "compile",
    "params": {},
    "id": 1
}

req_socket.send_json(message)
response = req_socket.recv_json()

8.3.4. Example 4

This example shows how to send multiple requests at once.

import zmq

context = zmq.Context()
req_socket = context.socket(zmq.REQ)
# Connect with the server
# In this example we assume that the server listens on the port 51357.
# Example 1 shows how to always get correct port number.
req_socket.connect("tcp://localhost:51357")

# Request message
message = [
    {
        "api": "1.0",
        "jsonrpc": "2.0",
        "method": "load",
        "params": {"filename": "abs_path_to_the_model"},
        "id": 1
    },
    {
        "api": "1.0",
        "jsonrpc": "2.0",
        "method": "compile",
        "params": {},
        "id": 1
    }
]

req_socket.send_json(message)
response = req_socket.recv_json()
print("Great success!")