workshop3.py

# =====================================================================================
# WORKSHOP 3: PARALLEL IMAGE PROCESSING WITH THREAD-SHARED MEMORY
# =====================================================================================
#
# HOW TO RUN THIS SCRIPT (Linux / macOS)
# ---------------------------------------------------------------------------------
# 1) Create a virtual environment (only needed once):
#
#       python3 -m venv venv
#
# 2) Activate it (needed every time you open a new terminal):
#
#       source venv/bin/activate
#
# 3) Install dependencies inside the virtual environment:
#
#       pip install opencv-python-headless numpy prettytable
#
#    (opencv-python-headless is used instead of opencv-python because this
#    script never opens a GUI window; it only reads/writes image files.)
#
# 4) Run the script, providing the path to an image (png, jpg, jpeg, tif,
#    tiff, bmp, webp). If you omit -i/--image, the script will ask for a path
#    interactively. Unlike a "fallback" scheme, NO substitute image is ever
#    auto-generated here: if no valid image is found, the script clearly
#    reports that and exits.
#
#       python3 workshop3_image_parallel.py -i photo.jpg
#
#    Optional flags:
#       -r / --repetitions      Repetitions per configuration (default: 3)
#       --sizes                 Comma-separated scale factors relative to the
#                               original image, e.g. "0.25,0.5,1.0" (default).
#                               Used to compare execution time across
#                               different image sizes, as required.
#       --threads               Comma-separated explicit thread counts to
#                               test, overriding the auto-detected schedule
#                               (e.g. "1,2,4,8"). By default the script
#                               detects the number of logical CPU threads
#                               available and builds the schedule itself
#                               (1, 2, 4, 8, ... up to the detected max) --
#                               no worker count is ever hardcoded.
#       -o / --output           Also save the full text report to a file
#       --output-dir            Directory to save the 6 result images
#                               (default: ./output_images)
#       -v / --verbose-progress Show live per-thread progress in the console
#                               (off by default; never written to --output)
#
#    Full example:
#
#       python3 workshop3_image_parallel.py -i photo.jpg -r 3 -o reporte.txt
#
# 5) When done:
#
#       deactivate
#
# WHAT THIS SCRIPT PRODUCES
# ---------------------------------------------------------------------------------
# - A text report (console, and optionally saved to --output) with a results
#   table (execution time, speedup, efficiency, overhead, correctness check)
#   for every combination of image size x thread count x repetition.
# - 6 output images, saved to --output-dir: the grayscale, edge-detected, and
#   blurred results, once from the serial run and once from the parallel run
#   (at the largest tested image size), so the two can be inspected/compared
#   directly.
# =====================================================================================

# =========== Auto-install dependencies ===========
# This block runs BEFORE any third-party import so that missing packages are
# installed transparently the first time the script is executed, on any OS
# (Linux, macOS, Windows) without the user needing to run pip manually.
import sys
import importlib
import importlib.util
import subprocess

_REQUIRED_PACKAGES = {
    # import_name       : pip_package_name
    "cv2"              : "opencv-python-headless",
    "numpy"            : "numpy",
    "prettytable"      : "prettytable",
}

def _ensure_dependencies() -> None:
    missing = [pip_name for import_name, pip_name in _REQUIRED_PACKAGES.items()
               if importlib.util.find_spec(import_name) is None]
    if not missing:
        return  # all good — nothing to do

    print("=" * 60)
    print("  Instalando dependencias faltantes, por favor espere...")
    print(f"  Paquetes: {', '.join(missing)}")
    print("=" * 60)
    try:
        subprocess.check_call(
            [sys.executable, "-m", "pip", "install", "--quiet", *missing]
        )
        print("  ✓ Dependencias instaladas correctamente.\n")
    except subprocess.CalledProcessError as exc:
        print(f"\n  ERROR: No se pudieron instalar las dependencias.\n  {exc}")
        print("  Ejecuta manualmente:  pip install " + " ".join(missing))
        sys.exit(1)

_ensure_dependencies()

# =========== Third-party imports (guaranteed available after bootstrap) ===========

import argparse
import logging
import os

import threading
from pathlib import Path
from time import perf_counter
from typing import Dict, List, Optional, Tuple, TypedDict

import cv2
import numpy as np
from prettytable import PrettyTable

# =========== Logging ===========

