production-scheduling

生产排程

你是生产排程和制造运营方面的专家。你的目标是帮助组织制定高效的生产计划，以缩短生产周期、减少在制品（WIP）、按期交付并最大限度地提高资源利用率。

初步评估

制定生产计划前，请先了解：

1. 制造环境

   • 车间类型？（作业车间、流水车间、批量生产、连续生产）
   • 有多少台机器／工作中心？
   • 产品种类和产量如何？
   • 是按订单生产、备货生产还是按订单装配？

2. 排程约束

   • 是否存在工序优先约束（工艺路线／顺序要求）？
   • 机器的能力和限制是什么？
   • 是否有准备时间和换型要求？
   • 是否存在资源约束（劳动力、工装、物料）？
   • 班次安排和可用性如何？

3. 排程目标

   • 首要目标是什么？（缩短生产周期、按期交付、减少在制品）
   • 交期承诺和违约罚款如何？
   • 如何为相互竞争的工单设定优先级？
   • 各目标之间需要如何权衡？

4. 当前状态

   • 当前采用什么排程方法？（人工、ERP、电子表格）
   • 排程绩效指标如何？
   • 是否已知瓶颈或问题？
   • 计划周期和重新计划频率如何？

---

生产排程框架

车间布局类型

1. 作业车间

• 多种产品具有各自独特的工艺路线
• 产品种类多、产量低
• 排程问题复杂（NP-hard）
• 机器灵活，工序顺序多样
• 示例：定制机械、工具与模具车间

2. 流水车间

• 线性生产流程
• 所有产品遵循相同的机器顺序
• 比作业车间更容易排程
• 重点是优化工序顺序
• 示例：电子产品组装、食品加工

3. 批量生产

• 按批次生产
• 排程受准备时间影响
• 尽量减少换型
• 平衡批量大小与需求
• 示例：化工、制药

4. 连续生产

• 全天候运行
• 尽量减少换型和停机时间
• 重点是最大限度地提高吞吐量
• 生产均衡十分重要
• 示例：炼油、造纸厂

排程目标

常见目标：

1. 缩短生产周期 - 所有工单的总完成时间
2. 减少延迟 - 相对于交期的逾期交付
3. 减少在制品 - 在制品库存
4. 提高吞吐量 - 单位时间内生产的产品数量
5. 提高利用率 - 设备和劳动力效率
6. 缩短准备时间 - 减少工单之间的换型

多目标优化：

• 通常需要平衡多个目标
• 使用加权目标函数或 Pareto 优化
• 分析相互竞争目标之间的权衡

---

作业车间排程

问题建模

作业车间排程问题（JSP）：

已知：

• n 个工单 J₁、J₂、...、Jₙ
• m 台机器 M₁、M₂、...、Mₘ
• 每个工单都包含一系列工序，且每道工序都有加工时间
• 每道工序都需要指定的机器
• 不允许抢占（工序不能中断）

目标：缩短总生产周期（Cmax = 最晚完工时间）

经典 JSP 算法

调度规则：

用于工单排序的简单优先级规则：

import pandas as pd
import numpy as np
from typing import List, Dict, Tuple

class DispatchingRules:
    """
    Common dispatching rules for production scheduling
    """

    @staticmethod
    def FCFS(jobs):
        """First Come First Served - arrival order"""
        return sorted(jobs, key=lambda j: j['arrival_time'])

    @staticmethod
    def SPT(jobs):
        """Shortest Processing Time first"""
        return sorted(jobs, key=lambda j: j['processing_time'])

    @staticmethod
    def LPT(jobs):
        """Longest Processing Time first"""
        return sorted(jobs, key=lambda j: j['processing_time'], reverse=True)

    @staticmethod
    def EDD(jobs):
        """Earliest Due Date first"""
        return sorted(jobs, key=lambda j: j['due_date'])

    @staticmethod
    def CR(jobs, current_time):
        """Critical Ratio = (due_date - current_time) / remaining_processing_time"""
        for job in jobs:
            slack = job['due_date'] - current_time
            job['cr'] = slack / job['remaining_time'] if job['remaining_time'] > 0 else float('inf')
        return sorted(jobs, key=lambda j: j['cr'])

    @staticmethod
    def SLACK(jobs, current_time):
        """Minimum Slack = due_date - current_time - remaining_processing_time"""
        for job in jobs:
            job['slack'] = job['due_date'] - current_time - job['remaining_time']
        return sorted(jobs, key=lambda j: j['slack'])

    @staticmethod
    def WSPT(jobs):
        """Weighted Shortest Processing Time"""
        for job in jobs:
            job['wspt'] = job['priority'] / job['processing_time']
        return sorted(jobs, key=lambda j: j['wspt'], reverse=True)

# Example usage
jobs = [
    {'id': 'J1', 'arrival_time': 0, 'processing_time': 5, 'due_date': 15,
     'remaining_time': 5, 'priority': 3},
    {'id': 'J2', 'arrival_time': 1, 'processing_time': 3, 'due_date': 10,
     'remaining_time': 3, 'priority': 2},
    {'id': 'J3', 'arrival_time': 2, 'processing_time': 8, 'due_date': 20,
     'remaining_time': 8, 'priority': 1},
    {'id': 'J4', 'arrival_time': 3, 'processing_time': 4, 'due_date': 12,
     'remaining_time': 4, 'priority': 2},
]

# Apply different rules
spt_sequence = DispatchingRules.SPT(jobs.copy())
edd_sequence = DispatchingRules.EDD(jobs.copy())
cr_sequence = DispatchingRules.CR(jobs.copy(), current_time=5)

print("SPT Sequence:", [j['id'] for j in spt_sequence])
print("EDD Sequence:", [j['id'] for j in edd_sequence])
print("CR Sequence:", [j['id'] for j in cr_sequence])

