def execute_rpc_call()

in polymetis/python/polysim/grpc_sim_client.py [0:0]

• 16 lines of code
• 5 McCabe index (conditional complexity)

    def execute_rpc_call(self, request_func: Callable, args=[], log_request_time=False):
        """Executes an RPC call and performs round trip time intervals checks and logging

        Args:
            request_func: A Python function to call an RPC.

            args: List of arguments to pass to `request_func`.

            log_request_time: Whether to log the debug messages.

        """
        # Execute RPC call
        prev_time = time.time_ns()
        ret = request_func(*args)
        round_trip_time = (time.time_ns() - prev_time) / 1000.0 / 1000.0

        # Check round trip time
        if self.max_ping > 0.0 and round_trip_time > self.max_ping:
            log.debug(
                f"\n==== Warning: round trip time takes {round_trip_time} ms! ====\n"
            )

        # Log round trip times
        if self.log_interval > 0:
            self.interval_log.append(round_trip_time)
            if log_request_time:
                log.debug(
                    f"\nTime per RPC request in ms: avg: {sum(self.interval_log) / len(self.interval_log)} max: {max(self.interval_log)} min: {min(self.interval_log)}"
                )
                self.interval_log = []

        return ret