logging.basicConfig(level=logging.INFO, format="[%(asctime)s] [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)

# Capture the real console stream before any --output redirection happens, so
# live progress messages never get written into the saved text report.
_CONSOLE_STDOUT = sys.stdout

# =========== IMPORTANT: isolate OUR threading experiment from OpenCV's own ===========
#
# OpenCV internally uses its own multi-threading (via TBB, OpenMP, or a
# built-in thread pool, depending on how it was built) for many of its
# functions, INCLUDING the ones used here (cvtColor, Sobel, GaussianBlur).
#
# If we left this enabled, then even our "1 thread" / "serial" runs could
# secretly be using multiple CPU cores internally inside OpenCV, which would
# make our speedup/efficiency measurements meaningless: we would not be
# measuring the effect of OUR OWN threading strategy (dividing the image into
# subimages and using Python's `threading` module), but a confusing mixture
# of that plus OpenCV's own internal parallelism.
#
# Setting this to 1 forces every OpenCV call to run single-threaded
# internally, so any speedup we observe when using multiple Python threads is
# attributable ONLY to our own thread-based parallelization strategy -- a
# clean, fair, and correct experiment.
cv2.setNumThreads(1)

# =========== Constants ===========

VALID_IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".tif", ".tiff", ".bmp", ".webp"}

# Kernel sizes used by the edge-detection and blur effects. These determine
# how many rows of real neighboring pixel context ("halo") each subimage
# needs from its neighbors in order to produce results that are IDENTICAL to
# what would be produced by processing the whole image at once.
SOBEL_KSIZE = 3
BLUR_KSIZE = (5, 5)
SOBEL_RADIUS = SOBEL_KSIZE // 2  # 1
BLUR_RADIUS = BLUR_KSIZE[0] // 2  # 2

# HALO: how many extra rows of context each subimage needs on top and bottom.
# We use the largest radius among all effects applied per subimage (grayscale
# needs none since it is a pointwise operation; edge detection needs
# SOBEL_RADIUS; blur needs BLUR_RADIUS), so a single halo value safely covers
# every effect computed in RegionProcessor.process().
HALO = max(SOBEL_RADIUS, BLUR_RADIUS)

DEFAULT_SIZES = [0.25, 0.5, 1.0]
DEFAULT_REPETITIONS = 3
DEFAULT_OUTPUT_DIR = Path("output_images")


# =========== Image loading & validation ===========


class ImageValidationError(Exception):
    """Raised when no valid input image is available. This script NEVER
    fabricates a substitute image when this happens -- per the assignment's
    requirement, it must clearly report the problem instead."""


class ImageLoader:
    """Handles locating, validating, and loading the input image."""

    @staticmethod
    def prompt_for_path() -> str:
        """Ask the user, interactively via console, for the path to the image."""
        try:
            return input("Path: ").strip()
        except (EOFError, KeyboardInterrupt):
            # Non-interactive environment (no stdin available): treat as "no path".
            return ""

    @classmethod
    def load(cls, path: Optional[Path]) -> np.ndarray:
        """
        Load and validate the input image.

        Resolution order:
            1. Use `path` if it was given (e.g. via --image).
            2. Otherwise, ask for a path interactively via the console.
            3. If no usable path/image results from the above, raise
               ImageValidationError with a clear explanation. No fallback or
               synthetic image is ever created.

        Args:
            path (Optional[Path]): Path to the image file, or None to prompt.

        Returns:
            np.ndarray: The loaded image in BGR color format (OpenCV's
                default), shape (H, W, 3), dtype uint8.

        Raises:
            ImageValidationError: If no valid image could be located/loaded.
        """
        if path is None:
            typed_path = cls.prompt_for_path()
            path = Path(typed_path) if typed_path else None

        if path is None:
            raise ImageValidationError(
                "No image path was provided or typed. This script requires a real "
                "input image (it will never generate one automatically) -- please "
                "supply one via --image/-i or when prompted."
            )

        if not path.exists() or not path.is_file():
            raise ImageValidationError(f"Image file not found: {path}")

        if path.suffix.lower() not in VALID_IMAGE_EXTENSIONS:
            raise ImageValidationError(
                f"Unsupported image format '{path.suffix}'. Supported formats: "
                f"{sorted(VALID_IMAGE_EXTENSIONS)}"
            )

        image = cv2.imread(str(path), cv2.IMREAD_COLOR)
        if image is None:
            raise ImageValidationError(
                f"OpenCV could not decode the image at '{path}'. The file may be "
                "corrupted, empty, or an unsupported variant of the format."
            )

        height, width = image.shape[:2]
        logger.info(f"Loaded image '{path}': {width}x{height} px, {image.shape[2]} channel(s)")
        return image


# =========== Image effects (the 3 required transformations) ===========