使用约束编程进行作业车间排程

from ortools.sat.python import cp_model
import pandas as pd

class JobShopScheduler:
    """
    Solve Job Shop Scheduling Problem using CP-SAT
    Minimize makespan
    """

    def __init__(self, jobs_data):
        """
        jobs_data: list of jobs, each job is list of (machine, duration) tuples

        Example:
        jobs_data = [
            [(0, 3), (1, 2), (2, 2)],  # Job 0: M0(3h) -> M1(2h) -> M2(2h)
            [(0, 2), (2, 1), (1, 4)],  # Job 1: M0(2h) -> M2(1h) -> M1(4h)
            [(1, 4), (2, 3)]           # Job 2: M1(4h) -> M2(3h)
        ]
        """
        self.jobs_data = jobs_data
        self.num_jobs = len(jobs_data)
        self.num_machines = max(task[0] for job in jobs_data for task in job) + 1
        self.all_machines = range(self.num_machines)

        # Compute horizon (upper bound on makespan)
        self.horizon = sum(task[1] for job in jobs_data for task in job)

        self.model = cp_model.CpModel()
        self.solver = cp_model.CpSolver()

    def create_model(self):
        """Build the constraint programming model"""

        # Create variables
        self.task_type = {}
        self.starts = {}
        self.ends = {}
        self.intervals = {}

        for job_id, job in enumerate(self.jobs_data):
            for task_id, task in enumerate(job):
                machine, duration = task
                suffix = f'_{job_id}_{task_id}'

                # Start time variable
                start_var = self.model.NewIntVar(0, self.horizon, f'start{suffix}')
                # End time variable
                end_var = self.model.NewIntVar(0, self.horizon, f'end{suffix}')
                # Interval variable
                interval_var = self.model.NewIntervalVar(
                    start_var, duration, end_var, f'interval{suffix}'
                )

                self.task_type[(job_id, task_id)] = (machine, duration)
                self.starts[(job_id, task_id)] = start_var
                self.ends[(job_id, task_id)] = end_var
                self.intervals[(job_id, task_id)] = interval_var

        # Precedence constraints (within each job)
        for job_id, job in enumerate(self.jobs_data):
            for task_id in range(len(job) - 1):
                self.model.Add(
                    self.ends[(job_id, task_id)] <=
                    self.starts[(job_id, task_id + 1)]
                )

        # No overlap constraints (for each machine)
        for machine in self.all_machines:
            intervals_on_machine = [
                self.intervals[(job_id, task_id)]
                for job_id, job in enumerate(self.jobs_data)
                for task_id, task in enumerate(job)
                if task[0] == machine
            ]
            if intervals_on_machine:
                self.model.AddNoOverlap(intervals_on_machine)

        # Objective: minimize makespan
        self.makespan = self.model.NewIntVar(0, self.horizon, 'makespan')
        self.model.AddMaxEquality(
            self.makespan,
            [self.ends[(job_id, len(job) - 1)]
             for job_id, job in enumerate(self.jobs_data)]
        )
        self.model.Minimize(self.makespan)

    def solve(self, time_limit_seconds=10):
        """Solve the scheduling problem"""

        self.solver.parameters.max_time_in_seconds = time_limit_seconds
        status = self.solver.Solve(self.model)

        if status == cp_model.OPTIMAL or status == cp_model.FEASIBLE:
            return self._extract_solution()
        else:
            return None

    def _extract_solution(self):
        """Extract schedule from solved model"""

        schedule = []

        for job_id, job in enumerate(self.jobs_data):
            for task_id, task in enumerate(job):
                machine, duration = task
                start = self.solver.Value(self.starts[(job_id, task_id)])
                end = self.solver.Value(self.ends[(job_id, task_id)])

                schedule.append({
                    'job': job_id,
                    'task': task_id,
                    'machine': machine,
                    'start': start,
                    'end': end,
                    'duration': duration
                })

        makespan = self.solver.Value(self.makespan)

        return {
            'makespan': makespan,
            'schedule': pd.DataFrame(schedule),
            'objective': self.solver.ObjectiveValue(),
            'solve_time': self.solver.WallTime()
        }

    def print_schedule(self, solution):
        """Print the schedule in readable format"""

        if solution is None:
            print("No solution found")
            return

        print(f"\nOptimal Makespan: {solution['makespan']}")
        print(f"Solve Time: {solution['solve_time']:.2f} seconds\n")

        df = solution['schedule'].sort_values(['machine', 'start'])

        for machine in sorted(df['machine'].unique()):
            print(f"Machine {machine}:")
            machine_tasks = df[df['machine'] == machine]
            for _, row in machine_tasks.iterrows():
                print(f"  Job {row['job']}, Task {row['task']}: "
                      f"[{row['start']}, {row['end']}] (duration={row['duration']})")
            print()

# Example: Classic 3x3 Job Shop Problem (Fisher & Thompson)
jobs_data = [
    [(0, 3), (1, 2), (2, 2)],  # Job 0
    [(0, 2), (2, 1), (1, 4)],  # Job 1
    [(1, 4), (2, 3)]           # Job 2
]

jsp = JobShopScheduler(jobs_data)
jsp.create_model()
solution = jsp.solve(time_limit_seconds=10)
jsp.print_schedule(solution)

甘特图可视化

import matplotlib.pyplot as plt
import matplotlib.patches as patches
import numpy as np

