Estimate total effort from 3-point task estimates using Monte Carlo sampling.

estimates.py

#!/usr/bin/env python3
"""Estimate total effort from 3-point task estimates using Monte Carlo sampling."""

import argparse
import csv
import math
import random
import sys
from pathlib import Path
from typing import List, Tuple


def parse_args() -> argparse.Namespace:
    """Parse command-line arguments and return the parsed namespace."""
    parser = argparse.ArgumentParser(
        description="Estimate total duration using Monte Carlo simulation from triangular "
                    "distributions."
    )
    parser.add_argument(
        "csv_file",
        type=Path,
        help="Path to a CSV file with columns: name, minimum_duration, mode_duration, "
             "max_duration.",
    )
    parser.add_argument(
        "--runs",
        type=int,
        default=100000,
        help="Number of Monte Carlo runs to perform. Default is 100000.",
    )
    parser.add_argument(
        "--seed",
        type=int,
        default=0,
        help="Random seed for reproducible sampling. Default is 0.",
    )
    return parser.parse_args()


def read_tasks(csv_path: Path) -> List[Tuple[str, float, float, float]]:
    """Read task definitions from a CSV and return a list of duration tuples.

    Each task is represented as a tuple: (name, minimum, mode, maximum).
    """
    tasks: List[Tuple[str, float, float, float]] = []
    if not csv_path.exists():
        raise FileNotFoundError(f"CSV file not found: {csv_path}")

    with csv_path.open(newline="", encoding="utf-8") as csvfile:
        reader = csv.DictReader(csvfile)
        expected_fields = {"name", "minimum_duration", "mode_duration", "max_duration"}
        if not expected_fields.issubset(reader.fieldnames or []):
            raise ValueError(
                "CSV file must contain columns: name, minimum_duration, mode_duration, max_duration"
            )

        for row in reader:
            try:
                name = row["name"].strip()
                minimum = float(row["minimum_duration"])
                mode = float(row["mode_duration"])
                maximum = float(row["max_duration"])
            except (TypeError, ValueError) as exc:
                raise ValueError(f"Invalid numeric values in row: {row}") from exc

            if minimum > mode or mode > maximum:
                raise ValueError(
                    f"Invalid triangular parameters for task '{name}': "
                    f"minimum_duration <= mode_duration <= max_duration is required."
                )

            tasks.append((name, minimum, mode, maximum))

    if not tasks:
        raise ValueError("No tasks found in the CSV file.")

    return tasks


def sample_total_duration(tasks: List[Tuple[str, float, float, float]]) -> float:
    """Sample a total duration by drawing from each task's triangular distribution."""
    total = 0.0
    for _, minimum, mode, maximum in tasks:
        total += random.triangular(minimum, maximum, mode)
    return total


def percentile(values: List[float], percent: float) -> float:
    """Return the given percentile from a *sorted* list of numeric values.

    This function uses the linear interpolation method on the sorted sample values.
    It computes the fractional index into the sorted list and interpolates between
    the nearest lower and upper values when needed.
    """
    if not values:
        raise ValueError("Cannot compute percentiles of an empty list.")

    #sorted_values = sorted(values)
    # Use zero-based indexing. For example, the 50th percentile of 4 values maps to index 1.5.
    index = (len(values) - 1) * percent / 100.0
    lower = math.floor(index)
    upper = math.ceil(index)

    # If the index is an integer, return the exact ranked value.
    if lower == upper:
        return values[int(index)]

    # Otherwise, interpolate between the two surrounding values.
    lower_value = values[lower]
    upper_value = values[upper]
    weight = index - lower
    return lower_value + (upper_value - lower_value) * weight


def run_simulation(tasks: List[Tuple[str, float, float, float]],
                   runs: int) -> Tuple[float, float, float, float, float]:
    """Run the Monte Carlo simulation and return summary statistics.

    Returns a tuple of (min, 10th percentile, average, 90th percentile, max).
    """
    totals = [sample_total_duration(tasks) for _ in range(runs)]
    totals.sort()
    return (
        min(totals),
        percentile(totals, 10.0),
        sum(totals) / len(totals),
        percentile(totals, 90.0),
        max(totals),
    )


def format_duration(value: float) -> str:
    """Format a duration as a two-decimal numeric string."""
    return f"{value:.2f}"


def main() -> int:
    """Execute script entrypoint: parse args, run simulation, and print results."""
    args = parse_args()
    random.seed(args.seed)

    try:
        tasks = read_tasks(args.csv_file)
    except (FileNotFoundError, ValueError) as exc:
        print(f"Error: {exc}", file=sys.stderr)
        return 1

    if args.runs <= 0:
        print("Error: --runs must be a positive integer.", file=sys.stderr)
        return 1

    min_total, p10_total, avg_total, p90_total, max_total,  = run_simulation(tasks, args.runs)

    print("Monte Carlo effort estimate")
    print(f"CSV file: {args.csv_file}")
    print(f"Runs: {args.runs}")
    print(f"Seed: {args.seed}")
    print("----------------------------------------")
    print("Total duration estimates:")
    print(f"Minimum         : {format_duration(min_total)}")
    print(f"10th percentile : {format_duration(p10_total)}")
    print(f"Average         : {format_duration(avg_total)}")
    print(f"90th percentile : {format_duration(p90_total)}")
    print(f"Maximum         : {format_duration(max_total)}")


    return 0


if __name__ == "__main__":
    raise SystemExit(main())