class ImageEffects:
    """
    The 3 required image-processing effects, each implemented with OpenCV so
    the underlying computation happens in C++ and releases Python's GIL,
    which is what makes real multi-threaded speedup possible for this
    CPU-bound workload.

    Each effect is applied independently to the given BGR region (not chained
    through one another), producing 3 separate results: a grayscale image, an
    edge map, and a blurred image.
    """

    @staticmethod
    def to_grayscale(bgr: np.ndarray) -> np.ndarray:
        """Pointwise color-space conversion; needs no neighboring pixel
        context, so it never requires any halo margin."""
        return cv2.cvtColor(bgr, cv2.COLOR_BGR2GRAY)

    @staticmethod
    def detect_edges(bgr: np.ndarray) -> np.ndarray:
        """Sobel gradient magnitude on the grayscale version of the region.
        Requires SOBEL_RADIUS rows of real neighboring context on each side
        to be correct at region boundaries."""
        gray = cv2.cvtColor(bgr, cv2.COLOR_BGR2GRAY)
        sobel_x = cv2.Sobel(gray, cv2.CV_64F, 1, 0, ksize=SOBEL_KSIZE)
        sobel_y = cv2.Sobel(gray, cv2.CV_64F, 0, 1, ksize=SOBEL_KSIZE)
        magnitude = cv2.magnitude(sobel_x, sobel_y)
        return cv2.convertScaleAbs(magnitude)

    @staticmethod
    def blur(bgr: np.ndarray) -> np.ndarray:
        """Gaussian blur applied directly to the color image. Requires
        BLUR_RADIUS rows of real neighboring context on each side to be
        correct at region boundaries."""
        return cv2.GaussianBlur(bgr, BLUR_KSIZE, 0)


# =========== Region processing (shared by BOTH serial and parallel paths) ===========


class RegionProcessor:
    """
    Applies all 3 effects to a horizontal region (a range of rows) of the
    image.

    CRITICAL DESIGN POINT: this exact same function is used both by the
    serial run (called once, covering the entire image as a single "region")
    and by every thread in the parallel run (called once per subimage). This
    is what GUARANTEES -- by construction, not by coincidence -- that the
    serial and parallel implementations do identical work and therefore
    produce identical results.
    """

    @staticmethod
    def process(
        padded_bgr: np.ndarray, start: int, end: int, halo: int
    ) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
        """
        Args:
            padded_bgr (np.ndarray): The FULL image, already padded with
                `halo` extra rows of real/reflected context on top and
                bottom (see cv2.copyMakeBorder in main()). Using one shared,
                pre-padded array lets every region -- including the very
                first and very last ones, which touch the true image edges
                -- use the exact same slicing logic without special-casing.
            start (int): First row (inclusive) of the ORIGINAL (unpadded)
                image that this region is responsible for producing.
            end (int): Last row (exclusive) of the ORIGINAL image that this
                region is responsible for producing.
            halo (int): Rows of extra context on each side, already present
                in `padded_bgr`.

        Returns:
            Tuple[np.ndarray, np.ndarray, np.ndarray]: (grayscale, edges,
                blurred) results for rows [start, end) of the original image.
                grayscale and edges have shape (end-start, W); blurred has
                shape (end-start, W, 3).
        """
        # Slice out this region PLUS its halo margin on each side. Because
        # `padded_bgr` already has `halo` extra rows prepended, row `start`
        # of the original image is at row `start + halo` of the padded
        # array; slicing [start : end + 2*halo] therefore yields exactly
        # `halo` rows of context before `start` and after `end`.
        chunk = padded_bgr[start : end + 2 * halo]

        gray_full = ImageEffects.to_grayscale(chunk)
        edges_full = ImageEffects.detect_edges(chunk)
        blur_full = ImageEffects.blur(chunk)

        length = end - start
        gray = gray_full[halo : halo + length]
        edges = edges_full[halo : halo + length]
        blur_img = blur_full[halo : halo + length]

        return gray, edges, blur_img


# =========== Thread count detection & scheduling ===========


def detect_cpu_threads() -> int:
    """
    Detect the number of logical CPU threads available to this process.

    Uses os.sched_getaffinity(0) when available (Linux), which correctly
    reflects CPU limits imposed by containers/cgroups/taskset, unlike
    os.cpu_count() which reports the total machine core count regardless of
    any restriction actually placed on this process. Falls back to
    os.cpu_count() on platforms where sched_getaffinity is not available
    (e.g. macOS, Windows).
    """
    try:
        return len(os.sched_getaffinity(0))  # type: ignore[attr-defined]
    except AttributeError:
        return os.cpu_count() or 1


def get_thread_schedule(max_threads: int) -> List[int]:
    """
    Build the list of thread counts to benchmark: 1, 2, 4, 8, ... doubling,
    up to (and including) `max_threads`, WITHOUT ever exceeding it (no
    oversubscription, and no hardcoded worker count -- it is entirely derived
    from the detected CPU thread count).

    Example: max_threads=12 -> [1, 2, 4, 8, 12]
             max_threads=8  -> [1, 2, 4, 8]
             max_threads=4  -> [1, 2, 4]
             max_threads=1  -> [1]

    Args:
        max_threads (int): The detected number of logical CPU threads.

    Returns:
        List[int]: The thread counts to test.
    """
    schedule = [1]
    n = 1
    while True:
        n *= 2
        if n >= max_threads:
            if max_threads not in schedule:
                schedule.append(max_threads)
            break
        schedule.append(n)
    return schedule


