Look Ahead Scheduler

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)