kishorkukreja/awesome-supply-chain/assembly-line-balancing

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from collections import defaultdict
class LineBalancingMetrics:
    """
    Calculate assembly line balancing metrics
    """
    def __init__(self, workstation_times, cycle_time):
        """
        Parameters:
        - workstation_times: list of total times at each workstation
        - cycle_time: target cycle time (takt time)
        """
        self.workstation_times = np.array(workstation_times)
        self.cycle_time = cycle_time
        self.num_workstations = len(workstation_times)
    def calculate_metrics(self):
        """
        Calculate comprehensive line balancing metrics
        """
        # Total task time
        total_time = self.workstation_times.sum()
        # Theoretical minimum workstations
        min_workstations = np.ceil(total_time / self.cycle_time)
        # Line efficiency (balance efficiency)
        line_efficiency = (total_time / (self.num_workstations * self.cycle_time)) * 100
        # Balance delay (idle time %)
        balance_delay = 100 - line_efficiency
        # Smoothness index (variability in workstation times)
        # Lower is better
        max_time = self.workstation_times.max()
        smoothness_index = np.sqrt(
            np.sum((max_time - self.workstation_times) ** 2)
        )
        # Idle time at each workstation
        idle_times = self.cycle_time - self.workstation_times
        # Bottleneck identification
        bottleneck_station = np.argmax(self.workstation_times)
        bottleneck_time = self.workstation_times[bottleneck_station]
        return {
            'total_task_time': total_time,
            'cycle_time': self.cycle_time,
            'num_workstations': self.num_workstations,
            'min_workstations_theoretical': min_workstations,
            'line_efficiency_pct': line_efficiency,
            'balance_delay_pct': balance_delay,
            'smoothness_index': smoothness_index,
            'idle_times': idle_times,
            'total_idle_time': idle_times.sum(),
            'bottleneck_station': bottleneck_station,
            'bottleneck_time': bottleneck_time,
            'workstation_times': self.workstation_times
        }
    def calculate_takt_time(self, demand_per_day, available_time_minutes):
        """
        Calculate takt time = available time / customer demand
        Parameters:
        - demand_per_day: required production volume
        - available_time_minutes: working time available per day
        Returns takt time in minutes
        """
        takt_time = available_time_minutes / demand_per_day
        return {
            'demand_per_day': demand_per_day,
            'available_time_minutes': available_time_minutes,
            'takt_time_minutes': takt_time,
            'takt_time_seconds': takt_time * 60,
            'max_units_per_day': available_time_minutes / takt_time
        }
    def plot_balance_chart(self, metrics):
        """
        Visualize line balance with bar chart
        """
        fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(14, 5))
        # Workstation times vs cycle time
        stations = [f'WS{i+1}' for i in range(self.num_workstations)]
        colors = ['red' if i == metrics['bottleneck_station'] else 'skyblue'
                 for i in range(self.num_workstations)]
        bars = ax1.bar(stations, self.workstation_times, color=colors,
                      edgecolor='black', linewidth=1.5, alpha=0.7)
        # Cycle time line
        ax1.axhline(self.cycle_time, color='green', linestyle='--',
                   linewidth=2, label=f'Cycle Time ({self.cycle_time:.1f} min)')
        # Add value labels
        for i, (bar, time) in enumerate(zip(bars, self.workstation_times)):
            height = bar.get_height()
            ax1.text(bar.get_x() + bar.get_width()/2., height,
                    f'{time:.1f}', ha='center', va='bottom', fontweight='bold')
            # Add idle time annotation
            idle = self.cycle_time - time
            if idle > 0:
                ax1.text(bar.get_x() + bar.get_width()/2., height + 0.1,
                        f'(idle: {idle:.1f})', ha='center', va='bottom',
                        fontsize=9, color='red')
        ax1.set_xlabel('Workstation', fontsize=12, fontweight='bold')
        ax1.set_ylabel('Time (minutes)', fontsize=12, fontweight='bold')
        ax1.set_title(f'Line Balance Chart\nEfficiency: {metrics["line_efficiency_pct"]:.1f}% (Red = Bottleneck)',
                     fontsize=13, fontweight='bold')
        ax1.legend()
        ax1.grid(True, alpha=0.3, axis='y')
        # Idle time distribution
        idle_times = metrics['idle_times']
        ax2.bar(stations, idle_times, color='lightcoral', edgecolor='black',
               linewidth=1.5, alpha=0.7)
        ax2.set_xlabel('Workstation', fontsize=12, fontweight='bold')
        ax2.set_ylabel('Idle Time (minutes)', fontsize=12, fontweight='bold')
        ax2.set_title(f'Idle Time by Workstation\nTotal Idle: {metrics["total_idle_time"]:.1f} min',
                     fontsize=13, fontweight='bold')
        ax2.grid(True, alpha=0.3, axis='y')
        plt.tight_layout()
        return fig