# =========== Parallel processing with thread-shared memory ===========


class ParallelImageProcessor:
    """
    Divides the image into equal-sized (row-based) subimages and processes
    each one in a separate thread, with every thread writing its results
    directly into SHARED, preallocated output arrays (gray_out, edges_out,
    blur_out) -- no per-thread copies of the full output are made, which is
    the "thread-shared memory" the assignment asks for.
    """

    def __init__(self, padded_bgr: np.ndarray, height: int, width: int, halo: int):
        self.padded_bgr = padded_bgr
        self.height = height
        self.width = width
        self.halo = halo
        # A lock protecting the writes into the shared output arrays. Because
        # each thread is assigned a disjoint row range, these writes never
        # actually overlap in memory, so the lock is not strictly required
        # for correctness in THIS specific access pattern. It is included
        # explicitly anyway because: (a) the assignment requires "proper
        # synchronization" to be demonstrated, and (b) it is a safety net --
        # if the row-splitting logic ever had a bug that produced overlapping
        # ranges, this lock is what would prevent a real race condition from
        # corrupting the shared arrays (see the discussion section printed
        # at the end of the report for more on this).
        self._lock = threading.Lock()

    def run(
        self, n_threads: int, *, show_progress: bool = False
    ) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
        """
        Run the parallel pipeline with `n_threads` worker threads.

        Returns:
            Tuple[np.ndarray, np.ndarray, np.ndarray]: (grayscale, edges,
                blurred) results for the FULL image, assembled from every
                thread's contribution.
        """
        # Never use more threads than there are rows to split; a thread
        # assigned zero rows would do no useful work.
        effective_threads = max(1, min(n_threads, self.height))

        gray_out = np.empty((self.height, self.width), dtype=np.uint8)
        edges_out = np.empty((self.height, self.width), dtype=np.uint8)
        blur_out = np.empty((self.height, self.width, 3), dtype=np.uint8)

        boundaries = self._split_rows(self.height, effective_threads)

        threads = []
        for idx, (start, end) in enumerate(boundaries):
            t = threading.Thread(
                target=self._worker,
                args=(idx + 1, len(boundaries), start, end, gray_out, edges_out, blur_out, show_progress),
            )
            threads.append(t)
            t.start()
        for t in threads:
            t.join()

        return gray_out, edges_out, blur_out

    def _worker(
        self,
        thread_index: int,
        total_threads: int,
        start: int,
        end: int,
        gray_out: np.ndarray,
        edges_out: np.ndarray,
        blur_out: np.ndarray,
        show_progress: bool,
    ) -> None:
        if show_progress:
            _print_progress(
                f"  [thread {thread_index}/{total_threads}] processing rows {start}-{end}"
            )

        gray, edges, blur_img = RegionProcessor.process(self.padded_bgr, start, end, self.halo)

        with self._lock:
            gray_out[start:end] = gray
            edges_out[start:end] = edges
            blur_out[start:end] = blur_img

        if show_progress:
            _print_progress(f"  [thread {thread_index}/{total_threads}] done")

    @staticmethod
    def _split_rows(height: int, n_threads: int) -> List[Tuple[int, int]]:
        """
        Split `height` rows into `n_threads` contiguous, disjoint, roughly
        equal-sized ranges. Any remainder rows are distributed one at a time
        to the first few threads, so ranges differ in size by at most 1 row.
        """
        base = height // n_threads
        remainder = height % n_threads
        boundaries = []
        start = 0
        for i in range(n_threads):
            extra = 1 if i < remainder else 0
            end = start + base + extra
            boundaries.append((start, end))
            start = end
        return boundaries


# =========== Console progress reporting (off by default) ===========

_print_lock = threading.Lock()


def _print_progress(message: str) -> None:
    """Thread-safe console print for progress messages. Always writes to the
    real console stream (captured before any --output redirection), so
    progress feedback never gets written into the saved report file."""
    with _print_lock:
        print(message, file=_CONSOLE_STDOUT, flush=True)


# =========== Statistics helpers ===========


def mean(values: List[float]) -> float:
    return sum(values) / len(values) if values else 0.0


def std(values: List[float], mean_value: Optional[float] = None) -> float:
    if not values:
        return 0.0
    if mean_value is None:
        mean_value = mean(values)
    return (sum((x - mean_value) ** 2 for x in values) / len(values)) ** 0.5


def cv_stat(values: List[float], mean_value: Optional[float] = None, std_value: Optional[float] = None) -> float:
    mean_value = mean_value if mean_value is not None else mean(values)
    std_value = std_value if std_value is not None else std(values, mean_value)
    return std_value / mean_value if mean_value > 0 else 0.0


