Skip to content

Enhanced Transport Planning Constraints

Jump to bottom
Agilasoft Cloud Technologies edited this page Jan 8, 2026 · 8 revisions

Design Document: Enhanced Transport Planning Constraints

Executive Summary

This document outlines the design for implementing additional constraints in the Transport Planning system to improve vehicle assignment accuracy and compliance. The constraints include:

  1. Time Window Constraints: Pick and Drop window start/end times from addresses, with travel time calculations using vehicle average speed and routing provider distances, including loading and unloading times
  2. Address Day Availability: Day-of-week restrictions for pick and drop operations at addresses
  3. Vehicle Plate Number Coding: Restrictions based on license plate numbers (e.g., odd/even days, last digit rules)
  4. Truck Ban Constraints: Area and time-based restrictions on vehicle movement
  5. Ad-Hoc Factors: User-entered temporary constraints like road closures, port congestion, weather events, etc.

Key Requirements:

  • Vehicle Average Speed: Transport Vehicle must have avg_speed data to calculate travel time
  • Distance Calculation: Distance should be calculated by routing provider (with fallback to existing distance_km field)
  • Loading/Unloading Time: Must be considered in time window calculations (stored in Pick and Drop Mode or Address)

Current State Analysis

Existing Infrastructure

  1. Time Windows:

    • Fields exist on Transport Leg: pick_window_start, pick_window_end, drop_window_start, drop_window_end
    • Fields are fetched from Address custom fields: custom_pickup_window_start, custom_pickup_window_end, custom_drop_window_start, custom_drop_windows_end
    • Currently used for display and basic validation, but NOT used in vehicle selection logic

  2. Address Day Availability:

    • Fields exist on Address: custom_pick_monday through custom_pick_sunday (Check fields)
    • Fields exist on Address: custom_drop_monday through custom_drop_sunday (Check fields)
    • User-defined checkboxes to indicate which days of the week pick/drop operations are allowed
    • Currently NOT used in vehicle selection logic

  3. Vehicle Selection Logic:

    • _find_candidate_vehicle() in transport_plan.py - main vehicle selection function
    • _vehicle_free_on_date() - checks vehicle availability by date
    • Currently checks: vehicle type, capacity, availability (estimated_return_datetime)

Gaps Identified

  1. Time windows are not considered when selecting vehicles
  2. Address day availability is not checked during vehicle assignment
  3. Plate number coding logic is not implemented
  4. Truck ban constraints are not checked during vehicle assignment
  5. Ad-hoc factors (road closures, port congestion, etc.) are not tracked or considered
  6. No unified constraint checking system

Proposed Solution Architecture

1. Constraint Checking Framework

Create a unified constraint checking system that validates vehicles against all applicable constraints before assignment.

1.1 Constraint Validator Module

Location: logistics/transport/constraint_validator.py

Purpose: Centralized constraint checking logic

Key Functions:

def validate_vehicle_constraints(
    vehicle: Dict[str, Any],
    leg: Dict[str, Any],
    scheduled_datetime: datetime,
    debug: Optional[List[str]] = None
) -> Tuple[bool, Optional[str], int, List[Dict[str, Any]]]:
    """
    Validate vehicle against all applicable constraints.
    
    Returns:
        (is_valid, reason_if_invalid, delay_minutes, alternative_routes)
        - is_valid: True if vehicle can be assigned, False if blocked
        - reason_if_invalid: Error message if blocked, warning message if delays
        - delay_minutes: Total estimated delay in minutes (0 if no delays)
        - alternative_routes: List of alternative route suggestions
    """
    # Check time windows
    # Check address day availability
    # Check plate number coding
    # Check truck bans
    # Check ad-hoc factors
    # Return combined result
    pass

def check_time_window_constraints(
    vehicle: Dict[str, Any],
    leg: Dict[str, Any],
    scheduled_datetime: datetime
) -> Tuple[bool, Optional[str]]:
    """Check if vehicle can meet pick/drop time windows"""
    pass

def check_plate_coding_constraints(
    vehicle: Dict[str, Any],
    scheduled_datetime: datetime,
    address: Optional[str] = None
) -> Tuple[bool, Optional[str]]:
    """Check if vehicle plate number is allowed on scheduled date/time"""
    pass

def check_truck_ban_constraints(
    vehicle: Dict[str, Any],
    scheduled_datetime: datetime,
    pick_address: Optional[str] = None,
    drop_address: Optional[str] = None
) -> Tuple[bool, Optional[str]]:
    """Check if vehicle is banned from area/route at scheduled time"""
    pass

def check_address_day_availability(
    address: str,
    scheduled_date: date,
    operation_type: str  # "pick" or "drop"
) -> Tuple[bool, Optional[str]]:
    """Check if address is available for pick/drop on the scheduled day"""
    pass

def check_adhoc_factors(
    scheduled_datetime: datetime,
    address: Optional[str] = None,
    route: Optional[List[str]] = None  # List of addresses in route
) -> Tuple[bool, Optional[str]]:
    """Check for ad-hoc factors affecting transport (road closures, port congestion, etc.)"""
    pass

2. Time Window Constraints

2.1 Data Model

Existing Fields (no changes needed):

  • Transport Leg.pick_window_start (Time)
  • Transport Leg.pick_window_end (Time)
  • Transport Leg.drop_window_start (Time)
  • Transport Leg.drop_window_end (Time)

2.2 Data Model Extensions

Existing Fields:

  • Transport Vehicle.avg_speed (Float) - Average speed in km/h (already exists)
  • Transport Leg.distance_km (Float) - Distance in kilometers (already exists)
  • Transport Leg.routing_provider (Link) - Routing provider used (already exists)

New Fields Required:

On Pick and Drop Mode (user-defined loading/unloading time calculation):

{
  "fieldname": "base_loading_time_minutes",
  "fieldtype": "Float",
  "label": "Base Loading Time (minutes)",
  "description": "Base time required for loading operations (minimum time regardless of volume/weight)",
  "default": 15
},
{
  "fieldname": "loading_time_calculation_method",
  "fieldtype": "Select",
  "label": "Loading Time Calculation Method",
  "options": "Fixed Time\nVolume-Based\nWeight-Based\nVolume and Weight Combined",
  "default": "Volume-Based",
  "description": "How to calculate loading time based on cargo"
},
{
  "fieldname": "loading_time_per_volume_m3",
  "fieldtype": "Float",
  "label": "Loading Time per m³ (minutes)",
  "description": "Additional minutes per cubic meter of volume",
  "depends_on": "eval:in_list(['Volume-Based', 'Volume and Weight Combined'], doc.loading_time_calculation_method)",
  "default": 5
},
{
  "fieldname": "loading_time_per_weight_kg",
  "fieldtype": "Float",
  "label": "Loading Time per 100kg (minutes)",
  "description": "Additional minutes per 100kg of weight",
  "depends_on": "eval:in_list(['Weight-Based', 'Volume and Weight Combined'], doc.loading_time_calculation_method)",
  "default": 2
},
{
  "fieldname": "max_loading_time_minutes",
  "fieldtype": "Float",
  "label": "Maximum Loading Time (minutes)",
  "description": "Maximum loading time cap (0 = no limit)"
},
{
  "fieldname": "base_unloading_time_minutes",
  "fieldtype": "Float",
  "label": "Base Unloading Time (minutes)",
  "description": "Base time required for unloading operations (minimum time regardless of volume/weight)",
  "default": 15
},
{
  "fieldname": "unloading_time_calculation_method",
  "fieldtype": "Select",
  "label": "Unloading Time Calculation Method",
  "options": "Fixed Time\nVolume-Based\nWeight-Based\nVolume and Weight Combined",
  "default": "Volume-Based",
  "description": "How to calculate unloading time based on cargo"
},
{
  "fieldname": "unloading_time_per_volume_m3",
  "fieldtype": "Float",
  "label": "Unloading Time per m³ (minutes)",
  "description": "Additional minutes per cubic meter of volume",
  "depends_on": "eval:in_list(['Volume-Based', 'Volume and Weight Combined'], doc.unloading_time_calculation_method)",
  "default": 5
},
{
  "fieldname": "unloading_time_per_weight_kg",
  "fieldtype": "Float",
  "label": "Unloading Time per 100kg (minutes)",
  "description": "Additional minutes per 100kg of weight",
  "depends_on": "eval:in_list(['Weight-Based', 'Volume and Weight Combined'], doc.unloading_time_calculation_method)",
  "default": 2
},
{
  "fieldname": "max_unloading_time_minutes",
  "fieldtype": "Float",
  "label": "Maximum Unloading Time (minutes)",
  "description": "Maximum unloading time cap (0 = no limit)"
}

