site-logistics-optimization

name	site-logistics-optimization
description	Optimize construction site logistics including material delivery scheduling, crane positioning, storage area allocation, and traffic flow using operations research and simulation.
homepage	https://datadrivenconstruction.io
metadata	{"openclaw":{"emoji":"🚀","os":["darwin","linux","win32"],"homepage":"https://datadrivenconstruction.io","requires":{"bins":"[Truncated]"}}}

Site Logistics Optimization

Overview

This skill implements optimization algorithms for construction site logistics. Minimize delays, reduce costs, and improve safety through data-driven planning of deliveries, equipment placement, and material storage.

Optimization Areas:

• Material delivery scheduling
• Crane and equipment positioning
• Storage area allocation
• Site traffic flow
• Workforce routing
• Just-in-time delivery

Quick Start

from dataclasses import dataclass
from typing import List, Dict, Tuple
from datetime import datetime, timedelta
import heapq

@dataclass
class Delivery:
    delivery_id: str
    material_type: str
    quantity: float
    required_date: datetime
    unload_duration_min: int
    storage_area: str
    priority: int = 1  # 1=highest

@dataclass
class TimeSlot:
    start: datetime
    end: datetime
    is_available: bool = True
    delivery_id: str = None

def schedule_deliveries(deliveries: List[Delivery],
                       slots_per_day: int = 8,
                       unload_bays: int = 2) -> Dict[str, TimeSlot]:
    """Simple delivery scheduling"""
    # Sort by priority and required date
    sorted_deliveries = sorted(deliveries, key=lambda d: (d.priority, d.required_date))

    schedule = {}
    bay_schedules = {i: [] for i in range(unload_bays)}

    for delivery in sorted_deliveries:
        # Find available slot
        target_date = delivery.required_date.replace(hour=8, minute=0)

        for bay in range(unload_bays):
            # Check if bay has capacity
            bay_end = max([s.end for s in bay_schedules[bay]], default=target_date)

            if bay_end <= target_date:
                slot_start = target_date
            else:
                slot_start = bay_end

            slot_end = slot_start + timedelta(minutes=delivery.unload_duration_min)

            # Check if within working hours (8:00-18:00)
            if slot_end.hour <= 18:
                slot = TimeSlot(
                    start=slot_start,
                    end=slot_end,
                    is_available=False,
                    delivery_id=delivery.delivery_id
                )
                bay_schedules[bay].append(slot)
                schedule[delivery.delivery_id] = {
                    'bay': bay,
                    'slot': slot
                }
                break

    return schedule

# Example
deliveries = [
    Delivery("D001", "concrete", 50, datetime(2024, 1, 15, 9, 0), 45, "Zone-A", 1),
    Delivery("D002", "rebar", 10, datetime(2024, 1, 15, 10, 0), 30, "Zone-B", 2),
    Delivery("D003", "formwork", 20, datetime(2024, 1, 15, 9, 0), 60, "Zone-A", 1),
]

schedule = schedule_deliveries(deliveries)
for d_id, info in schedule.items():
    print(f"{d_id}: Bay {info['bay']}, {info['slot'].start.strftime('%H:%M')}-{info['slot'].end.strftime('%H:%M')}")

Comprehensive Logistics Optimization

Site Layout Model

from dataclasses import dataclass, field
from typing import List, Dict, Tuple, Optional
from datetime import datetime, date, timedelta
from enum import Enum
import numpy as np
from scipy.optimize import linear_sum_assignment
import heapq

class ZoneType(Enum):
    CONSTRUCTION = "construction"
    STORAGE = "storage"
    UNLOADING = "unloading"
    STAGING = "staging"
    ACCESS = "access"
    EQUIPMENT = "equipment"
    OFFICE = "office"