# =========== Metrics data structures ===========


class RunStats(TypedDict):
    results: List[float]
    mean: float
    std: float
    cv: float


class Metrics(RunStats):
    size_scale: float
    n_threads: Optional[int]  # None means "serial baseline"
    speedup: float
    efficiency: float
    overhead: float
    correctness_ok: Optional[bool]  # None for the serial baseline itself


# =========== Benchmark orchestration ===========


def run_benchmark(
    original_bgr: np.ndarray,
    sizes: List[float],
    thread_schedule: List[int],
    repetitions: int,
    show_progress: bool = False,
) -> Tuple[List[Metrics], "PrettyTable", Dict[str, np.ndarray]]:
    """
    Run the full serial + parallel benchmark across every (size, thread
    count) combination, `repetitions` times each.

    Returns:
        - The list of all collected Metrics.
        - A PrettyTable with a human-readable summary.
        - A dict with the 6 output images (3 serial + 3 parallel) computed
          at the LARGEST tested size, ready to be saved to disk.
    """
    results: List[Metrics] = []
    table = PrettyTable(
        ["Size", "Type", "Threads", "ExecTime", "Speedup", "Efficiency", "Overhead", "Correctness"]
    )

    saved_images: Dict[str, np.ndarray] = {}
    largest_size = max(sizes)

    for scale in sizes:
        if scale == 1.0:
            scaled = original_bgr
        else:
            new_w = max(1, int(round(original_bgr.shape[1] * scale)))
            new_h = max(1, int(round(original_bgr.shape[0] * scale)))
            scaled = cv2.resize(original_bgr, (new_w, new_h), interpolation=cv2.INTER_AREA)

        height, width = scaled.shape[:2]
        _print_progress(f"\n=== Image size {scale:.2f}x -> {width}x{height} px ===")

        # Pad the WHOLE region once with real/reflected context on top and
        # bottom. Every region -- serial (the whole image) and every
        # parallel subimage, including the very first/last ones that touch
        # the true image boundary -- slices its context out of this SAME
        # padded array, using the SAME border-handling logic
        # (cv2.BORDER_REFLECT_101). This is what guarantees bit-exact
        # equality between the serial and parallel results: there is no
        # separate/different border handling path for either case.
        padded = cv2.copyMakeBorder(scaled, HALO, HALO, 0, 0, cv2.BORDER_REFLECT_101)

        # ---- Serial baseline ----
        serial_times: List[float] = []
        serial_gray = serial_edges = serial_blur = None
        for rep in range(1, repetitions + 1):
            _print_progress(f"  [SERIAL] repetition {rep}/{repetitions} starting...")
            t0 = perf_counter()
            gray, edges, blur_img = RegionProcessor.process(padded, 0, height, HALO)
            elapsed = perf_counter() - t0
            serial_times.append(elapsed)
            serial_gray, serial_edges, serial_blur = gray, edges, blur_img
            _print_progress(f"  [SERIAL] repetition {rep}/{repetitions} finished in {elapsed:.3f}s")

        s_mean = mean(serial_times)
        s_std = std(serial_times, s_mean)
        s_cv = cv_stat(serial_times, s_mean, s_std)

        results.append(
            Metrics(
                size_scale=scale,
                n_threads=None,
                results=serial_times,
                mean=s_mean,
                std=s_std,
                cv=s_cv,
                speedup=1.0,
                efficiency=1.0,
                overhead=0.0,
                correctness_ok=None,
            )
        )
        table.add_row(
            [
                f"{scale:.2f}",
                "SERIAL",
                "N/A",
                f"{s_mean:.3f}s +/- {s_std:.3f}s (CV {s_cv:.1%})",
                "1.00",
                "100.00%",
                "0.000s",
                "N/A",
            ]
        )

        # ---- Parallel runs, one per thread count in the schedule ----
        parallel_processor = ParallelImageProcessor(padded, height, width, HALO)

        par_gray = par_edges = par_blur = None
        for n_threads in thread_schedule:
            par_times: List[float] = []
            for rep in range(1, repetitions + 1):
                _print_progress(
                    f"  [PARALLEL x{n_threads}] repetition {rep}/{repetitions} starting..."
                )
                t0 = perf_counter()
                gray, edges, blur_img = parallel_processor.run(n_threads, show_progress=show_progress)
                elapsed = perf_counter() - t0
                par_times.append(elapsed)
                par_gray, par_edges, par_blur = gray, edges, blur_img
                _print_progress(
                    f"  [PARALLEL x{n_threads}] repetition {rep}/{repetitions} finished in {elapsed:.3f}s"
                )

            p_mean = mean(par_times)
            p_std = std(par_times, p_mean)
            p_cv = cv_stat(par_times, p_mean, p_std)
            speedup = s_mean / p_mean if p_mean > 0 else float("inf")
            efficiency = speedup / n_threads
            overhead = n_threads * p_mean - s_mean

            correctness_ok = (
                np.array_equal(serial_gray, par_gray)
                and np.array_equal(serial_edges, par_edges)
                and np.array_equal(serial_blur, par_blur)
            )
            if not correctness_ok:
                logger.warning(
                    f"Correctness check FAILED for size={scale}, threads={n_threads} "
                    "-- serial and parallel results are NOT identical."
                )

            results.append(
                Metrics(
                    size_scale=scale,
                    n_threads=n_threads,
                    results=par_times,
                    mean=p_mean,
                    std=p_std,
                    cv=p_cv,
                    speedup=speedup,
                    efficiency=efficiency,
                    overhead=overhead,
                    correctness_ok=correctness_ok,
                )
            )
            table.add_row(
                [
                    f"{scale:.2f}",
                    "PARALLEL",
                    n_threads,
                    f"{p_mean:.3f}s +/- {p_std:.3f}s (CV {p_cv:.1%})",
                    f"{speedup:.2f}",
                    f"{efficiency:.2%}",
                    f"{overhead:.3f}s",
                    "PASS" if correctness_ok else "FAIL",
                ]
            )

        if scale == largest_size:
            saved_images = {
                "serial_grayscale": serial_gray,
                "serial_edges": serial_edges,
                "serial_blur": serial_blur,
                "parallel_grayscale": par_gray,
                "parallel_edges": par_edges,
                "parallel_blur": par_blur,
            }

    return results, table, saved_images