Alternative: On Address (if loading/unloading time calculation varies by address):

{
  "fieldname": "custom_base_loading_time_minutes",
  "fieldtype": "Float",
  "label": "Base Loading Time (minutes)"
},
{
  "fieldname": "custom_loading_time_calculation_method",
  "fieldtype": "Select",
  "options": "Use Pick Mode Settings\nFixed Time\nVolume-Based\nWeight-Based\nVolume and Weight Combined"
},
{
  "fieldname": "custom_loading_time_per_volume_m3",
  "fieldtype": "Float",
  "label": "Loading Time per m³ (minutes)"
},
{
  "fieldname": "custom_loading_time_per_weight_kg",
  "fieldtype": "Float",
  "label": "Loading Time per 100kg (minutes)"
},
{
  "fieldname": "custom_base_unloading_time_minutes",
  "fieldtype": "Float",
  "label": "Base Unloading Time (minutes)"
},
{
  "fieldname": "custom_unloading_time_calculation_method",
  "fieldtype": "Select",
  "options": "Use Drop Mode Settings\nFixed Time\nVolume-Based\nWeight-Based\nVolume and Weight Combined"
},
{
  "fieldname": "custom_unloading_time_per_volume_m3",
  "fieldtype": "Float",
  "label": "Unloading Time per m³ (minutes)"
},
{
  "fieldname": "custom_unloading_time_per_weight_kg",
  "fieldtype": "Float",
  "label": "Unloading Time per 100kg (minutes)"
}

On Transport Settings (fallback defaults):

{
  "fieldname": "default_base_loading_time_minutes",
  "fieldtype": "Float",
  "label": "Default Base Loading Time (minutes)",
  "default": 15
},
{
  "fieldname": "default_loading_time_per_volume_m3",
  "fieldtype": "Float",
  "label": "Default Loading Time per m³ (minutes)",
  "default": 5
},
{
  "fieldname": "default_loading_time_per_weight_kg",
  "fieldtype": "Float",
  "label": "Default Loading Time per 100kg (minutes)",
  "default": 2
},
{
  "fieldname": "default_base_unloading_time_minutes",
  "fieldtype": "Float",
  "label": "Default Base Unloading Time (minutes)",
  "default": 15
},
{
  "fieldname": "default_unloading_time_per_volume_m3",
  "fieldtype": "Float",
  "label": "Default Unloading Time per m³ (minutes)",
  "default": 5
},
{
  "fieldname": "default_unloading_time_per_weight_kg",
  "fieldtype": "Float",
  "label": "Default Unloading Time per 100kg (minutes)",
  "default": 2
}

2.3 Logic Implementation

Travel Time Calculation Formula:

Travel Time (minutes) = (Distance in km / Average Speed in km/h) × 60
Total Time = Travel Time + Loading Time + Unloading Time

Loading/Unloading Time Calculation Formula:

Base Time = base_loading_time_minutes (from Pick/Drop Mode, Address, or default)

If calculation_method = "Fixed Time":
    Loading/Unloading Time = Base Time

If calculation_method = "Volume-Based":
    Loading/Unloading Time = Base Time + (Volume in m³ × time_per_volume_m3)

If calculation_method = "Weight-Based":
    Loading/Unloading Time = Base Time + (Weight in kg / 100 × time_per_weight_kg)

If calculation_method = "Volume and Weight Combined":
    Loading/Unloading Time = Base Time + 
                              (Volume in m³ × time_per_volume_m3) + 
                              (Weight in kg / 100 × time_per_weight_kg)

If max_time_minutes > 0:
    Final Time = min(Calculated Time, max_time_minutes)

Algorithm:

  1. For each leg in a Run Sheet, calculate required arrival times:

    • Pick Arrival: Must arrive at pick address between pick_window_start and pick_window_end
    • Pick Departure: Pick arrival + loading time
    • Drop Arrival: Pick departure + travel time (pick to drop)
    • Drop Departure: Drop arrival + unloading time
    • Drop Completion: Must complete within drop_window_end

  2. When selecting a vehicle:

    • Get vehicle's average speed from Transport Vehicle.avg_speed
    • Get distance from routing provider or use Transport Leg.distance_km if available
    • Get cargo volume and weight from Transport Leg (from cargo_weight_kg and calculated volume from packages)
    • Calculate loading time based on:
      • Pick Mode settings (base time + volume/weight-based calculation)
      • Address custom settings (if override specified)
      • Default settings (fallback)
    • Calculate unloading time based on:
      • Drop Mode settings (base time + volume/weight-based calculation)
      • Address custom settings (if override specified)
      • Default settings (fallback)
    • Calculate estimated travel times considering all factors
    • Verify that vehicle can arrive within time windows
    • Consider multiple legs in a Run Sheet (sequential time windows)

  3. Time Window Validation with Full Time Calculation:

    def can_meet_time_windows(vehicle, legs, estimated_dispatch_datetime):
    """
    Check if vehicle can meet all time windows considering:
    - Travel time (distance / avg_speed)
    - Loading time at pick location
    - Unloading time at drop location
    """
    from frappe.utils import get_datetime, add_to_date
    
    current_time = estimated_dispatch_datetime
    vehicle_avg_speed = vehicle.get("avg_speed", 0)  # km/h
    
    # Fallback to default speed if not set
    if not vehicle_avg_speed or vehicle_avg_speed <= 0:
        vehicle_avg_speed = frappe.db.get_single_value(
            "Transport Settings", 
            "routing_default_avg_speed_kmh"
        ) or 50  # Default 50 km/h
    
    for leg in legs:
        # Get cargo volume and weight from leg
        cargo_weight_kg = leg.get("cargo_weight_kg", 0) or 0
        cargo_volume_m3 = calculate_leg_volume(leg)  # Sum from packages or use leg field
        
        # Get loading time (calculated based on volume/weight and pick_mode)
        loading_time_minutes = calculate_loading_time(
            leg, cargo_weight_kg, cargo_volume_m3
        )
        
        # Get unloading time (calculated based on volume/weight and drop_mode)
        unloading_time_minutes = calculate_unloading_time(
            leg, cargo_weight_kg, cargo_volume_m3
        )
        
        # Calculate travel time to pick location
        # Get distance from routing provider or use existing distance_km
        distance_to_pick = get_distance(
            vehicle.current_location or vehicle.base_facility,
            leg.pick_address,
            routing_provider=leg.get("routing_provider")
        )
        
        travel_to_pick_minutes = calculate_travel_time_minutes(
            distance_to_pick, 
            vehicle_avg_speed
        )
        
        # Estimated pick arrival
        estimated_pick_arrival = add_to_date(
            current_time, 
            minutes=travel_to_pick_minutes
        )
        
        # Check if within pick window
        pick_window_start = combine_datetime(leg.date, leg.pick_window_start)
        pick_window_end = combine_datetime(leg.date, leg.pick_window_end)
        
        if not (pick_window_start <= estimated_pick_arrival <= pick_window_end):
            return False, f"Cannot meet pick window for leg {leg.name}. Estimated arrival: {estimated_pick_arrival}, Window: {pick_window_start} - {pick_window_end}"
        
        # Pick departure = arrival + loading time
        estimated_pick_departure = add_to_date(
            estimated_pick_arrival,
            minutes=loading_time_minutes
        )
        
        # Calculate travel time from pick to drop
        distance_pick_to_drop = get_distance(
            leg.pick_address,
            leg.drop_address,
            routing_provider=leg.get("routing_provider")
        ) or leg.get("distance_km", 0)
        
        travel_pick_to_drop_minutes = calculate_travel_time_minutes(
            distance_pick_to_drop,
            vehicle_avg_speed
        )
        
        # Estimated drop arrival
        estimated_drop_arrival = add_to_date(
            estimated_pick_departure,
            minutes=travel_pick_to_drop_minutes
        )
        
        # Check if within drop window
        drop_window_start = combine_datetime(leg.date, leg.drop_window_start)
        drop_window_end = combine_datetime(leg.date, leg.drop_window_end)
        
        # Estimated drop completion = arrival + unloading time
        estimated_drop_completion = add_to_date(
            estimated_drop_arrival,
            minutes=unloading_time_minutes
        )
        
        if not (drop_window_start <= estimated_drop_completion <= drop_window_end):
            return False, f"Cannot meet drop window for leg {leg.name}. Estimated completion: {estimated_drop_completion}, Window: {drop_window_start} - {drop_window_end}"
        
        # Update current_time for next leg
        current_time = estimated_drop_completion
    
    return True, None

  4. Loading/Unloading Time Calculation Functions:

    def calculate_leg_volume(leg) -> float:
    """Calculate total volume from leg packages or use leg field"""
    # Try to get volume from leg field first
    if hasattr(leg, "cargo_volume_m3") and leg.cargo_volume_m3:
        return float(leg.cargo_volume_m3)
    
    # Otherwise, sum from packages
    volume = 0.0
    if hasattr(leg, "transport_job") and leg.transport_job:
        try:
            packages = frappe.get_all(
                "Transport Job Package",
                filters={"parent": leg.transport_job},
                fields=["volume", "length", "width", "height"]
            )
            for pkg in packages:
                if pkg.get("volume"):
                    volume += float(pkg.volume)
                elif all([pkg.get("length"), pkg.get("width"), pkg.get("height")]):
                    # Calculate volume: length × width × height (in meters)
                    volume += (float(pkg.length) * float(pkg.width) * float(pkg.height)) / 1000000  # Convert cm³ to m³
        except Exception:
            pass
    
    return volume