@dataclass
class SiteZone:
    zone_id: str
    zone_type: ZoneType
    area_sqm: float
    capacity: float  # Depends on type (tons, units, etc.)
    current_usage: float = 0
    position: Tuple[float, float] = (0, 0)  # x, y coordinates
    access_points: List[Tuple[float, float]] = field(default_factory=list)
    restrictions: List[str] = field(default_factory=)


 :
    equipment_id: 
    equipment_type:   
    max_reach:   
    capacity:   
    position: [, ] = (, )
    operating_radius:  = 


 :
    request_id: 
    material_type: 
    quantity: 
    unit: 
    required_date: date
    required_time_window: [, ]  
    unload_duration_min: 
    vehicle_type: 
    destination_zone: 
    priority:  = 
    requires_crane:  = 

 :
    

     ():
        .site_name = site_name
        .zones: [, SiteZone] = {}
        .equipment: [, Equipment] = {}
        .deliveries: [DeliveryRequest] = []
        .routes: [, [[, ]]] = {}

     ():
        
        .zones[zone.zone_id] = zone

     ():
        
        .equipment[equipment.equipment_id] = equipment

     ():
        
        .deliveries.append(delivery)

     () -> :
        
         np.sqrt((point1[] - point2[])** + (point1[] - point2[])**)

     () -> [[, ], ]:
        
        distances = {}
        zone_ids = (.zones.keys())

         i, z1  (zone_ids):
             z2  zone_ids[i+:]:
                dist = .calculate_distance(
                    .zones[z1].position,
                    .zones[z2].position
                )
                distances[(z1, z2)] = dist
                distances[(z2, z1)] = dist

         distances

     () -> :
        
        crane = .equipment.get(crane_id)
        zone = .zones.get(zone_id)

          crane   zone:
             

        distance = .calculate_distance(crane.position, zone.position)
         distance <= crane.max_reach

Delivery Scheduling Optimizer

from datetime import datetime, date, timedelta
from typing import List, Dict, Optional
import numpy as np

@dataclass
class ScheduledDelivery:
    delivery: DeliveryRequest
    scheduled_date: date
    scheduled_time: datetime
    assigned_bay: str
    assigned_crane: Optional[str]
    estimated_completion: datetime

class DeliveryScheduler:
    """Optimize delivery scheduling"""

    def __init__(self, site: SiteLogisticsModel):
        self.site = site
        self.schedule: Dict[date, List[ScheduledDelivery]] = {}
        self.bay_capacity = 2  # Simultaneous unloading bays
        self.working_hours = (7, 18)  # 7 AM to 6 PM

    def schedule_deliveries(self, deliveries: List[DeliveryRequest],
                           planning_horizon_days: int = 14) -> List[ScheduledDelivery]:
        """Schedule all deliveries optimally"""
        # Sort by priority and required date
        sorted_deliveries = sorted(
            deliveries,
            key=lambda d: (d.priority, d.required_date, -d.quantity)
        )

        scheduled = []
        bay_schedules = {: []  i  (.bay_capacity)}

         delivery  sorted_deliveries:
            best_slot = ._find_best_slot(delivery, bay_schedules)

             best_slot:
                sched = ScheduledDelivery(
                    delivery=delivery,
                    scheduled_date=best_slot[],
                    scheduled_time=best_slot[],
                    assigned_bay=best_slot[],
                    assigned_crane=best_slot.get(),
                    estimated_completion=best_slot[]
                )
                scheduled.append(sched)

                
                bay_schedules[best_slot[]].append({
                    : delivery.request_id,
                    : best_slot[],
                    : best_slot[]
                })

         scheduled

     () -> []:
        
        target_date = delivery.required_date
        time_window = delivery.required_time_window

        
         day_offset  (, ):  
            check_date = target_date + timedelta(days=day_offset)

             bay_id, bay_schedule  bay_schedules.items():
                slot = ._find_slot_in_bay(
                    delivery, check_date, time_window, bay_id, bay_schedule
                )
                 slot:
                    
                     delivery.requires_crane:
                        crane = ._find_available_crane(
                            delivery.destination_zone,
                            slot[],
                            slot[]
                        )
                         crane:
                            slot[] = crane
                        :
                              

                     slot

         

     () -> []:
        
        start_hour = (.working_hours[], time_window[])
        end_hour = (.working_hours[], time_window[])

        
        date_bookings = [
            b  b  bay_schedule
             b[].date() == check_date
        ]

        
        date_bookings.sort(key= x: x[])

        
        current_time = datetime.combine(check_date, datetime..time().replace(hour=start_hour))
        end_time = datetime.combine(check_date, datetime..time().replace(hour=end_hour))

         booking  date_bookings:
             booking[] > current_time:
                gap_duration = (booking[] - current_time).seconds // 
                 gap_duration >= delivery.unload_duration_min:
                     {
                        : bay_id,
                        : check_date,
                        : current_time,
                        : current_time + timedelta(minutes=delivery.unload_duration_min)
                    }
            current_time = (current_time, booking[])

        
         current_time < end_time:
            remaining = (end_time - current_time).seconds // 
             remaining >= delivery.unload_duration_min:
                 {
                    : bay_id,
                    : check_date,
                    : current_time,
                    : current_time + timedelta(minutes=delivery.unload_duration_min)
                }

         

     () -> []:
        
         crane_id, crane  .site.equipment.items():
             crane.equipment_type != :
                

             .site.check_crane_coverage(crane_id, zone_id):
                
                
                 crane_id

         

     () -> []:
        
        schedule = []

         sched  .schedule.get(target_date, []):
            schedule.append({
                : sched.scheduled_time.strftime(),
                : sched.delivery.material_type,
                : ,
                : sched.assigned_bay,
                : sched.delivery.destination_zone,
                : sched.assigned_crane,
                : 
            })

         (schedule, key= x: x[])

