(self, slot: dict[str, Any], script: str, timeout_seconds: float = 30.0, check: bool = True)
| 439 | raise RuntimeError("remote slot missing ssh target") |
| 440 | |
| 441 | ssh_options = slot.get("ssh_options") |
| 442 | ssh_cmd = [ssh_bin] |
| 443 | if isinstance(ssh_options, list): |
| 444 | ssh_cmd.extend(str(opt) for opt in ssh_options if str(opt).strip()) |
| 445 | ssh_cmd.extend([ssh_target, "bash", "-lc", script]) |
| 446 | return subprocess.run( |
| 447 | ssh_cmd, |
| 448 | check=check, |
| 449 | stdout=subprocess.PIPE, |
| 450 | stderr=subprocess.PIPE, |
| 451 | text=True, |
| 452 | timeout=timeout_seconds, |
| 453 | ) |
| 454 | |
| 455 | def _launch_remote_instance(self, slot: dict[str, Any], instance_type: str, cmd: list[str], cuda_device: str) -> tuple[dict[str, Any], dict[str, Any]]: |
| 456 | if self._is_truthy(slot.get("use_remote_proxy", False)): |
| 457 | return self._launch_remote_instance_via_proxy(slot, instance_type, cmd, cuda_device) |
| 458 | |
| 459 | host = str(slot["host"]) |
| 460 | engine_rank = int(slot["engine_rank"]) |
| 461 | python_executable = str(slot.get("python_executable", sys.executable)) |
| 462 | workdir = str(slot.get("workdir", Path(__file__).resolve().parents[3])) |
no test coverage detected