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_end and act.planned_finish >= week_start
        ]

        # Calculate totals
        total_hours = sum(act.labor_hours for act in week_activities)
        trades = list(set(act.trade for act in week_activities if act.trade))

        # Get constraints needing resolution this week
        constraints_due = [
            c for c in self.constraints.values()
            if c.status == ConstraintStatus.OPEN and c.needed_by <= week_end
        ]

        plan = WeeklyWorkPlan(
            week_start=week_start,
            week_end=week_end,
            activities=week_activities,
            total_labor_hours=total_hours,
            trades_involved=trades,
            constraints_to_resolve=constraints_due
        )

        self.weekly_plans.append(plan)
        return plan

    def generate_daily_plan(self, date: datetime) -> DailyPlan:
        """Generate daily coordination plan."""
        # Get activities for this day
        day_activities = [
            act for act in self.activities.values()
            if act.planned_start <= date <= act.planned_finish
            and act.can_start
        ]

        # Labor by trade
        labor_by_trade = defaultdict(int)
        for act in day_activities:
            labor_by_trade[act.trade] += act.crew_size

        # Equipment needed
        equipment = []
        for act in day_activities:
            for c in act.constraints:
                if c.constraint_type == ConstraintType.EQUIPMENT:
                    equipment.append(c.description)

        # Inspections
        inspections = [
            c.description for c in self.constraints.values()
            if c.constraint_type == ConstraintType.INSPECTION
            and c.needed_by.date() == date.date()
        ]

        # Safety focus based on activities
        safety_focus = self._determine_safety_focus(day_activities)

        return DailyPlan(
            date=date,
            activities=day_activities,
            labor_by_trade=dict(labor_by_trade),
            equipment_needed=equipment,
            inspections=inspections,
            safety_focus=safety_focus
        )

    def _determine_safety_focus(self, activities: List[LookAheadActivity]) -> str:
        """Determine daily safety focus based on activities."""
        # Keywords to safety topics
        keywords = {
            "excavation": "Excavation Safety - Shoring, sloping, access",
            "steel": "Steel Erection - Fall protection, crane safety",
            "concrete": "Concrete Pour - Silica, formwork, PPE",
            "roof": "Fall Protection - 100% tie-off, guardrails",
            "electrical": "Electrical Safety - LOTO, qualified personnel",
            "crane": "Crane Safety - Rigging, load charts, signaling",
            "welding": "Hot Work - Fire watch, permits, ventilation"
        }

        for act in activities:
            name_lower = act.name.lower()
            for keyword, safety in keywords.items():
                if keyword in name_lower:
                    return safety

        return "General Site Safety - PPE, housekeeping, awareness"

    def generate_look_ahead_report(self, weeks: int = 3) -> str:
        """Generate look-ahead schedule report."""
        self.analyze_make_ready()
        today = datetime.now()

        lines = [
            f"# {weeks}-Week Look-Ahead Schedule",
            f"",
            f"**Project:** {self.project_name}",
            f"**Generated:** {today.strftime('%Y-%m-%d')}",
            f"**Period:** {today.strftime('%Y-%m-%d')} to {(today + timedelta(weeks=weeks)).strftime('%Y-%m-%d')}",
            f"",
        ]

        # Summary
        make_ready = self.analyze_make_ready()
        lines.extend([
            f"## Summary",
            f"",
            f"- **Total Activities:** {len(self.activities)}",
            f"- **Ready to Start:** {len(make_ready['ready'])}",
            f"- **Awaiting Constraints:** {len(make_ready['not_ready'])}",
            f"- **Blocked:** {len(make_ready['blocked'])}",
            f""
        ])

        # Open constraints
        open_constraints = [c for c in self.constraints.values()
                          if c.status == ConstraintStatus.OPEN]
        if open_constraints:
            lines.extend([
                f"## Open Constraints ({len(open_constraints)})",
                f"",
                f"| Activity | Type | Description | Responsible | Needed By |",
                f"|----------|------|-------------|-------------|-----------|"
            ])
            for c in sorted(open_constraints, key=lambda x: x.needed_by):
                lines.append(
                    f"| {c.activity_id} | {c.constraint_type.value} | {c.description} | "
                    f"{c.responsible_party} | {c.needed_by.strftime('%Y-%m-%d')} |"
                )
            lines.append("")

        # Weekly breakdown
        for week_num in range(weeks):
            week_start = today + timedelta(weeks=week_num)
            week_start = week_start - timedelta(days=week_start.weekday())  # Monday

            plan = self.generate_weekly_plan(week_start)

            lines.extend([
                f"## Week {week_num + 1}: {plan.week_start.strftime('%b %d')} - {plan.week_end.strftime('%b %d')}",
                f"",
                f"**Activities:** {len(plan.activities)} | **Labor:** {plan.total_labor_hours:.0f} hrs | **Trades:** {', '.join(plan.trades_involved)}",
                f""
            ])

            if plan.activities:
                lines.append("| Activity | Trade | Location | Start | Duration | Status |")
                lines.append("|----------|-------|----------|-------|----------|--------|")
                for act in sorted(plan.activities, key=lambda a: a.planned_start):
                    status = "Ready" if act.can_start else "Constraint"
                    lines.append(
                        f"| {act.name} | {act.trade} | {act.location} | "
                        f"{act.planned_start.strftime('%m/%d')} | {act.duration}d | {status} |"
                    )
                lines.append("")

        return "\n".join(lines)

    def get_constraint_log(self) -> str:
        """Generate constraint log."""
        lines = [
            "# Constraint Log",
            "",
            "| ID | Activity | Type | Description | Responsible | Status | Needed By |",
            "|----|----------|------|-------------|-------------|--------|-----------|"
        ]

        for c in sorted(self.constraints.values(), key=lambda x: x.needed_by):
            lines.append(
                f"| {c.id} | {c.activity_id} | {c.constraint_type.value} | "
                f"{c.description} | {c.responsible_party} | {c.status.value} | "
                f"{c.needed_by.strftime('%Y-%m-%d')} |"
            )

        return "\n".join(lines)

