Time Slot Optimization

Optimize time slot allocation for dock scheduling and logistics operations.

Installation

pip install numpy ortools

Quick Start

from datetime import datetime, timedelta
from typing import List, Dict, Tuple

def create_time_slots(start: datetime, end: datetime,
                      slot_duration: int) -> List[Tuple[datetime, datetime]]:
    """
    Create time slots for a scheduling period.
    slot_duration: duration in minutes
    """
    slots = []
    current = start

    while current + timedelta(minutes=slot_duration) <= end:
        slot_end = current + timedelta(minutes=slot_duration)
        slots.append((current, slot_end))
        current = slot_end

    return slots

def assign_to_slots(requests: List[dict], slots: List[Tuple[datetime, datetime]],
                   max_per_slot: int) -> dict:
    """Simple slot assignment with capacity limits."""
    slot_assignments = {i: [] for i in range(len(slots))}
    assignments = {}

    for request in sorted(requests, key=lambda r: r.get('priority', 0), reverse=True):
        preferred = request.get('preferred_time')

        for i, (start, end) in enumerate(slots):
            if len(slot_assignments[i]) < max_per_slot:
                if preferred is None or start <= preferred < end:
                    slot_assignments[i].append(request['id'])
                    assignments[request['id']] = {
                        'slot': i,
                        'time': start
                    }
                    break

    return assignments

Common Patterns

Appointment Scheduling

class AppointmentScheduler:
    """Manage appointment scheduling with time slots."""

    def __init__(self, operating_hours: Tuple[int, int],
                 slot_duration: int, max_concurrent: int):
        self.operating_hours = operating_hours  # (start_hour, end_hour)
        self.slot_duration = slot_duration  # minutes
        self.max_concurrent = max_concurrent
        self.appointments = {}  # date -> {slot_index: [appointments]}

    def get_available_slots(self, date: datetime.date) -> List[dict]:
        """Get available time slots for a date."""
        slots = []
        current = datetime.combine(date, datetime.min.time().replace(
            hour=self.operating_hours[0]
        ))
        end = datetime.combine(date, datetime.min.time().replace(
            hour=self.operating_hours[1]
        ))

        slot_idx = 0
        while current < end:
            slot_end = current + timedelta(minutes=self.slot_duration)
            booked = len(self._get_slot_appointments(date, slot_idx))

            slots.append({
                'index': slot_idx,
                'start': current,
                'end': slot_end,
                'available': self.max_concurrent - booked,
                'is_available': booked < self.max_concurrent
            })

            current = slot_end
            slot_idx += 1

        return slots

    def book_appointment(self, date: datetime.date, slot_index: int,
                        appointment: dict) -> bool:
        """Book an appointment in a specific slot."""
        if date not in self.appointments:
            self.appointments[date] = {}

        current = self._get_slot_appointments(date, slot_index)
        if len(current) >= self.max_concurrent:
            return False

        if slot_index not in self.appointments[date]:
            self.appointments[date][slot_index] = []

        self.appointments[date][slot_index].append(appointment)
        return True

    def _get_slot_appointments(self, date: datetime.date, slot_index: int) -> List:
        """Get appointments for a specific slot."""
        if date not in self.appointments:
            return []
        return self.appointments[date].get(slot_index, [])

    def find_best_slot(self, date: datetime.date,
                       preferred_time: datetime = None) -> int:
        """Find the best available slot, optionally near preferred time."""
        available = self.get_available_slots(date)
        available_slots = [s for s in available if s['is_available']]

        if not available_slots:
            return None

        if preferred_time:
            # Find closest to preferred time
            return min(
                available_slots,
                key=lambda s: abs((s['start'] - preferred_time).total_seconds())
            )['index']

        # Return slot with most availability
        return max(available_slots, key=lambda s: s['available'])['index']

Workload Balancing

def balance_workload_across_slots(requests: List[dict],
                                  slots: List[Tuple[datetime, datetime]],
                                  resource_capacity: Dict[int, int]) -> dict:
    """
    Distribute requests across slots to balance workload.
    resource_capacity: {slot_index: max_capacity}
    """
    from ortools.sat.python import cp_model

    model = cp_model.CpModel()
    n_requests = len(requests)
    n_slots = len(slots)

    # Decision variables: request -> slot assignment
    assignment = {}
    for r in range(n_requests):
        for s in range(n_slots):
            assignment[(r, s)] = model.NewBoolVar(f'r{r}_s{s}')

    # Each request assigned to exactly one slot
    for r in range(n_requests):
        model.Add(sum(assignment[(r, s)] for s in range(n_slots)) == 1)

    # Slot capacity constraints
    for s in range(n_slots):
        capacity = resource_capacity.get(s, float('inf'))
        if capacity < float('inf'):
            model.Add(sum(assignment[(r, s)] for r in range(n_requests)) <= capacity)

    # Minimize workload imbalance
    slot_loads = []
    for s in range(n_slots):
        load = model.NewIntVar(0, n_requests, f'load_{s}')
        model.Add(load == sum(assignment[(r, s)] for r in range(n_requests)))
        slot_loads.append(load)

    max_load = model.NewIntVar(0, n_requests, 'max_load')
    min_load = model.NewIntVar(0, n_requests, 'min_load')
    model.AddMaxEquality(max_load, slot_loads)
    model.AddMinEquality(min_load, slot_loads)

    model.Minimize(max_load - min_load)

    solver = cp_model.CpSolver()
    status = solver.Solve(model)

    if status in [cp_model.OPTIMAL, cp_model.FEASIBLE]:
        result = {}
        for r in range(n_requests):
            for s in range(n_slots):
                if solver.Value(assignment[(r, s)]):
                    result[requests[r]['id']] = {
                        'slot': s,
                        'time_window': slots[s]
                    }
        return result
    return None

