Imagine you are building a tool to automate the processing of a large collection of images. Your goal is to apply a transformation—such as resizing each image to a standard dimension or applying a simple filter—across all files in a directory. Processing each image one by one would be slow, especially with hundreds or thousands of files. By leveraging **multiprocessing**, you can process many images at once, dramatically reducing the total time required. This approach is especially useful for tasks like preparing image datasets for machine learning or batch editing photographs.


import unittest
import user_code
import ast
import re   
import unittest
import importlib
import io
from contextlib import redirect_stdout

class TestTask(unittest.TestCase):
    def test_parallel_uppercase(self):
        import user_code
        importlib.reload(user_code)
        image_paths = ["cat.jpg", "dog.png", "bird.bmp"]
        result = user_code.process_images_parallel(image_paths, user_code.process_func, num_workers=2)
        expected = ["CAT.JPG", "DOG.PNG", "BIRD.BMP"]
        _dynamic_test(
            self,
            isinstance(result, list) and result == expected,
            f"Returns correct uppercase filenames: {expected}",
            f"Expected: {expected}, got: {result}",
        )

    def test_parallel_with_ext_func(self):
        import user_code
        importlib.reload(user_code)
        image_paths = ["cat.jpg", "dog.png", "bird.bmp"]
        result = user_code.process_images_parallel(image_paths, user_code.ext_func, num_workers=3)
        expected = ["jpg", "png", "bmp"]
        _dynamic_test(
            self,
            isinstance(result, list) and result == expected,
            f"Correctly applies custom function to extract extensions: {expected}",
            f"Expected: {expected}, got: {result}",
        )

    def test_parallel_with_one_worker(self):
        import user_code
        importlib.reload(user_code)
        image_paths = ["lion.jpeg", "tiger.tiff"]
        result = user_code.process_images_parallel(image_paths, user_code.process_func, num_workers=1)
        expected = ["LION.JPEG", "TIGER.TIFF"]
        _dynamic_test(
            self,
            isinstance(result, list) and result == expected,
            f"Works with a single worker: {expected}",
            f"Expected: {expected}, got: {result}",
        )

    def test_print_output(self):
        import user_code
        importlib.reload(user_code)
        f = io.StringIO()
        with redirect_stdout(f):
            image_paths = ["cat.jpg", "dog.png", "bird.bmp"]
            result = user_code.process_images_parallel(image_paths, user_code.process_func, num_workers=2)
            print(result)
        output = normalize_text(f.getvalue())
        expected = normalize_text("['CAT.JPG', 'DOG.PNG', 'BIRD.BMP']")
        _dynamic_test(
            self,
            expected in output,
            f"Prints the correct result: {expected}",
            f"Expected output: '{expected}' not found in: '{output}'",
        )

def _dynamic_test(test_case, condition, success_message, failure_message):
    if condition:
        test_case._testMethodName = success_message
        test_case.assertTrue(True, success_message)
    else:
        test_case._testMethodName = failure_message
        test_case.fail(failure_message)

def normalize_text(text):
    text = text.lower()
    text = re.sub(r"\s{2,}", " ", text)
    text = re.sub(r"\s*([,:?])\s*", r"\1 ", text)
    return text.strip()

def change_var(code: str, var_name: str, value: str) -> str:
    tree = ast.parse(code)
    lines = code.splitlines()
    for i, node in enumerate(tree.body):
        if isinstance(node, ast.Assign):
            for target in node.targets:
                if isinstance(target, ast.Name) and target.id == var_name:
                    start_line = node.lineno - 1
                    line = lines[start_line]
                    indent = ' ' * (len(line) - len(line.lstrip()))
                    lines[start_line] = f"{indent}{var_name} = {value}"
                    next_line = len(lines)
                    for next_node in tree.body[i+1:]:
                        if hasattr(next_node, 'lineno'):
                            next_line = next_node.lineno - 1
                            break
                    if next_line > start_line + 1:
                        lines[start_line+1:next_line] = []
                    
                    return '\\n'.join(lines)
    return code

if __name__ == "__main__":
    unittest.main()


test_main.py

A beginner-friendly course introducing the concepts, techniques, and practical applications of multithreading and multiprocessing in Python. Learn how to write concurrent programs, manage threads and processes, and solve real-world problems using parallel execution.

Explore the foundational concepts of concurrency, parallelism, and the differences between threads and processes in Python.

Learn how to create, start, and manage threads in Python, and understand thread synchronization.

Explore how to create and manage processes in Python, and understand inter-process communication.

Delve into advanced concurrency patterns, performance considerations, and best practices for writing robust concurrent code.

Challenge: Parallel Image Processing

Solução

Challenge: Parallel Image Processing

Solução