Send requests that execute third party functions from a remote package

[wigging]

I have a server computer that contains a third party package for interacting with devices connected to that computer. I have a client computer on the same network that sends requests to the server computer. The purpose of these requests is to run various functions from the third party package. The diagram below shows a basic example of this where I'm requesting a serial number from the third party package.

My example for accomplishing something like this with pyzmq is shown in the code given below. The Client class has two methods related to the functions provided by the third party package. I'm using a dictionary in each method where the call key is used by the server code to determine which package function to call.

Should I be sending a dictionary to the server to determine which functions to execute or does ZeroMQ have a feature that would be better suited to implement something like this? I'm new to ZeroMQ and pyzmq and I have read about topics with pub-sub patterns and multipart socket requests. But I don't see how to use any of these features for the request-reply pattern I'm using here. Any guidance on this would be greatly appreciated.

# client.py

import zmq
from typing import cast

class Client:
    def __init__(self):
        context = zmq.Context()
        socket = context.socket(zmq.REQ)
        socket.connect("tcp://localhost:5555")
        self.socket = socket

    def get_serial_number(self) -> str:
        d = {"call": "serial"}
        self.socket.send_json(d)

        sn = self.socket.recv_string()
        return sn

    def calc_motor_temp(self, speed: float) -> float:
        d = {"call": "motor", "speed": speed}
        self.socket.send_json(d)

        motor_data = cast(dict[str, float], self.socket.recv_json())
        return motor_data["temp"]

def main():
    """Run client."""

    client = Client()
    serial_num = client.get_serial_number()
    motor_temp = client.calc_motor_temp(800)

    print(f"Serial number: {serial_num}")
    print(f"Motor temperature: {motor_temp}")

if __name__ == "__main__":
    main()

# server.py

import zmq
import package
from typing import Any

def main():
    """Run server."""
    context = zmq.Context()
    socket = context.socket(zmq.REP)
    socket.bind("tcp://*:5555")
    print("Server is running...")

    while True:
        # Wait for requests from client
        request: Any = socket.recv_json()
        print(f"Received request: {request}")

        call = request["call"]

        if call == "serial":
            serial_num = package.serial_number()
            socket.send_string(serial_num)

        if call == "motor":
            speed = request["speed"]
            temp = package.motor_temperature(speed)
            socket.send_json({"temp": temp})

if __name__ == "__main__":
    main()

# package.py
# This is just a module of functions but in production this would be a third party Python package

def serial_number() -> str:
    """Get the serial number."""
    return "4234asdf1e"

def motor_temperature(speed: float) -> float:
    """Calculate motor temperature based on speed."""
    t = 1.5 * speed / 32.1
    return t

[wigging]

I was able to use socket.send_multipart and socket.recv_multipart as a way to provide a "topic" for the message. The server code uses the topic to select the function from the package. I am also using Python's struct to convert the motor speed to bytes for sending and receiving instead of the JSON serialization that was used in the original example. Anyway, I'm not sure if this is the best solution but using multipart seems to be another approach to accomplish this kind of pattern.

# client.py

import zmq
import struct

class Client:
    def __init__(self):
        context = zmq.Context()
        socket = context.socket(zmq.REQ)
        socket.connect("tcp://localhost:5555")
        self.socket = socket

    def get_serial_number(self) -> str:
        self.socket.send_multipart([b"serial"])
        sn = self.socket.recv_string()
        return sn

    def calc_motor_temp(self, speed: float) -> float:
        speed_bytes = struct.pack("f", speed)
        self.socket.send_multipart([b"motor", speed_bytes])

        temp_bytes = self.socket.recv()
        temp = struct.unpack("f", temp_bytes)[0]
        return temp

def main():
    """Run client."""
    client = Client()
    serial_num = client.get_serial_number()
    motor_temp = client.calc_motor_temp(800)

    print(f"Serial number: {serial_num}")
    print(f"Motor temperature: {motor_temp}")

if __name__ == "__main__":
    main()

# server.py

import zmq
import struct
import package