def plot_gantt_chart(schedule_df, title="Production Schedule"):
    """
    Create Gantt chart visualization of production schedule

    Parameters:
    - schedule_df: DataFrame with columns [job, machine, start, end]
    """

    fig, ax = plt.subplots(figsize=(14, 8))

    # Assign colors to jobs
    jobs = schedule_df['job'].unique()
    colors = plt.cm.Set3(np.linspace(0, 1, len(jobs)))
    job_colors = {job: colors[i] for i, job in enumerate(jobs)}

    # Create bars for each task
    for idx, row in schedule_df.iterrows():
        ax.barh(
            y=row['machine'],
            width=row['end'] - row['start'],
            left=row['start'],
            height=0.8,
            color=job_colors[row['job']],
            edgecolor='black',
            linewidth=1.5,
            label=f"Job {row['job']}" if idx == 0 or
                  row['job'] not in schedule_df.iloc[:idx]['job'].values else ""
        )

        # Add job label on bar
        ax.text(
            (row['start'] + row['end']) / 2,
            row['machine'],
            f"J{row['job']}",
            ha='center',
            va='center',
            fontweight='bold',
            fontsize=10
        )

    # Configure axes
    machines = sorted(schedule_df['machine'].unique())
    ax.set_yticks(machines)
    ax.set_yticklabels([f'Machine {m}' for m in machines])
    ax.set_xlabel('Time', fontsize=12, fontweight='bold')
    ax.set_ylabel('Machine', fontsize=12, fontweight='bold')
    ax.set_title(title, fontsize=14, fontweight='bold')

    # Add grid
    ax.grid(True, axis='x', alpha=0.3, linestyle='--')

    # Add legend (unique jobs only)
    handles, labels = ax.get_legend_handles_labels()
    by_label = dict(zip(labels, handles))
    ax.legend(by_label.values(), by_label.keys(),
              loc='upper right', fontsize=10)

    plt.tight_layout()
    return fig

# Use with previous solution
if solution:
    fig = plot_gantt_chart(solution['schedule'],
                          title=f"Job Shop Schedule (Makespan={solution['makespan']})")
    plt.show()

---

流水车间排程

NEH 算法（Nawaz-Enscore-Ham）

用于置换流水车间排程的最佳启发式算法：

import numpy as np
import itertools

class FlowShopNEH:
    """
    NEH Algorithm for Permutation Flow Shop Scheduling
    Minimize makespan
    """

    def __init__(self, processing_times):
        """
        processing_times: 2D array [n_jobs x m_machines]
        processing_times[j][m] = time for job j on machine m
        """
        self.processing_times = np.array(processing_times)
        self.n_jobs, self.n_machines = self.processing_times.shape

    def calculate_makespan(self, sequence):
        """
        Calculate makespan for given job sequence

        Parameters:
        - sequence: list of job indices

        Returns completion time (makespan)
        """
        n = len(sequence)
        m = self.n_machines

        # Completion times: C[j][m] = completion time of job j on machine m
        C = np.zeros((n, m))

        for j_idx, job in enumerate(sequence):
            for machine in range(m):
                if j_idx == 0 and machine == 0:
                    # First job, first machine
                    C[j_idx][machine] = self.processing_times[job][machine]
                elif j_idx == 0:
                    # First job, subsequent machines
                    C[j_idx][machine] = (C[j_idx][machine-1] +
                                        self.processing_times[job][machine])
                elif machine == 0:
                    # Subsequent jobs, first machine
                    C[j_idx][machine] = (C[j_idx-1][machine] +
                                        self.processing_times[job][machine])
                else:
                    # Subsequent jobs, subsequent machines
                    C[j_idx][machine] = (max(C[j_idx-1][machine],
                                            C[j_idx][machine-1]) +
                                        self.processing_times[job][machine])

        return C[n-1][m-1]  # Makespan = completion time of last job on last machine

    def neh_algorithm(self):
        """
        NEH heuristic algorithm

        Returns:
        - best_sequence: optimal job sequence
        - makespan: corresponding makespan
        """

        # Step 1: Calculate total processing time for each job
        total_times = self.processing_times.sum(axis=1)

        # Step 2: Sort jobs by descending total processing time
        sorted_jobs = np.argsort(-total_times)  # Descending order

        # Step 3: Build sequence iteratively
        sequence = [sorted_jobs[0]]  # Start with job with highest total time

        for job in sorted_jobs[1:]:
            best_makespan = float('inf')
            best_position = 0

            # Try inserting job at each position
            for pos in range(len(sequence) + 1):
                trial_sequence = sequence[:pos] + [job] + sequence[pos:]
                makespan = self.calculate_makespan(trial_sequence)

                if makespan < best_makespan:
                    best_makespan = makespan
                    best_position = pos

            # Insert job at best position
            sequence.insert(best_position, job)

        final_makespan = self.calculate_makespan(sequence)

        return {
            'sequence': sequence,
            'makespan': final_makespan,
            'job_labels': [f'Job_{j}' for j in sequence]
        }

# Example: 4 jobs x 3 machines
processing_times = [
    [5, 3, 4],  # Job 0
    [3, 4, 2],  # Job 1
    [6, 2, 3],  # Job 2
    [4, 5, 1]   # Job 3
]

fs = FlowShopNEH(processing_times)
result = fs.neh_algorithm()

print(f"Optimal Sequence: {result['job_labels']}")
print(f"Makespan: {result['makespan']}")

考虑准备时间的流水车间

