(self)
| 126 | self.r_fd: int | None = None |
| 127 | |
| 128 | def start(self) -> float: |
| 129 | r_fd, w_fd = os.pipe() |
| 130 | env = apply_backend_visible_devices( |
| 131 | self.backend, |
| 132 | visible_devices=self.visible_devices, |
| 133 | fallback_device=self.gpu, |
| 134 | base_env={**os.environ, **self.env}, |
| 135 | ) |
| 136 | cmd = [str(self.binary), str(self.drafter), f"--stream-fd={w_fd}"] |
| 137 | t0 = time.perf_counter() |
| 138 | self.proc = subprocess.Popen( |
| 139 | cmd, |
| 140 | stdin=subprocess.PIPE, |
| 141 | stdout=subprocess.PIPE, |
| 142 | stderr=subprocess.STDOUT, |
| 143 | pass_fds=(w_fd,), |
| 144 | env=env, |
| 145 | cwd=str(ROOT), |
| 146 | bufsize=0, |
| 147 | ) |
| 148 | os.close(w_fd) |
| 149 | self.r_fd = r_fd |
| 150 | self.log.attach(self.proc) |
| 151 | self.log.wait_for("[pflash-daemon] ready", 180) |
| 152 | return time.perf_counter() - t0 |
| 153 | |
| 154 | def compress(self, counted_ids: Path, *, keep_ratio: float, lookahead: int, |
| 155 | chunk_size: int, pool_kernel: int) -> tuple[list[int], float]: |
no test coverage detected