def save_output_images(images: Dict[str, np.ndarray], output_dir: Path) -> None:
    """Save the 6 result images (3 serial + 3 parallel) to disk as PNG
    (lossless, so no compression artifacts are introduced after the
    correctness check has already been performed in memory)."""
    output_dir.mkdir(parents=True, exist_ok=True)
    for name, arr in images.items():
        if arr is None:
            continue
        out_path = output_dir / f"{name}.png"
        cv2.imwrite(str(out_path), arr)
        logger.info(f"Saved {out_path}")


# =========== Analysis / discussion section ===========


def print_analysis(results: List[Metrics], n_cpu_threads: int) -> None:
    """
    Prints the discussion required by the assignment: comparison across
    sizes/thread counts, correctness summary, and an explanation of the
    memory-sharing challenges (the halo/boundary problem, GIL implications,
    and the role of the lock) plus proposed solutions -- all grounded in the
    actual measured numbers, not generic filler text.
    """
    print("\n" + "=" * 100)
    print("ANALYSIS AND DISCUSSION")
    print("=" * 100)

    parallel_results = [r for r in results if r["n_threads"] is not None]
    failed = [r for r in parallel_results if r["correctness_ok"] is False]
    passed = [r for r in parallel_results if r["correctness_ok"] is True]

    print("\n1. CORRECTNESS VERIFICATION")
    print("-" * 100)
    print(
        f"   {len(passed)}/{len(parallel_results)} (size, thread count) configurations produced "
        "results IDENTICAL to the serial baseline (grayscale, edges, and blur compared with "
        "numpy array equality, not just visual similarity)."
    )
    if failed:
        print("   FAILED configurations:")
        for r in failed:
            print(f"     - size={r['size_scale']:.2f}, threads={r['n_threads']}")
        print(
            "   A failure here would indicate a bug in the row-splitting or halo/padding logic "
            "(e.g. insufficient halo for the kernel sizes used, or an off-by-one error at chunk "
            "boundaries), NOT normal floating point noise -- these operations are deterministic."
        )
    else:
        print(
            "   All configurations PASSED. This is expected: RegionProcessor.process() is the "
            "exact same function used by both the serial run (called once for the whole image) "
            "and every thread in the parallel run (called once per subimage), and the halo "
            "margin used (HALO={}) covers the maximum context needed by any of the 3 "
            "effects (Sobel radius={}, blur radius={}). Because both paths draw their pixel "
            "context from the SAME pre-padded array with the SAME border handling "
            "(cv2.BORDER_REFLECT_101), there is no source of discrepancy left.".format(
                HALO, SOBEL_RADIUS, BLUR_RADIUS
            )
        )

    print("\n2. SPEEDUP AND EFFICIENCY BY IMAGE SIZE")
    print("-" * 100)
    sizes = sorted({r["size_scale"] for r in results})
    for scale in sizes:
        size_parallel = [r for r in parallel_results if r["size_scale"] == scale]
        if not size_parallel:
            continue
        best = max(size_parallel, key=lambda r: r["speedup"])
        print(
            f"   Size {scale:.2f}x: best speedup = {best['speedup']:.2f}x at "
            f"{best['n_threads']} threads (efficiency = {best['efficiency']:.1%})"
        )

    print("\n3. THE HALO / BOUNDARY PROBLEM (the real thread-shared-memory challenge here)")
    print("-" * 100)
    print(
        "   Dividing an image into row-based subimages is trivial for a POINTWISE operation "
        "like grayscale conversion (each output pixel depends only on the corresponding input "
        "pixel). It is NOT trivial for edge detection or blurring, because both are convolutions: "
        "an output pixel near the top or bottom of a subimage depends on input pixels that fall "
        "in the NEIGHBORING subimage."
    )
    print(
        "   If each thread only had access to its own disjoint row range, the rows near every "
        "internal split would be computed using incorrect (e.g. zero-padded or reflected) "
        "context instead of the real neighboring pixels, producing visible seams in the edge "
        "and blur outputs -- a correctness bug, not a performance one."
    )
    print(
        f"   Solution implemented: every thread reads a 'halo' of {HALO} extra rows of real "
        "pixel data from its neighbors (sliced from the single, shared, pre-padded image array) "
        "before processing, and only WRITES its own disjoint core rows back into the shared "
        "output arrays. Reads of the halo region are safe without locking because that data is "
        "never mutated (read-only shared memory); only the final write step is synchronized."
    )

    print("\n4. RACE CONDITIONS AND SYNCHRONIZATION")
    print("-" * 100)
    print(
        "   In this specific design, each thread writes to a disjoint row range of the shared "
        "output arrays (gray_out, edges_out, blur_out), so there is no actual overlapping write "
        "and therefore no real race condition in normal operation."
    )
    print(
        "   A threading.Lock is still used around each thread's write step, for two reasons: "
        "(1) it is a required, explicit synchronization primitive per the assignment, and "
        "(2) it acts as a safety net -- if a future change to the row-splitting logic introduced "
        "an off-by-one error that produced OVERLAPPING ranges, concurrent unsynchronized writes "
        "to the same array elements from different threads could interleave unpredictably and "
        "silently corrupt the output (a genuine race condition), and the lock prevents that."
    )
    print(
        "   A related risk that was deliberately avoided: if a thread accidentally wrote its "
        "halo rows (not just its core rows) into the shared output, it WOULD overlap with its "
        "neighbor's write range, creating a real race. This is why RegionProcessor.process() "
        "always crops the halo off before returning, and the caller only ever writes the "
        "cropped, disjoint core range."
    )

    print("\n5. GIL IMPLICATIONS")
    print("-" * 100)
    print(
        "   This workload is CPU-bound (pixel math), unlike a network-bound benchmark. Ordinary "
        "pure-Python CPU-bound code would NOT benefit from Python threads, because the GIL "
        "prevents more than one thread from executing Python bytecode at a time."
    )
    print(
        "   OpenCV's functions (cvtColor, Sobel, GaussianBlur) are implemented in C++ and "
        "release the GIL while they run, which is what allows multiple threads to do real, "
        "concurrent CPU work here. cv2.setNumThreads(1) was set explicitly at startup so that "
        "OpenCV's OWN internal multi-threading does not confound this experiment -- any speedup "
        "measured here comes only from OUR thread-based subimage division, not from OpenCV "
        "parallelizing internally regardless of how many Python threads we use."
    )

    print("\n6. THREAD MANAGEMENT OVERHEAD vs IMAGE SIZE TRADE-OFF")
    print("-" * 100)
    print(
        f"   This machine exposes {n_cpu_threads} logical CPU thread(s); the schedule tested was "
        "derived from that value, never hardcoded."
    )
    for scale in sizes:
        size_parallel = [r for r in parallel_results if r["size_scale"] == scale]
        worst = min(size_parallel, key=lambda r: r["speedup"]) if size_parallel else None
        if worst and worst["speedup"] < 1.0:
            print(
                f"   Size {scale:.2f}x: {worst['n_threads']} threads was SLOWER than serial "
                f"(speedup {worst['speedup']:.2f}x) -- for small images, the fixed cost of "
                "creating/joining threads and acquiring the lock can exceed the actual work "
                "saved by parallelizing so little data."
            )
    print(
        "   In general, thread management overhead is a roughly fixed per-thread cost, while "
        "the work available to parallelize grows with image size; larger images therefore "
        "amortize that overhead better and should show higher efficiency, up to the point where "
        "the thread count approaches or exceeds the number of physical CPU cores available."
    )

    print("\n" + "=" * 100)
    print("END OF ANALYSIS")
    print("=" * 100 + "\n")