class FlowShopWithSetups:
    """
    Flow shop scheduling with sequence-dependent setup times
    """

    def __init__(self, processing_times, setup_times):
        """
        processing_times: 2D array [n_jobs x m_machines]
        setup_times: 3D array [n_jobs x n_jobs x m_machines]
                     setup_times[i][j][m] = setup time from job i to job j on machine m
        """
        self.processing_times = np.array(processing_times)
        self.setup_times = np.array(setup_times)
        self.n_jobs, self.n_machines = self.processing_times.shape

    def calculate_makespan_with_setups(self, sequence):
        """Calculate makespan including setup times"""

        n = len(sequence)
        m = self.n_machines
        C = np.zeros((n, m))

        for j_idx, job in enumerate(sequence):
            for machine in range(m):
                # Get setup time
                if j_idx == 0:
                    setup_time = 0  # No setup for first job
                else:
                    prev_job = sequence[j_idx - 1]
                    setup_time = self.setup_times[prev_job][job][machine]

                processing_time = self.processing_times[job][machine]

                if j_idx == 0 and machine == 0:
                    C[j_idx][machine] = processing_time
                elif j_idx == 0:
                    C[j_idx][machine] = C[j_idx][machine-1] + setup_time + processing_time
                elif machine == 0:
                    C[j_idx][machine] = C[j_idx-1][machine] + setup_time + processing_time
                else:
                    C[j_idx][machine] = (max(C[j_idx-1][machine], C[j_idx][machine-1]) +
                                        setup_time + processing_time)

        return C[n-1][m-1]

    def genetic_algorithm(self, population_size=50, generations=100,
                         mutation_rate=0.1, crossover_rate=0.8):
        """
        Genetic algorithm for flow shop with setups

        Returns best sequence found
        """

        # Initialize population with random sequences
        population = [np.random.permutation(self.n_jobs).tolist()
                     for _ in range(population_size)]

        best_solution = None
        best_makespan = float('inf')

        for gen in range(generations):
            # Evaluate fitness (lower makespan = better)
            fitness = []
            for seq in population:
                makespan = self.calculate_makespan_with_setups(seq)
                fitness.append(1.0 / makespan)  # Inverse for maximization

                if makespan < best_makespan:
                    best_makespan = makespan
                    best_solution = seq.copy()

            # Selection: tournament selection
            new_population = []
            fitness = np.array(fitness)

            for _ in range(population_size):
                # Tournament of size 3
                tournament = np.random.choice(population_size, 3, replace=False)
                winner = tournament[np.argmax(fitness[tournament])]
                new_population.append(population[winner].copy())

            # Crossover: order crossover (OX)
            for i in range(0, population_size-1, 2):
                if np.random.random() < crossover_rate:
                    parent1 = new_population[i]
                    parent2 = new_population[i+1]
                    child1, child2 = self._order_crossover(parent1, parent2)
                    new_population[i] = child1
                    new_population[i+1] = child2

            # Mutation: swap mutation
            for i in range(population_size):
                if np.random.random() < mutation_rate:
                    new_population[i] = self._swap_mutation(new_population[i])

            population = new_population

        return {
            'sequence': best_solution,
            'makespan': best_makespan
        }

    def _order_crossover(self, parent1, parent2):
        """Order crossover (OX) for permutation"""
        size = len(parent1)
        start, end = sorted(np.random.choice(size, 2, replace=False))

        child1 = [-1] * size
        child2 = [-1] * size

        # Copy segment
        child1[start:end] = parent1[start:end]
        child2[start:end] = parent2[start:end]

        # Fill remaining from other parent
        self._fill_remaining(child1, parent2, start, end)
        self._fill_remaining(child2, parent1, start, end)

        return child1, child2

    def _fill_remaining(self, child, parent, start, end):
        """Helper for order crossover"""
        size = len(child)
        parent_idx = end
        child_idx = end

        while -1 in child:
            if parent[parent_idx % size] not in child:
                child[child_idx % size] = parent[parent_idx % size]
                child_idx += 1
            parent_idx += 1

    def _swap_mutation(self, sequence):
        """Swap two random positions"""
        seq = sequence.copy()
        i, j = np.random.choice(len(seq), 2, replace=False)
        seq[i], seq[j] = seq[j], seq[i]
        return seq

# Example with setup times
processing_times = [
    [5, 3, 4],
    [3, 4, 2],
    [6, 2, 3],
    [4, 5, 1]
]

# Setup times: [from_job][to_job][machine]
n_jobs, n_machines = 4, 3
setup_times = np.random.randint(1, 3, size=(n_jobs, n_jobs, n_machines))
# No setup from job to itself
for i in range(n_jobs):
    setup_times[i][i][:] = 0

fs_setup = FlowShopWithSetups(processing_times, setup_times)
result = fs_setup.genetic_algorithm(population_size=50, generations=100)

print(f"Best Sequence: {result['sequence']}")
print(f"Makespan: {result['makespan']}")

---

高级排程技术

有限产能排程（FCS）

from datetime import datetime, timedelta
import pandas as pd