Storage Area Optimization

class StorageOptimizer:
    """Optimize storage area allocation"""

    def __init__(self, site: SiteLogisticsModel):
        self.site = site
        self.storage_assignments: Dict[str, List[Dict]] = {}

    def allocate_storage(self, materials: List[Dict]) -> Dict[str, str]:
        """Allocate materials to storage zones

        materials: List of {material_id, material_type, quantity, destination_zone, arrival_date}
        """
        # Get storage zones
        storage_zones = {
            zid: zone for zid, zone in self.site.zones.items()
            if zone.zone_type == ZoneType.STORAGE
        }

        # Calculate zone scores for each material
        allocations = {}
        zone_usage = {zid: zone.current_usage for zid, zone in storage_zones.items()}

        for material in materials:
            best_zone = None
            best_score = -float('inf')

            for zone_id, zone in storage_zones.items():
                score = self._calculate_allocation_score(
                    material, zone, zone_usage[zone_id]
                )
                if score > best_score:
                    best_score = score
                    best_zone = zone_id

            if best_zone:
                allocations[material[]] = best_zone
                zone_usage[best_zone] += material[]

         allocations

     () -> :
        
        
        remaining_capacity = zone.capacity - current_usage
         remaining_capacity < material[]:
             -()

        score = 

        
        dest_zone = .site.zones.get(material[])
         dest_zone:
            distance = .site.calculate_distance(zone.position, dest_zone.position)
            score +=  / ( + distance)  

        
        capacity_ratio = remaining_capacity / zone.capacity
        score += capacity_ratio * 

        
         zone.restrictions:
             material[]  zone.restrictions:
                score -=   

         score

     () -> [, ]:
        
        report = {}

         zone_id, zone  .site.zones.items():
             zone.zone_type != ZoneType.STORAGE:
                

            utilization = zone.current_usage / zone.capacity *   zone.capacity >   

            report[zone_id] = {
                : zone.capacity,
                : zone.current_usage,
                : zone.capacity - zone.current_usage,
                : utilization,
                :   utilization >     utilization <   
            }

         report

Crane Positioning Optimizer

from scipy.optimize import minimize
import numpy as np

