diff --git a/setup.cfg b/setup.cfg index a64169c..af57b7c 100644 --- a/setup.cfg +++ b/setup.cfg @@ -18,12 +18,11 @@ package_dir = = src packages = find: include_package_data = True -python_requires = >=3.8 +python_requires = >= 3.7 install_requires = scikit-learn matplotlib numpy - tqdm >= 4.0.0 unidecode >= 1.3.0 syntok >= 1.4.0 langcodes[data] >= 3.3.0 diff --git a/src/great_ai/utilities/parallel_map/get_config.py b/src/great_ai/utilities/parallel_map/get_config.py index d7d8f8a..f1f6173 100644 --- a/src/great_ai/utilities/parallel_map/get_config.py +++ b/src/great_ai/utilities/parallel_map/get_config.py @@ -16,7 +16,6 @@ def get_config( input_values: Union[Sequence, Iterable], chunk_size: Optional[int], concurrency: Optional[int], - disable_logging: bool, ) -> ParallelMapConfiguration: is_input_sequence = hasattr(input_values, "__len__") @@ -57,8 +56,4 @@ def get_config( function_name=function.__name__, ) - if not disable_logging: - logger.info("Parallel map: configured ✅") - config.pretty_print() - return config diff --git a/src/great_ai/utilities/parallel_map/manage_communication.py b/src/great_ai/utilities/parallel_map/manage_communication.py index 2d36beb..1558bd2 100644 --- a/src/great_ai/utilities/parallel_map/manage_communication.py +++ b/src/great_ai/utilities/parallel_map/manage_communication.py @@ -1,9 +1,7 @@ import multiprocessing as mp import queue import traceback -from typing import Any, Dict, Iterable, TypeVar, Union - -from tqdm.cli import tqdm +from typing import Dict, Iterable, TypeVar, Union from ..chunk import chunk from ..logger import get_logger @@ -16,7 +14,6 @@ V = TypeVar("V") def manage_communication( *, - tqdm_options: Dict[str, Any], input_values: Iterable[T], chunk_size: int, input_queue: Union[mp.Queue, queue.Queue], @@ -24,54 +21,49 @@ def manage_communication( unordered: bool, ignore_exceptions: bool, ) -> Iterable[V]: - progress = tqdm(**tqdm_options) chunks = iter(chunk(enumerate(input_values), chunk_size=chunk_size)) indexed_results: Dict[int, V] = {} next_output_index = 0 read_input_length = 0 is_iteration_over = False - try: - while not is_iteration_over or next_output_index < read_input_length: - if not is_iteration_over: - try: - next_chunk = next(chunks) - input_queue.put(next_chunk) - read_input_length += len(next_chunk) - except StopIteration: - is_iteration_over = True - except Exception as e: + while not is_iteration_over or next_output_index < read_input_length: + if not is_iteration_over: + try: + next_chunk = next(chunks) + input_queue.put(next_chunk) + read_input_length += len(next_chunk) + except StopIteration: + is_iteration_over = True + except Exception as e: + if not ignore_exceptions: + raise e + else: + logger.error( + f"Exception {e} encountered in input, traceback:\n{traceback.format_exc()}" + ) + + try: + result_chunk = output_queue.get_nowait() + + for index, value, exception in result_chunk: + if exception is not None: + e, tb = exception if not ignore_exceptions: raise e else: logger.error( - f"Exception {e} encountered in input, traceback:\n{traceback.format_exc()}" + f"Exception {e} encountered in worker, traceback:\n{tb}" ) + if unordered: + yield value + next_output_index += 1 + else: + indexed_results[index] = value - try: - result_chunk = output_queue.get_nowait() - progress.update(len(result_chunk)) - - for index, value, exception in result_chunk: - if exception is not None: - e, tb = exception - if not ignore_exceptions: - raise e - else: - logger.error( - f"Exception {e} encountered in worker, traceback:\n{tb}" - ) - if unordered: - yield value - next_output_index += 1 - else: - indexed_results[index] = value - - if not unordered: - while next_output_index in indexed_results: - yield indexed_results[next_output_index] - del indexed_results[next_output_index] - next_output_index += 1 - except queue.Empty: - pass - finally: - progress.close() + if not unordered: + while next_output_index in indexed_results: + yield indexed_results[next_output_index] + del indexed_results[next_output_index] + next_output_index += 1 + except queue.Empty: + pass diff --git a/src/great_ai/utilities/parallel_map/manage_serial.py b/src/great_ai/utilities/parallel_map/manage_serial.py index 6faf6eb..ff0677d 100644 --- a/src/great_ai/utilities/parallel_map/manage_serial.py +++ b/src/great_ai/utilities/parallel_map/manage_serial.py @@ -1,7 +1,5 @@ import traceback -from typing import Any, Callable, Dict, Iterable, TypeVar - -from tqdm.cli import tqdm +from typing import Callable, Iterable, TypeVar from ..logger import get_logger @@ -14,12 +12,11 @@ V = TypeVar("V") def manage_serial( *, function: Callable[[T], V], - tqdm_options: Dict[str, Any], input_values: Iterable[T], ignore_exceptions: bool, ) -> Iterable[V]: try: - for v in tqdm(input_values, **tqdm_options): + for v in input_values: try: yield function(v) except Exception as e: diff --git a/src/great_ai/utilities/parallel_map/parallel_map.py b/src/great_ai/utilities/parallel_map/parallel_map.py index 22a59fb..555842b 100644 --- a/src/great_ai/utilities/parallel_map/parallel_map.py +++ b/src/great_ai/utilities/parallel_map/parallel_map.py @@ -1,5 +1,5 @@ import multiprocessing as mp -from typing import Callable, Iterable, Optional, Sequence, TypeVar, overload +from typing import Callable, Iterable, Literal, Optional, Sequence, TypeVar, overload from .get_config import get_config from .manage_communication import manage_communication @@ -17,9 +17,8 @@ def parallel_map( *, chunk_size: Optional[int], concurrency: Optional[int], - disable_logging: Optional[bool], unordered: Optional[bool], - ignore_exceptions: Optional[bool], + ignore_exceptions: Optional[Literal[False]], ) -> Iterable[V]: ... @@ -31,20 +30,44 @@ def parallel_map( *, chunk_size: int, concurrency: Optional[int], - disable_logging: Optional[bool], unordered: Optional[bool], - ignore_exceptions: Optional[bool], + ignore_exceptions: Optional[Literal[False]], ) -> Iterable[V]: ... +@overload +def parallel_map( + function: Callable[[T], V], + input_values: Sequence[T], + *, + chunk_size: Optional[int], + concurrency: Optional[int], + unordered: Optional[bool], + ignore_exceptions: True, +) -> Iterable[Optional[V]]: + ... + + +@overload +def parallel_map( + function: Callable[[T], V], + input_values: Iterable[T], + *, + chunk_size: int, + concurrency: Optional[int], + unordered: Optional[bool], + ignore_exceptions: True, +) -> Iterable[Optional[V]]: + ... + + def parallel_map( function, input_values, *, chunk_size=None, concurrency=None, - disable_logging=False, unordered=False, ignore_exceptions=False, ): @@ -53,21 +76,11 @@ def parallel_map( input_values=input_values, chunk_size=chunk_size, concurrency=concurrency, - disable_logging=disable_logging, - ) - - tqdm_options = dict( - desc=f"Parallel map {config.function_name}", - disable=disable_logging, - total=config.input_length, - miniters=1, - dynamic_ncols=True, ) if config.concurrency == 1: yield from manage_serial( function=function, - tqdm_options=tqdm_options, input_values=input_values, ignore_exceptions=ignore_exceptions, ) @@ -98,7 +111,6 @@ def parallel_map( try: yield from manage_communication( - tqdm_options=tqdm_options, input_values=input_values, chunk_size=config.chunk_size, input_queue=input_queue, diff --git a/src/great_ai/utilities/parallel_map/parallel_map_configuration.py b/src/great_ai/utilities/parallel_map/parallel_map_configuration.py index 384c642..4f06f14 100644 --- a/src/great_ai/utilities/parallel_map/parallel_map_configuration.py +++ b/src/great_ai/utilities/parallel_map/parallel_map_configuration.py @@ -14,13 +14,3 @@ class ParallelMapConfiguration(BaseModel): input_length: Optional[int] serialized_map_function: bytes function_name: str - - def pretty_print(self, prefix=" ⚙️ "): - logger.info(f"{prefix} concurrency: {self.concurrency}") - logger.info(f"{prefix} chunk size: {self.chunk_size}") - logger.info( - f"{prefix} chunk count: {self.chunk_count if self.chunk_count else 'unknown'}" - ) - logger.info( - f"{prefix} function size: {len(self.serialized_map_function) / 1024:.0f} kB" - ) diff --git a/src/great_ai/utilities/parallel_map/threaded_parallel_map.py b/src/great_ai/utilities/parallel_map/threaded_parallel_map.py index 92d4d2d..655ac4f 100644 --- a/src/great_ai/utilities/parallel_map/threaded_parallel_map.py +++ b/src/great_ai/utilities/parallel_map/threaded_parallel_map.py @@ -18,7 +18,6 @@ def threaded_parallel_map( *, chunk_size: Optional[int], concurrency: Optional[int], - disable_logging: Optional[bool], unordered: Optional[bool], ignore_exceptions: Optional[bool], ) -> Iterable[V]: @@ -32,7 +31,6 @@ def threaded_parallel_map( *, chunk_size: int, concurrency: Optional[int], - disable_logging: Optional[bool], unordered: Optional[bool], ignore_exceptions: Optional[bool], ) -> Iterable[V]: @@ -45,7 +43,6 @@ def threaded_parallel_map( *, chunk_size=None, concurrency=None, - disable_logging=False, unordered=False, ignore_exceptions=False, ): @@ -54,21 +51,11 @@ def threaded_parallel_map( input_values=input_values, chunk_size=chunk_size, concurrency=concurrency, - disable_logging=disable_logging, - ) - - tqdm_options = dict( - desc=f"Threaded parallel map {config.function_name}", - disable=disable_logging, - total=config.input_length, - miniters=1, - dynamic_ncols=True, ) if config.concurrency == 1: yield from manage_serial( function=function, - tqdm_options=tqdm_options, input_values=input_values, ignore_exceptions=ignore_exceptions, ) @@ -96,7 +83,6 @@ def threaded_parallel_map( t.start() yield from manage_communication( - tqdm_options=tqdm_options, input_values=input_values, chunk_size=config.chunk_size, input_queue=input_queue, diff --git a/tests/utilities/test_parallel_map.py b/tests/utilities/test_parallel_map.py index cab163e..07a14c8 100644 --- a/tests/utilities/test_parallel_map.py +++ b/tests/utilities/test_parallel_map.py @@ -29,11 +29,9 @@ class TestParallelMap(unittest.TestCase): ) def test_simple_case_without_progress_bar(self) -> None: - assert list( - parallel_map( - lambda v: v**2, range(COUNT), disable_logging=True, concurrency=2 - ) - ) == [v**2 for v in range(COUNT)] + assert list(parallel_map(lambda v: v**2, range(COUNT), concurrency=2)) == [ + v**2 for v in range(COUNT) + ] def test_simple_case_invalid_values(self) -> None: with pytest.raises(AssertionError): @@ -113,19 +111,6 @@ class TestParallelMap(unittest.TestCase): ) def test_no_op(self) -> None: - assert list(parallel_map(lambda v: v**2, [], disable_logging=True)) == [] - - assert ( - list( - parallel_map(lambda v: v**2, [], disable_logging=True, chunk_size=100) - ) - == [] - ) - assert ( - list( - parallel_map( - lambda v: v**2, [], disable_logging=True, concurrency=100 - ) - ) - == [] - ) + assert list(parallel_map(lambda v: v**2, [])) == [] + assert list(parallel_map(lambda v: v**2, [], chunk_size=100)) == [] + assert list(parallel_map(lambda v: v**2, [], concurrency=100)) == [] diff --git a/tests/utilities/test_threaded_parallel_map.py b/tests/utilities/test_threaded_parallel_map.py index c2e170b..0bf84a0 100644 --- a/tests/utilities/test_threaded_parallel_map.py +++ b/tests/utilities/test_threaded_parallel_map.py @@ -30,9 +30,7 @@ class TestParallelMap(unittest.TestCase): def test_simple_case_without_progress_bar(self) -> None: assert list( - threaded_parallel_map( - lambda v: v**2, range(COUNT), disable_logging=True, concurrency=2 - ) + threaded_parallel_map(lambda v: v**2, range(COUNT), concurrency=2) ) == [v**2 for v in range(COUNT)] def test_simple_case_invalid_values(self) -> None: @@ -121,24 +119,6 @@ class TestParallelMap(unittest.TestCase): ) def test_no_op(self) -> None: - assert ( - list(threaded_parallel_map(lambda v: v**2, [], disable_logging=True)) - == [] - ) - - assert ( - list( - threaded_parallel_map( - lambda v: v**2, [], disable_logging=True, chunk_size=100 - ) - ) - == [] - ) - assert ( - list( - threaded_parallel_map( - lambda v: v**2, [], disable_logging=True, concurrency=100 - ) - ) - == [] - ) + assert list(threaded_parallel_map(lambda v: v**2, [])) == [] + assert list(threaded_parallel_map(lambda v: v**2, [], chunk_size=100)) == [] + assert list(threaded_parallel_map(lambda v: v**2, [], concurrency=100)) == []