class FiniteCapacityScheduler:
    """
    Finite capacity scheduler with resource constraints
    Forward and backward scheduling
    """

    def __init__(self, work_centers):
        """
        work_centers: dict {wc_id: {'capacity_hours': X, 'efficiency': Y}}
        """
        self.work_centers = work_centers
        self.schedule = []

    def forward_schedule(self, orders, start_date):
        """
        Forward scheduling from start date
        Schedule as early as possible

        Parameters:
        - orders: list of dicts with 'order_id', 'routing', 'quantity'
        - routing is list of (work_center, hours_per_unit)
        """

        # Track work center availability
        wc_availability = {wc: start_date for wc in self.work_centers}

        # Sort orders by priority (if available) or order
        orders = sorted(orders, key=lambda x: x.get('priority', 0), reverse=True)

        results = []

        for order in orders:
            order_start = start_date

            for operation_idx, (wc, hours_per_unit) in enumerate(order['routing']):
                # Calculate required hours
                required_hours = order['quantity'] * hours_per_unit
                efficiency = self.work_centers[wc]['efficiency']
                actual_hours = required_hours / efficiency

                # Earliest start is max of order flow and resource availability
                operation_start = max(order_start, wc_availability[wc])
                operation_end = operation_start + timedelta(hours=actual_hours)

                # Update work center availability
                wc_availability[wc] = operation_end

                # Record operation
                results.append({
                    'order_id': order['order_id'],
                    'operation': operation_idx,
                    'work_center': wc,
                    'start': operation_start,
                    'end': operation_end,
                    'hours': actual_hours
                })

                # Next operation starts after this one
                order_start = operation_end

        return pd.DataFrame(results)

    def backward_schedule(self, orders, due_dates):
        """
        Backward scheduling from due dates
        Schedule as late as possible (just-in-time)

        Parameters:
        - orders: list of dicts with 'order_id', 'routing', 'quantity'
        - due_dates: dict {order_id: due_date}
        """

        results = []

        for order in orders:
            order_due = due_dates[order['order_id']]

            # Process routing in reverse
            routing = list(reversed(order['routing']))

            for operation_idx, (wc, hours_per_unit) in enumerate(routing):
                required_hours = order['quantity'] * hours_per_unit
                efficiency = self.work_centers[wc]['efficiency']
                actual_hours = required_hours / efficiency

                operation_end = order_due
                operation_start = operation_end - timedelta(hours=actual_hours)

                results.append({
                    'order_id': order['order_id'],
                    'operation': len(routing) - operation_idx - 1,
                    'work_center': wc,
                    'start': operation_start,
                    'end': operation_end,
                    'hours': actual_hours
                })

                order_due = operation_start

        return pd.DataFrame(results).sort_values(['order_id', 'operation'])

    def check_capacity_feasibility(self, schedule, planning_horizon):
        """
        Check if schedule is feasible given capacity constraints

        Returns overload periods
        """

        overloads = []

        for wc in self.work_centers:
            wc_schedule = schedule[schedule['work_center'] == wc]
            capacity = self.work_centers[wc]['capacity_hours']

            # Check each day/period
            current_date = schedule['start'].min().date()
            end_date = schedule['end'].max().date()

            while current_date <= end_date:
                day_start = datetime.combine(current_date, datetime.min.time())
                day_end = day_start + timedelta(days=1)

                # Calculate load for this day
                daily_ops = wc_schedule[
                    (wc_schedule['start'] < day_end) &
                    (wc_schedule['end'] > day_start)
                ]

                load = 0
                for _, op in daily_ops.iterrows():
                    # Calculate overlap with this day
                    overlap_start = max(op['start'], day_start)
                    overlap_end = min(op['end'], day_end)
                    overlap_hours = (overlap_end - overlap_start).total_seconds() / 3600
                    load += overlap_hours

                if load > capacity:
                    overloads.append({
                        'work_center': wc,
                        'date': current_date,
                        'load': load,
                        'capacity': capacity,
                        'overload': load - capacity
                    })

                current_date += timedelta(days=1)

        return pd.DataFrame(overloads) if overloads else None

# Example usage
work_centers = {
    'Cutting': {'capacity_hours': 16, 'efficiency': 0.90},
    'Welding': {'capacity_hours': 16, 'efficiency': 0.85},
    'Assembly': {'capacity_hours': 16, 'efficiency': 0.92}
}

orders = [
    {
        'order_id': 'ORD001',
        'quantity': 100,
        'priority': 1,
        'routing': [('Cutting', 0.5), ('Welding', 0.8), ('Assembly', 1.0)]
    },
    {
        'order_id': 'ORD002',
        'quantity': 75,
        'priority': 2,
        'routing': [('Cutting', 0.6), ('Assembly', 1.2)]
    }
]

fcs = FiniteCapacityScheduler(work_centers)

# Forward schedule
start_date = datetime(2025, 2, 1, 8, 0)
forward_schedule = fcs.forward_schedule(orders, start_date)
print("Forward Schedule:")
print(forward_schedule)

# Check capacity
overloads = fcs.check_capacity_feasibility(forward_schedule, planning_horizon=30)
if overloads is not None:
    print("\nCapacity Overloads:")
    print(overloads)
else:
    print("\nSchedule is feasible - no capacity violations")

使用遗传算法优化排程

import random

