cwicr-schedule-integrator

# CWICR Schedule Integrator

## Business Case

### Problem Statement
Project planning requires:
- Linking costs to schedule activities
- Generating cost-loaded schedules
- Projecting cash flow requirements
- Tracking earned value

### Solution
Integrate CWICR cost data with project schedules to create cost-loaded Gantt charts, cash flow curves, and earned value tracking.

### Business Value
- **Cost visibility** - See when costs occur
- **Cash flow** - Project funding requirements
- **Earned value** - Track cost/schedule performance
- **Integration** - Connect cost and schedule data

## Technical Implementation

```python
import pandas as pd
import numpy as np
from typing import Dict, Any, List, Optional, Tuple
from dataclasses import dataclass, field
from datetime import datetime, timedelta, date
from enum import Enum
from collections import defaultdict


class CostDistribution(Enum):
    """Methods for distributing costs over time."""
    UNIFORM = "uniform"          # Even distribution
    FRONT_LOADED = "front_loaded"  # More at start
    BACK_LOADED = "back_loaded"    # More at end
    S_CURVE = "s_curve"          # S-curve distribution


@dataclass
class ScheduleActivity:
    """Project schedule activity."""
    activity_id: str
    description: str
    start_date: date
    end_date: date
    duration_days: int
    work_items: List[str]
    budgeted_cost: float
    predecessors: List[str] = field(default_factory=list)


@dataclass
class CostLoadedActivity:
    """Activity with daily cost distribution."""
    activity: ScheduleActivity
    daily_costs: Dict[date, float]
    cumulative_costs: Dict[date, float]


@dataclass
class CashFlowProjection:
    """Cash flow projection."""
    project_name: str
    start_date: date
    end_date: date
    total_cost: float
    daily_costs: Dict[date, float]
    weekly_costs: Dict[str, float]
    monthly_costs: Dict[str, float]
    cumulative: Dict[date, float]


@dataclass
class EarnedValueMetrics:
    """Earned value metrics at point in time."""
    data_date: date
    planned_value: float  # PV / BCWS
    earned_value: float   # EV / BCWP
    actual_cost: float    # AC / ACWP
    schedule_variance: float  # SV = EV - PV
    cost_variance: float      # CV = EV - AC
    spi: float               # Schedule Performance Index
    cpi: float               # Cost Performance Index
    eac: float               # Estimate at Completion
    etc: float               # Estimate to Complete


class CWICRScheduleIntegrator:
    """Integrate CWICR costs with project schedules."""

    def __init__(self, cwicr_data: pd.DataFrame):
        self.cost_data = cwicr_data
        self._index_data()

    def _index_data(self):
        """Index cost data."""
        if 'work_item_code' in self.cost_data.columns:
            self._code_index = self.cost_data.set_index('work_item_code')
        else:
            self._code_index = None

    def get_work_item_cost(self, code: str, quantity: float) -> float:
        """Get total cost for work item."""
        if self._code_index is None or code not in self._code_index.index:
            return 0

        item = self._code_index.loc[code]
        labor = float(item.get('labor_cost', 0) or 0)
        material = float(item.get('material_cost', 0) or 0)
        equipment = float(item.get('equipment_cost', 0) or 0)

        return (labor + material + equipment) * quantity

    def create_schedule_activity(self,
                                  activity_id: str,
                                  description: str,
                                  start_date: date,
                                  duration_days: int,
                                  work_items: List[Dict[str, Any]],
                                  predecessors: List[str] = None) -> ScheduleActivity:
        """Create schedule activity with linked work items."""

        # Calculate budgeted cost
        total_cost = 0
        codes = []
        for item in work_items:
            code = item.get('work_item_code', item.get('code'))
            qty = item.get('quantity', 0)
            total_cost += self.get_work_item_cost(code, qty)
            codes.append(code)

        end_date = start_date + timedelta(days=duration_days)

        return ScheduleActivity(
            activity_id=activity_id,
            description=description,
            start_date=start_date,
            end_date=end_date,
            duration_days=duration_days,
            work_items=codes,
            budgeted_cost=round(total_cost, 2),
            predecessors=predecessors or []
        )

    def distribute_cost(self,
                        activity: ScheduleActivity,
                        method: CostDistribution = CostDistribution.UNIFORM) -> CostLoadedActivity:
        """Distribute activity cost over duration."""

        daily_costs = {}
        days = activity.duration_days

        if days == 0:
            daily_costs[activity.start_date] = activity.budgeted_cost
        else:
            if method == CostDistribution.UNIFORM:
                daily_amount = activity.budgeted_cost / days
                for i in range(days):
                    day = activity.start_date + timedelta(days=i)
                    daily_costs[day] = daily_amount

            elif method == CostDistribution.FRONT_LOADED:
                # Higher at start, decreasing
                total_weight = sum(range(days, 0, -1))
                for i in range(days):
                    day = activity.start_date + timedelta(days=i)
                    weight = (days - i) / total_weight
                    daily_costs[day] = activity.budgeted_cost * weight

            elif method == CostDistribution.BACK_LOADED:
                # Lower at start, increasing
                total_weight = sum(range(1, days + 1))
                for i in range(days):
                    day = activity.start_date + timedelta(days=i)
                    weight = (i + 1) / total_weight
                    daily_costs[day] =