def calculate_loading_time(leg, cargo_weight_kg: float, cargo_volume_m3: float) -> float:
    """
    Calculate loading time based on Pick Mode settings, volume, and weight.
    
    Priority:
    1. Address custom settings (if override specified)
    2. Pick Mode settings
    3. Transport Settings defaults
    """
    pick_mode = leg.get("pick_mode")
    pick_address = leg.get("pick_address")
    
    # Check if address has custom settings
    if pick_address:
        try:
            address_doc = frappe.get_doc("Address", pick_address)
            if hasattr(address_doc, "custom_loading_time_calculation_method") and \
               address_doc.custom_loading_time_calculation_method and \
               address_doc.custom_loading_time_calculation_method != "Use Pick Mode Settings":
                return _calculate_time_from_settings(
                    address_doc,
                    cargo_weight_kg,
                    cargo_volume_m3,
                    prefix="custom_loading_time",
                    operation="loading"
                )
        except Exception:
            pass
    
    # Use Pick Mode settings
    if pick_mode:
        try:
            mode_doc = frappe.get_doc("Pick and Drop Mode", pick_mode)
            return _calculate_time_from_settings(
                mode_doc,
                cargo_weight_kg,
                cargo_volume_m3,
                prefix="loading_time",
                operation="loading"
            )
        except Exception:
            pass
    
    # Fallback to Transport Settings defaults
    return _calculate_time_from_defaults(
        cargo_weight_kg,
        cargo_volume_m3,
        operation="loading"
    )

def calculate_unloading_time(leg, cargo_weight_kg: float, cargo_volume_m3: float) -> float:
    """
    Calculate unloading time based on Drop Mode settings, volume, and weight.
    
    Priority:
    1. Address custom settings (if override specified)
    2. Drop Mode settings
    3. Transport Settings defaults
    """
    drop_mode = leg.get("drop_mode")
    drop_address = leg.get("drop_address")
    
    # Check if address has custom settings
    if drop_address:
        try:
            address_doc = frappe.get_doc("Address", drop_address)
            if hasattr(address_doc, "custom_unloading_time_calculation_method") and \
               address_doc.custom_unloading_time_calculation_method and \
               address_doc.custom_unloading_time_calculation_method != "Use Drop Mode Settings":
                return _calculate_time_from_settings(
                    address_doc,
                    cargo_weight_kg,
                    cargo_volume_m3,
                    prefix="custom_unloading_time",
                    operation="unloading"
                )
        except Exception:
            pass
    
    # Use Drop Mode settings
    if drop_mode:
        try:
            mode_doc = frappe.get_doc("Pick and Drop Mode", drop_mode)
            return _calculate_time_from_settings(
                mode_doc,
                cargo_weight_kg,
                cargo_volume_m3,
                prefix="unloading_time",
                operation="unloading"
            )
        except Exception:
            pass
    
    # Fallback to Transport Settings defaults
    return _calculate_time_from_defaults(
        cargo_weight_kg,
        cargo_volume_m3,
        operation="unloading"
    )

def _calculate_time_from_settings(doc, cargo_weight_kg: float, cargo_volume_m3: float, 
                                  prefix: str, operation: str) -> float:
    """Calculate time from document settings (Mode or Address)"""
    # Get base time
    base_time = getattr(doc, f"{prefix}_base_{operation}_minutes", None) or \
                getattr(doc, f"base_{operation}_time_minutes", None) or 0
    
    # Get calculation method
    method = getattr(doc, f"{prefix}_calculation_method", None) or \
             getattr(doc, f"{operation}_time_calculation_method", None) or \
             "Volume-Based"
    
    calculated_time = float(base_time)
    
    if method == "Fixed Time":
        # Just use base time
        pass
    elif method == "Volume-Based":
        time_per_volume = getattr(doc, f"{prefix}_per_volume_m3", None) or \
                         getattr(doc, f"{operation}_time_per_volume_m3", None) or 0
        calculated_time += cargo_volume_m3 * float(time_per_volume)
    elif method == "Weight-Based":
        time_per_weight = getattr(doc, f"{prefix}_per_weight_kg", None) or \
                         getattr(doc, f"{operation}_time_per_weight_kg", None) or 0
        # time_per_weight is per 100kg
        calculated_time += (cargo_weight_kg / 100.0) * float(time_per_weight)
    elif method == "Volume and Weight Combined":
        time_per_volume = getattr(doc, f"{prefix}_per_volume_m3", None) or \
                         getattr(doc, f"{operation}_time_per_volume_m3", None) or 0
        time_per_weight = getattr(doc, f"{prefix}_per_weight_kg", None) or \
                         getattr(doc, f"{operation}_time_per_weight_kg", None) or 0
        calculated_time += (cargo_volume_m3 * float(time_per_volume)) + \
                         ((cargo_weight_kg / 100.0) * float(time_per_weight))
    
    # Apply maximum cap if set
    max_time = getattr(doc, f"max_{operation}_time_minutes", None) or 0
    if max_time and max_time > 0:
        calculated_time = min(calculated_time, float(max_time))
    
    return max(calculated_time, 0)  # Ensure non-negative

def _calculate_time_from_defaults(cargo_weight_kg: float, cargo_volume_m3: float, 
                                  operation: str) -> float:
    """Calculate time from Transport Settings defaults"""
    settings = frappe.get_single("Transport Settings")
    
    base_time = getattr(settings, f"default_base_{operation}_time_minutes", 15) or 15
    time_per_volume = getattr(settings, f"default_{operation}_time_per_volume_m3", 5) or 5
    time_per_weight = getattr(settings, f"default_{operation}_time_per_weight_kg", 2) or 2
    
    # Use volume-based calculation as default
    calculated_time = float(base_time) + (cargo_volume_m3 * float(time_per_volume))
    
    return max(calculated_time, 0)

  5. Example Calculations:

    • Small cargo (0.5 m³, 50 kg):

      • Base: 15 min
      • Volume component: 0.5 × 5 = 2.5 min
      • Total: 17.5 min

    • Medium cargo (5 m³, 500 kg):

      • Base: 15 min
      • Volume component: 5 × 5 = 25 min
      • Total: 40 min

    • Large cargo (20 m³, 2000 kg) with max cap of 120 min:

      • Base: 15 min
      • Volume component: 20 × 5 = 100 min
      • Total: 115 min (capped at 120, so 115 min)

    • Heavy cargo (2 m³, 5000 kg) using weight-based:

      • Base: 15 min
      • Weight component: (5000 / 100) × 2 = 100 min
      • Total: 115 min

    def calculate_travel_time_minutes(distance_km, avg_speed_kmh): """Calculate travel time in minutes from distance and average speed""" if not distance_km or distance_km <= 0: return 0 if not avg_speed_kmh or avg_speed_kmh <= 0: return 0

    # Time in hours = distance / speed
# Convert to minutes
return (distance_km / avg_speed_kmh) * 60

    def get_distance(from_address, to_address, routing_provider=None): """ Get distance between two addresses. Priority: 1. Use routing provider if available 2. Use cached distance if available 3. Use existing distance_km from leg if available 4. Calculate using routing service """ # Try to get from routing provider if routing_provider: # Call routing provider API # This would integrate with existing routing logic pass

    # Fallback: use existing distance calculation
# This should integrate with existing routing infrastructure
return None  # Placeholder - integrate with routing system

