def execute_script()

in evals/elsuite/hr_ml_agent_bench/low_level_actions.py [0:0]
45 lines of code
12 McCabe index (conditional complexity)

def execute_script(script_name, work_dir=".", **kwargs):
    if not os.path.exists(os.path.join(work_dir, script_name)):
        raise EnvException(f"The file {script_name} does not exist.")
    try:
        script_path = script_name
        python = kwargs["python"]
        device = get_device()
        cmd = f"CUDA_VISIBLE_DEVICES={device} {python} -u {script_path}"
        process = subprocess.Popen(
            cmd,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            shell=True,
            cwd=work_dir,
        )

        stdout_lines = []
        stderr_lines = []

        selector = selectors.DefaultSelector()
        selector.register(process.stdout, selectors.EVENT_READ)
        selector.register(process.stderr, selectors.EVENT_READ)

        while process.poll() is None and selector.get_map():
            events = selector.select(timeout=1)

            for key, _ in events:
                line = key.fileobj.readline()
                if key.fileobj == process.stdout:
                    stdout_lines.append(line)
                else:
                    stderr_lines.append(line)

        for line in process.stdout:
            stdout_lines.append(line)

        for line in process.stderr:
            stderr_lines.append(line)

        return_code = process.returncode

        if return_code != 0:
            observation = "".join(stderr_lines)
        else:
            observation = "".join(stdout_lines)

        if observation == "" and return_code == 0:
            observation = "".join(stderr_lines)
        return observation
    except Exception as e:
        raise EnvException(
            f"Something went wrong in executing {script_name}: {e}. Please check if it is ready to be executed."
        )