INNER CODE UNIT · Python

_dispatch

onnxsim/onnxsim · onnxsim/rpc/server.py:380

    def _dispatch(
        self, op, header, blobs, models, counter
    ) -> Tuple[Dict[str, Any], List[bytes]]:
        server = self.server
        server.stats[op] = server.stats.get(op, 0) + 1
        if op == "info":
            return {"info": server.info()}, []
        if op == "upload":
            if len(blobs) != 1:
                raise proto.RPCError("upload takes exactly one blob")
            name = _sanitize(header["name"])
            path = os.path.join(server.work_dir, name)
            with open(path, "wb") as f:
                f.write(blobs[0])
            return {"name": name, "size": len(blobs[0])}, []
        if op == "load_model":
            if blobs:
                data = blobs[0]

View source record →