| 20 | return function_(*args) |
| 21 | |
| 22 | class ParallelTask(): |
| 23 | def __init__(self, data_, function_, total=None, num_process=81, passing_indices=False, return_list=True): |
| 24 | function_ = partial(exec, function_=function_) |
| 25 | if passing_indices: |
| 26 | data_ = gen_with_indices(data_) |
| 27 | else: |
| 28 | data_ = gen(data_) |
| 29 | self.t = IterableSource(data_) |
| 30 | self.total = total |
| 31 | self.num_process = num_process |
| 32 | self.function_ = function_ |
| 33 | self.return_list = return_list |
| 34 | |
| 35 | def _collect_list(self, handler, tqdm_args): |
| 36 | collections_ = [] |
| 37 | for flag, line in tqdm(handler, **tqdm_args): |
| 38 | if flag: |
| 39 | collections_.append(line) |
| 40 | return collections_ |
| 41 | |
| 42 | def _collect_genetrator(self, handler, tqdm_args): |
| 43 | for flag, line in tqdm(handler, **tqdm_args): |
| 44 | if flag: |
| 45 | yield line |
| 46 | |
| 47 | def run_and_collect(self, tqdm_args=None, buffer=None): |
| 48 | if tqdm_args is None: |
| 49 | tqdm_args = {} |
| 50 | if buffer is None: |
| 51 | buffer = max(100, self.num_process*3) |
| 52 | handler = self.t.async_map(self.function_, max_workers=self.num_process, buffer=buffer) |
| 53 | |
| 54 | if self.total is not None and 'total' not in tqdm_args: |
| 55 | tqdm_args['total'] = self.total |
| 56 | if self.return_list: |
| 57 | return self._collect_list(handler, tqdm_args) |
| 58 | else: |
| 59 | return self._collect_genetrator(handler, tqdm_args) |
no outgoing calls
no test coverage detected