# =========== CLI ===========


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description=(
            "Benchmark serial vs. thread-based parallel image processing "
            "(grayscale, edge detection, blur) using OpenCV."
        )
    )
    parser.add_argument(
        "-i",
        "--image",
        type=Path,
        default=None,
        help="Path to the input image (png, jpg, jpeg, tif, tiff, bmp, webp). If omitted, "
        "you will be prompted for a path. No image is ever auto-generated: if none is "
        "found, the script reports the error and exits.",
    )
    parser.add_argument(
        "-r",
        "--repetitions",
        type=int,
        default=DEFAULT_REPETITIONS,
        help=f"Repetitions per configuration, used to compute mean/std/CV (default: {DEFAULT_REPETITIONS}).",
    )
    parser.add_argument(
        "--sizes",
        type=str,
        default=",".join(str(s) for s in DEFAULT_SIZES),
        help="Comma-separated scale factors relative to the original image's resolution, "
        f"e.g. '0.25,0.5,1.0' (default: {','.join(str(s) for s in DEFAULT_SIZES)}). Used "
        "to compare execution time across different image sizes, as required.",
    )
    parser.add_argument(
        "--threads",
        type=str,
        default=None,
        help="Comma-separated explicit thread counts to test, e.g. '1,2,4,8', overriding "
        "the auto-detected schedule. By default, the number of logical CPU threads is "
        "detected automatically and a doubling schedule (1,2,4,...,max) is built from "
        "it -- no worker count is ever hardcoded.",
    )
    parser.add_argument(
        "-o",
        "--output",
        type=Path,
        default=None,
        help="Optional path to also save the full text report to a file.",
    )
    parser.add_argument(
        "--output-dir",
        type=Path,
        default=DEFAULT_OUTPUT_DIR,
        help=f"Directory to save the 6 result images (default: {DEFAULT_OUTPUT_DIR}).",
    )
    parser.add_argument(
        "-v",
        "--verbose-progress",
        action="store_true",
        help="Show live, per-thread console progress. Off by default to keep console "
        "output concise; never written into the --output report file either way.",
    )

    return parser.parse_args()


