Zone Picking

Implement zone-based picking strategies for efficient warehouse operations.

Installation

pip install numpy pandas

Quick Start

# Define warehouse zones
zones = {
    'A': {'aisles': [1, 2, 3], 'type': 'fast-moving'},
    'B': {'aisles': [4, 5, 6], 'type': 'medium'},
    'C': {'aisles': [7, 8, 9, 10], 'type': 'slow-moving'}
}

# Assign order items to zones
order = [
    {'sku': 'A100', 'aisle': 2},  # Zone A
    {'sku': 'B200', 'aisle': 5},  # Zone B
    {'sku': 'C300', 'aisle': 8}   # Zone C
]

Zone Configuration

class WarehouseZone:
    def __init__(self, zone_id, aisle_start, aisle_end, zone_type='standard'):
        self.zone_id = zone_id
        self.aisle_start = aisle_start
        self.aisle_end = aisle_end
        self.zone_type = zone_type
        self.pickers = []
        self.staging_location = None

    def contains_location(self, aisle):
        """Check if aisle is in this zone."""
        return self.aisle_start <= aisle <= self.aisle_end

    def assign_picker(self, picker_id):
        """Assign a picker to this zone."""
        self.pickers.append(picker_id)

    def set_staging(self, location):
        """Set consolidation/staging area for zone."""
        self.staging_location = location


class ZonedWarehouse:
    def __init__(self):
        self.zones = {}
        self.conveyor_stations = {}

    def add_zone(self, zone):
        """Add a zone to the warehouse."""
        self.zones[zone.zone_id] = zone

    def get_zone_for_location(self, aisle):
        """Find which zone contains a given aisle."""
        for zone_id, zone in self.zones.items():
            if zone.contains_location(aisle):
                return zone
        return None

    def split_order_by_zone(self, order_items):
        """Split order items by zone."""
        zone_items = {}
        for item in order_items:
            zone = self.get_zone_for_location(item['aisle'])
            if zone:
                if zone.zone_id not in zone_items:
                    zone_items[zone.zone_id] = []
                zone_items[zone.zone_id].append(item)
        return zone_items

Zone Picking Strategies

Progressive Zone Picking

def progressive_zone_picking(order, zones, zone_sequence):
    """
    Order passes through zones sequentially.
    Each picker adds items and passes to next zone.
    """
    zone_picks = {z: [] for z in zone_sequence}

    # Assign items to zones
    for item in order['items']:
        zone = get_zone_for_aisle(item['aisle'], zones)
        if zone in zone_picks:
            zone_picks[zone].append(item)

    pick_sequence = []
    for zone in zone_sequence:
        if zone_picks[zone]:
            pick_sequence.append({
                'zone': zone,
                'items': zone_picks[zone],
                'action': 'pick_and_pass'
            })

    return pick_sequence

def get_zone_for_aisle(aisle, zones):
    """Find zone containing the aisle."""
    for zone_id, zone_config in zones.items():
        if aisle in zone_config['aisles']:
            return zone_id
    return None

Parallel Zone Picking

def parallel_zone_picking(order, zones):
    """
    All zones pick simultaneously, consolidate at end.
    Faster but requires consolidation step.
    """
    zone_assignments = {}

    for item in order['items']:
        zone = get_zone_for_aisle(item['aisle'], zones)
        if zone not in zone_assignments:
            zone_assignments[zone] = {
                'items': [],
                'picker': zones[zone].get('assigned_picker')
            }
        zone_assignments[zone]['items'].append(item)

    # All zones can start simultaneously
    parallel_tasks = []
    for zone_id, assignment in zone_assignments.items():
        parallel_tasks.append({
            'zone': zone_id,
            'picker': assignment['picker'],
            'items': assignment['items'],
            'status': 'ready'
        })

    return {
        'parallel_tasks': parallel_tasks,
        'consolidation_required': len(parallel_tasks) > 1
    }

Dynamic Zone Assignment

