Look-Ahead Scheduler

Overview

Generate rolling look-ahead schedules from the master schedule. Create actionable short-term plans with constraint analysis, crew assignments, and daily coordination.

"Look-ahead planning catches 80% of schedule problems before they occur" — DDC Community

Look-Ahead Hierarchy

┌─────────────────────────────────────────────────────────────────┐
│                    LOOK-AHEAD PLANNING                           │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│  Master Schedule (12+ months)                                    │
│       ↓                                                         │
│  Phase Schedule (3-6 months)                                    │
│       ↓                                                         │
│  6-Week Look-Ahead (make-ready)                                 │
│       ↓                                                         │
│  3-Week Look-Ahead (constraint removal)                         │
│       ↓                                                         │
│  Weekly Work Plan (commitment)                                  │
│       ↓                                                         │
│  Daily Coordination                                              │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘

Technical Implementation

from dataclasses import dataclass, field
from typing import List, Dict, Optional, Set
from datetime import datetime, timedelta
from enum import Enum
from collections import defaultdict

class ConstraintType(Enum):
    PREDECESSOR = "predecessor"
    LABOR = "labor"
    MATERIAL = "material"
    EQUIPMENT = "equipment"
    INFORMATION = "information"
    PERMIT = "permit"
    INSPECTION = "inspection"
    SPACE = "space"
    WEATHER = "weather"

class ConstraintStatus(Enum):
    OPEN = "open"
    IN_PROGRESS = "in_progress"
    RESOLVED = "resolved"
    BLOCKED = "blocked"

class ActivityStatus(Enum):
    NOT_STARTED = "not_started"
    IN_PROGRESS = "in_progress"
    COMPLETE = "complete"
    DELAYED = "delayed"

@dataclass
class Constraint:
    id: str
    activity_id: str
    constraint_type: ConstraintType
    description: str
    responsible_party: str
    needed_by: datetime
    status: ConstraintStatus = ConstraintStatus.OPEN
    resolution_notes: str = ""
    resolved_date: Optional[datetime] = None

@dataclass
class LookAheadActivity:
    id: str
    name: str
    trade: str
    location: str
    planned_start: datetime
    planned_finish: datetime
    duration: int
    labor_hours: float
    crew_size: int
    predecessors: List[str] = field(default_factory=list)
    constraints: List[Constraint] = field(default_factory=list)
    status: ActivityStatus = ActivityStatus.NOT_STARTED
    percent_complete: float = 0.0
    notes: str = ""
    can_start: bool = False

@dataclass
class WeeklyWorkPlan:
    week_start: datetime
    week_end: datetime
    activities: List[LookAheadActivity]
    total_labor_hours: float
    trades_involved: List[str]
    constraints_to_resolve: List[Constraint]

@dataclass
class DailyPlan:
    date: datetime
    activities: List[LookAheadActivity]
    labor_by_trade: Dict[str, int]
    equipment_needed: List[str]
    inspections: List[str]
    safety_focus: str

class LookAheadScheduler:
    """Generate rolling look-ahead schedules."""

    def __init__(self, project_name: str):
        self.project_name = project_name
        self.activities: Dict[str, LookAheadActivity] = {}
        self.constraints: Dict[str, Constraint] = {}
        self.weekly_plans: List[WeeklyWorkPlan] = []

    def import_from_master(self, master_activities: List[Dict],
                          look_ahead_start: datetime,
                          look_ahead_weeks: int = 6) -> int:
        """Import activities from master schedule for look-ahead period."""
        look_ahead_end = look_ahead_start + timedelta(weeks=look_ahead_weeks)
        count = 0

        for act in master_activities:
            start = datetime.fromisoformat(act['planned_start']) if isinstance(act['planned_start'], str) else act['planned_start']
            finish = datetime.fromisoformat(act['planned_finish']) if isinstance(act['planned_finish'], str) else act['planned_finish']

            # Include if overlaps look-ahead period
            if start <= look_ahead_end and finish >= look_ahead_start:
                activity = LookAheadActivity(
                    id=act['id'],
                    name=act['name'],
                    trade=act.get('trade', ''),
                    location=act.get('location', ''),
                    planned_start=start,
                    planned_finish=finish,
                    duration=act.get('duration', (finish - start).days),
                    labor_hours=act.get('labor_hours', 0),
                    crew_size=act.get('crew_size', 0),
                    predecessors=act.get('predecessors', [])
                )
                self.activities[activity.id] = activity
                count += 1

        return count

    def add_constraint(self, activity_id: str, constraint_type: ConstraintType,
                      description: str, responsible_party: str,
                      needed_by: datetime) -> Constraint:
        """Add constraint to activity."""
        constraint_id = f"CON-{len(self.constraints)+1:04d}"

        constraint = Constraint(
            id=constraint_id,
            activity_id=activity_id,
            constraint_type=constraint_type,
            description=description,
            responsible_party=responsible_party,
            needed_by=needed_by
        )

        self.constraints[constraint_id] = constraint

        if activity_id in self.activities:
            self.activities[activity_id].constraints.append(constraint)

        return constraint

    def update_constraint(self, constraint_id: str, status: ConstraintStatus,
                         notes: str = "") -> Constraint:
        """Update constraint status."""
        if constraint_id not in self.constraints:
            raise ValueError(f"Constraint {constraint_id} not found")

        constraint = self.constraints[constraint_id]
        constraint.status = status
        constraint.resolution_notes = notes

        if status == ConstraintStatus.RESOLVED:
            constraint.resolved_date = datetime.now()

        return constraint

    def analyze_make_ready(self) -> Dict[str, List[str]]:
        """Analyze which activities are 'make-ready' (constraints resolved)."""
        ready = []
        not_ready = []
        blocked = []

        for act in self.activities.values():
            # Check predecessors complete
            pred_complete = all(
                self.activities.get(p, {}).status == ActivityStatus.COMPLETE
                for p in act.predecessors if p in self.activities
            )

            # Check constraints resolved
            open_constraints = [c for c in act.constraints
                              if c.status != ConstraintStatus.RESOLVED]

            if not pred_complete:
                blocked.append(act.id)
                act.can_start = False
            elif open_constraints:
                not_ready.append(act.id)
                act.can_start = False
            else:
                ready.append(act.id)
                act.can_start = True

        return {
            "ready": ready,
            "not_ready": not_ready,
            "blocked": blocked
        }

    def generate_weekly_plan(self, week_start: datetime) -> WeeklyWorkPlan:
        """Generate weekly work plan."""
        week_end = week_start + timedelta(days=6)

        # Get activities for this week
        week_activities = [
            act for act in self.activities.values()
            if act.planned_start <= week_