Track material deliveries, manage inventory, and coordinate logistics. Monitor delivery schedules and site storage.
Material logistics cause project delays:
Centralized material delivery tracking system that manages schedules, monitors status, and coordinates site logistics.
import pandas as pd
from datetime import datetime, date, timedelta
from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from enum import Enum
class DeliveryStatus(Enum):
"""Delivery status."""
SCHEDULED = "scheduled"
IN_TRANSIT = "in_transit"
DELIVERED = "delivered"
PARTIAL = "partial"
DELAYED = "delayed"
CANCELLED = "cancelled"
class DeliveryPriority(Enum):
"""Delivery priority."""
CRITICAL = "critical"
HIGH = "high"
NORMAL = "normal"
LOW = "low"
class MaterialCategory(Enum):
"""Material categories."""
STRUCTURAL = "structural"
CONCRETE = "concrete"
MEP = "mep"
FINISHES = "finishes"
EQUIPMENT = "equipment"
OTHER = "other"
@dataclass
class MaterialItem:
"""Material item in delivery."""
item_id: str
description: str
quantity_ordered: float
quantity_received: float
unit: str
category: MaterialCategory
spec_section: str = ""
notes: str = ""
@property
def is_complete(self) -> bool:
return self.quantity_received >= self.quantity_ordered
@dataclass
class Delivery:
"""Material delivery record."""
delivery_id: str
po_number: str
vendor: str
vendor_contact: str
vendor_phone: str
scheduled_date: date
priority: DeliveryPriority
status: DeliveryStatus
items: List[MaterialItem]
delivery_location: str
storage_area: str
receiver: str = ""
actual_date: Optional[date] = None
tracking_number: str = ""
carrier: str = ""
notes: str = ""
delay_reason: str = ""
def to_dict(self) -> Dict[str, Any]:
return {
'delivery_id': self.delivery_id,
'po_number': self.po_number,
'vendor': self.vendor,
'scheduled_date': self.scheduled_date.isoformat(),
'actual_date': self.actual_date.isoformat() if self.actual_date else None,
'status': self.status.value,
'priority': self.priority.value,
'items_count': len(self.items),
'location': self.delivery_location,
'storage': self.storage_area
}
@dataclass
class StorageArea:
"""Site storage area."""
area_id: str
name: str
location: str
capacity_sqm: float
current_usage_sqm: float
material_types: List[str]
is_covered: bool
access_restrictions: str = ""
class MaterialDeliveryTracker:
"""Track material deliveries and logistics."""
def __init__(self, project_name: str):
self.project_name = project_name
self.deliveries: Dict[str, Delivery] = {}
self.storage_areas: Dict[str, StorageArea] = {}
self._delivery_counter = 0
def schedule_delivery(self,
po_number: str,
vendor: str,
scheduled_date: date,
delivery_location: str,
storage_area: str,
priority: DeliveryPriority = DeliveryPriority.NORMAL,
vendor_contact: str = "",
vendor_phone: str = "") -> Delivery:
"""Schedule new delivery."""
self._delivery_counter += 1
delivery_id = f"DEL-{self._delivery_counter:05d}"
delivery = Delivery(
delivery_id=delivery_id,
po_number=po_number,
vendor=vendor,
vendor_contact=vendor_contact,
vendor_phone=vendor_phone,
scheduled_date=scheduled_date,
priority=priority,
status=DeliveryStatus.SCHEDULED,
items=[],
delivery_location=delivery_location,
storage_area=storage_area
)
self.deliveries[delivery_id] = delivery
return delivery
def add_item(self, delivery_id: str,
description: str,
quantity: float,
unit: str,
category: MaterialCategory,
spec_section: str = "") -> MaterialItem:
"""Add item to delivery."""
if delivery_id not in self.deliveries:
raise ValueError(f"Delivery {delivery_id} not found")
delivery = self.deliveries[delivery_id]
item_id = f"{delivery_id}-{len(delivery.items) + 1:03d}"
item = MaterialItem(
item_id=item_id,
description=description,
quantity_ordered=quantity,
quantity_received=0,
unit=unit,
category=category,
spec_section=spec_section
)
delivery.items.append(item)
return item
def update_status(self, delivery_id: str, status: DeliveryStatus,
tracking_number: str = "", carrier: str = "",
delay_reason: str = ""):
"""Update delivery status."""
if delivery_id not in self.deliveries:
raise ValueError(f"Delivery {delivery_id} not found")
delivery = self.deliveries[delivery_id]
delivery.status = status
if tracking_number:
delivery.tracking_number = tracking_number
if carrier:
delivery.carrier = carrier
if delay_reason:
delivery.delay_reason = delay_reason
def receive_delivery(self, delivery_id: str, receiver: str,
received_quantities: Dict[str, float] = None,
actual_date: date = None):
"""Record delivery receipt."""
if delivery_id not in self.deliveries:
raise ValueError(f"Delivery {delivery_id} not found")
delivery = self.deliveries[delivery_id]
delivery.receiver = receiver
delivery.actual_date = actual_date or date.today()
# Update received quantities
if received_quantities:
for item in delivery.items:
if item.item_id in received_quantities:
item.quantity_received = received_quantities[item.item_id]
else:
# Assume full receipt
for item in delivery.items:
item.quantity_received = item.quantity_ordered
# Check if all items complete
all_complete = all(item.is_complete for item in delivery.items)
if all_complete:
delivery.status = DeliveryStatus.DELIVERED
else:
delivery.status = DeliveryStatus.PARTIAL
def add_storage_area(self, area_id: str, name: str, location: str,
capacity_sqm: float, material_types: List[str],
is_covered: bool = False) -> StorageArea:
"""Add storage area."""
area = StorageArea(
area_id=area_id,
name=name,
location=location,
capacity_sqm=capacity_sqm,
current_usage_sqm=0,
material_types=material_types,
is_covered=is_covered
)
self.storage_areas[area_id] = area
return area
def get_upcoming_deliveries(self, days: int = 7) -> List[Delivery]:
"""Get deliveries scheduled within specified days."""
cutoff = date.today() + timedelta(days=days)
return [d for d in self.deliveries.values()
if d.status in [DeliveryStatus.SCHEDULED, DeliveryStatus.IN_TRANSIT]
and d.scheduled_date <= cutoff]
def get_delayed_deliveries(self) -> List[Delivery]:
"""Get overdue deliveries."""
today = date.today()
return [d for d in self.deliveries.values()
if d.status in [DeliveryStatus.SCHEDULED, DeliveryStatus.IN_TRANSIT, DeliveryStatus.DELAYED]
and d.scheduled_date < today]
def get_deliveries_by_vendor(self, vendor: str) -> List[Delivery]:
"""Get all deliveries from a vendor."""
return [d for d in self.deliveries.values()
if vendor.lower() in d.vendor.lower()]
def get_summary(self) -> Dict[str, Any]:
"""Generate delivery summary."""
deliveries = list(self.deliveries.values())
by_status = {}
by_priority = {}
by_vendor = {}
for d in deliveries:
status = d.status.value
by_status[status] = by_status.get(status, 0) + 1
priority = d.priority.value
by_priority[priority] = by_priority.get(priority, 0) + 1
by_vendor[d.vendor] = by_vendor.get(d.vendor, 0) + 1
# Upcoming this week
upcoming = self.get_upcoming_deliveries(7)
delayed = self.get_delayed_deliveries()
# On-time delivery rate
completed = [d for d in deliveries if d.status == DeliveryStatus.DELIVERED]
on_time = sum(1 for d in completed
if d.actual_date and d.actual_date <= d.scheduled_date)
otd_rate = (on_time / len(completed) * 100) if completed else 0
return {
'project': self.project_name,
'total_deliveries': len(deliveries),
'by_status': by_status,
'by_priority': by_priority,
'by_vendor': by_vendor,
'upcoming_7_days': len(upcoming),
'overdue': len(delayed),
'on_time_delivery_rate': round(otd_rate, 1),
'storage_areas': len(self.storage_areas)
}
def export_schedule(self, output_path: str):
"""Export delivery schedule to Excel."""
with pd.ExcelWriter(output_path, engine='openpyxl') as writer:
# Delivery schedule
schedule_data = [d.to_dict() for d in self.deliveries.values()]
schedule_df = pd.DataFrame(schedule_data)
if not schedule_df.empty:
schedule_df.to_excel(writer, sheet_name='Schedule', index=False)
# Item details
items_data = []
for delivery in self.deliveries.values():
for item in delivery.items:
items_data.append({
'Delivery ID': delivery.delivery_id,
'PO': delivery.po_number,
'Item': item.description,
'Ordered': item.quantity_ordered,
'Received': item.quantity_received,
'Unit': item.unit,
'Category': item.category.value,
'Complete': item.is_complete
})
if items_data:
pd.DataFrame(items_data).to_excel(writer, sheet_name='Items', index=False)
# Upcoming
upcoming = self.get_upcoming_deliveries(14)
if upcoming:
upcoming_df = pd.DataFrame([d.to_dict() for d in upcoming])
upcoming_df.to_excel(writer, sheet_name='Upcoming', index=False)
return output_path
from datetime import date, timedelta
# Initialize tracker
tracker = MaterialDeliveryTracker("Office Tower")
# Schedule delivery
delivery = tracker.schedule_delivery(
po_number="PO-2024-001",
vendor="ABC Steel Supply",
scheduled_date=date.today() + timedelta(days=7),
delivery_location="Site Gate A",
storage_area="Laydown Area 1",
priority=DeliveryPriority.HIGH
)
# Add items
tracker.add_item(delivery.delivery_id, "W8x31 Beams", 50, "EA", MaterialCategory.STRUCTURAL)
tracker.add_item(delivery.delivery_id, "Connection Plates", 200, "EA", MaterialCategory.STRUCTURAL)
# Check upcoming
upcoming = tracker.get_upcoming_deliveries(7)
print(f"Deliveries this week: {len(upcoming)}")
tracker.update_status(delivery.delivery_id, DeliveryStatus.IN_TRANSIT,
tracking_number="TRACK123", carrier="XYZ Freight")
tracker.receive_delivery(delivery.delivery_id, receiver="John Smith")
delayed = tracker.get_delayed_deliveries()
for d in delayed:
print(f"{d.delivery_id}: {d.vendor} - was due {d.scheduled_date}")