def dynamic_zone_assignment(orders, zones, picker_capacity):
    """
    Dynamically assign orders to zones based on workload.
    Balance work across zones.
    """
    zone_workload = {z: 0 for z in zones}
    zone_orders = {z: [] for z in zones}

    for order in orders:
        # Find zone with most items for this order
        zone_item_counts = {}
        for item in order['items']:
            zone = get_zone_for_aisle(item['aisle'], zones)
            zone_item_counts[zone] = zone_item_counts.get(zone, 0) + 1

        # Assign to primary zone
        primary_zone = max(zone_item_counts.keys(),
                          key=lambda z: zone_item_counts[z])

        zone_orders[primary_zone].append(order)
        zone_workload[primary_zone] += len(order['items'])

    return zone_orders, zone_workload

Zone Hand-off Management

class ZoneHandoffManager:
    def __init__(self, zones):
        self.zones = zones
        self.handoff_points = {}
        self.pending_handoffs = []

    def setup_handoff_point(self, from_zone, to_zone, location):
        """Define hand-off location between zones."""
        self.handoff_points[(from_zone, to_zone)] = location

    def create_handoff(self, order_id, from_zone, to_zone, items):
        """Create a pending hand-off."""
        handoff = {
            'order_id': order_id,
            'from_zone': from_zone,
            'to_zone': to_zone,
            'items': items,
            'status': 'pending',
            'location': self.handoff_points.get((from_zone, to_zone))
        }
        self.pending_handoffs.append(handoff)
        return handoff

    def complete_handoff(self, order_id, from_zone):
        """Mark hand-off as complete."""
        for handoff in self.pending_handoffs:
            if (handoff['order_id'] == order_id and
                handoff['from_zone'] == from_zone):
                handoff['status'] = 'complete'
                return True
        return False

    def get_pending_for_zone(self, zone_id):
        """Get all pending hand-offs for a zone."""
        return [h for h in self.pending_handoffs
                if h['to_zone'] == zone_id and h['status'] == 'pending']

Zone Balancing

def analyze_zone_balance(zones, historical_picks):
    """Analyze pick distribution across zones."""
    zone_stats = {}

    for zone_id in zones:
        zone_picks = [p for p in historical_picks
                     if p['zone'] == zone_id]

        zone_stats[zone_id] = {
            'total_picks': len(zone_picks),
            'avg_picks_per_hour': len(zone_picks) / 24,  # Assuming daily data
            'unique_skus': len(set(p['sku'] for p in zone_picks)),
            'avg_travel_time': np.mean([p['travel_time'] for p in zone_picks])
        }

    # Calculate imbalance
    pick_counts = [s['total_picks'] for s in zone_stats.values()]
    imbalance_ratio = max(pick_counts) / min(pick_counts) if min(pick_counts) > 0 else float('inf')

    return zone_stats, imbalance_ratio

def recommend_zone_adjustments(zone_stats, target_balance=1.2):
    """Recommend zone boundary adjustments for better balance."""
    avg_picks = np.mean([s['total_picks'] for s in zone_stats.values()])

    recommendations = []
    for zone_id, stats in zone_stats.items():
        ratio = stats['total_picks'] / avg_picks

        if ratio > target_balance:
            recommendations.append({
                'zone': zone_id,
                'action': 'reduce_size',
                'reason': f'Zone has {ratio:.1%} of average workload'
            })
        elif ratio < 1 / target_balance:
            recommendations.append({
                'zone': zone_id,
                'action': 'expand',
                'reason': f'Zone has only {ratio:.1%} of average workload'
            })

    return recommendations

Zone Performance Metrics

def calculate_zone_metrics(zone_id, picks, time_period):
    """Calculate performance metrics for a zone."""
    zone_picks = [p for p in picks if p['zone'] == zone_id]

    if not zone_picks:
        return None

    return {
        'zone_id': zone_id,
        'total_picks': len(zone_picks),
        'picks_per_hour': len(zone_picks) / time_period,
        'avg_pick_time': np.mean([p['pick_time'] for p in zone_picks]),
        'travel_time_ratio': (
            sum(p['travel_time'] for p in zone_picks) /
            sum(p['total_time'] for p in zone_picks)
        ),
        'unique_orders': len(set(p['order_id'] for p in zone_picks)),
        'items_per_order': len(zone_picks) / len(set(p['order_id'] for p in zone_picks))
    }