# Example usage
workstation_times = [5.2, 6.8, 5.5, 6.9, 5.1, 6.5]  # minutes
cycle_time = 7.0  # target cycle time
metrics_calc = LineBalancingMetrics(workstation_times, cycle_time)
# Calculate metrics
metrics = metrics_calc.calculate_metrics()
print("Line Balancing Metrics:")
print(f"  Number of Workstations: {metrics['num_workstations']}")
print(f"  Theoretical Minimum: {metrics['min_workstations_theoretical']:.0f}")
print(f"  Cycle Time: {metrics['cycle_time']:.1f} minutes")
print(f"  Total Task Time: {metrics['total_task_time']:.1f} minutes")
print(f"  Line Efficiency: {metrics['line_efficiency_pct']:.1f}%")
print(f"  Balance Delay: {metrics['balance_delay_pct']:.1f}%")
print(f"  Smoothness Index: {metrics['smoothness_index']:.2f}")
print(f"  Total Idle Time: {metrics['total_idle_time']:.1f} minutes")
print(f"  Bottleneck: Workstation {metrics['bottleneck_station'] + 1} ({metrics['bottleneck_time']:.1f} min)")
# Takt time calculation
takt = metrics_calc.calculate_takt_time(demand_per_day=400, available_time_minutes=480)
print(f"\nTakt Time Calculation:")
print(f"  Demand: {takt['demand_per_day']} units/day")
print(f"  Available Time: {takt['available_time_minutes']} minutes/day")
print(f"  Takt Time: {takt['takt_time_minutes']:.2f} minutes ({takt['takt_time_seconds']:.0f} seconds)")
# Plot
fig = metrics_calc.plot_balance_chart(metrics)
plt.show()

class LineBalancer:
    """
    Assembly line balancing using heuristic algorithms
    """
    def __init__(self, tasks, precedence_graph, cycle_time):
        """
        Parameters:
        - tasks: dict {task_id: processing_time}
        - precedence_graph: dict {task_id: [list of immediate predecessors]}
        - cycle_time: target cycle time (takt time)
        Example:
        tasks = {'A': 3.0, 'B': 2.5, 'C': 4.0, ...}
        precedence_graph = {'A': [], 'B': ['A'], 'C': ['A'], ...}
        """
        self.tasks = tasks
        self.precedence = precedence_graph
        self.cycle_time = cycle_time
    def calculate_positional_weights(self):
        """
        Calculate positional weight for each task
        Positional weight = task time + sum of all following tasks' times
        """
        # Build successor relationships
        successors = defaultdict(list)
        for task, predecessors in self.precedence.items():
            for pred in predecessors:
                successors[pred].append(task)
        # Calculate positional weights (recursive)
        positional_weights = {}
        def calc_weight(task):
            if task in positional_weights:
                return positional_weights[task]
            # Weight = own time + sum of all successors' weights
            weight = self.tasks[task]
            for successor in successors[task]:
                weight += calc_weight(successor)
            positional_weights[task] = weight
            return weight
        # Calculate for all tasks
        for task in self.tasks:
            calc_weight(task)
        return positional_weights
    def rpw_method(self):
        """
        Ranked Positional Weight method for line balancing
        Returns workstation assignments
        """
        # Calculate positional weights
        weights = self.calculate_positional_weights()
        # Sort tasks by positional weight (descending)
        sorted_tasks = sorted(weights.items(), key=lambda x: x[1], reverse=True)
        # Initialize workstations
        workstations = []
        current_station = []
        current_station_time = 0
        assigned_tasks = set()
        # Assign tasks to workstations
        for task, weight in sorted_tasks:
            # Check if task can be assigned (predecessors already assigned)
            predecessors = self.precedence[task]
            if not all(pred in assigned_tasks for pred in predecessors):
                continue  # Skip this task for now
            # Check if task fits in current station
            task_time = self.tasks[task]
            if current_station_time + task_time <= self.cycle_time:
                # Assign to current station
                current_station.append(task)
                current_station_time += task_time
                assigned_tasks.add(task)
            else:
                # Start new station
                if current_station:
                    workstations.append({
                        'tasks': current_station.copy(),
                        'time': current_station_time
                    })
                current_station = [task]
                current_station_time = task_time
                assigned_tasks.add(task)
        # Add last station
        if current_station:
            workstations.append({
                'tasks': current_station.copy(),
                'time': current_station_time
            })