class ScheduleOptimizer:
    """
    Genetic algorithm for multi-objective scheduling
    Minimize: makespan, tardiness, setup costs
    """

    def __init__(self, jobs, machines, setup_matrix):
        self.jobs = jobs  # List of job objects
        self.machines = machines
        self.setup_matrix = setup_matrix  # Setup time between jobs

    def create_individual(self):
        """Create random job sequence (chromosome)"""
        return random.sample(self.jobs, len(self.jobs))

    def calculate_fitness(self, sequence):
        """
        Multi-objective fitness function
        Lower is better
        """
        makespan = self._calculate_makespan(sequence)
        tardiness = self._calculate_tardiness(sequence)
        setup_cost = self._calculate_setup_cost(sequence)

        # Weighted sum (normalize and weight)
        fitness = (
            0.4 * (makespan / 100) +     # Normalize to reasonable scale
            0.4 * (tardiness / 100) +
            0.2 * (setup_cost / 50)
        )

        return {
            'fitness': fitness,
            'makespan': makespan,
            'tardiness': tardiness,
            'setup_cost': setup_cost
        }

    def _calculate_makespan(self, sequence):
        """Calculate total completion time"""
        time = 0
        for job in sequence:
            time += job['processing_time']
        return time

    def _calculate_tardiness(self, sequence):
        """Calculate total tardiness (late deliveries)"""
        time = 0
        total_tardiness = 0

        for job in sequence:
            time += job['processing_time']
            tardiness = max(0, time - job['due_date'])
            total_tardiness += tardiness

        return total_tardiness

    def _calculate_setup_cost(self, sequence):
        """Calculate total setup time/cost"""
        total_setup = 0

        for i in range(len(sequence) - 1):
            current_job = sequence[i]['id']
            next_job = sequence[i + 1]['id']
            total_setup += self.setup_matrix.get((current_job, next_job), 0)

        return total_setup

    def optimize(self, population_size=100, generations=200):
        """Run genetic algorithm"""

        # Initialize population
        population = [self.create_individual() for _ in range(population_size)]

        best_solution = None
        best_fitness = float('inf')
        history = []

        for generation in range(generations):
            # Evaluate fitness
            fitness_scores = []
            for individual in population:
                metrics = self.calculate_fitness(individual)
                fitness_scores.append(metrics['fitness'])

                if metrics['fitness'] < best_fitness:
                    best_fitness = metrics['fitness']
                    best_solution = individual.copy()
                    best_metrics = metrics

            history.append({
                'generation': generation,
                'best_fitness': best_fitness,
                'avg_fitness': sum(fitness_scores) / len(fitness_scores)
            })

            # Selection: tournament
            selected = []
            for _ in range(population_size):
                tournament = random.sample(list(zip(population, fitness_scores)), 5)
                winner = min(tournament, key=lambda x: x[1])[0]
                selected.append(winner)

            # Crossover and mutation
            next_generation = []
            for i in range(0, population_size, 2):
                parent1 = selected[i]
                parent2 = selected[i + 1] if i + 1 < population_size else selected[0]

                if random.random() < 0.8:  # Crossover rate
                    child1, child2 = self._pmx_crossover(parent1, parent2)
                else:
                    child1, child2 = parent1.copy(), parent2.copy()

                if random.random() < 0.2:  # Mutation rate
                    child1 = self._swap_mutation(child1)
                if random.random() < 0.2:
                    child2 = self._swap_mutation(child2)

                next_generation.extend([child1, child2])

            population = next_generation[:population_size]

        return {
            'best_sequence': [j['id'] for j in best_solution],
            'metrics': best_metrics,
            'history': pd.DataFrame(history)
        }

    def _pmx_crossover(self, parent1, parent2):
        """Partially Mapped Crossover for permutation"""
        size = len(parent1)
        child1 = [None] * size
        child2 = [None] * size

        # Select crossover points
        cx_point1, cx_point2 = sorted(random.sample(range(size), 2))

        # Copy segments
        child1[cx_point1:cx_point2] = parent1[cx_point1:cx_point2]
        child2[cx_point1:cx_point2] = parent2[cx_point1:cx_point2]

        # Fill remaining positions
        for i in range(size):
            if i < cx_point1 or i >= cx_point2:
                # Fill from parent2 for child1
                gene = parent2[i]
                while gene in child1:
                    idx = parent2.index(gene)
                    if cx_point1 <= idx < cx_point2:
                        gene = parent1[idx]
                    else:
                        break
                child1[i] = gene

                # Fill from parent1 for child2
                gene = parent1[i]
                while gene in child2:
                    idx = parent1.index(gene)
                    if cx_point1 <= idx < cx_point2:
                        gene = parent2[idx]
                    else:
                        break
                child2[i] = gene

        return child1, child2

    def _swap_mutation(self, sequence):
        """Swap two random jobs"""
        seq = sequence.copy()
        i, j = random.sample(range(len(seq)), 2)
        seq[i], seq[j] = seq[j], seq[i]
        return seq

---

MRP 与生产排程集成

物料需求计划（MRP）

import pandas as pd
import numpy as np
from datetime import datetime, timedelta

