| 100 | return result_dict |
| 101 | |
| 102 | def predict_all(self, agent: Agent, inputs: List[T_INPUT], already_runs: List[Any]=None) -> List[T_OUTPUT]: |
| 103 | print(f"Start Predicting All ...") |
| 104 | assert already_runs is None or len(already_runs) == len(inputs) |
| 105 | |
| 106 | thread_count = self.workers |
| 107 | if self.worker_limit: |
| 108 | thread_count = min(self.workers, self.worker_limit) |
| 109 | |
| 110 | executor = ThreadPoolExecutor(max_workers=thread_count) |
| 111 | |
| 112 | threads = [] |
| 113 | results = [None] * len(inputs) |
| 114 | |
| 115 | def call_wrap(data_item, index): |
| 116 | try: |
| 117 | session = agent.create_session() |
| 118 | result = self.predict_single(session, data_item) |
| 119 | self.save_single(index, data_item, result, session) |
| 120 | except Exception as e: |
| 121 | import traceback |
| 122 | traceback.print_exc() |
| 123 | pass |
| 124 | results[index] = result |
| 125 | |
| 126 | for idx, item in enumerate(inputs): |
| 127 | if already_runs is not None and already_runs[idx] is not None: |
| 128 | results[idx] = already_runs[idx] |
| 129 | continue |
| 130 | future = executor.submit(call_wrap, item, idx) |
| 131 | threads.append(future) |
| 132 | |
| 133 | with tqdm(total=len(inputs)) as pbar: |
| 134 | for thread in as_completed(threads): |
| 135 | pbar.update(1) |
| 136 | |
| 137 | return results |
| 138 | |
| 139 | def save_single(self, index: int, input: T_INPUT, output: T_OUTPUT, session: Session=None): |
| 140 | save_obj = { |