annotations/Azaion.Inference/remote_command.pyx

import msgpack

cdef class RemoteCommand:
    def __init__(self, CommandType command_type, bytes data):
        self.command_type = command_type
        self.data = data

    def __str__(self):
        command_type_names = {
            10: "LOGIN",
            20: "LOAD",
            30: "INFERENCE",
            40: "STOP_INFERENCE",
            100: "EXIT"
        }
        data_str = f'{len(self.data)} bytes' if self.data else ''
        return f'{command_type_names[self.command_type]} ({data_str})'

    @staticmethod
    cdef from_msgpack(bytes data):
        unpacked = msgpack.unpackb(data, strict_map_key=False)
        return RemoteCommand(unpacked.get("CommandType"), unpacked.get("Data"))

    cdef bytes serialize(self):
        return msgpack.packb({
            "CommandType": self.command_type,
            "Data": self.data
        })