Conflict Resolution

def resolve_time_conflicts(requests: List[dict],
                          conflict_threshold: int = 0) -> List[List[dict]]:
    """
    Identify and group conflicting time requests.
    conflict_threshold: minimum minutes of overlap to be a conflict
    """
    conflicts = []

    for i, r1 in enumerate(requests):
        for j, r2 in enumerate(requests[i+1:], i+1):
            overlap = calculate_overlap(
                r1['start'], r1['end'],
                r2['start'], r2['end']
            )
            if overlap > conflict_threshold:
                conflicts.append({
                    'requests': [r1['id'], r2['id']],
                    'overlap_minutes': overlap
                })

    # Group conflicting requests
    from collections import defaultdict
    conflict_graph = defaultdict(set)
    for c in conflicts:
        r1, r2 = c['requests']
        conflict_graph[r1].add(r2)
        conflict_graph[r2].add(r1)

    # Find connected components (conflict groups)
    visited = set()
    groups = []

    for request_id in conflict_graph:
        if request_id not in visited:
            group = []
            stack = [request_id]
            while stack:
                current = stack.pop()
                if current not in visited:
                    visited.add(current)
                    group.append(current)
                    stack.extend(conflict_graph[current] - visited)
            groups.append(group)

    return groups


def calculate_overlap(start1: datetime, end1: datetime,
                      start2: datetime, end2: datetime) -> int:
    """Calculate overlap in minutes between two time ranges."""
    overlap_start = max(start1, start2)
    overlap_end = min(end1, end2)

    if overlap_end > overlap_start:
        return int((overlap_end - overlap_start).total_seconds() / 60)
    return 0

Slot Utilization Analysis

def analyze_slot_utilization(assignments: dict, slots: List[Tuple[datetime, datetime]],
                            capacity_per_slot: int) -> dict:
    """Analyze time slot utilization."""
    slot_counts = {i: 0 for i in range(len(slots))}

    for assignment in assignments.values():
        slot_idx = assignment.get('slot')
        if slot_idx is not None:
            slot_counts[slot_idx] += 1

    utilization = {}
    for i, (start, end) in enumerate(slots):
        count = slot_counts[i]
        utilization[i] = {
            'time_window': (start, end),
            'count': count,
            'utilization': count / capacity_per_slot if capacity_per_slot > 0 else 0,
            'status': 'full' if count >= capacity_per_slot else
                     'high' if count >= capacity_per_slot * 0.8 else
                     'normal' if count >= capacity_per_slot * 0.3 else 'low'
        }

    return {
        'slot_utilization': utilization,
        'total_assigned': sum(slot_counts.values()),
        'avg_utilization': sum(u['utilization'] for u in utilization.values()) / len(utilization),
        'peak_slot': max(utilization.items(), key=lambda x: x[1]['count'])[0],
        'empty_slots': sum(1 for u in utilization.values() if u['count'] == 0)
    }

Dynamic Time Slot Adjustment

def adjust_slots_for_demand(current_slots: List[Tuple[datetime, datetime]],
                           demand_forecast: Dict[int, int],
                           min_slot_duration: int = 15,
                           max_slot_duration: int = 120) -> List[Tuple[datetime, datetime]]:
    """
    Adjust time slot durations based on demand forecast.
    High demand -> shorter slots, Low demand -> longer slots
    """
    adjusted_slots = []
    avg_demand = sum(demand_forecast.values()) / len(demand_forecast) if demand_forecast else 1

    for i, (start, end) in enumerate(current_slots):
        current_duration = (end - start).total_seconds() / 60
        demand = demand_forecast.get(i, avg_demand)

        # Adjust duration inversely with demand
        if demand > avg_demand * 1.5:
            new_duration = max(min_slot_duration, current_duration * 0.75)
        elif demand < avg_demand * 0.5:
            new_duration = min(max_slot_duration, current_duration * 1.5)
        else:
            new_duration = current_duration

        adjusted_slots.append((start, start + timedelta(minutes=new_duration)))

    return adjusted_slots