def main():
    """Run server."""
    context = zmq.Context()
    socket = context.socket(zmq.REP)
    socket.bind("tcp://localhost:5555")
    print("Server is running")

    while True:
        # Wait for requests from client
        message = socket.recv_multipart()
        print(f"Received message: {message}")

        topic = message[0]

        if topic == b"serial":
            serial_num = package.serial_number()
            socket.send_string(serial_num)

        if topic == b"motor":
            speed_bytes = message[1]
            speed = struct.unpack("f", speed_bytes)[0]

            temp = package.motor_temperature(speed)
            temp_bytes = struct.pack("f", temp)
            socket.send(temp_bytes)

if __name__ == "__main__":
    main()

[minrk]

I would generally recommend defining a "Message" as something in your application, and then defining serialization/deserialization so that you have two functions:

def send_message(socket: zmq.Socket, message: Message): ...

def recv_message(socket: zmq.Socket) -> Message: ...

Then you can try different approaches to serialization and it won't show up anywhere in your code except in these two functions.

Since a zmq message is one or more Frames (bytestrings), you might have a reason to split your message into multiple blobs (e.g. if they contain large blobs of data to avoid unnecessary copies, or to include message signatures, etc.). msgspec is a great serialization library which will get you validation and performance, but Python dicts via JSON work just fine, too, for simple cases (this is what Jupyter uses, for example). There are some docs on the topic.

[wigging]

I looked at the docs you linked to and it just talks about how to do your own serialization. I think I understand the basics of serializing a message. What I would like help with is how I should use that message to call different functions on a remote package and optionally pass arguments to that remote package function. There seems to be different ways to accomplish this but can you provide a simple example code that demonstrates this pattern?

[minrk]

Ah, sorry. I don't have a simple example for RPC messages, but fields in a dict make sense. This is how the Jupyter protocol is implemented, but it's not a simple example.

The bulk of the logic can be found here where messages are deserialized and the msg_type field is retrieved, which is then mapped to a method in a dict.

A much simpler version arbitrary RPC message could be a message that looked like:

method_name: str
args: list
kwargs: dict

and called

lookup_method(method_name)(*args, **kwargs)

[wigging]

Based on your suggestions, here is my client and server code:

# client.py

import zmq
from typing import Any

class Client:
    """Client for sending/receiving messages."""

    def __init__(self, address="tcp://localhost:5555"):
        context = zmq.Context()
        socket = context.socket(zmq.REQ)
        socket.connect(address)
        self.socket = socket

    def _send_message(self, function: str, *args: Any):
        msg = {"func": function, "args": args}
        self.socket.send_json(msg)

    def _recv_message(self) -> Any:
        msg: Any = self.socket.recv_json()
        return msg["result"]

    def get_serial_number(self) -> str:
        """Get serial number from remote function."""
        self._send_message("serialnum")
        serial_num = self._recv_message()
        return serial_num

    def calc_motor_temp(self, speed: float) -> float:
        """Calculate motor temperature using remote function."""
        self._send_message("motortemp", speed)
        temp = self._recv_message()
        return temp

    def close(self):
        self.socket.close()

# server.py

import zmq
import package
from typing import Any

FUNCTIONS = {
    "serialnum": package.serial_number,
    "motortemp": package.motor_temperature
}

class Server:
    """Sever for receiving/sending messages."""

    def __init__(self, address="tcp://localhost:5555"):
        context = zmq.Context()
        socket = context.socket(zmq.REP)
        socket.bind(address)
        self.socket = socket
        print("Server is running")

    def _recv_message(self) -> Any:
        msg: Any = self.socket.recv_json()
        print("\nReceived message:", msg)
        return msg

    def _send_message(self, result: Any):
        msg = {"result": result}
        self.socket.send_json(msg)
        print("Sent message:", msg)

    def run(self):
        """Run the server."""
        while True:
            # Wait for requests from client
            message = self._recv_message()

            # Get result from function
            name: str = message["func"]
            args = message["args"]
            result = FUNCTIONS[name](*args)

            # Send result to client
            self._send_message(result)

The client sends a message that contains the function name and the input arguments for the function. The server uses the function name and input arguments to execute the function from the package. The result is sent back to the client. So far, this is working well for what I need to do but let me know if you have any suggestions for improvement.