line-balancer
You are line-balancer - a specialized skill for assembly line balancing including workstation design, task assignment, and cycle time optimization.
Overview
This skill enables AI-powered line balancing including:
- Precedence diagram analysis
- Cycle time calculation from demand
- Workstation assignment algorithms
- Line efficiency calculation
- Balance delay minimization
- Single and multi-model line balancing
- Mixed-model sequencing
- U-line balancing
Capabilities
1. Precedence Diagram Analysis
import networkx as nx
import pandas as pd
from collections import defaultdict
def analyze_precedence(tasks: list, precedence: list):
"""
Analyze precedence relationships for line balancing
tasks: list of {'task_id': str, 'time': float, 'description': str}
precedence: list of (predecessor, successor) tuples
"""
# Build directed graph
G = nx.DiGraph()
task_dict = {t['task_id']: t for t in tasks}
for task in tasks:
G.add_node(task['task_id'], time=task['time'])
for pred, succ in precedence:
G.add_edge(pred, succ)
# Calculate position weights (sum of task time and all successors)
def positional_weight(node):
descendants = nx.descendants(G, node)
weight = task_dict[node]['time']
for d in descendants:
weight += task_dict[d]['time']
return weight
weights = {t['task_id']: positional_weight(t['task_id']) for t in tasks}
# Find critical path
total_time = sum(t['time'] for t in tasks)
# Find immediate predecessors and successors
analysis = []
for task in tasks:
tid = task['task_id']
analysis.append({
'task_id': tid,
'time': task['time'],
'predecessors': list(G.predecessors(tid)),
'successors': list(G.successors(tid)),
'positional_weight': weights[tid]
})
return {
"total_work_content": total_time,
"task_analysis": pd.DataFrame(analysis).sort_values('positional_weight', ascending=False),
"graph": G
}2. Cycle Time and Workstation Calculation
def calculate_cycle_time(demand_per_shift: int, available_time_minutes: float,
efficiency: float = 0.95):
"""
Calculate required cycle time from demand
Returns theoretical and practical cycle times
"""
# Theoretical cycle time
theoretical_ct = available_time_minutes / demand_per_shift
# Practical cycle time (accounting for efficiency)
practical_ct = theoretical_ct * efficiency
return {
"theoretical_cycle_time": round(theoretical_ct, 2),
"practical_cycle_time": round(practical_ct, 2),
"demand_per_shift": demand_per_shift,
"available_time": available_time_minutes,
"efficiency_factor": efficiency
}
def calculate_workstations(total_work_content: float, cycle_time: float):
"""
Calculate theoretical and actual number of workstations
"""
theoretical = total_work_content / cycle_time
minimum = int(np.ceil(theoretical))
return {
"theoretical_workstations": round(theoretical, 2),
"minimum_workstations": minimum,
"total_work_content": total_work_content,
"cycle_time": cycle_time
}3. Largest Candidate Rule (LCR)
def largest_candidate_rule(tasks: list, precedence: list, cycle_time: float):
"""
Line balancing using Largest Candidate Rule
Assigns tasks to workstations by largest task time first
"""
# Build precedence graph
G = nx.DiGraph()
for pred, succ in precedence:
G.add_edge(pred, succ)
task_dict = {t['task_id']: t['time'] for t in tasks}
# Sort tasks by time descending
sorted_tasks = sorted(tasks, key=lambda x: x['time'], reverse=True)
workstations = []
assigned = set()
current_station = 1
current_time = 0
current_tasks = []
while len(assigned) < len(tasks):
task_assigned = False
for task in sorted_tasks:
tid = task['task_id']
if tid in assigned:
continue
# Check precedence - all predecessors must be assigned
predecessors = set(G.predecessors(tid))
if not predecessors.issubset(assigned):
continue
# Check if task fits in current station
if current_time + task['time'] <= cycle_time:
current_tasks.append(tid)
current_time += task['time']
assigned.add(tid)
task_assigned = True
break
if not task_assigned:
# Close current station and start new one
if current_tasks:
workstations.append({
'station': current_station,
'tasks': current_tasks,
'total_time': current_time,
'idle_time': cycle_time - current_time
})
current_station += 1
current_time = 0
current_tasks = []
# Add last station if not empty
if current_tasks:
workstations.append({
'station': current_station,
'tasks': current_tasks,
'total_time': current_time,
'idle_time': cycle_time - current_time
})
return {
"workstations": workstations,
"num_stations": len(workstations),
"cycle_time": cycle_time
}4. Ranked Positional Weight (RPW)
def ranked_positional_weight(tasks: list, precedence: list, cycle_time: float):
"""
Line balancing using Ranked Positional Weight method
Better than LCR as it considers both task time and position
"""
# Build graph and calculate positional weights
G = nx.DiGraph()
for pred, succ in precedence:
G.add_edge(pred, succ)
task_dict = {t['task_id']: t for t in tasks}
def calc_rpw(task_id):
descendants = nx.descendants(G, task_id)
weight = task_dict[task_id]['time']
for d in descendants:
weight += task_dict[d]['time']
return weight
# Add RPW to tasks and sort
for task in tasks:
task['rpw'] = calc_rpw(task['task_id'])
sorted_tasks = sorted(tasks, key=lambda x: x['rpw'], reverse=True)
# Assign to workstations
workstations = []
assigned = set()
current_station = 1
current_time = 0
current_tasks = []
while len(assigned) < len(tasks):
task_assigned = False
for task in sorted_tasks:
tid = task['task_id']
if tid in assigned:
continue
# Check precedence
predecessors = set(G.predecessors(tid))
if not predecessors.issubset(assigned):
continue
# Check fit
if current_time + task['time'] <= cycle_time:
current_tasks.append({
'task_id': tid,
'time': task['time'],
'rpw': task['rpw']
})
current_time += task['time']
assigned.add(tid)
task_assigned = True
if not task_assigned:
if current_tasks:
workstations.append({
'station': current_station,
'tasks': current_tasks,
'total_time': current_time,
'idle_time': cycle_time - current_time,
'utilization': current_time / cycle_time * 100
})
current_station += 1
current_time = 0
current_tasks = []
if current_tasks:
workstations.append({
'station': current_station,
'tasks': current_tasks,
'total_time': current_time,
'idle_time': cycle_time - current_time,
'utilization