Running multiple `predict_dataflow` in multiple threads, and aggregate the results. Args: dataflows: a list of DataFlow to be used in :func:`predict_dataflow` model_funcs: a list of callable to be used in :func:`predict_dataflow` Returns: list of dict, in the f
(dataflows, model_funcs)
| 183 | |
| 184 | |
| 185 | def multithread_predict_dataflow(dataflows, model_funcs): |
| 186 | """ |
| 187 | Running multiple `predict_dataflow` in multiple threads, and aggregate the results. |
| 188 | |
| 189 | Args: |
| 190 | dataflows: a list of DataFlow to be used in :func:`predict_dataflow` |
| 191 | model_funcs: a list of callable to be used in :func:`predict_dataflow` |
| 192 | |
| 193 | Returns: |
| 194 | list of dict, in the format used by |
| 195 | `DatasetSplit.eval_inference_results` |
| 196 | """ |
| 197 | num_worker = len(model_funcs) |
| 198 | assert len(dataflows) == num_worker |
| 199 | if num_worker == 1: |
| 200 | return predict_dataflow(dataflows[0], model_funcs[0]) |
| 201 | kwargs = {'thread_name_prefix': 'EvalWorker'} if sys.version_info.minor >= 6 else {} |
| 202 | with ThreadPoolExecutor(max_workers=num_worker, **kwargs) as executor, \ |
| 203 | tqdm.tqdm(total=sum(df.size() for df in dataflows)) as pbar: |
| 204 | futures = [] |
| 205 | for dataflow, pred in zip(dataflows, model_funcs): |
| 206 | futures.append(executor.submit(predict_dataflow, dataflow, pred, pbar)) |
| 207 | all_results = list(itertools.chain(*[fut.result() for fut in futures])) |
| 208 | return all_results |
| 209 | |
| 210 | |
| 211 | class EvalCallback(Callback): |
no test coverage detected