用 Python 做 IE 数据分析：从工时数据到线平衡计算

# 核心库
import pandas as pd        # 数据处理
import numpy as np         # 数值计算
from scipy import stats    # 统计检验
import matplotlib.pyplot as plt   # 可视化

# 优化与仿真（按需安装）
# pip install pulp         # 线性规划
# pip install simpy        # 离散事件仿真
# pip install ortools      # 运筹优化

import numpy as np
import pandas as pd

def calc_standard_time(observations, rating=1.0, allowance=0.15, verbose=True):
    """标准工时计算：剔除异常值 + 评比 + 宽放"""
    obs = np.array(observations, dtype=float)

    # 1. 三倍标准差法剔除异常值
    while True:
        mean = obs.mean()
        std = obs.std(ddof=1)
        lower, upper = mean - 3 * std, mean + 3 * std
        mask = (obs >= lower) & (obs <= upper)
        if mask.all():
            break
        removed = obs[~mask]
        if verbose:
            print(f"剔除异常值: {removed.round(2).tolist()}")
        obs = obs[mask]

    observed_time = obs.mean()
    normal_time = observed_time * rating
    standard_time = normal_time * (1 + allowance)

    if verbose:
        print(f"有效观测次数: {len(obs)}")
        print(f"观测平均时间: {observed_time:.2f} 秒")
        print(f"评比系数:     {rating}")
        print(f"正常时间:     {normal_time:.2f} 秒")
        print(f"宽放率:       {allowance:.1%}")
        print(f"标准时间:     {standard_time:.2f} 秒")
        print(f"标准差:       {obs.std(ddof=1):.3f}  变异系数: {obs.std(ddof=1)/observed_time:.2%}")

    return standard_time

# 示例：15 次观测，含一个异常值
data = [42.1, 43.5, 41.8, 42.6, 42.9, 43.1, 41.9, 42.4,
        42.7, 68.2, 43.0, 42.3, 41.7, 42.8, 42.5]
st = calc_standard_time(data, rating=1.10, allowance=0.18)

# 从 Excel 读取多工位测时数据
df = pd.read_excel("工时观测.xlsx")
# 假设格式：工位 | 观测序列（多列）

results = []
for _, row in df.iterrows():
    obs = row.drop("工位").dropna().astype(float).tolist()
    st = calc_standard_time(obs, rating=1.05, allowance=0.15, verbose=False)
    results.append({"工位": row["工位"], "标准工时": round(st, 2)})

result_df = pd.DataFrame(results)
print(result_df)

import pandas as pd
import numpy as np

def line_balance(task_times, takt_time=None, available_time=27600, demand=None):
    """产线平衡分析
    task_times: 各工位作业时间列表（秒）
    takt_time: 节拍（秒），与 demand 二选一
    available_time: 每日有效工作时间（秒），默认 460 分钟
    demand: 日需求数量
    """
    tt = np.array(task_times, dtype=float)

    if takt_time is None:
        if demand is None:
            raise ValueError("需提供 takt_time 或 demand")
        takt_time = available_time / demand

    total_time = tt.sum()
    bottleneck = tt.max()
    bottleneck_idx = int(tt.argmax()) + 1
    n_stations = len(tt)

    balance_rate = total_time / (bottleneck * n_stations)
    idle_rate = 1 - balance_rate

    # 理论最少工位数（向上取整）
    min_stations = int(np.ceil(total_time / takt_time))

    # 日产能（受瓶颈限制）
    capacity = int(available_time / bottleneck)

    # 各工位与节拍的差
    diff = tt - takt_time

    return {
        "节拍(秒)": round(takt_time, 2),
        "工位数": n_stations,
        "总作业时间(秒)": round(total_time, 2),
        "瓶颈工位": bottleneck_idx,
        "瓶颈时间(秒)": round(bottleneck, 2),
        "平衡率": f"{balance_rate:.1%}",
        "失衡率": f"{idle_rate:.1%}",
        "理论最少工位数": min_stations,
        "日产能(件)": capacity,
        "超节拍工位": [i + 1 for i, d in enumerate(diff) if d > 0],
        "各工位与节拍差": [round(d, 2) for d in diff],
    }

# 示例
task_times = [32, 41, 28, 25, 39]
res = line_balance(task_times, demand=800, available_time=27600)
for k, v in res.items():
    print(f"{k}: {v}")

def simulate_improvement(task_times, improvements, demand=800, available_time=27600):
    """模拟改善后的平衡效果
    improvements: {工位索引(0起): 改善后的时间}
    """
    tt = np.array(task_times, dtype=float).copy()
    for idx, new_time in improvements.items():
        tt[idx] = new_time
    return line_balance(tt.tolist(), demand=demand, available_time=available_time)

# 假设：W2 通过布局改善降至 30.7，W5 通过电动工具降至 35，
# 并把 1.5 秒作业从 W5 转到 W4
before = line_balance([32, 41, 28, 25, 39], demand=800)
after = line_balance([32, 30.7, 34, 26.5, 33.5], demand=800)

print("\n改善前:", before["平衡率"], "日产能", before["日产能(件)"])
print("改善后:", after["平衡率"], "日产能", after["日产能(件)"])

def assign_stations(elements, takt_time):
    """LPT 启发式工位分配
    elements: [(要素名, 时间, 前置要素列表)]
    返回：各工位包含的要素
    """
    # 按时间降序排列
    sorted_elem = sorted(elements, key=lambda x: -x[1])
    stations = [[]]

    for name, time_val, preds in sorted_elem:
        placed = False
        for st in stations:
            station_time = sum(e[1] for e in st)
            assigned_names = [e[0] for e in st]
            # 检查前置约束 & 容量约束
            if all(p in assigned_names for p in preds) and station_time + time_val <= takt_time:
                st.append((name, time_val, preds))
                placed = True
                break
        if not placed:
            stations.append([(name, time_val, preds)])

    return stations