2. Address custom_standard_unloading_time_minutes
3. Transport Settings default_unloading_time_minutes
4. Default 30 minutes
"""
# Check Drop Mode
if leg.get("drop_mode"):
    drop_mode_doc = frappe.get_doc("Pick and Drop Mode", leg.drop_mode)
    if hasattr(drop_mode_doc, "standard_unloading_time_minutes") and drop_mode_doc.standard_unloading_time_minutes:
        return drop_mode_doc.standard_unloading_time_minutes

# Check Address
if leg.get("drop_address"):
    address_doc = frappe.get_doc("Address", leg.drop_address)
    if hasattr(address_doc, "custom_standard_unloading_time_minutes") and address_doc.custom_standard_unloading_time_minutes:
        return address_doc.custom_standard_unloading_time_minutes

# Check Transport Settings
default_unloading = frappe.db.get_single_value(
    "Transport Settings",
    "default_unloading_time_minutes"
)
if default_unloading:
    return default_unloading

# Default
return 30

2.4 Integration with Routing Provider

Distance Calculation:

  • Primary: Use routing provider API to get distance between addresses
  • Fallback: Use Transport Leg.distance_km if already calculated
  • Cache: Store calculated distances to avoid repeated API calls

Routing Provider Integration:

  • Check if Transport Leg.routing_provider is set
  • If available, use routing provider's distance calculation
  • If not available, use existing distance calculation logic
  • Consider routing profile settings (fastest route, shortest route, etc.)

Performance Optimization:

  • Cache distance calculations for common address pairs
  • Batch distance requests when possible
  • Use historical distance data if routing provider unavailable

2.6 Validation Requirements

Vehicle Average Speed:

  • Transport Vehicle.avg_speed should be set for accurate travel time calculations
  • If not set, system will use default from Transport Settings.routing_default_avg_speed_kmh
  • If default is also not set, use 50 km/h as fallback
  • Recommendation: Validate that vehicles have avg_speed set, or prompt user to set it

Distance Calculation:

  • Priority 1: Use routing provider if Transport Leg.routing_provider is set
  • Priority 2: Use existing Transport Leg.distance_km if available
  • Priority 3: Calculate using routing service API
  • Priority 4: Use cached distance if available
  • Fallback: If all fail, skip time window validation for that leg (log warning)

Loading/Unloading Time:

  • Calculation Method: User-defined in Pick and Drop Mode (Fixed Time, Volume-Based, Weight-Based, or Combined)
  • Base Time: Minimum time required regardless of cargo size (set in Pick/Drop Mode or Address)
  • Volume/Weight Rates: User-defined rates for calculating additional time based on cargo volume (m³) and/or weight (kg)
  • Priority:
    1. Address custom settings (if override specified)
    2. Pick/Drop Mode settings
    3. Transport Settings defaults
  • Cargo Data:
    • Weight: From Transport Leg.cargo_weight_kg
    • Volume: Calculated from Transport Job Package (sum of package volumes) or from leg field if available
  • Maximum Cap: Optional maximum time limit per Pick/Drop Mode
  • Minimum: Base time (typically 15 minutes as safety buffer)
  • User Control: All rates and methods are user-defined, allowing flexibility for different operational requirements

2.7 Integration Points

  • Modify _find_candidate_vehicle() to call time window validation with full time calculations
  • Add time window checks in _find_vehicle_for_trip() for consolidated trips
  • Ensure avg_speed is fetched when querying vehicles (add to v_fields)
  • Integrate with routing provider for distance calculations
  • Add volume/weight-based loading/unloading time calculation fields to Pick and Drop Mode:
    • Base time, calculation method, rates per volume/weight, maximum cap
  • Add address-level overrides for loading/unloading time calculation (optional)
  • Add default loading/unloading time settings to Transport Settings
  • Implement calculate_loading_time() and calculate_unloading_time() functions
  • Implement calculate_leg_volume() to get volume from packages or leg field
  • Ensure cargo_weight_kg is available on Transport Leg
  • Update _vehicle_free_on_date() to consider time windows (optional enhancement)
  • Add validation warnings if vehicle doesn't have avg_speed set

2A. Address Day Availability Constraints

2A.1 Data Model

Existing Fields (already exist on Address):

  • Address.custom_pick_monday through custom_pick_sunday (Check fields)
  • Address.custom_drop_monday through custom_drop_sunday (Check fields)

These are user-defined checkboxes indicating which days of the week pick/drop operations are allowed at each address.

2A.2 Logic Implementation

Algorithm:

  1. When selecting a vehicle for a leg:

    • Get the scheduled date for the leg
    • Determine the day of week (Monday-Sunday)
    • Check the corresponding day checkbox on the pick address (e.g., custom_pick_monday)
    • Check the corresponding day checkbox on the drop address (e.g., custom_drop_monday)
    • If either address is not available on that day, skip the vehicle/leg combination

  2. Day Availability Validation:

    def check_address_day_availability(address_name, scheduled_date, operation_type):
    """
    Check if address allows pick/drop on the scheduled day.
    
    Args:
        address_name: Name of the Address document
        scheduled_date: Date object for the scheduled operation
        operation_type: "pick" or "drop"
    
    Returns:
        (is_available, reason_if_not_available)
    """
    if not address_name:
        return True, None  # No address specified, assume available
    
    # Get day of week (0=Monday, 6=Sunday)
    day_of_week = scheduled_date.weekday()
    
    # Map to day names
    day_names = ["monday", "tuesday", "wednesday", "thursday", "friday", "saturday", "sunday"]
    day_name = day_names[day_of_week]
    
    # Get address document
    try:
        address_doc = frappe.get_doc("Address", address_name)
    except frappe.DoesNotExistError:
        return True, None  # Address not found, assume available
    
    # Check the appropriate field
    field_name = f"custom_{operation_type}_{day_name}"
    
    if not address_doc.meta.has_field(field_name):
        return True, None  # Field doesn't exist, assume available
    
    is_available = getattr(address_doc, field_name, False)
    
    if not is_available:
        return False, f"Address {address_name} does not allow {operation_type} operations on {day_name.capitalize()}"
    
    return True, None

  3. Implementation Notes:

    • If checkbox is unchecked (False), the address is NOT available on that day
    • If checkbox is checked (True), the address IS available on that day
    • If field doesn't exist or is None, assume available (backward compatibility)
    • Check both pick and drop addresses for each leg

2A.3 Integration Points

  • Add address day availability check in validate_vehicle_constraints()
  • Check both pick and drop addresses for each leg
  • Provide clear error messages indicating which address and day is unavailable
  • Consider day availability when grouping legs for consolidation

3. Plate Number Coding Constraints

3.1 Data Model Extensions

New Fields on Transport Vehicle:

{
  "fieldname": "license_plate_number",
  "fieldtype": "Data",
  "label": "License Plate Number"
}

New Fields on Vehicle Type:

{
  "fieldname": "exempt_from_plate_coding",
  "fieldtype": "Check",
  "label": "Exempt from Plate Number Coding",
  "default": 0,
  "description": "If checked, vehicles of this type are exempt from plate number coding restrictions (e.g., emergency vehicles, government vehicles)"
}

New Doctype: Plate Coding Rule

{
  "doctype": "Plate Coding Rule",
  "fields": [
    {
      "fieldname": "rule_name",
      "fieldtype": "Data",
      "label": "Rule Name",
      "reqd": 1
    },
    {
      "fieldname": "coding_type",
      "fieldtype": "Select",
      "label": "Coding Type",
      "options": "Last Digit\nFirst Digit\nOdd/Even\nCustom Pattern",
      "reqd": 1
    },
    {
      "fieldname": "restricted_digits",
      "fieldtype": "Table",
      "options": "Plate Coding Restricted Digits",
      "label": "Restricted Digits"
    },
    {
      "fieldname": "restricted_days",
      "fieldtype": "Table MultiSelect",
      "options": "Weekday",
      "label": "Restricted Days"
    },
    {
      "fieldname": "time_restriction",
      "fieldtype": "Check",
      "label": "Time Restriction"
    },
    {
      "fieldname": "restricted_start_time",
      "fieldtype": "Time",
      "depends_on": "eval:doc.time_restriction"
    },
    {
      "fieldname": "restricted_end_time",
      "fieldtype": "Time",
      "depends_on": "eval:doc.time_restriction"
    },
    {
      "fieldname": "scope_level",
      "fieldtype": "Select",
      "label": "Scope Level",
      "options": "Nationwide\nRegion\nProvince\nCity / Municipality"
    },
    {
      "fieldname": "scope_location",
      "fieldtype": "Dynamic Link",
      "options": "scope_level",
      "label": "Scope Location"
    },
    {
      "fieldname": "start_date",
      "fieldtype": "Date",
      "label": "Start Date"
    },
    {
      "fieldname": "end_date",
      "fieldtype": "Date",
      "label": "End Date"
    },
    {
      "fieldname": "is_active",
      "fieldtype": "Check",
      "default": 1,
      "label": "Is Active"
    }
  ]
}

Child Table: Plate Coding Restricted Digits

{
  "doctype": "Plate Coding Restricted Digits",
  "istable": 1,
  "fields": [
    {
      "fieldname": "restricted_digit",
      "fieldtype": "Int",
      "label": "Restricted Digit",
      "reqd": 1
    },
    {
      "fieldname": "restricted_day",
      "fieldtype": "Link",
      "options": "Weekday",
      "label": "Restricted Day"
    }
  ]
}

3.2 Logic Implementation

Algorithm:

  1. Check if vehicle type is exempt from plate coding restrictions
  2. If exempt, skip all plate coding checks (return allowed)
  3. Extract last digit (or specified digit) from vehicle's license plate number
  4. Check scheduled date's day of week
  5. Query active Plate Coding Rules for the scheduled date/time and location
  6. Check if vehicle's plate digit is restricted for that day/time/location

Example Logic:

def check_plate_coding(vehicle, scheduled_datetime, address=None):
    # First, check if vehicle type is exempt from plate coding
    vehicle_type = vehicle.get("vehicle_type")
    if vehicle_type:
        try:
            vehicle_type_doc = frappe.get_doc("Vehicle Type", vehicle_type)
            if hasattr(vehicle_type_doc, "exempt_from_plate_coding") and vehicle_type_doc.exempt_from_plate_coding:
                return True, None  # Vehicle type is exempt, no restrictions apply
        except frappe.DoesNotExistError:
            pass  # Vehicle type not found, continue with normal checks
    
    plate_number = vehicle.get("license_plate_number")
    if not plate_number:
        return True, None  # No plate number, assume allowed
    
    # Extract last digit
    last_digit = int(plate_number[-1]) if plate_number[-1].isdigit() else None
    if last_digit is None:
        return True, None  # No numeric digit found
    
    # Get day of week
    day_of_week = scheduled_datetime.strftime("%A")
    scheduled_date = scheduled_datetime.date()
    scheduled_time = scheduled_datetime.time()
    
    # Build filters for active coding rules
    filters = [
        ["is_active", "=", 1],
        ["start_date", "<=", scheduled_date],
        ["end_date", ">=", scheduled_date]
    ]
    
    # Add location scope if address provided
    if address:
        # Get address location details (city, province, etc.)
        address_doc = frappe.get_doc("Address", address)
        # Add scope filters based on address location
        pass
    
    # Get applicable coding rules
    coding_rules = frappe.get_all(
        "Plate Coding Rule",
        filters=filters,
        fields=["name", "coding_type", "time_restriction", 
                "restricted_start_time", "restricted_end_time"]
    )
    
    for rule in coding_rules:
        # Check if time restriction applies
        if rule.time_restriction:
            if not (rule.restricted_start_time <= scheduled_time <= rule.restricted_end_time):
                continue  # Time restriction doesn't apply
        
        # Get restricted digits for this rule
        restricted_digits = frappe.get_all(
            "Plate Coding Restricted Digits",
            filters={"parent": rule.name},
            fields=["restricted_digit", "restricted_day"]
        )
        
        for rd in restricted_digits:
            # Check if digit matches and day matches
            if rd.restricted_digit == last_digit:
                if not rd.restricted_day or rd.restricted_day == day_of_week:
                    return False, f"Vehicle plate {plate_number} (last digit {last_digit}) is restricted on {day_of_week} by rule {rule.name}"
    
    return True, None

3.3 Integration Points

  • Add exempt_from_plate_coding field to Vehicle Type doctype
  • Check vehicle type exemption before checking plate coding rules
  • Add plate coding check in validate_vehicle_constraints()
  • Query rules based on scheduled date/time and address location
  • Cache active rules for performance
  • When fetching vehicle, also fetch vehicle_type to check exemption

4. Truck Ban Constraints

4.1 Data Model

New Doctype: Truck Ban Constraint

A standalone doctype for managing truck ban restrictions (area and time-based restrictions on vehicle movement).

{
  "doctype": "Truck Ban Constraint",
  "fields": [
    {
      "fieldname": "ban_name",
      "fieldtype": "Data",
      "label": "Ban Name",
      "reqd": 1,
      "in_list_view": 1,
      "unique": 1
    },
    {
      "fieldname": "ban_type",
      "fieldtype": "Select",
      "label": "Ban Type",
      "options": "Area Ban\nRoute Ban\nTime-Based Ban\nWeight-Based Ban\nVehicle Type Ban",
      "reqd": 1
    },
    {
      "fieldname": "start_date",
      "fieldtype": "Date",
      "label": "Start Date",
      "reqd": 1
    },
    {
      "fieldname": "end_date",
      "fieldtype": "Date",
      "label": "End Date",
      "reqd": 1
    },
    {
      "fieldname": "all_day",
      "fieldtype": "Check",
      "label": "All Day",
      "default": 1
    },
    {
      "fieldname": "start_time",
      "fieldtype": "Time",
      "label": "Start Time",
      "depends_on": "eval:!doc.all_day"
    },
    {
      "fieldname": "end_time",
      "fieldtype": "Time",
      "label": "End Time",
      "depends_on": "eval:!doc.all_day"
    },
    {
      "fieldname": "scope_level",
      "fieldtype": "Select",
      "label": "Scope Level",
      "options": "Nationwide\nRegion\nProvince\nCity / Municipality\nBarangay\nPort\nRoad / Corridor"
    },
    {
      "fieldname": "scope_location",
      "fieldtype": "Dynamic Link",
      "options": "scope_level",
      "label": "Scope Location"
    },
    {
      "fieldname": "restricted_addresses",
      "fieldtype": "Table",
      "options": "Truck Ban Restricted Addresses",
      "label": "Restricted Addresses/Areas"
    },
    {
      "fieldname": "restricted_routes",
      "fieldtype": "Table",
      "options": "Truck Ban Restricted Routes",
      "label": "Restricted Routes"
    },
    {
      "fieldname": "restricted_vehicle_types",
      "fieldtype": "Table",
      "options": "Truck Ban Constraint Vehicle Types",
      "label": "Restricted Vehicle Types",
      "description": "If specified, ban applies only to these vehicle types. If empty, applies to all vehicle types."
    },
    {
      "fieldname": "min_vehicle_weight_restriction",
      "fieldtype": "Float",
      "label": "Minimum Vehicle Weight (kg)",
      "description": "Vehicles with weight above this value are banned. Leave empty if not applicable."
    },
    {
      "fieldname": "is_active",
      "fieldtype": "Check",
      "default": 1,
      "label": "Is Active"
    },
    {
      "fieldname": "description",
      "fieldtype": "Text Editor",
      "label": "Description"
    }
  ]
}

Child Table: Truck Ban Restricted Addresses

{
  "doctype": "Truck Ban Restricted Addresses",
  "istable": 1,
  "fields": [
    {
      "fieldname": "address",
      "fieldtype": "Link",
      "options": "Address",
      "label": "Address",
      "reqd": 1
    },
    {
      "fieldname": "radius_km",
      "fieldtype": "Float",
      "label": "Radius (km)",
      "default": 0,
      "description": "0 = exact address, >0 = radius around address"
    }
  ]
}

Child Table: Truck Ban Restricted Routes

{
  "doctype": "Truck Ban Restricted Routes",
  "istable": 1,
  "fields": [
    {
      "fieldname": "from_address",
      "fieldtype": "Link",
      "options": "Address",
      "label": "From Address",
      "reqd": 1
    },
    {
      "fieldname": "to_address",
      "fieldtype": "Link",
      "options": "Address",
      "label": "To Address",
      "reqd": 1
    },
    {
      "fieldname": "route_name",
      "fieldtype": "Data",
      "label": "Route Name",
      "description": "Optional name for the route (e.g., 'Highway 1', 'Main Street')"
    }
  ]
}

Child Table: Truck Ban Constraint Vehicle Types

{
  "doctype": "Truck Ban Constraint Vehicle Types",
  "istable": 1,
  "fields": [
    {
      "fieldname": "vehicle_type",
      "fieldtype": "Link",
      "options": "Vehicle Type",
      "label": "Vehicle Type",
      "reqd": 1
    }
  ]
}

4.2 Logic Implementation

Algorithm:

  1. For scheduled date/time and addresses, query active Truck Ban constraints
  2. Check if constraint applies to:
    • The scheduled date/time (within start/end date and time range if not all_day)
    • The vehicle type (if restricted vehicle types are specified)
    • The vehicle weight/size (if min_vehicle_weight_restriction is set)
    • The address locations (pick and/or drop addresses, exact or within radius)
    • The route (if both pick and drop addresses match a restricted route)
  3. Return violation if any constraint matches
  4. If no vehicle types are specified in the ban, it applies to all vehicle types

Example Logic:

def check_truck_ban(vehicle, scheduled_datetime, pick_address=None, drop_address=None):
    """
    Check if vehicle is banned from area/route at scheduled time.
    
    Args:
        vehicle: Vehicle dictionary with vehicle_type, capacity_weight, etc.
        scheduled_datetime: Scheduled datetime for the operation
        pick_address: Pick address name (optional)
        drop_address: Drop address name (optional)
    
    Returns:
        (is_allowed, reason_if_blocked)
    """
    scheduled_date = scheduled_datetime.date()
    scheduled_time = scheduled_datetime.time()
    
    # Build filters for active truck ban constraints
    filters = [
        ["is_active", "=", 1],
        ["start_date", "<=", scheduled_date],
        ["end_date", ">=", scheduled_date]
    ]
    
    # Get active truck ban constraints
    truck_bans = frappe.get_all(
        "Truck Ban Constraint",
        filters=filters,
        fields=["name", "ban_name", "ban_type", "all_day", "start_time", "end_time",
                "scope_level", "scope_location", "min_vehicle_weight_restriction"]
    )
    
    for ban in truck_bans:
        # Check time restriction
        if not ban.all_day:
            if not (ban.start_time <= scheduled_time <= ban.end_time):
                continue
        
        # Get restricted vehicle types for this ban
        restricted_vehicle_types = frappe.get_all(
            "Truck Ban Constraint Vehicle Types",
            filters={"parent": ban.name},
            pluck="vehicle_type"
        )
        
        # Check vehicle type restriction
        if restricted_vehicle_types:
            # Ban applies only to specified vehicle types
            if vehicle.vehicle_type not in restricted_vehicle_types:
                continue  # This ban doesn't apply to this vehicle type
        # If no vehicle types specified, ban applies to all vehicle types
        
        # Check vehicle weight restriction
        if ban.min_vehicle_weight_restriction:
            vehicle_weight = vehicle.get("capacity_weight", 0)
            if vehicle_weight >= ban.min_vehicle_weight_restriction:
                return False, f"Vehicle weight {vehicle_weight}kg exceeds ban limit {ban.min_vehicle_weight_restriction}kg for constraint {ban.ban_name or ban.name}"
        
        # Check address/location restriction
        addresses_to_check = []
        if pick_address:
            addresses_to_check.append(("pick", pick_address))
        if drop_address:
            addresses_to_check.append(("drop", drop_address))
        
        for addr_type, address in addresses_to_check:
            # Check restricted addresses
            restricted_addresses = frappe.get_all(
                "Truck Ban Restricted Addresses",
                filters={"parent": ban.name},
                fields=["address", "radius_km"]
            )
            
            for ra in restricted_addresses:
                if ra.address == address:
                    return False, f"{addr_type.capitalize()} address {address} is in banned area for constraint {ban.ban_name or ban.name}"
                
                # Check radius if specified
                if ra.radius_km > 0:
                    distance = calculate_distance(address, ra.address)
                    if distance <= ra.radius_km:
                        return False, f"{addr_type.capitalize()} address {address} is within {ra.radius_km}km of banned area for constraint {ban.ban_name or ban.name}"
        
        # Check restricted routes (if both pick and drop addresses are provided)
        if pick_address and drop_address:
            restricted_routes = frappe.get_all(
                "Truck Ban Restricted Routes",
                filters={"parent": ban.name},
                fields=["from_address", "to_address", "route_name"]
            )
            
            for route in restricted_routes:
                # Check if route matches (bidirectional)
                if (route.from_address == pick_address and route.to_address == drop_address) or \
                   (route.from_address == drop_address and route.to_address == pick_address):
                    route_name = route.route_name or f"{route.from_address} to {route.to_address}"
                    return False, f"Route {route_name} is banned by constraint {ban.ban_name or ban.name}"
    
    return True, None

4.3 Integration Points

  • Create new Truck Ban Constraint doctype (standalone, not part of Scheduling Constraint)
  • Create child tables: Truck Ban Restricted Addresses and Truck Ban Restricted Routes
  • Add truck ban check in validate_vehicle_constraints()
  • Query constraints based on date/time and address
  • Support both address-based and route-based bans

4A. Ad-Hoc Factors Constraints

4A.1 Data Model

New Doctype: Ad-Hoc Transport Factor

For user-entered temporary constraints that affect transport planning (road closures, port congestion, weather events, etc.).

{
  "doctype": "Ad-Hoc Transport Factor",
  "fields": [
    {
      "fieldname": "factor_name",
      "fieldtype": "Data",
      "label": "Factor Name",
      "reqd": 1,
      "in_list_view": 1
    },
    {
      "fieldname": "factor_type",
      "fieldtype": "Select",
      "label": "Factor Type",
      "options": "Road Closure\nPort Congestion\nWeather Event\nTraffic Incident\nConstruction\nSpecial Event\nOther",
      "reqd": 1,
      "in_list_view": 1
    },
    {
      "fieldname": "severity",
      "fieldtype": "Select",
      "label": "Severity",
      "options": "Low\nMedium\nHigh\nCritical",
      "default": "Medium"
    },
    {
      "fieldname": "start_datetime",
      "fieldtype": "Datetime",
      "label": "Start Date/Time",
      "reqd": 1
    },
    {
      "fieldname": "end_datetime",
      "fieldtype": "Datetime",
      "label": "End Date/Time",
      "reqd": 1
    },
    {
      "fieldname": "affected_addresses",
      "fieldtype": "Table",
      "options": "Ad-Hoc Factor Affected Addresses",
      "label": "Affected Addresses/Areas"
    },
    {
      "fieldname": "affected_routes",
      "fieldtype": "Table",
      "options": "Ad-Hoc Factor Affected Routes",
      "label": "Affected Routes"
    },
    {
      "fieldname": "impact_description",
      "fieldtype": "Text Editor",
      "label": "Impact Description"
    },
    {
      "fieldname": "delay_estimate_minutes",
      "fieldtype": "Int",
      "label": "Estimated Delay (minutes)",
      "description": "Expected delay caused by this factor"
    },
    {
      "fieldname": "alternative_routes",
      "fieldtype": "Table",
      "options": "Ad-Hoc Factor Alternative Routes",
      "label": "Alternative Routes"
    },
    {
      "fieldname": "restricted_vehicle_types",
      "fieldtype": "Table MultiSelect",
      "options": "Vehicle Type",
      "label": "Restricted Vehicle Types",
      "description": "Vehicle types affected by this factor (leave empty for all)"
    },
    {
      "fieldname": "is_active",
      "fieldtype": "Check",
      "default": 1,
      "label": "Is Active"
    },
    {
      "fieldname": "created_by",
      "fieldtype": "Link",
      "options": "User",
      "label": "Created By",
      "read_only": 1
    },
    {
      "fieldname": "last_updated",
      "fieldtype": "Datetime",
      "label": "Last Updated",
      "read_only": 1
    }
  ]
}

Child Table: Ad-Hoc Factor Affected Addresses

{
  "doctype": "Ad-Hoc Factor Affected Addresses",
  "istable": 1,
  "fields": [
    {
      "fieldname": "address",
      "fieldtype": "Link",
      "options": "Address",
      "label": "Address",
      "reqd": 1
    },
    {
      "fieldname": "radius_km",
      "fieldtype": "Float",
      "label": "Radius (km)",
      "default": 0,
      "description": "0 = exact address, >0 = radius around address"
    },
    {
      "fieldname": "impact_type",
      "fieldtype": "Select",
      "label": "Impact Type",
      "options": "Complete Blockage\nPartial Blockage\nDelays Only\nAccess Restricted",
      "default": "Delays Only"
    }
  ]
}

Child Table: Ad-Hoc Factor Affected Routes

{
  "doctype": "Ad-Hoc Factor Affected Routes",
  "istable": 1,
  "fields": [
    {
      "fieldname": "from_address",
      "fieldtype": "Link",
      "options": "Address",
      "label": "From Address",
      "reqd": 1
    },
    {
      "fieldname": "to_address",
      "fieldtype": "Link",
      "options": "Address",
      "label": "To Address",
      "reqd": 1
    },
    {
      "fieldname": "impact_type",
      "fieldtype": "Select",
      "label": "Impact Type",
      "options": "Complete Blockage\nPartial Blockage\nDelays Only\nAccess Restricted",
      "default": "Delays Only"
    }
  ]
}

Child Table: Ad-Hoc Factor Alternative Routes

{
  "doctype": "Ad-Hoc Factor Alternative Routes",
  "istable": 1,
  "fields": [
    {
      "fieldname": "from_address",
      "fieldtype": "Link",
      "options": "Address",
      "label": "From Address",
      "reqd": 1
    },
    {
      "fieldname": "to_address",
      "fieldtype": "Link",
      "options": "Address",
      "label": "To Address",
      "reqd": 1
    },
    {
      "fieldname": "alternative_from_address",
      "fieldtype": "Link",
      "options": "Address",
      "label": "Alternative From Address",
      "reqd": 1
    },
    {
      "fieldname": "alternative_to_address",
      "fieldtype": "Link",
      "options": "Address",
      "label": "Alternative To Address",
      "reqd": 1
    },
    {
      "fieldname": "additional_time_minutes",
      "fieldtype": "Int",
      "label": "Additional Time (minutes)",
      "description": "Extra time required for alternative route"
    }
  ]
}

4A.2 Logic Implementation

Algorithm:

  1. For scheduled date/time and route, query active Ad-Hoc Factors
  2. Check if factor applies to:
    • The scheduled date/time (within start/end datetime)
    • The addresses in the route (pick, drop, or intermediate)
    • The specific route segments
    • The vehicle type (if restricted)
  3. Determine impact:
    • Complete Blockage: Route/address is completely unavailable
    • Partial Blockage: Route/address has limited access
    • Delays Only: Route/address is accessible but with delays
    • Access Restricted: Route/address has access restrictions
  4. Return violation or delay estimate

Example Logic:

def check_adhoc_factors(scheduled_datetime, pick_address=None, drop_address=None, 
                        route_addresses=None, vehicle_type=None):
    """
    Check for ad-hoc factors affecting transport.
    
    Args:
        scheduled_datetime: Scheduled datetime for the operation
        pick_address: Pick address name
        drop_address: Drop address name
        route_addresses: List of addresses in the route (optional)
        vehicle_type: Vehicle type (optional)
    
    Returns:
        (is_allowed, reason_if_blocked, delay_minutes, alternative_routes)
    """
    scheduled_date = scheduled_datetime.date()
    scheduled_time = scheduled_datetime.time()
    
    # Build filters for active factors
    filters = [
        ["is_active", "=", 1],
        ["start_datetime", "<=", scheduled_datetime],
        ["end_datetime", ">=", scheduled_datetime]
    ]
    
    # Get active ad-hoc factors
    factors = frappe.get_all(
        "Ad-Hoc Transport Factor",
        filters=filters,
        fields=["name", "factor_type", "severity", "delay_estimate_minutes",
                "impact_description", "restricted_vehicle_types"]
    )
    
    total_delay = 0
    blocking_factors = []
    warnings = []
    alternative_routes = []
    
    for factor in factors:
        # Check vehicle type restriction
        if factor.restricted_vehicle_types:
            restricted_types = [vt.vehicle_type for vt in factor.restricted_vehicle_types]
            if vehicle_type and vehicle_type in restricted_types:
                continue  # Factor doesn't apply to this vehicle type
        
        # Check affected addresses
        affected_addresses = frappe.get_all(
            "Ad-Hoc Factor Affected Addresses",
            filters={"parent": factor.name},
            fields=["address", "radius_km", "impact_type"]
        )
        
        # Check if pick or drop address is affected
        for aa in affected_addresses:
            if pick_address and (aa.address == pick_address or 
                (aa.radius_km > 0 and is_within_radius(pick_address, aa.address, aa.radius_km))):
                if aa.impact_type == "Complete Blockage":
                    blocking_factors.append(f"{factor.factor_name} - Pick address blocked")
                elif aa.impact_type == "Partial Blockage":
                    warnings.append(f"{factor.factor_name} - Pick address partially blocked")
                    total_delay += factor.delay_estimate_minutes or 0
                else:
                    total_delay += factor.delay_estimate_minutes or 0
            
            if drop_address and (aa.address == drop_address or 
                (aa.radius_km > 0 and is_within_radius(drop_address, aa.address, aa.radius_km))):
                if aa.impact_type == "Complete Blockage":
                    blocking_factors.append(f"{factor.factor_name} - Drop address blocked")
                elif aa.impact_type == "Partial Blockage":
                    warnings.append(f"{factor.factor_name} - Drop address partially blocked")
                    total_delay += factor.delay_estimate_minutes or 0
                else:
                    total_delay += factor.delay_estimate_minutes or 0
        
        # Check affected routes
        if pick_address and drop_address:
            affected_routes = frappe.get_all(
                "Ad-Hoc Factor Affected Routes",
                filters={"parent": factor.name},
                fields=["from_address", "to_address", "impact_type"]
            )
            
            for ar in affected_routes:
                if (ar.from_address == pick_address and ar.to_address == drop_address) or \
                   (ar.from_address == drop_address and ar.to_address == pick_address):
                    if ar.impact_type == "Complete Blockage":
                        blocking_factors.append(f"{factor.factor_name} - Route blocked")
                    elif ar.impact_type == "Partial Blockage":
                        warnings.append(f"{factor.factor_name} - Route partially blocked")
                        total_delay += factor.delay_estimate_minutes or 0
                    else:
                        total_delay += factor.delay_estimate_minutes or 0
                    
                    # Get alternative routes
                    alt_routes = frappe.get_all(
                        "Ad-Hoc Factor Alternative Routes",
                        filters={"parent": factor.name,
                                "from_address": pick_address,
                                "to_address": drop_address},
                        fields=["alternative_from_address", "alternative_to_address",
                               "additional_time_minutes"]
                    )
                    alternative_routes.extend(alt_routes)
    
    # If there are blocking factors, route is not allowed
    if blocking_factors:
        return False, "; ".join(blocking_factors), total_delay, alternative_routes
    
    # If there are only delays, route is allowed but with delays
    if total_delay > 0 or warnings:
        warning_msg = "; ".join(warnings) if warnings else f"Expected delay: {total_delay} minutes"
        return True, warning_msg, total_delay, alternative_routes
    
    return True, None, 0, []

4A.3 Integration Points

  • Create new Ad-Hoc Transport Factor doctype
  • Add ad-hoc factor check in validate_vehicle_constraints()
  • Consider delays when calculating time windows
  • Suggest alternative routes when available
  • Display warnings for delays without blocking

4A.4 User Interface

Ad-Hoc Transport Factor Form:

  • Quick entry form for common scenarios (road closure, port congestion, etc.)
  • Map view for selecting affected addresses/areas
  • Calendar view showing active factors
  • Impact preview (how many legs/routes affected)
  • Quick actions: Extend duration, Deactivate, Create similar

Transport Plan Integration:

  • Show ad-hoc factor warnings in allocation results
  • Display estimated delays
  • Suggest alternative routes when available
  • Allow manual override with justification

5. Integration with Vehicle Selection

5.1 Modified Vehicle Selection Flow

Current Flow:

_find_candidate_vehicle()
  ├─ Filter by vehicle_type
  ├─ Filter by capacity
  ├─ Check _vehicle_free_on_date()
  └─ Return first available vehicle

New Flow:

_find_candidate_vehicle()
  ├─ Filter by vehicle_type
  ├─ Filter by capacity
  ├─ Check _vehicle_free_on_date()
  ├─ For each candidate vehicle:
  │   ├─ validate_vehicle_constraints()
  │   │   ├─ check_time_window_constraints()
  │   │   ├─ check_address_day_availability() (pick address)
  │   │   ├─ check_address_day_availability() (drop address)
  │   │   ├─ check_plate_coding_constraints()
  │   │   ├─ check_truck_ban_constraints() (with pick and drop addresses)
  │   │   └─ check_adhoc_factors()
  │   └─ If valid, add to available list
  └─ Return first available vehicle (or best match)

5.2 Code Changes

File: logistics/transport/doctype/transport_plan/transport_plan.py

Function: _find_candidate_vehicle()

Changes:

def _find_candidate_vehicle(leg: Dict[str, Any], debug: Optional[List[str]] = None, 
                            vehicle_to_runsheet: Optional[Dict[str, str]] = None, 
                            target_runsheet: Optional[str] = None) -> Optional[Dict[str, Any]]:
    # ... existing code ...
    
    # IMPORTANT: Ensure avg_speed and vehicle_type are included in vehicle fields query
    # vehicle_type is needed for plate coding exemption checks
    v_fields = ["name", "vehicle_name", "company_owned", "avg_speed", "vehicle_type"]
    for vf in ["transport_company", "capacity_weight", "capacity_volume", "capacity_pallets"]:
        if _has_field("Transport Vehicle", vf):
            v_fields.append(vf)
    
    # ... existing vehicle query code ...
    
    for v in available_vehicles:
        if not _vehicle_free_on_date(v["name"], sched):
            debug.append(f"Internal vehicle busy on {sched}: {v.get('vehicle_name') or v['name']}")
            continue

        # NEW: Check constraints
        from logistics.transport.constraint_validator import validate_vehicle_constraints
        
        # Get scheduled datetime (use estimated_dispatch_datetime or run_date)
        scheduled_datetime = _get_scheduled_datetime(leg)
        
        is_valid, constraint_reason, delay_minutes, alternatives = validate_vehicle_constraints(
            v, leg, scheduled_datetime, debug
        )
        
        if not is_valid:
            debug.append(f"Vehicle {v.get('vehicle_name') or v['name']} failed constraint: {constraint_reason}")
            if alternatives:
                debug.append(f"Alternative routes available: {len(alternatives)}")
            continue
        
        # If there are delays but route is still valid, log warning
        if delay_minutes and delay_minutes > 0:
            debug.append(f"Vehicle {v.get('vehicle_name') or v['name']} will experience {delay_minutes} minutes delay due to ad-hoc factors")

        # ... existing capacity checks ...
        
        if ok:
            debug.append(f"Internal vehicle selected: {v.get('vehicle_name') or v['name']}")
            return v

6. Performance Considerations

6.1 Caching Strategy

  1. Active Constraints Cache:

    • Cache active Plate Coding Rules and Truck Ban Constraints
    • Refresh cache daily or on constraint updates
    • Key: (date, location_scope)

  2. Distance and Travel Time Cache:

    • Cache calculated distances between common address pairs
    • Cache calculated travel times (distance + avg_speed combinations)
    • Use routing service or historical data
    • TTL: 1 hour (configurable)
    • Key: (from_address, to_address, avg_speed)

  3. Query Optimization:

    • Use indexed fields for constraint queries
    • Batch constraint checks for multiple vehicles
    • Pre-filter constraints by date range before detailed checks

6.2 Lazy Evaluation

  • Only check constraints when necessary (e.g., skip time windows if not set)
  • Early exit on first constraint violation
  • Parallel constraint checking for multiple vehicles (if performance critical)

7. User Interface Enhancements

7.1 Constraint Display

Run Sheet Form:

  • Show constraint violations/warnings when assigning vehicle
  • Display time window requirements for each leg
  • Show plate coding restrictions for selected date
  • Display exemption status if vehicle type is exempt from plate coding

Transport Plan Form:

  • Show summary of constraint violations in allocation results
  • Highlight legs that couldn't be assigned due to constraints
  • Provide suggestions for alternative vehicles/dates

7.2 Constraint Management

Plate Coding Rule Form:

  • Calendar view showing restricted days
  • Preview of affected vehicles (excluding exempt vehicle types)
  • Test constraint against specific vehicle/date
  • Show list of exempt vehicle types

Vehicle Type Form:

  • Display exemption status prominently
  • Show which plate coding rules would apply if not exempt
  • Help text explaining exemption use cases (emergency vehicles, government vehicles, etc.)

Truck Ban Constraint Form:

  • Map view for banned areas (if coordinates available)
  • Preview of affected addresses and routes
  • Impact analysis (how many vehicles/legs affected)
  • Calendar view showing active ban periods
  • Quick actions: Extend duration, Deactivate, Create similar

8. Error Handling and Logging

8.1 Constraint Violation Messages

  • Clear, actionable error messages
  • Include: constraint type, reason, affected vehicle/leg, suggested alternatives

8.2 Debug Information

  • Log all constraint checks in debug mode
  • Track constraint violation statistics
  • Performance metrics for constraint checking

9. Testing Strategy

9.1 Unit Tests

  • Test each constraint type independently
  • Test edge cases (missing data, invalid dates, etc.)
  • Test constraint combinations

9.2 Integration Tests

  • Test vehicle selection with all constraints enabled
  • Test with real-world scenarios
  • Performance tests with large datasets

9.3 User Acceptance Tests

  • Test with actual users and real constraints
  • Validate constraint rules match business requirements
  • Test UI feedback and error messages

10. Migration and Rollout Plan

10.1 Phase 1: Foundation (Week 1-2)

  • Create constraint validator module
  • Add license_plate_number field to Transport Vehicle
  • Add exempt_from_plate_coding field to Vehicle Type
  • Create Plate Coding Rule doctype
  • Create Truck Ban Constraint doctype (standalone)

10.2 Phase 2: Time Windows (Week 3-4)

  • Add loading/unloading time fields to Pick and Drop Mode (or Address)
  • Add default loading/unloading time settings to Transport Settings
  • Implement travel time calculation using vehicle avg_speed
  • Integrate with routing provider for distance calculation
  • Implement time window constraint checking with full time calculations
  • Integrate into vehicle selection
  • Testing and refinement

10.3 Phase 3: Plate Coding (Week 5-6)

  • Implement plate coding logic
  • Create UI for managing coding rules
  • Integrate into vehicle selection
  • Testing and refinement

10.4 Phase 4: Truck Bans (Week 7-8)

  • Create Truck Ban Constraint doctype and child tables
  • Implement location-based and route-based checking
  • Integrate into vehicle selection
  • Testing and refinement

10.5 Phase 5: Address Day Availability (Week 9-10)

  • Implement address day availability checking
  • Integrate into vehicle selection
  • Add UI indicators for day availability
  • Testing and refinement

10.6 Phase 6: Ad-Hoc Factors (Week 11-12)

  • Create Ad-Hoc Transport Factor doctype
  • Implement ad-hoc factor checking logic
  • Add UI for managing ad-hoc factors
  • Integrate into vehicle selection with delay calculations
  • Alternative route suggestions
  • Testing and refinement

10.7 Phase 7: Polish and Optimization (Week 13-14)

  • Performance optimization
  • UI improvements
  • Documentation
  • User training

11. Configuration and Settings

11.1 Transport Settings

Add new settings:

  • enable_time_window_constraints (Check)
  • enable_address_day_availability (Check)
  • enable_plate_coding_constraints (Check)
  • enable_truck_ban_constraints (Check)
  • enable_adhoc_factors (Check)
  • constraint_checking_mode (Select: "Strict" / "Warning" / "Disabled")
  • routing_default_avg_speed_kmh (Float, default: 50) - Default average speed if vehicle doesn't have avg_speed
  • default_loading_time_minutes (Int, default: 30) - Default loading time if not specified in Pick Mode or Address
  • default_unloading_time_minutes (Int, default: 30) - Default unloading time if not specified in Drop Mode or Address
  • use_routing_service_for_distance (Check) - Use routing provider for distance calculation
  • cache_route_distances (Check) - Cache calculated distances for performance
  • distance_cache_ttl_hours (Int, default: 24) - Time to live for distance cache
  • adhoc_factor_delay_threshold_minutes (Int, default: 60) - Maximum delay before blocking route

12. Future Enhancements

  1. Machine Learning: Learn from historical data to predict travel times
  2. Dynamic Constraints: Real-time constraint updates (e.g., traffic, weather)
  3. Multi-Objective Optimization: Balance constraints with cost, distance, etc.
  4. Constraint Conflict Resolution: Automatic suggestions when constraints conflict
  5. Constraint Templates: Pre-defined constraint sets for common scenarios

Approval Checklist

  • Architecture and approach approved
  • Data model changes reviewed
  • Integration points confirmed
  • Performance considerations acceptable
  • UI/UX approach approved
  • Testing strategy acceptable
  • Migration plan feasible
  • Timeline and resources allocated

Questions for Discussion

  1. Should time window violations be hard errors or warnings?
  2. How should we handle travel time calculations? Use routing service or estimates?
  3. Should plate coding rules be location-specific or global?
  4. How granular should truck ban areas be? (exact addresses vs. zones)
  5. Should constraints be configurable per Transport Plan or global?
  6. What is the priority order when multiple constraints conflict?
  7. How should ad-hoc factor delays be handled? Should they block routes or just add delays?
  8. Should address day availability be strict (block if unavailable) or flexible (warn only)?
  9. How should alternative routes from ad-hoc factors be presented to users?
  10. Should ad-hoc factors automatically expire or require manual deactivation?
  11. Where should loading/unloading times be stored? Pick and Drop Mode, Address, or both?
  12. What should happen if vehicle doesn't have avg_speed set? Use default or skip vehicle?
  13. How should we handle distance calculation failures? Fallback to estimates or block assignment?
  14. Should loading/unloading times vary by cargo type/weight? (Future enhancement)

Document Version: 1.0
Date: 2026-01-08
Author: AI Assistant
Status: Draft - Awaiting Approval

Getting Started

Setup and Settings

Sea Freight

Air Freight

Transport

Customs

Warehousing

Pricing Center

Job Management

Sustainability

Intercompany

Special Projects

Pages

Features

Reports

Glossary

Clone this wiki locally