class MRPScheduler:
    """
    MRP logic with production scheduling
    Calculate material requirements and schedule production
    """

    def __init__(self, bom, inventory, lead_times):
        """
        Parameters:
        - bom: bill of materials dict {parent: [(child, quantity)]}
        - inventory: current on-hand inventory {item: quantity}
        - lead_times: production/procurement lead times {item: days}
        """
        self.bom = bom
        self.inventory = inventory
        self.lead_times = lead_times

    def explode_bom(self, item, quantity, level=0):
        """
        Recursively explode BOM
        Returns list of (item, quantity, level) tuples
        """
        requirements = [(item, quantity, level)]

        if item in self.bom:
            for component, qty_per in self.bom[item]:
                requirements.extend(
                    self.explode_bom(component, quantity * qty_per, level + 1)
                )

        return requirements

    def calculate_net_requirements(self, gross_requirements):
        """
        Calculate net requirements considering on-hand inventory

        Parameters:
        - gross_requirements: dict {item: quantity}

        Returns net requirements and planned orders
        """
        net_requirements = {}
        planned_orders = {}

        for item, gross_qty in gross_requirements.items():
            on_hand = self.inventory.get(item, 0)
            net_qty = max(0, gross_qty - on_hand)

            if net_qty > 0:
                net_requirements[item] = net_qty
                planned_orders[item] = net_qty

                # Update inventory (consumption)
                self.inventory[item] = max(0, on_hand - gross_qty)
            else:
                # Sufficient inventory
                self.inventory[item] = on_hand - gross_qty

        return net_requirements, planned_orders

    def time_phase_mrp(self, master_schedule, planning_horizon_days=90):
        """
        Time-phased MRP calculation

        Parameters:
        - master_schedule: list of dicts {'item', 'quantity', 'due_date'}
        - planning_horizon_days: planning horizon

        Returns time-phased requirements and production schedule
        """

        start_date = min(order['due_date'] for order in master_schedule)
        end_date = start_date + timedelta(days=planning_horizon_days)

        # Create time buckets (weekly)
        time_buckets = pd.date_range(start=start_date, end=end_date, freq='W')

        # Initialize requirements by time bucket
        all_items = set()
        for order in master_schedule:
            all_items.add(order['item'])
            requirements = self.explode_bom(order['item'], order['quantity'])
            for item, qty, level in requirements:
                all_items.add(item)

        mrp_records = []

        for item in all_items:
            # Gross requirements by time bucket
            gross_req = pd.Series(0, index=time_buckets)

            for order in master_schedule:
                if order['item'] == item:
                    # Find appropriate time bucket
                    bucket = time_buckets[time_buckets >= order['due_date']][0]
                    gross_req[bucket] += order['quantity']
                else:
                    # Check if item is a component
                    requirements = self.explode_bom(order['item'], order['quantity'])
                    for req_item, qty, level in requirements:
                        if req_item == item:
                            # Offset by parent lead time
                            parent_lead = self.lead_times.get(order['item'], 0)
                            offset_date = order['due_date'] - timedelta(days=parent_lead)
                            if offset_date >= time_buckets[0]:
                                bucket = time_buckets[time_buckets >= offset_date][0]
                                gross_req[bucket] += qty

            # Calculate net requirements and planned orders
            on_hand = self.inventory.get(item, 0)

            for bucket in time_buckets:
                gross = gross_req[bucket]

                if gross > 0:
                    net = max(0, gross - on_hand)

                    if net > 0:
                        # Planned order release (offset by lead time)
                        lead_time = self.lead_times.get(item, 0)
                        release_date = bucket - timedelta(days=lead_time)

                        mrp_records.append({
                            'item': item,
                            'bucket': bucket,
                            'gross_requirement': gross,
                            'on_hand_start': on_hand,
                            'net_requirement': net,
                            'planned_order': net,
                            'release_date': release_date
                        })

                        on_hand = 0  # Inventory consumed
                    else:
                        # Sufficient inventory
                        mrp_records.append({
                            'item': item,
                            'bucket': bucket,
                            'gross_requirement': gross,
                            'on_hand_start': on_hand,
                            'net_requirement': 0,
                            'planned_order': 0,
                            'release_date': None
                        })

                        on_hand -= gross

        return pd.DataFrame(mrp_records)

# Example usage
bom = {
    'Product_A': [('Subassembly_X', 2), ('Component_Y', 4)],
    'Subassembly_X': [('Part_1', 3), ('Part_2', 1)],
}

inventory = {
    'Product_A': 10,
    'Subassembly_X': 5,
    'Component_Y': 20,
    'Part_1': 50,
    'Part_2': 30
}

lead_times = {
    'Product_A': 5,
    'Subassembly_X': 7,
    'Component_Y': 3,
    'Part_1': 10,
    'Part_2': 14
}

mrp = MRPScheduler(bom, inventory, lead_times)

# Master production schedule
master_schedule = [
    {'item': 'Product_A', 'quantity': 100, 'due_date': datetime(2025, 3, 1)},
    {'item': 'Product_A', 'quantity': 150, 'due_date': datetime(2025, 3, 15)},
]

mrp_plan = mrp.time_phase_mrp(master_schedule)
print(mrp_plan[mrp_plan['planned_order'] > 0])

---

绩效指标

关键排程指标

def calculate_scheduling_metrics(schedule_df, jobs_df):
    """
    Calculate comprehensive scheduling performance metrics

    Parameters:
    - schedule_df: DataFrame with actual schedule [job, start, end, machine]
    - jobs_df: DataFrame with job details [job, processing_time, due_date, priority]

    Returns dict of metrics
    """

    # Merge schedule with job details
    df = schedule_df.merge(jobs_df, on='job')

    # 1. Makespan
    makespan = df['end'].max() - df['start'].min()

    # 2. Mean flow time
    flow_times = df.groupby('job')['end'].max() - df.groupby('job')['start'].min()
    mean_flow_time = flow_times.mean()

    # 3. Tardiness
    completion_times = df.groupby('job')['end'].max()
    due_dates = df.groupby('job')['due_date'].first()
    tardiness = (completion_times - due_dates).clip(lower=0)
    total_tardiness = tardiness.sum()
    mean_tardiness = tardiness.mean()
    num_tardy = (tardiness > 0).sum()

    # 4. Utilization
    total_processing_time = df['processing_time'].sum()
    num_machines = df['machine'].nunique()
    utilization = (total_processing_time / (makespan * num_machines)) * 100

    # 5. WIP (Work in Process)
    # Average number of jobs in system
    all_times = sorted(set(df['start'].tolist() + df['end'].tolist()))
    wip_samples = []

    for t in all_times[:-1]:
        jobs_in_process = ((df['start'] <= t) & (df['end'] > t)).sum()
        wip_samples.append(jobs_in_process)

    avg_wip = np.mean(wip_samples) if wip_samples else 0

    # 6. On-time delivery rate
    otd_rate = ((tardiness == 0).sum() / len(tardiness)) * 100

    return {
        'makespan': makespan,
        'mean_flow_time': mean_flow_time,
        'total_tardiness': total_tardiness,
        'mean_tardiness': mean_tardiness,
        'num_tardy_jobs': num_tardy,
        'utilization_pct': utilization,
        'avg_wip': avg_wip,
        'on_time_delivery_pct': otd_rate
    }

---

工具与库

Python 库

优化与排程：

• ortools：Google 的约束规划与优化库（CP-SAT 求解器）
• pulp：用于排程问题的线性规划
• pyomo：高级优化建模
• gekko：动态优化与排程
• docplex：IBM CPLEX Python API
• simanneal：模拟退火元启发式算法

仿真：

• simpy：用于制造业的离散事件仿真
• salabim：动画与仿真
• numpy：数值计算

可视化：