# 示例
elements = [
    ("A1 取外壳", 4.2, []),
    ("A2 贴标签", 5.8, ["A1 取外壳"]),
    ("A3 装主板", 8.5, ["A1 取外壳"]),
    ("A4 锁螺丝", 9.3, ["A3 装主板"]),
    ("A5 通电测试", 6.1, ["A4 锁螺丝"]),
    ("A6 外观检查", 1.5, ["A5 通电测试"]),
]
stations = assign_stations(elements, takt_time=20)
for i, st in enumerate(stations, 1):
    t = sum(e[1] for e in st)
    print(f"工位{i} (负荷 {t:.1f}s): {[e[0] for e in st]}")

import simpy
import random
import statistics

class ProductionLine:
    def __init__(self, env, num_machines, process_time_mean, process_time_std):
        self.env = env
        self.machines = simpy.Resource(env, num_machines)
        self.pt_mean = process_time_mean
        self.pt_std = process_time_std
        self.wait_times = []
        self.throughput = 0

    def process(self, item):
        """单个工件的加工流程"""
        arrive = self.env.now
        with self.machines.request() as req:
            yield req
            self.wait_times.append(self.env.now - arrive)
            pt = max(0.1, random.gauss(self.pt_mean, self.pt_std))
            yield self.env.timeout(pt)
            self.throughput += 1

def source(env, line, arrival_interval, num_items):
    """工件发生器"""
    for i in range(num_items):
        env.process(line.process(f"Item-{i}"))
        yield env.timeout(random.expovariate(1.0 / arrival_interval))

def run_simulation(num_machines=3, pt_mean=10, pt_std=2,
                   arrival_interval=4, num_items=500, sim_time=2000):
    random.seed(42)
    env = simpy.Environment()
    line = ProductionLine(env, num_machines, pt_mean, pt_std)
    env.process(source(env, line, arrival_interval, num_items))
    env.run(until=sim_time)

    utilization = (pt_mean * line.throughput) / (num_machines * sim_time)
    return {
        "吞吐量": line.throughput,
        "平均等待时间": round(statistics.mean(line.wait_times), 2),
        "最大等待时间": round(max(line.wait_times), 2),
        "设备利用率": f"{utilization:.1%}",
    }

# 对比不同设备数量的效果
for n in [2, 3, 4, 5]:
    r = run_simulation(num_machines=n)
    print(f"设备数={n}: 吞吐={r['吞吐量']}, 平均等待={r['平均等待时间']}s, "
          f"利用率={r['设备利用率']}")

import pulp

def production_planning(products, resources, profit, usage, capacity):
    """产品组合优化
    products: 产品列表
    resources: 资源（工序）列表
    profit: {产品: 单位利润}
    usage: {产品: {资源: 单位消耗}}
    capacity: {资源: 可用能力}
    """
    prob = pulp.LpProblem("ProductMix", pulp.LpMaximize)
    x = {p: pulp.LpVariable(f"x_{p}", lowBound=0, cat="Integer") for p in products}

    # 目标函数：最大化利润
    prob += pulp.lpSum(profit[p] * x[p] for p in products)

    # 约束：资源能力限制
    for r in resources:
        prob += pulp.lpSum(usage[p][r] * x[p] for p in products) <= capacity[r]

    prob.solve(pulp.PULP_CBC_CMD(msg=False))

    return {p: int(x[p].value()) for p in products}, pulp.value(prob.objective)

# 示例
products = ["A", "B", "C"]
resources = ["机加", "装配", "检验"]
profit = {"A": 120, "B": 180, "C": 150}
usage = {
    "A": {"机加": 2, "装配": 3, "检验": 1},
    "B": {"机加": 4, "装配": 3, "检验": 2},
    "C": {"机加": 3, "装配": 4, "检验": 1},
}
capacity = {"机加": 800, "装配": 900, "检验": 400}

plan, total = production_planning(products, resources, profit, usage, capacity)
print("最优生产计划:", plan)
print("最大利润:", round(total, 2))

import matplotlib.pyplot as plt
import numpy as np

def plot_yamazumi(task_times, takt_time, labels=None):
    """山积图：工位负荷可视化"""
    n = len(task_times)
    if labels is None:
        labels = [f"W{i+1}" for i in range(n)]

    fig, ax = plt.subplots(figsize=(10, 5))
    colors = ["#d62728" if t > takt_time else "#2ca02c" for t in task_times]
    bars = ax.bar(labels, task_times, color=colors, alpha=0.85)

    ax.axhline(takt_time, color="black", linestyle="--", linewidth=1.5,
               label=f"节拍 {takt_time}s")

    balance = sum(task_times) / (max(task_times) * n)
    ax.set_title(f"产线平衡山积图（平衡率 {balance:.1%}）", fontsize=13)
    ax.set_ylabel("作业时间 (秒)")
    ax.legend()

    for bar, t in zip(bars, task_times):
        ax.text(bar.get_x() + bar.get_width()/2, t + 0.5,
                f"{t}", ha="center", fontsize=10)

    plt.tight_layout()
    plt.show()

plot_yamazumi([32, 41, 28, 25, 39], takt_time=34.5)

plt.rcParams["font.sans-serif"] = ["SimHei", "Microsoft YaHei"]
plt.rcParams["axes.unicode_minus"] = False