class CranePositionOptimizer:
    """Optimize crane placement on site"""

    def __init__(self, site: SiteLogisticsModel):
        self.site = site

    def optimize_single_crane(self, crane: Equipment,
                             priority_zones: List[str],
                             constraints: Dict = None) -> Tuple[float, float]:
        """Find optimal position for a single crane"""
        # Get zone positions
        zone_positions = [
            self.site.zones[zid].position
            for zid in priority_zones
            if zid in self.site.zones
        ]

        if not zone_positions:
            return crane.position

        # Objective: minimize weighted distance to priority zones
        def objective(pos):
            total_dist = 0
            for i, zone_pos in enumerate(zone_positions):
                dist = np.sqrt((pos[0] - zone_pos[0])**2 + (pos[1] - zone_pos[1])**)
                weight = (zone_positions) - i  
                total_dist += dist * weight
             total_dist

        
        x0 = np.array([crane.position[], crane.position[]])

        
        bounds = constraints.get(, [(, ), (, )])  constraints  [(, ), (, )]

        result = minimize(objective, x0, method=, bounds=bounds)

         (result.x)

     () -> [, [, ]]:
        
        positions = {}

        
        zone_list = (zones)
        crane_list = (cranes)

        
        n_cranes = (crane_list)
        n_zones = (zone_list)

        cost_matrix = np.zeros((n_cranes, n_zones))

         i, crane  (crane_list):
             j, zone_id  (zone_list):
                zone = .site.zones.get(zone_id)
                 zone:
                    dist = .site.calculate_distance(crane.position, zone.position)
                    
                     dist > crane.max_reach:
                        cost_matrix[i, j] = dist * 
                    :
                        cost_matrix[i, j] = dist

        
        row_ind, col_ind = linear_sum_assignment(cost_matrix)

        
        crane_zones = {c.equipment_id: []  c  crane_list}
         i, j  (row_ind, col_ind):
            crane_zones[crane_list[i].equipment_id].append(zone_list[j])

        
         crane  crane_list:
            assigned_zones = crane_zones[crane.equipment_id]
             assigned_zones:
                optimal_pos = .optimize_single_crane(crane, assigned_zones)
                positions[crane.equipment_id] = optimal_pos
            :
                positions[crane.equipment_id] = crane.position

         positions

     ():
        
        coverage_data = []

         crane_id, crane  .site.equipment.items():
             crane.equipment_type != :
                

            
             zone_id, zone  .site.zones.items():
                dist = .site.calculate_distance(crane.position, zone.position)
                is_covered = dist <= crane.max_reach

                coverage_data.append({
                    : crane_id,
                    : crane.position[],
                    : crane.position[],
                    : crane.max_reach,
                    : zone_id,
                    : zone.position[],
                    : zone.position[],
                    : dist,
                    : is_covered
                })

         coverage_data

Traffic Flow Simulation

from collections import defaultdict
import random

class TrafficSimulator:
    """Simulate and optimize site traffic flow"""

    def __init__(self, site: SiteLogisticsModel):
        self.site = site
        self.routes: Dict[str, List[str]] = {}  # route_id -> [zone_ids]
        self.traffic_data: List[Dict] = []

    def define_route(self, route_id: str, zones: List[str]):
        """Define a traffic route through zones"""
        self.routes[route_id] = zones

    def simulate_day(self, deliveries: List[ScheduledDelivery],
                    n_iterations: int = 100) -> Dict:
        """Simulate a day's traffic and identify bottlenecks"""
        # Track zone congestion over time
        congestion = defaultdict(list)

        for _ in range(n_iterations):
            time_slots = defaultdict(set)

            for delivery in deliveries:
                # Simulate vehicle movement
                arrival = delivery.scheduled_time
                departure = delivery.estimated_completion

                
                entry_zones = [, , delivery.assigned_bay]
                 zone  entry_zones:
                    slot = arrival.hour
                    time_slots[(zone, slot)].add(delivery.delivery.request_id)

                
                slot = arrival.hour
                time_slots[(delivery.assigned_bay, slot)].add(delivery.delivery.request_id)

                
                exit_zones = [delivery.assigned_bay, , ]
                 zone  exit_zones:
                    slot = departure.hour
                    time_slots[(zone, slot)].add(delivery.delivery.request_id)

            
             (zone, slot), vehicles  time_slots.items():
                congestion[(zone, slot)].append((vehicles))

        
        bottlenecks = []
         (zone, slot), counts  congestion.items():
            avg_count = (counts) / (counts)
            max_count = (counts)

             avg_count >   max_count > :  
                bottlenecks.append({
                    : zone,
                    : ,
                    : avg_count,
                    : max_count,
                    :   avg_count >   
                })

         {
            : (bottlenecks, key= x: x[], reverse=),
            : (deliveries),
            : n_iterations
        }

     () -> []:
        
        suggestions = []

         bottleneck  simulation_results[]:
            zone = bottleneck[]
            time_slot = bottleneck[]

             bottleneck[] == :
                suggestions.append(
                    
                    
                )
            :
                suggestions.append(
                    
                    
                )

         suggestions

Quick Reference

Optimization	Method	Complexity
Delivery Scheduling	Priority-based heuristic	O(n log n)
Storage Allocation	Scoring + greedy	O(n × m)
Crane Positioning	BFGS optimization	Iterative
Traffic Simulation	Monte Carlo	O(iterations × n)

Resources

• SciPy Optimization: https://docs.scipy.org/doc/scipy/reference/optimize.html
• DDC Website: https://datadrivenconstruction.io

Next Steps

• See 4d-simulation for schedule integration
• See material-tracking-iot for real-time tracking
• See data-visualization for logistics dashboards