Quick Start

from datetime import datetime, timedelta

# Initialize scheduler
scheduler = LookAheadScheduler("Office Tower Construction")

# Import activities from master schedule
master_activities = [
    {
        "id": "A100",
        "name": "Level 5 MEP Rough-in",
        "trade": "Mechanical",
        "location": "Level 5",
        "planned_start": datetime.now(),
        "planned_finish": datetime.now() + timedelta(days=10),
        "labor_hours": 400,
        "crew_size": 5
    },
    {
        "id": "A101",
        "name": "Level 5 Framing",
        "trade": "Carpentry",
        "location": "Level 5",
        "planned_start": datetime.now() + timedelta(days=5),
        "planned_finish": datetime.now() + timedelta(days=15),
        "labor_hours": 320,
        "crew_size": 4,
        "predecessors": ["A100"]
    }
]

scheduler.import_from_master(master_activities, datetime.now(), look_ahead_weeks=6)

# Add constraints
scheduler.add_constraint(
    "A100",
    ConstraintType.MATERIAL,
    "Ductwork delivery",
    "ABC Mechanical",
    datetime.now() + timedelta(days=2)
)

scheduler.add_constraint(
    "A101",
    ConstraintType.INFORMATION,
    "Framing layout drawings",
    "Architect",
    datetime.now() + timedelta(days=4)
)

# Analyze make-ready status
status = scheduler.analyze_make_ready()
print(f"Ready: {len(status['ready'])}, Not Ready: {len(status['not_ready'])}")

# Generate look-ahead report
print(scheduler.generate_look_ahead_report(weeks=3))

# Generate daily plan
daily = scheduler.generate_daily_plan(datetime.now())
print(f"Today's safety focus: {daily.safety_focus}")

Requirements

pip install (no external dependencies)