class _Tee:
    """Duplicates stdout to a file while still printing to the console."""

    def __init__(self, *streams):
        self.streams = streams

    def write(self, data):
        for s in self.streams:
            s.write(data)

    def flush(self):
        for s in self.streams:
            s.flush()


def get_system_info() -> str:
    n_cpu = detect_cpu_threads()
    return (
        "Machine Specifications:\n"
        f"- Logical CPU threads detected: {n_cpu}\n"
        f"- OpenCV version: {cv2.__version__}\n"
        f"- OpenCV internal threading: disabled (cv2.setNumThreads(1))\n"
        f"- NumPy version: {np.__version__}"
    )


if __name__ == "__main__":
    args = parse_args()

    try:
        original_image = ImageLoader.load(args.image)
    except ImageValidationError as e:
        logger.error(str(e))
        print(f"\nERROR: {e}\n")
        sys.exit(1)

    try:
        sizes = sorted(float(s.strip()) for s in args.sizes.split(",") if s.strip())
    except ValueError:
        logger.error(f"Invalid --sizes value: '{args.sizes}'. Expected comma-separated numbers.")
        sys.exit(1)
    if not sizes:
        logger.error("--sizes resolved to an empty list.")
        sys.exit(1)

    n_cpu_threads = detect_cpu_threads()
    if args.threads:
        try:
            thread_schedule = sorted({int(t.strip()) for t in args.threads.split(",") if t.strip()})
        except ValueError:
            logger.error(f"Invalid --threads value: '{args.threads}'. Expected comma-separated integers.")
            sys.exit(1)
    else:
        thread_schedule = get_thread_schedule(n_cpu_threads)

    output_file = None
    original_stdout = sys.stdout
    if args.output:
        output_file = open(args.output, "w", encoding="utf-8")
        sys.stdout = _Tee(original_stdout, output_file)

    try:
        print("=" * 100)
        print("WORKSHOP 3: PARALLEL IMAGE PROCESSING WITH THREAD-SHARED MEMORY")
        print("=" * 100)
        print(get_system_info())
        print("=" * 100)
        print("Experimental Configuration:")
        print(f"  Image sizes tested (scale factors): {sizes}")
        print(f"  Thread schedule tested: {thread_schedule}")
        print(f"  Repetitions per configuration: {args.repetitions}")
        print("=" * 100)

        results, table, saved_images = run_benchmark(
            original_image,
            sizes,
            thread_schedule,
            args.repetitions,
            show_progress=args.verbose_progress,
        )

        print("\n" + "=" * 100)
        print("RESULTS TABLE")
        print("=" * 100)
        print(table)

        print_analysis(results, n_cpu_threads)
    finally:
        if output_file:
            sys.stdout = original_stdout
            output_file.close()
            print(f"\nReport also saved to: {args.output}")

    save_output_images(saved_images, args.output_dir)
    print(f"\nOutput images saved to: {args.output_dir.resolve()}")