Operations Manager

#!/usr/bin/env python3
"""
Capacity Planner - Resource capacity planning and demand-supply analysis.

Reads resource data and demand forecast CSVs to compute capacity utilization,
identify shortfalls, and model staffing scenarios across planning horizons.

Usage:
    python capacity_planner.py --file resources.csv --forecast demand.csv
    python capacity_planner.py --file resources.csv --forecast demand.csv --json
    python capacity_planner.py --file resources.csv --forecast demand.csv --productivity 0.85

Input - resources.csv columns:
    team            - Team or department name
    role            - Role type
    headcount       - Number of FTEs
    hours_per_day   - Available hours per person per day
    days_per_week   - Working days per week
    productivity    - Productivity factor 0-1 (optional, default from --productivity flag)

Input - demand.csv columns:
    period          - Time period (e.g., Week 1, Jan, Q1)
    team            - Team or department name
    demand_hours    - Required hours for the period
    priority        - Priority level: critical, high, medium, low (optional)
    project         - Project or workstream name (optional)

Output: Capacity analysis with utilization, gaps, and staffing recommendations.
"""

import argparse
import csv
import json
import math
import os
import sys
from collections import defaultdict


def read_csv(path: str) -> list:
    if not os.path.isfile(path):
        print(f"Error: File not found: {path}", file=sys.stderr)
        sys.exit(1)
    with open(path, "r", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        return list(reader)


def safe_float(val: str, default: float = 0.0) -> float:
    try:
        return float(val)
    except (ValueError, TypeError):
        return default


def safe_int(val: str, default: int = 0) -> int:
    try:
        return int(float(val))
    except (ValueError, TypeError):
        return default


def compute_capacity(resources: list, default_productivity: float) -> dict:
    """Compute available capacity by team."""
    team_capacity = defaultdict(lambda: {
        "headcount": 0,
        "weekly_hours": 0,
        "roles": defaultdict(int),
    })

    for r in resources:
        team = r.get("team", "Unknown").strip()
        role = r.get("role", "General").strip()
        hc = safe_int(r.get("headcount", 1))
        hours_day = safe_float(r.get("hours_per_day", 8))
        days_week = safe_float(r.get("days_per_week", 5))
        productivity = safe_float(r.get("productivity")) or default_productivity

        weekly_hours = hc * hours_day * days_week * productivity

        team_capacity[team]["headcount"] += hc
        team_capacity[team]["weekly_hours"] += weekly_hours
        team_capacity[team]["roles"][role] += hc

    return dict(team_capacity)


def compute_demand(forecast: list) -> dict:
    """Compute demand by team and period."""
    team_demand = defaultdict(lambda: defaultdict(lambda: {
        "demand_hours": 0,
        "projects": [],
        "critical_hours": 0,
    }))

    periods = []
    for f in forecast:
        period = f.get("period", "Unknown").strip()
        team = f.get("team", "Unknown").strip()
        demand = safe_float(f.get("demand_hours", 0))
        priority = f.get("priority", "medium").strip().lower()
        project = f.get("project", "").strip()

        if period not in periods:
            periods.append(period)

        team_demand[team][period]["demand_hours"] += demand
        if project and project not in team_demand[team][period]["projects"]:
            team_demand[team][period]["projects"].append(project)
        if priority == "critical":
            team_demand[team][period]["critical_hours"] += demand

    return dict(team_demand), periods


def analyze_capacity(capacity: dict, demand: dict, periods: list) -> dict:
    """Analyze capacity vs demand."""
    all_teams = sorted(set(list(capacity.keys()) + list(demand.keys())))

    team_analysis = []
    for team in all_teams:
        cap = capacity.get(team, {"headcount": 0, "weekly_hours": 0, "roles": {}})
        weekly_capacity = cap["weekly_hours"]

        period_details = []
        total_demand = 0
        total_surplus = 0
        total_deficit = 0
        max_utilization = 0

        for period in periods:
            period_demand = demand.get(team, {}).get(period, {"demand_hours": 0, "critical_hours": 0, "projects": []})
            demand_hrs = period_demand["demand_hours"]
            total_demand += demand_hrs

            gap = weekly_capacity - demand_hrs
            utilization = round(demand_hrs / max(1, weekly_capacity) * 100, 1)
            max_utilization = max(max_utilization, utilization)

            if gap >= 0:
                total_surplus += gap
            else:
                total_deficit += abs(gap)

            # Status
            if utilization > 100:
                status = "OVER_CAPACITY"
            elif utilization > 90:
                status = "NEAR_CAPACITY"
            elif utilization > 70:
                status = "OPTIMAL"
            elif utilization > 50:
                status = "UNDER_UTILIZED"
            else:
                status = "SIGNIFICANTLY_UNDER"

            period_details.append({
                "period": period,
                "capacity_hours": round(weekly_capacity, 1),
                "demand_hours": round(demand_hrs, 1),
                "gap_hours": round(gap, 1),
                "utilization_pct": utilization,
                "status": status,
                "critical_hours": period_demand["critical_hours"],
                "projects": period_demand["projects"],
            })

        avg_utilization = round(total_demand / max(1, weekly_capacity * len(periods)) * 100, 1)

        # FTE gap calculation
        fte_surplus_deficit = 0
        if weekly_capacity > 0 and len(periods) > 0:
            avg_demand = total_demand / len(periods)
            fte_surplus_deficit = round((avg_demand - weekly_capacity) / (weekly_capacity / max(1, cap["headcount"])), 1)

        team_analysis.append({
            "team": team,
            "headcount": cap["headcount"],
            "weekly_capacity_hours": round(weekly_capacity, 1),
            "avg_utilization_pct": avg_utilization,
            "peak_utilization_pct": max_utilization,
            "total_demand_hours": round(total_demand, 1),
            "total_surplus_hours": round(total_surplus, 1),
            "total_deficit_hours": round(total_deficit, 1),
            "fte_adjustment_needed": fte_surplus_deficit,
            "roles": dict(cap.get("roles", {})),
            "periods": period_details,
        })

    return team_analysis


def compute_org_summary(analysis: list, periods: list) -> dict:
    """Compute organization-level summary."""
    total_hc = sum(a["headcount"] for a in analysis)
    total_capacity = sum(a["weekly_capacity_hours"] for a in analysis)
    total_demand = sum(a["total_demand_hours"] for a in analysis)
    total_deficit = sum(a["total_deficit_hours"] for a in analysis)
    total_surplus = sum(a["total_surplus_hours"] for a in analysis)

    over_capacity = [a for a in analysis if a["peak_utilization_pct"] > 100]
    under_utilized = [a for a in analysis if a["avg_utilization_pct"] < 60]

    avg_util = round(total_demand / max(1, total_capacity * len(periods)) * 100, 1)

    # Additional FTEs needed
    total_fte_adj = sum(max(0, a["fte_adjustment_needed"]) for a in analysis)

    return {
        "total_headcount": total_hc,
        "total_weekly_capacity_hours": round(total_capacity, 1),
        "planning_periods": len(periods),
        "avg_utilization_pct": avg_util,
        "total_deficit_hours": round(total_deficit, 1),
        "total_surplus_hours": round(total_surplus, 1),
        "teams_over_capacity": len(over_capacity),
        "teams_under_utilized": len(under_utilized),
        "additional_ftes_needed": round(total_fte_adj, 1),
    }


def build_recommendations(analysis: list, summary: dict) -> list:
    """Generate staffing recommendations."""
    recs = []

    # Over-capacity teams
    over = [a for a in analysis if a["peak_utilization_pct"] > 100]
    for team in over:
        recs.append(
            f"{team['team']}: Peak utilization {team['peak_utilization_pct']:.0f}% exceeds capacity. "
            f"Need {max(0, team['fte_adjustment_needed']):.1f} additional FTEs or redistribute "
            f"{team['total_deficit_hours']:.0f} hours of demand to other teams."
        )

    # Near-capacity teams at risk
    near = [a for a in analysis if 90 <= a["peak_utilization_pct"] <= 100 and a["avg_utilization_pct"] > 85]
    for team in near:
        recs.append(
            f"{team['team']}: Running near capacity (avg {team['avg_utilization_pct']:.0f}%, "
            f"peak {team['peak_utilization_pct']:.0f}%). Any demand increase or attrition will cause overload. "
            "Consider cross-training or contingent staffing."
        )

    # Under-utilized teams
    under = [a for a in analysis if a["avg_utilization_pct"] < 50]
    for team in under:
        recs.append(
            f"{team['team']}: Utilization at {team['avg_utilization_pct']:.0f}%. "
            f"Surplus of {team['total_surplus_hours']:.0f} hours available for redeployment."
        )

    # Cross-team rebalancing opportunity
    if over and under:
        recs.append(
            "Cross-team rebalancing opportunity: redistribute work from over-capacity teams "
            f"({', '.join(a['team'] for a in over)}) to under-utilized teams "
            f"({', '.join(a['team'] for a in under)}) where skill overlap permits."
        )

    if not recs:
        recs.append("Capacity is well-balanced across all teams. Continue monitoring for demand changes.")

    return recs


def format_human(analysis: list, summary: dict, periods: list, recommendations: list) -> str:
    """Format results for human-readable output."""
    lines = []
    lines.append("=" * 75)
    lines.append("CAPACITY PLANNING REPORT")
    lines.append("=" * 75)
    lines.append("")
    lines.append(f"  Total Headcount:           {summary['total_headcount']}")
    lines.append(f"  Weekly Capacity:           {summary['total_weekly_capacity_hours']:.0f} hours")
    lines.append(f"  Planning Periods:          {summary['planning_periods']}")
    lines.append(f"  Avg Utilization:           {summary['avg_utilization_pct']:.1f}%")
    lines.append(f"  Teams Over Capacity:       {summary['teams_over_capacity']}")
    lines.append(f"  Teams Under-Utilized:      {summary['teams_under_utilized']}")
    lines.append(f"  Additional FTEs Needed:    {summary['additional_ftes_needed']:.1f}")
    lines.append(f"  Total Deficit Hours:       {summary['total_deficit_hours']:.0f}")
    lines.append(f"  Total Surplus Hours:       {summary['total_surplus_hours']:.0f}")

    # Team summary
    lines.append("")
    lines.append("-" * 75)
    lines.append("TEAM CAPACITY SUMMARY")
    lines.append("-" * 75)
    lines.append(f"  {'Team':<20} {'HC':>4} {'Cap/wk':>8} {'Avg Util':>9} {'Peak':>6} {'FTE +/-':>8} {'Status':>15}")
    lines.append(f"  {'-'*20} {'-'*4} {'-'*8} {'-'*9} {'-'*6} {'-'*8} {'-'*15}")

    for a in sorted(analysis, key=lambda x: -x["peak_utilization_pct"]):
        if a["peak_utilization_pct"] > 100:
            status = "OVER_CAPACITY"
        elif a["peak_utilization_pct"] > 90:
            status = "NEAR_CAPACITY"
        elif a["avg_utilization_pct"] > 70:
            status = "OPTIMAL"
        elif a["avg_utilization_pct"] > 50:
            status = "UNDER_USED"
        else:
            status = "LOW_USAGE"

        lines.append(
            f"  {a['team']:<20} {a['headcount']:>4} {a['weekly_capacity_hours']:>7.0f}h "
            f"{a['avg_utilization_pct']:>8.1f}% {a['peak_utilization_pct']:>5.0f}% "
            f"{a['fte_adjustment_needed']:>+7.1f} {status:>15}"
        )

    # Period-by-period for flagged teams
    flagged = [a for a in analysis if a["peak_utilization_pct"] > 90 or a["avg_utilization_pct"] < 50]
    if flagged:
        lines.append("")
        lines.append("-" * 75)
        lines.append("PERIOD DETAIL (Flagged Teams)")
        lines.append("-" * 75)
        for a in flagged:
            lines.append(f"\n  {a['team']} (HC: {a['headcount']}, Cap: {a['weekly_capacity_hours']:.0f}h/wk)")
            lines.append(f"  {'Period':<12} {'Demand':>8} {'Gap':>8} {'Util':>7} {'Status':>18}")
            lines.append(f"  {'-'*12} {'-'*8} {'-'*8} {'-'*7} {'-'*18}")
            for p in a["periods"]:
                lines.append(
                    f"  {p['period']:<12} {p['demand_hours']:>7.0f}h {p['gap_hours']:>+7.0f}h "
                    f"{p['utilization_pct']:>6.1f}% {p['status']:>18}"
                )

    # Recommendations
    lines.append("")
    lines.append("-" * 75)
    lines.append("RECOMMENDATIONS")
    lines.append("-" * 75)
    for i, rec in enumerate(recommendations, 1):
        lines.append(f"  {i}. {rec}")

    lines.append("")
    return "\n".join(lines)


def main():
    parser = argparse.ArgumentParser(
        description="Resource capacity planning and demand-supply analysis."
    )
    parser.add_argument("--file", required=True, help="Path to resources CSV")
    parser.add_argument("--forecast", required=True, help="Path to demand forecast CSV")
    parser.add_argument("--productivity", type=float, default=0.80, help="Default productivity factor 0-1 (default: 0.80)")
    parser.add_argument("--json", action="store_true", help="Output results as JSON")
    args = parser.parse_args()

    resources = read_csv(args.file)
    forecast = read_csv(args.forecast)

    if not resources:
        print("Error: No data in resources file.", file=sys.stderr)
        sys.exit(1)
    if not forecast:
        print("Error: No data in forecast file.", file=sys.stderr)
        sys.exit(1)

    capacity = compute_capacity(resources, args.productivity)
    demand, periods = compute_demand(forecast)
    analysis = analyze_capacity(capacity, demand, periods)
    summary = compute_org_summary(analysis, periods)
    recommendations = build_recommendations(analysis, summary)

    if args.json:
        output = {
            "summary": summary,
            "teams": analysis,
            "recommendations": recommendations,
        }
        print(json.dumps(output, indent=2))
    else:
        print(format_human(analysis, summary, periods, recommendations))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Process Mapper - Map and analyze business processes for optimization.

Reads a CSV of process steps and produces a process analysis including
cycle time calculation, bottleneck identification, handoff analysis,
value-add vs non-value-add classification, and improvement recommendations.

Usage:
    python process_mapper.py --file process_steps.csv
    python process_mapper.py --file process_steps.csv --json
    python process_mapper.py --file process_steps.csv --target-cycle 480

Input CSV columns:
    step_id         - Step sequence number or ID
    step_name       - Name/description of the step
    owner           - Role or team responsible
    duration_min    - Average duration in minutes
    wait_time_min   - Average wait/queue time before this step (minutes)
    type            - Step type: process, decision, handoff, rework, inspection
    value_add       - Is this step value-adding? (yes/no)
    error_rate_pct  - Error/rework rate at this step (percentage, optional)
    automation      - Automation level: manual, semi-auto, automated (optional)
    notes           - Additional notes (optional)

Output: Process analysis with cycle time, bottlenecks, efficiency metrics, and recommendations.
"""

import argparse
import csv
import json
import os
import sys
from collections import defaultdict


def read_csv(path: str) -> list:
    if not os.path.isfile(path):
        print(f"Error: File not found: {path}", file=sys.stderr)
        sys.exit(1)
    with open(path, "r", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        rows = list(reader)
    required = {"step_id", "step_name", "duration_min"}
    if rows:
        missing = required - set(rows[0].keys())
        if missing:
            print(f"Error: Missing required columns: {', '.join(missing)}", file=sys.stderr)
            sys.exit(1)
    return rows


def safe_float(val: str, default: float = 0.0) -> float:
    try:
        return float(val)
    except (ValueError, TypeError):
        return default


def analyze_process(steps: list, target_cycle: float = None) -> dict:
    """Analyze process steps."""
    total_duration = 0
    total_wait = 0
    total_steps = len(steps)
    value_add_time = 0
    non_value_add_time = 0
    handoffs = 0
    decisions = 0
    rework_steps = 0
    manual_steps = 0
    errors_weighted = 0

    step_details = []
    owners = set()
    prev_owner = None

    for step in steps:
        duration = safe_float(step.get("duration_min", 0))
        wait = safe_float(step.get("wait_time_min", 0))
        error_rate = safe_float(step.get("error_rate_pct", 0))
        step_type = step.get("type", "process").strip().lower()
        is_value_add = step.get("value_add", "yes").strip().lower() in ("yes", "true", "1", "y")
        automation = step.get("automation", "manual").strip().lower()
        owner = step.get("owner", "Unknown").strip()

        total_duration += duration
        total_wait += wait

        if is_value_add:
            value_add_time += duration
        else:
            non_value_add_time += duration

        if step_type == "handoff" or (prev_owner and owner != prev_owner):
            handoffs += 1
        if step_type == "decision":
            decisions += 1
        if step_type == "rework":
            rework_steps += 1
        if automation == "manual":
            manual_steps += 1

        errors_weighted += duration * error_rate / 100
        owners.add(owner)
        prev_owner = owner

        # Cycle time contribution
        step_cycle = duration + wait
        cycle_pct = 0  # Will calculate after totals

        step_details.append({
            "step_id": step.get("step_id", ""),
            "step_name": step.get("step_name", ""),
            "owner": owner,
            "duration_min": duration,
            "wait_time_min": wait,
            "total_time_min": step_cycle,
            "type": step_type,
            "value_add": is_value_add,
            "error_rate_pct": error_rate,
            "automation": automation,
        })

    total_cycle = total_duration + total_wait

    # Calculate percentages
    for sd in step_details:
        sd["cycle_pct"] = round(sd["total_time_min"] / max(1, total_cycle) * 100, 1)

    # Process efficiency
    process_efficiency = round(value_add_time / max(1, total_duration) * 100, 1)
    flow_efficiency = round(value_add_time / max(1, total_cycle) * 100, 1)

    return {
        "total_steps": total_steps,
        "total_duration_min": round(total_duration, 1),
        "total_wait_time_min": round(total_wait, 1),
        "total_cycle_time_min": round(total_cycle, 1),
        "total_cycle_time_hours": round(total_cycle / 60, 2),
        "target_cycle_min": target_cycle,
        "cycle_gap_min": round(total_cycle - target_cycle, 1) if target_cycle else None,
        "cycle_gap_pct": round((total_cycle - target_cycle) / target_cycle * 100, 1) if target_cycle else None,
        "value_add_time_min": round(value_add_time, 1),
        "non_value_add_time_min": round(non_value_add_time, 1),
        "process_efficiency_pct": process_efficiency,
        "flow_efficiency_pct": flow_efficiency,
        "handoffs": handoffs,
        "decisions": decisions,
        "rework_steps": rework_steps,
        "manual_steps": manual_steps,
        "unique_owners": len(owners),
        "owners": sorted(owners),
        "steps": step_details,
    }


def identify_bottlenecks(analysis: dict) -> list:
    """Identify process bottlenecks."""
    bottlenecks = []
    steps = analysis["steps"]
    avg_duration = analysis["total_duration_min"] / max(1, analysis["total_steps"])
    avg_wait = analysis["total_wait_time_min"] / max(1, analysis["total_steps"])

    for step in steps:
        issues = []

        # Long duration (> 2x average)
        if step["duration_min"] > avg_duration * 2:
            issues.append(f"Duration {step['duration_min']:.0f} min is {step['duration_min']/max(1,avg_duration):.1f}x the average")

        # Long wait time (> 2x average)
        if step["wait_time_min"] > avg_wait * 2 and step["wait_time_min"] > 30:
            issues.append(f"Wait time {step['wait_time_min']:.0f} min is {step['wait_time_min']/max(1,avg_wait):.1f}x the average")

        # High error rate
        if step["error_rate_pct"] > 5:
            issues.append(f"Error rate {step['error_rate_pct']:.1f}% exceeds 5% threshold")

        # Large cycle contribution
        if step["cycle_pct"] > 25:
            issues.append(f"Accounts for {step['cycle_pct']:.1f}% of total cycle time")

        if issues:
            severity = "HIGH" if len(issues) >= 2 or step["cycle_pct"] > 30 else "MEDIUM"
            bottlenecks.append({
                "step_id": step["step_id"],
                "step_name": step["step_name"],
                "owner": step["owner"],
                "severity": severity,
                "issues": issues,
                "duration_min": step["duration_min"],
                "wait_time_min": step["wait_time_min"],
                "cycle_pct": step["cycle_pct"],
            })

    bottlenecks.sort(key=lambda x: x["cycle_pct"], reverse=True)
    return bottlenecks


def compute_owner_analysis(analysis: dict) -> list:
    """Analyze workload by owner/team."""
    owner_data = defaultdict(lambda: {"steps": 0, "duration": 0, "wait": 0, "errors_weighted": 0})

    for step in analysis["steps"]:
        owner = step["owner"]
        owner_data[owner]["steps"] += 1
        owner_data[owner]["duration"] += step["duration_min"]
        owner_data[owner]["wait"] += step["wait_time_min"]
        owner_data[owner]["errors_weighted"] += step["duration_min"] * step["error_rate_pct"] / 100

    results = []
    for owner, data in sorted(owner_data.items()):
        total_time = data["duration"] + data["wait"]
        results.append({
            "owner": owner,
            "steps": data["steps"],
            "duration_min": round(data["duration"], 1),
            "wait_time_min": round(data["wait"], 1),
            "total_time_min": round(total_time, 1),
            "pct_of_cycle": round(total_time / max(1, analysis["total_cycle_time_min"]) * 100, 1),
        })

    results.sort(key=lambda x: x["total_time_min"], reverse=True)
    return results


def build_recommendations(analysis: dict, bottlenecks: list) -> list:
    """Generate improvement recommendations."""
    recs = []

    # Flow efficiency
    if analysis["flow_efficiency_pct"] < 25:
        recs.append(
            f"Flow efficiency is {analysis['flow_efficiency_pct']:.1f}% (target: >25%). "
            f"Total wait time ({analysis['total_wait_time_min']:.0f} min) exceeds processing time. "
            "Focus on reducing queue and handoff delays between steps."
        )

    # Handoffs
    if analysis["handoffs"] > analysis["total_steps"] * 0.3:
        recs.append(
            f"Process has {analysis['handoffs']} handoffs across {analysis['unique_owners']} owners. "
            "Consider consolidating steps under fewer owners to reduce handoff delays and communication overhead."
        )

    # Manual steps
    manual_pct = analysis["manual_steps"] / max(1, analysis["total_steps"]) * 100
    if manual_pct > 60:
        recs.append(
            f"{analysis['manual_steps']}/{analysis['total_steps']} steps ({manual_pct:.0f}%) are manual. "
            "Identify repetitive, rule-based steps for automation to reduce cycle time and error rates."
        )

    # Top bottlenecks
    high_bottlenecks = [b for b in bottlenecks if b["severity"] == "HIGH"]
    for bn in high_bottlenecks[:2]:
        recs.append(
            f"Bottleneck at '{bn['step_name']}' ({bn['cycle_pct']:.1f}% of cycle): "
            + "; ".join(bn["issues"][:2]) + ". "
            "Apply 5 Whys root cause analysis to this step."
        )

    # Non-value-add
    nva_pct = analysis["non_value_add_time_min"] / max(1, analysis["total_duration_min"]) * 100
    if nva_pct > 30:
        recs.append(
            f"Non-value-add activities account for {nva_pct:.0f}% of processing time. "
            "Review each non-value-add step: can it be eliminated, combined, or automated?"
        )

    # Target gap
    if analysis.get("cycle_gap_min") and analysis["cycle_gap_min"] > 0:
        recs.append(
            f"Current cycle time ({analysis['total_cycle_time_min']:.0f} min) exceeds target "
            f"({analysis['target_cycle_min']:.0f} min) by {analysis['cycle_gap_pct']:.1f}%. "
            "Address the top bottlenecks above to close the gap."
        )

    if not recs:
        recs.append("Process metrics are within acceptable ranges. Consider incremental improvements through Kaizen events.")

    return recs


def format_human(analysis: dict, bottlenecks: list, owners: list, recommendations: list) -> str:
    """Format results for human-readable output."""
    lines = []
    lines.append("=" * 70)
    lines.append("PROCESS ANALYSIS REPORT")
    lines.append("=" * 70)
    lines.append("")
    lines.append(f"  Total Steps:           {analysis['total_steps']}")
    lines.append(f"  Processing Time:       {analysis['total_duration_min']:.0f} min ({analysis['total_duration_min']/60:.1f} hrs)")
    lines.append(f"  Wait/Queue Time:       {analysis['total_wait_time_min']:.0f} min ({analysis['total_wait_time_min']/60:.1f} hrs)")
    lines.append(f"  Total Cycle Time:      {analysis['total_cycle_time_min']:.0f} min ({analysis['total_cycle_time_hours']:.1f} hrs)")
    if analysis.get("target_cycle_min"):
        lines.append(f"  Target Cycle Time:     {analysis['target_cycle_min']:.0f} min")
        lines.append(f"  Gap:                   {analysis['cycle_gap_min']:+.0f} min ({analysis['cycle_gap_pct']:+.1f}%)")
    lines.append(f"  Flow Efficiency:       {analysis['flow_efficiency_pct']:.1f}%")
    lines.append(f"  Process Efficiency:    {analysis['process_efficiency_pct']:.1f}%")
    lines.append(f"  Handoffs:              {analysis['handoffs']}")
    lines.append(f"  Unique Owners:         {analysis['unique_owners']}")
    lines.append(f"  Manual Steps:          {analysis['manual_steps']}/{analysis['total_steps']}")

    # Step-by-step map
    lines.append("")
    lines.append("-" * 70)
    lines.append("PROCESS MAP")
    lines.append("-" * 70)
    lines.append(f"  {'#':<4} {'Step':<25} {'Owner':<15} {'Dur':>5} {'Wait':>5} {'Err%':>5} {'%Cyc':>5} {'VA':>3}")
    lines.append(f"  {'-'*4} {'-'*25} {'-'*15} {'-'*5} {'-'*5} {'-'*5} {'-'*5} {'-'*3}")
    for step in analysis["steps"]:
        va = "Y" if step["value_add"] else "N"
        lines.append(
            f"  {step['step_id']:<4} {step['step_name']:<25} {step['owner']:<15} "
            f"{step['duration_min']:>5.0f} {step['wait_time_min']:>5.0f} {step['error_rate_pct']:>5.1f} "
            f"{step['cycle_pct']:>5.1f} {va:>3}"
        )

    # Owner analysis
    lines.append("")
    lines.append("-" * 70)
    lines.append("WORKLOAD BY OWNER")
    lines.append("-" * 70)
    lines.append(f"  {'Owner':<20} {'Steps':>6} {'Dur (min)':>10} {'Wait (min)':>10} {'% Cycle':>8}")
    lines.append(f"  {'-'*20} {'-'*6} {'-'*10} {'-'*10} {'-'*8}")
    for o in owners:
        lines.append(f"  {o['owner']:<20} {o['steps']:>6} {o['duration_min']:>10.0f} {o['wait_time_min']:>10.0f} {o['pct_of_cycle']:>7.1f}%")

    # Bottlenecks
    if bottlenecks:
        lines.append("")
        lines.append("-" * 70)
        lines.append("BOTTLENECKS")
        lines.append("-" * 70)
        for bn in bottlenecks:
            lines.append(f"\n  [{bn['severity']}] Step {bn['step_id']}: {bn['step_name']} ({bn['cycle_pct']:.1f}% of cycle)")
            for issue in bn["issues"]:
                lines.append(f"    - {issue}")

    # Recommendations
    lines.append("")
    lines.append("-" * 70)
    lines.append("RECOMMENDATIONS")
    lines.append("-" * 70)
    for i, rec in enumerate(recommendations, 1):
        lines.append(f"  {i}. {rec}")

    lines.append("")
    return "\n".join(lines)


def main():
    parser = argparse.ArgumentParser(
        description="Map and analyze business processes for optimization opportunities."
    )
    parser.add_argument("--file", required=True, help="Path to process steps CSV")
    parser.add_argument("--target-cycle", type=float, default=None, help="Target cycle time in minutes")
    parser.add_argument("--json", action="store_true", help="Output results as JSON")
    args = parser.parse_args()

    steps = read_csv(args.file)
    if not steps:
        print("Error: No data found in CSV file.", file=sys.stderr)
        sys.exit(1)

    analysis = analyze_process(steps, args.target_cycle)
    bottlenecks = identify_bottlenecks(analysis)
    owners = compute_owner_analysis(analysis)
    recommendations = build_recommendations(analysis, bottlenecks)

    if args.json:
        output = {
            "process_summary": {k: v for k, v in analysis.items() if k != "steps"},
            "steps": analysis["steps"],
            "bottlenecks": bottlenecks,
            "owner_analysis": owners,
            "recommendations": recommendations,
        }
        print(json.dumps(output, indent=2))
    else:
        print(format_human(analysis, bottlenecks, owners, recommendations))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
SLA Tracker - Track and analyze SLA compliance across service tickets.

Reads a CSV of service tickets with timestamps and SLA targets, then computes
compliance rates, breach analysis, trend detection, and priority-level breakdowns.

Usage:
    python sla_tracker.py --file tickets.csv
    python sla_tracker.py --file tickets.csv --threshold 95 --json
    python sla_tracker.py --file tickets.csv --period month

Input CSV columns:
    ticket_id       - Unique ticket identifier
    category        - Service category or type
    priority        - Priority level (P1/Critical, P2/High, P3/Medium, P4/Low)
    team            - Assigned team
    created_date    - Ticket creation date (YYYY-MM-DD)
    resolved_date   - Ticket resolution date (YYYY-MM-DD, blank if open)
    sla_target_hrs  - SLA target in hours for this ticket
    actual_hrs      - Actual resolution time in hours (blank if open)
    status          - Status: open, resolved, breached, escalated

Output: SLA compliance report with breach analysis, team performance, and trends.
"""

import argparse
import csv
import json
import os
import sys
from collections import defaultdict
from datetime import datetime


PRIORITY_MAP = {
    "p1": "P1-Critical",
    "critical": "P1-Critical",
    "p2": "P2-High",
    "high": "P2-High",
    "p3": "P3-Medium",
    "medium": "P3-Medium",
    "p4": "P4-Low",
    "low": "P4-Low",
}


def read_csv(path: str) -> list:
    if not os.path.isfile(path):
        print(f"Error: File not found: {path}", file=sys.stderr)
        sys.exit(1)
    with open(path, "r", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        rows = list(reader)
    required = {"ticket_id", "sla_target_hrs"}
    if rows:
        missing = required - set(rows[0].keys())
        if missing:
            print(f"Error: Missing required columns: {', '.join(missing)}", file=sys.stderr)
            sys.exit(1)
    return rows


def safe_float(val: str, default: float = 0.0) -> float:
    try:
        return float(val)
    except (ValueError, TypeError):
        return default


def parse_date(val: str) -> datetime:
    if not val or not val.strip():
        return None
    for fmt in ("%Y-%m-%d", "%m/%d/%Y", "%d/%m/%Y", "%Y-%m-%d %H:%M:%S"):
        try:
            return datetime.strptime(val.strip(), fmt)
        except ValueError:
            continue
    return None


def normalize_priority(val: str) -> str:
    """Normalize priority value."""
    if not val:
        return "P3-Medium"
    return PRIORITY_MAP.get(val.strip().lower(), val.strip())


def get_period_key(dt: datetime, period_type: str) -> str:
    """Get period key from datetime."""
    if not dt:
        return "Unknown"
    if period_type == "week":
        return f"{dt.year}-W{dt.isocalendar()[1]:02d}"
    elif period_type == "month":
        return f"{dt.year}-{dt.month:02d}"
    elif period_type == "quarter":
        q = (dt.month - 1) // 3 + 1
        return f"{dt.year}-Q{q}"
    else:
        return f"{dt.year}-{dt.month:02d}"


def analyze_tickets(rows: list, period_type: str) -> dict:
    """Analyze all tickets for SLA compliance."""
    tickets = []
    for row in rows:
        sla_target = safe_float(row.get("sla_target_hrs"))
        actual = safe_float(row.get("actual_hrs"))
        status = row.get("status", "").strip().lower()
        priority = normalize_priority(row.get("priority", ""))
        team = row.get("team", "Unknown").strip() or "Unknown"
        category = row.get("category", "Unknown").strip() or "Unknown"
        created = parse_date(row.get("created_date", ""))

        is_resolved = status in ("resolved", "closed") or actual > 0
        is_open = status in ("open", "in_progress", "escalated") and actual <= 0

        if is_resolved and sla_target > 0:
            met_sla = actual <= sla_target
            breach_hours = max(0, actual - sla_target) if not met_sla else 0
            breach_pct = round((actual - sla_target) / sla_target * 100, 1) if not met_sla else 0
        else:
            met_sla = None
            breach_hours = 0
            breach_pct = 0

        period = get_period_key(created, period_type) if created else "Unknown"

        tickets.append({
            "ticket_id": row["ticket_id"],
            "category": category,
            "priority": priority,
            "team": team,
            "period": period,
            "sla_target_hrs": sla_target,
            "actual_hrs": actual,
            "status": status,
            "is_resolved": is_resolved,
            "is_open": is_open,
            "met_sla": met_sla,
            "breach_hours": round(breach_hours, 1),
            "breach_pct": breach_pct,
        })

    return tickets


def compute_overall_metrics(tickets: list) -> dict:
    """Compute overall SLA metrics."""
    total = len(tickets)
    resolved = [t for t in tickets if t["is_resolved"]]
    open_tickets = [t for t in tickets if t["is_open"]]
    sla_measured = [t for t in resolved if t["met_sla"] is not None]
    met = [t for t in sla_measured if t["met_sla"]]
    breached = [t for t in sla_measured if not t["met_sla"]]

    compliance_rate = round(len(met) / max(1, len(sla_measured)) * 100, 1)

    # Average resolution time
    actual_times = [t["actual_hrs"] for t in resolved if t["actual_hrs"] > 0]
    avg_resolution = round(sum(actual_times) / max(1, len(actual_times)), 1) if actual_times else 0

    # Average breach severity
    breach_amounts = [t["breach_hours"] for t in breached]
    avg_breach = round(sum(breach_amounts) / max(1, len(breach_amounts)), 1) if breach_amounts else 0

    return {
        "total_tickets": total,
        "resolved": len(resolved),
        "open": len(open_tickets),
        "sla_measured": len(sla_measured),
        "met_sla": len(met),
        "breached_sla": len(breached),
        "compliance_rate_pct": compliance_rate,
        "avg_resolution_hrs": avg_resolution,
        "avg_breach_hrs": avg_breach,
        "max_breach_hrs": round(max(breach_amounts, default=0), 1),
    }


def compute_priority_breakdown(tickets: list) -> list:
    """Compute SLA metrics by priority."""
    priority_data = defaultdict(lambda: {"total": 0, "met": 0, "breached": 0, "actual_times": [], "breach_times": []})

    for t in tickets:
        if t["met_sla"] is not None:
            priority_data[t["priority"]]["total"] += 1
            if t["met_sla"]:
                priority_data[t["priority"]]["met"] += 1
            else:
                priority_data[t["priority"]]["breached"] += 1
                priority_data[t["priority"]]["breach_times"].append(t["breach_hours"])
            if t["actual_hrs"] > 0:
                priority_data[t["priority"]]["actual_times"].append(t["actual_hrs"])

    results = []
    for priority in sorted(priority_data.keys()):
        data = priority_data[priority]
        compliance = round(data["met"] / max(1, data["total"]) * 100, 1)
        avg_time = round(sum(data["actual_times"]) / max(1, len(data["actual_times"])), 1) if data["actual_times"] else 0
        results.append({
            "priority": priority,
            "total": data["total"],
            "met": data["met"],
            "breached": data["breached"],
            "compliance_pct": compliance,
            "avg_resolution_hrs": avg_time,
        })

    return results


def compute_team_breakdown(tickets: list) -> list:
    """Compute SLA metrics by team."""
    team_data = defaultdict(lambda: {"total": 0, "met": 0, "breached": 0, "actual_times": []})

    for t in tickets:
        if t["met_sla"] is not None:
            team_data[t["team"]]["total"] += 1
            if t["met_sla"]:
                team_data[t["team"]]["met"] += 1
            else:
                team_data[t["team"]]["breached"] += 1
            if t["actual_hrs"] > 0:
                team_data[t["team"]]["actual_times"].append(t["actual_hrs"])

    results = []
    for team in sorted(team_data.keys()):
        data = team_data[team]
        compliance = round(data["met"] / max(1, data["total"]) * 100, 1)
        avg_time = round(sum(data["actual_times"]) / max(1, len(data["actual_times"])), 1) if data["actual_times"] else 0
        results.append({
            "team": team,
            "total": data["total"],
            "met": data["met"],
            "breached": data["breached"],
            "compliance_pct": compliance,
            "avg_resolution_hrs": avg_time,
        })

    results.sort(key=lambda x: x["compliance_pct"])
    return results


def compute_trend(tickets: list) -> list:
    """Compute SLA compliance trend by period."""
    period_data = defaultdict(lambda: {"total": 0, "met": 0})

    for t in tickets:
        if t["met_sla"] is not None:
            period_data[t["period"]]["total"] += 1
            if t["met_sla"]:
                period_data[t["period"]]["met"] += 1

    results = []
    for period in sorted(period_data.keys()):
        data = period_data[period]
        compliance = round(data["met"] / max(1, data["total"]) * 100, 1)
        results.append({
            "period": period,
            "total": data["total"],
            "met": data["met"],
            "breached": data["total"] - data["met"],
            "compliance_pct": compliance,
        })

    return results


def find_worst_breaches(tickets: list, top_n: int = 10) -> list:
    """Find the worst SLA breaches."""
    breached = [t for t in tickets if t["met_sla"] is False]
    breached.sort(key=lambda x: x["breach_hours"], reverse=True)
    return breached[:top_n]


def build_recommendations(overall: dict, priority: list, teams: list, trend: list, threshold: float) -> list:
    """Generate recommendations."""
    recs = []

    if overall["compliance_rate_pct"] < threshold:
        recs.append(
            f"Overall SLA compliance ({overall['compliance_rate_pct']}%) is below the {threshold}% target. "
            f"{overall['breached_sla']} tickets breached SLA with an average overshoot of {overall['avg_breach_hrs']} hours."
        )

    # Priority-specific issues
    for p in priority:
        if "Critical" in p["priority"] and p["compliance_pct"] < 95:
            recs.append(
                f"{p['priority']} compliance at {p['compliance_pct']}% is below 95% target. "
                f"{p['breached']} critical tickets breached SLA. Implement immediate escalation triggers."
            )
        elif "High" in p["priority"] and p["compliance_pct"] < 90:
            recs.append(
                f"{p['priority']} compliance at {p['compliance_pct']}% is below 90% target. "
                "Review triage and assignment processes for high-priority tickets."
            )

    # Underperforming teams
    low_teams = [t for t in teams if t["compliance_pct"] < threshold and t["total"] >= 5]
    for t in low_teams[:3]:
        recs.append(
            f"Team '{t['team']}' compliance at {t['compliance_pct']}% ({t['breached']} breaches). "
            "Investigate capacity, skill gaps, or process issues."
        )

    # Trend analysis
    if len(trend) >= 2:
        recent = trend[-1]["compliance_pct"]
        previous = trend[-2]["compliance_pct"]
        if recent < previous - 5:
            recs.append(
                f"Compliance trending down: {previous}% -> {recent}% in the most recent period. "
                "Investigate whether this is driven by volume increase, staffing changes, or process degradation."
            )

    if not recs:
        recs.append(f"SLA compliance is meeting the {threshold}% target. Continue monitoring and look for incremental improvements.")

    return recs


def format_human(overall: dict, priority: list, teams: list, trend: list,
                 worst: list, recommendations: list, threshold: float) -> str:
    """Format results for human-readable output."""
    lines = []
    lines.append("=" * 70)
    lines.append("SLA COMPLIANCE REPORT")
    lines.append("=" * 70)
    lines.append("")

    status = "MEETING TARGET" if overall["compliance_rate_pct"] >= threshold else "BELOW TARGET"
    lines.append(f"  SLA Compliance:        {overall['compliance_rate_pct']}% ({status}, target: {threshold}%)")
    lines.append(f"  Total Tickets:         {overall['total_tickets']}")
    lines.append(f"  Resolved:              {overall['resolved']}")
    lines.append(f"  Open:                  {overall['open']}")
    lines.append(f"  Met SLA:               {overall['met_sla']}")
    lines.append(f"  Breached SLA:          {overall['breached_sla']}")
    lines.append(f"  Avg Resolution:        {overall['avg_resolution_hrs']} hrs")
    lines.append(f"  Avg Breach Overshoot:  {overall['avg_breach_hrs']} hrs")

    # Priority breakdown
    lines.append("")
    lines.append("-" * 70)
    lines.append("BY PRIORITY")
    lines.append("-" * 70)
    lines.append(f"  {'Priority':<18} {'Total':>6} {'Met':>6} {'Breach':>7} {'Compl%':>7} {'Avg Hrs':>8}")
    lines.append(f"  {'-'*18} {'-'*6} {'-'*6} {'-'*7} {'-'*7} {'-'*8}")
    for p in priority:
        flag = " <<<" if p["compliance_pct"] < threshold else ""
        lines.append(
            f"  {p['priority']:<18} {p['total']:>6} {p['met']:>6} {p['breached']:>7} "
            f"{p['compliance_pct']:>6.1f}% {p['avg_resolution_hrs']:>7.1f}h{flag}"
        )

    # Team breakdown
    lines.append("")
    lines.append("-" * 70)
    lines.append("BY TEAM")
    lines.append("-" * 70)
    lines.append(f"  {'Team':<22} {'Total':>6} {'Met':>6} {'Breach':>7} {'Compl%':>7} {'Avg Hrs':>8}")
    lines.append(f"  {'-'*22} {'-'*6} {'-'*6} {'-'*7} {'-'*7} {'-'*8}")
    for t in teams:
        flag = " <<<" if t["compliance_pct"] < threshold else ""
        lines.append(
            f"  {t['team']:<22} {t['total']:>6} {t['met']:>6} {t['breached']:>7} "
            f"{t['compliance_pct']:>6.1f}% {t['avg_resolution_hrs']:>7.1f}h{flag}"
        )

    # Trend
    if trend:
        lines.append("")
        lines.append("-" * 70)
        lines.append("TREND")
        lines.append("-" * 70)
        for t in trend:
            bar_len = int(t["compliance_pct"] / 5)
            bar = "#" * bar_len + "." * (20 - bar_len)
            lines.append(f"  {t['period']:<12} {t['compliance_pct']:>6.1f}% [{bar}] ({t['met']}/{t['total']})")

    # Worst breaches
    if worst:
        lines.append("")
        lines.append("-" * 70)
        lines.append("WORST BREACHES")
        lines.append("-" * 70)
        for w in worst:
            lines.append(
                f"  {w['ticket_id']} | {w['priority']} | {w['team']} | "
                f"Target: {w['sla_target_hrs']}h | Actual: {w['actual_hrs']}h | "
                f"Breach: +{w['breach_hours']}h ({w['breach_pct']:+.1f}%)"
            )

    # Recommendations
    lines.append("")
    lines.append("-" * 70)
    lines.append("RECOMMENDATIONS")
    lines.append("-" * 70)
    for i, rec in enumerate(recommendations, 1):
        lines.append(f"  {i}. {rec}")

    lines.append("")
    return "\n".join(lines)


def main():
    parser = argparse.ArgumentParser(
        description="Track and analyze SLA compliance across service tickets."
    )
    parser.add_argument("--file", required=True, help="Path to tickets CSV")
    parser.add_argument("--threshold", type=float, default=95, help="SLA compliance target percentage (default: 95)")
    parser.add_argument("--period", default="month", choices=["week", "month", "quarter"], help="Trend period grouping (default: month)")
    parser.add_argument("--json", action="store_true", help="Output results as JSON")
    args = parser.parse_args()

    rows = read_csv(args.file)
    if not rows:
        print("Error: No data found in CSV file.", file=sys.stderr)
        sys.exit(1)

    tickets = analyze_tickets(rows, args.period)
    overall = compute_overall_metrics(tickets)
    priority = compute_priority_breakdown(tickets)
    teams = compute_team_breakdown(tickets)
    trend = compute_trend(tickets)
    worst = find_worst_breaches(tickets)
    recommendations = build_recommendations(overall, priority, teams, trend, args.threshold)

    if args.json:
        output = {
            "overall": overall,
            "by_priority": priority,
            "by_team": teams,
            "trend": trend,
            "worst_breaches": worst,
            "recommendations": recommendations,
        }
        print(json.dumps(output, indent=2))
    else:
        print(format_human(overall, priority, teams, trend, worst, recommendations, args.threshold))


if __name__ == "__main__":
    main()