(self, counted_ids: Path, *, keep_ratio: float, lookahead: int,
chunk_size: int, pool_kernel: int)
| 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]: |
| 156 | if self.proc is None or self.proc.stdin is None or self.r_fd is None: |
| 157 | raise RuntimeError("pflash daemon is not running") |
| 158 | keep_x1000 = int(round(keep_ratio * 1000)) |
| 159 | cmd = f"compress {keep_x1000} {lookahead} {chunk_size} {pool_kernel} {counted_ids}\n" |
| 160 | t0 = time.perf_counter() |
| 161 | self.proc.stdin.write(cmd.encode("utf-8")) |
| 162 | self.proc.stdin.flush() |
| 163 | tokens = read_stream_until_sentinel(self.r_fd) |
| 164 | return tokens, time.perf_counter() - t0 |
| 165 | |
| 166 | def stop(self) -> None: |
| 167 | if self.proc is None: |
no test coverage detected