• matplotlib、plotly：甘特图与排程
• seaborn：统计可视化
• gantt：甘特图库

商业排程软件

高级计划系统（APS）：

• SAP APO (PP/DS)：生产计划与详细排程
• Oracle Advanced Supply Chain Planning (ASCP)：基于约束的计划
• Blue Yonder (JDA)：制造计划与排程
• Kinaxis RapidResponse：并发计划与排程
• Dassault Systèmes DELMIA：生产排程与仿真

专业排程软件：

• Asprova：面向车间作业型与混合模式制造的 APS
• Preactor：有限产能排程（Siemens）
• Opcenter APS：高级计划与排程（Siemens）
• Ortems：面向复杂制造的 APS
• FELIOS：制造执行与排程

MES（制造执行系统）：

• SAP ME/MII：制造执行
• Rockwell FactoryTalk：生产排程模块
• Wonderware MES：实时排程
• Parsec TrakSYS：包含排程功能的 MES

---

常见挑战与解决方案

挑战：排程不稳定

问题：

• 频繁更改排程
• 扰乱车间生产
• 降低排程的可信度

解决方案：

• 设置冻结区（近期不重新排程）
• 制定排程稳定性指标
• 采用滚动时域方法
• 为不确定因素预留缓冲时间
• 建立变更沟通流程

挑战：意外中断

问题：

• 机器故障
• 物料短缺
• 质量问题
• 紧急订单

解决方案：

• 具备动态重新排程能力
• 在瓶颈处预留缓冲产能
• 采用替代工艺路线
• 制定加急处理流程
• 实时监控（与 MES 集成）
• 采用响应式排程算法

挑战：优化准备时间

问题：

• 准备／换线时间较长
• 序列优化与交期之间存在取舍
• 准备工作之间的依赖关系复杂

解决方案：

• 采用批次排程（将相似作业安排在一起）
• 优化准备时间矩阵
• 采用依赖序列的排程算法
• 缩短准备时间（SMED - 单分钟换模）
• 条件允许时并行进行准备工作

挑战：多目标冲突

问题：

• 缩短完工时间与满足交期之间存在冲突
• 减少在制品与保持机器满负荷运行之间存在冲突
• 优先级相互冲突

解决方案：

• 明确定义优先级层次
• 使用加权多目标函数
• 采用 Pareto 优化（权衡曲线）
• 通过仿真测试不同场景
• 协同计划（S&OP）

挑战：瓶颈处产能有限

问题：

• 瓶颈机器限制吞吐量
• 非瓶颈机器缺料或受阻
• 对整个系统产生影响

解决方案：

• 约束理论（TOC）／鼓－缓冲－绳
• 在瓶颈处设置保护性缓冲
• 使非瓶颈排程服从瓶颈排程
• 将工作从瓶颈处转移（外包、替代路线）
• 提升瓶颈产能（加班、增加设备）

挑战：数据质量与准确性

问题：

• 加工时间不准确
• 系统中的工艺路线有误
• 提前期数据过时
• 排程表现不佳

解决方案：

• 定期审计和更新数据
• 收集车间反馈（MES）
• 对实际工时与标准工时进行统计分析
• 建立数据治理流程
• 使用学习算法调整参数

---

输出格式

生产排程报告

执行摘要：

• 排程周期和规划范围
• 关键绩效指标（总工期、利用率、准时交付率）
• 关键瓶颈或约束
• 与上一版本相比的排程变更

按工作中心划分的详细排程：

工作中心	作业	工序	开始时间	结束时间	持续时间	准备时间	状态
切割	J001	OP10	2025-02-01 08:00	2025-02-01 11:30	3.5h	0.5h	已排程
切割	J003	OP10	2025-02-01 12:00	2025-02-01 16:00	4.0h	0.5h	已排程
焊接	J001	OP20	2025-02-01 12:00	2025-02-01 18:00	6.0h	1.0h	已排程

产能分析：

工作中心	可用工时	已排程工时	利用率 %	超负荷工时
切割	160	152	95%	0
焊接	160	168	105%	8
装配	160	128	80%	0

排程绩效指标：

• 总工期：12 天
• 平均流动时间：6.5 天
• 总延迟时间：45 小时
• 准时交付率：87%
• 平均在制品：23 个作业
• 设备利用率：88%

甘特图： [排程的可视化表示]

行动事项：

• 第 2 周焊接工作中心超负荷——建议安排加班或外包
• 作业 J005、J012、J018 有无法按期交付的风险
• 通过将相似作业安排在一起，可缩短切割准备时间
• J023 的物料要到 2 月 5 日才能到货——暂停排程

---

可提出的问题

如果需要更多背景信息：

1. 属于哪种制造环境？（作业车间、流水车间、批量生产、连续生产）
2. 有多少台机器／工作中心？工艺路线是什么？
3. 主要排程目标是什么？（总工期、交期、在制品、利用率）
4. 是否存在准备时间或换型约束？
5. 当前采用什么排程方法？已知问题有哪些？
6. 规划范围和重新排程频率是什么？
7. 除机器外，是否还存在资源约束（劳动力、工装、物料）？
8. 与 ERP／MES 系统的集成要求是什么？

---

相关技能

• master-production-scheduling：用于主生产计划（MPS）和需求驱动排程
• capacity-planning：用于产能分析和需求计划
• job-shop-scheduling：用于详细的 JSP 算法
• flow-shop-scheduling：用于流水车间专用方法
• assembly-line-balancing：用于生产线平衡和节拍时间
• lean-manufacturing：用于拉动式排程和 JIT
• inventory-optimization：用于在制品和缓冲库存优化
• optimization-modeling：用于数学排程模型
• constraint-programming：用于基于约束的高级排程