def run()

in polymetis/python/polymetis/robot_client/executable_robot_client.py [0:0]

• 23 lines of code
• 3 McCabe index (conditional complexity)

    def run(self):
        """Connects to gRPC server, and passes executable client required files.

        Note:
            Creates two temporary files: an executable configuration file,
            and a metadata protobuf message. The configuration file contains
            the path to the metadata file. The path of the configuration is
            passed to the executable.

        """
        with tempfile.NamedTemporaryFile(mode="w") as cfg_file:
            with tempfile.NamedTemporaryFile(mode="wb") as metadata_file:
                # Write metadata to temp file
                metadata_file.write(self.metadata.serialize())
                metadata_file.flush()

                # Save metadata file path
                self.executable_cfg.robot_client_metadata_path = metadata_file.name

                # Write configuration to temporary file
                cfg_pretty = OmegaConf.to_yaml(self.executable_cfg, resolve=True)
                log.info(f"=== Config: ===\n{cfg_pretty}")
                cfg_file.write(cfg_pretty)
                cfg_file.flush()

                # Find path to executable
                path_to_exec = which(self.executable_cfg.exec)
                assert path_to_exec, f"Unable to find binary {self.executable_cfg.exec}"

                # Add sudo if realtime; also, inherit $PATH variable
                command_list = [path_to_exec, cfg_file.name]
                if self.use_real_time:
                    command_list = ["sudo", "env", '"PATH=$PATH"'] + command_list

                # Run
                log.info(f"=== Executing client at {path_to_exec} ===")
                subprocess.run(
                    command_list,
                    stdin=sys.stdin,
                    stdout=sys.stdout,
                    stderr=sys.stderr,
                    check=True,
                )