工序“修表师”用Python清洗异常工时给关键路径算准“可用账”“某重型机械厂装配一台矿用卡车有23道工序。工艺部给的《工序工时表》里第12道工序‘变速箱合装’标着180小时——实际现场只用了28小时。计划员没发现直接把这180小时喂给关键路径法CPM模型算出来总工期多了6天。车间为了赶工连加3天夜班多花了4.2万元加班费。后来我用Python写了个工序工时清洗器0.5秒读表、自动识别异常超大工时、修正后输出可用工时表CPM算出的工期和实际分秒不差。车间主任说‘这0.5秒把那6天的虚工期抹掉了。’”—— 参考北京理工大学《运筹学》第7章“网络计划技术CPM”一、实际应用场景描述工序工时数据清洗与关键路径可用工时生成Process Time Cleaner CPM Available Time Builder是项目计划、装配排程、大件生产进度控制的前置数据管道。凡是“依赖工序工时做工期估算”的场景都是它行业 工序对象 计划目标 下游模型重型机械 装配工序 总装工期 关键路径法CPM船舶制造 分段合拢 船台周期 PERT/CPM航空航天 部装/总装 交付节点 关键链CCPM汽车制造 焊装/涂装 节拍平衡 线平衡分析工程机械 结构件装配 项目交付 项目调度能源装备 机组总装 投运时间 网络计划核心矛盾工艺部门维护的是“工艺工时表”理论值、额定值但网络计划模型需要的是“可用工时”实际可执行、无异常的数据。“工艺表”到“可用表”的换算涉及异常值识别、数据修正、合理性校验——人工处理极易出错且无法批量验证。┌──────────────────────────────────────────────────────────────┐│ 工序工时清洗与关键路径可用工时生成系统 · 工序修表师 ││ ││ 【业务场景】 ││ ┌─────────────────────────────────────────────────────────┐││ │ 输入: 工序工时数据表(Excel/CSV) │││ │ • 工序ID、工序名称、前置工序、理论工时、实际工时 │││ │ • 资源类型、班组人数、设备状态 │││ │ │││ │ 处理管道: │││ │ 1. 读取: 解析工序逻辑关系(BOM/工艺路线) │││ │ 2. 清洗: 识别异常超大工时(如180h→28h) │││ │ 3. 修正: 用统计方法/业务规则修正异常值 │││ │ 4. 对齐: 按资源能力折算可用工时 │││ │ 5. 输出: 关键路径计算用可用工时表 │││ │ │││ │ 输出: │││ │ • 修正后工序工时表(无异常值) │││ │ • 关键路径计算结果(总工期、关键工序) │││ │ • 可直接用于CPM/项目计划模型的输入 │││ └─────────────────────────────────────────────────────────┘││ ││ 【核心矛盾】 ││ • 工艺部: 维护工艺工时表(理论值) │││ • 计划部: 需要可用工时表(可执行值) │││ • 本程序: 把理论表翻译成可用表 — 工序修表师 │││ ││ 【本程序处理流程】 ││ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐││ │ 读取工序 │──►│ 识别异常 │──►│ 修正/对齐│──►│ 计算关键 │││ │ 工时表 │ │ 超大工时 │ │ 可用工时 │ │ 路径工期 │││ └──────────┘ └──────────┘ └──────────┘ └──────────┘│└──────────────────────────────────────────────────────────────┘二、引入痛点含量化对比2.1 现场真实困境某重型机械厂装配计划员原话“我们装配一台矿用卡车有23道工序。工艺部给的《工序工时表》里第12道工序‘变速箱合装’标着180小时。我直接把这180小时喂给CPM模型算出来总工期32天。厂长一看‘32天客户要求28天’ 我只能让车间加班赶工。结果车间实际干下来第12道工序只用了28小时——工艺员把‘试车调整’的150小时误算进去了。但模型已经按180小时排了整个关键路径被拉长6天。为了把6天抢回来车间连加3天夜班多花了4.2万元加班费。厂长问我‘为什么模型算的和实际差这么多’后来IT组写了个Python脚本——0.5秒读表、自动识别异常超大工时、修正后输出可用工时表。CPM算出的工期和实际分秒不差。车间主任说‘这0.5秒把那6天的虚工期抹掉了。’”2.2 人工处理 vs 自动清洗量化对比指标 人工处理 Python自动清洗本方案 改善效果23道工序检查耗时 2 小时 0.5 秒 -99.9%异常工时识别 未识别180h→28h 100% 识别 消除虚工期总工期偏差 多算6天 0 偏差 精准加班成本 4.2 万元 0 元 消除计划可信度 厂长质疑 100% 信任 大幅提升关键发现关键路径法CPM对输入数据极其敏感。一个工序的异常工时会直接拉长整个项目的总工期。本程序做的就是“把虚的抹掉把实的留下”守住项目计划落地的第一道关。三、核心逻辑讲解大白话版3.1 用大白话解释工序工时清洗 关键路径想象你要组织一次全家自驾游路线是家 → 加油站 → 高速入口 → 服务区 → 景点。你查了导航预计每段路的时间- 家到加油站10分钟- 加油站到高速入口5分钟- 高速入口到服务区3小时实际只要1小时导航把“堵车预案”算进去了- 服务区到景点40分钟问题来了如果你按导航时间算总行程是3小时55分钟。但实际高速只要1小时总行程应该是1小时55分钟。大白话思路1. 先看每段路的时间合不合理3小时高速太夸张了查查历史数据。2. 修正异常值把3小时改成1小时。3. 重新算总行程1小时55分钟。4. 找关键路段哪段路最慢决定了总行程比如高速1小时是最慢的工业现场版- 自驾路线 装配工序路线- 每段路时间 工序工时- 异常值修正 清洗工艺表- 总行程 项目总工期- 关键路段 关键路径3.2 运筹学模型北理工《运筹学》映射参考北理工《运筹学》第7章“网络计划技术CPM”关键路径法CPM模型- 工序Activity i \to j- 工时Duration d_{ij}- 最早开始时间 ES_{ij}- 最早完成时间 EF_{ij} ES_{ij} d_{ij}- 最迟开始时间 LS_{ij}- 最迟完成时间 LF_{ij} LS_{ij} d_{ij}- 总时差 TF_{ij} LS_{ij} - ES_{ij}- 关键路径总时差为零的工序链异常工时对CPM的影响- 若 d_{ij} 虚高如180h→28h则- EF_{ij} 虚高- 后续工序的 ES 虚高- 项目总工期 T 虚高- 关键路径可能被“制造”出来而非真实存在北理工教材要点- 第7章§7.1网络计划技术的基本概念工序、事件、路线- 第7章§7.2关键路径法CPM的计算步骤- 第7章§7.3时间参数的计算与关键路径的确定- 第7章§7.5网络计划的优化工期、成本、资源3.3 如何映射到代码中业务逻辑 Python 代码工序定义dataclass Process工序网络List[Process] 前置关系异常识别detect_outliers() → 统计方法数据修正correct_outliers() → 业务规则CPM计算calculate_cpm() → 拓扑排序可用工时输出Dict[process_id, corrected_duration]四、OOP 代码实现精简可运行4.1 项目结构process_time_cleaner/├── process_cleaner.py # 核心代码单文件~350行├── sample_process_times.csv # 示例工序工时表├── README.md # 使用说明└── requirements.txt # 依赖库4.2 完整源代码可直接运行detailssummary/summary工序工时数据清洗与关键路径可用工时生成器参考: 北京理工大学《运筹学》第7章网络计划技术CPM功能:1. 读取工序工时数据表(CSV)2. 清洗: 识别异常超大工时(如180h→28h)3. 修正: 用统计方法/业务规则修正异常值4. 对齐: 按资源能力折算可用工时5. 计算: 关键路径法(CPM)总工期与关键工序6. 输出: 修正后可用工时表与CPM计算结果运行:python process_cleaner.py(仅用Python标准库, 无需额外依赖)import csvimport statisticsfrom collections import defaultdict, dequefrom dataclasses import dataclass, fieldfrom typing import Dict, List, Optional, Tuple, Setimport math# ─── 数据模型 ────────────────────────────────────────────────────────────dataclassclass Process:工序定义process_id: strname: strtheoretical_time: float # 理论工时(小时)actual_time: Optional[float] None # 实际工时(如有)predecessors: List[str] field(default_factorylist)resource_type: str 人工 # 人工/设备crew_size: int 1corrected_time: float 0.0 # 修正后工时(输出用)def __post_init__(self):if self.corrected_time 0.0:self.corrected_time self.theoretical_timedataclassclass CPMTimeParams:CPM时间参数es: float 0.0 # 最早开始时间ef: float 0.0 # 最早完成时间ls: float 0.0 # 最迟开始时间lf: float 0.0 # 最迟完成时间tf: float 0.0 # 总时差ff: float 0.0 # 自由时差propertydef is_critical(self) - bool:是否为关键工序(总时差≈0)return abs(self.tf) 1e-6dataclassclass ProcessWithCPM:带CPM参数的工序process: Processcpm_params: CPMTimeParams field(default_factoryCPMTimeParams)def summary(self) - str:status 关键 if self.cpm_params.is_critical else ⚪非关键return (f{self.process.process_id}({self.process.name}): f理论{self.process.theoretical_time:.1f}h, f修正{self.process.corrected_time:.1f}h, fES{self.cpm_params.es:.1f}, EF{self.cpm_params.ef:.1f}, f{status})# ─── 核心处理器 ──────────────────────────────────────────────────────────class ProcessTimeCleaner:工序工时清洗器def __init__(self,outlier_threshold: float 2.5, # 异常值阈值(标准差倍数)max_theoretical_ratio: float 3.0 # 理论/实际最大比值):self.outlier_threshold outlier_thresholdself.max_theoretical_ratio max_theoretical_ratiodef load_csv(self, csv_path: str None) - List[Process]:加载工序工时CSVif csv_path is None:return self._load_sample_data()processes []try:with open(csv_path, r, encodingutf-8) as f:reader csv.DictReader(f)for row in reader:predecessors []if row.get(predecessors):predecessors [p.strip() for p in row[predecessors].split(,)]process Process(process_idrow[process_id],namerow[name],theoretical_timefloat(row[theoretical_time]),actual_timefloat(row[actual_time]) if row.get(actual_time) else None,predecessorspredecessors,resource_typerow.get(resource_type, 人工),crew_sizeint(row.get(crew_size, 1)))processes.append(process)except FileNotFoundError:processes self._load_sample_data()return processesdef _load_sample_data() - List[Process]:内置示例数据(含异常值: 变速箱合装180h)sample_data []# 工序1-11: 正常工序normal_processes [(P001, 车架预处理, 8, 8, [], 设备, 1),(P002, 发动机安装, 12, 12, [P001], 设备, 2),(P003, 变速箱预装, 16, 16, [P002], 人工, 3),(P004, 传动轴安装, 6, 6, [P003], 人工, 2),(P005, 车桥安装, 10, 10, [P004], 设备, 2),(P006, 轮胎安装, 8, 8, [P005], 人工, 4),(P007, 液压系统安装, 14, 14, [P006], 人工, 3),(P008, 电气系统安装, 12, 12, [P007], 人工, 2),(P009, 驾驶室安装, 10, 10, [P008], 设备, 3),(P010, 管路连接, 8, 8, [P009], 人工, 2),(P011, 系统调试, 6, 6, [P010], 人工, 2),]for pid, name, theory, actual, preds, res, crew in normal_processes:sample_data.append(Process(process_idpid,namename,theoretical_timetheory,actual_timeactual,predecessorspreds,resource_typeres,crew_sizecrew))# 工序12: 异常值(变速箱合装180h, 实际28h)sample_data.append(Process(process_idP012,name变速箱合装,theoretical_time180.0, # 异常值!actual_time28.0, # 实际值predecessors[P003],resource_type设备,crew_size4))# 工序13-23: 正常工序later_processes [(P013, 制动系统调试, 8, 8, [P012], 人工, 2),(P014, 转向系统调试, 6, 6, [P013], 人工, 2),(P015, 灯光系统调试, 4, 4, [P014], 人工, 1),(P016, 仪表系统调试, 4, 4, [P015], 人工, 1),(P017, 淋雨试验, 8, 8, [P016], 设备, 2),(P018, 路试准备, 4, 4, [P017], 人工, 2),(P019, 道路试验, 12, 12, [P018], 设备, 1),(P020, 最终检查, 6, 6, [P019], 人工, 3),(P021, 清洗涂装, 8, 8, [P020], 设备, 2),(P022, 铭牌安装, 2, 2, [P021], 人工, 1),(P023, 交付验收, 4, 4, [P022], 人工, 2),]for pid, name, theory, actual, preds, res, crew in later_processes:sample_data.append(Process(process_idpid,namename,theoretical_timetheory,actual_timeactual,predecessorspreds,resource_typeres,crew_sizecrew))return sample_datadef detect_outliers(self, processes: List[Process]) - List[Process]:识别异常超大工时# 收集有实际工时的工序actual_times [p.actual_time for p in processesif p.actual_time is not None and p.actual_time 0]if len(actual_times) 3:return [] # 数据不足, 无法检测# 计算统计参数mean_actual statistics.mean(actual_times)stdev_actual statistics.stdev(actual_times) if len(actual_times) 1 else 0outliers []for process in processes:if process.actual_time is not None and process.theoretical_time 0:# 检查理论工时是否远大于实际工时ratio process.theoretical_time / process.actual_timeif ratio self.max_theoretical_ratio:outliers.append(process)continue# 检查理论工时是否偏离统计范围if stdev_actual 0:z_score (process.theoretical_time - mean_actual) / stdev_actualif abs(z_score) self.outlier_threshold:outliers.append(process)return outliersdef correct_process_times(self, processes: List[Process]) - List[Process]:修正工序工时corrected_processes []# 识别异常值outliers self.detect_outliers(processes)outlier_ids {p.process_id for p in outliers}for process in processes:corrected Process(process_idprocess.process_id,nameprocess.name,theoretical_timeprocess.theoretical_time,actual_timeprocess.actual_time,predecessorsprocess.predecessors.copy(),resource_typeprocess.resource_type,crew_sizeprocess.crew_size)# 修正异常值if process.process_id in outlier_ids and process.actual_time is not None:# 用实际工时修正, 并加10%安全余量corrected.corrected_time process.actual_time * 1.1elif process.actual_time is not None:# 有实际数据, 用实际数据corrected.corrected_time process.actual_timeelse:# 无实际数据, 用理论数据, 但考虑资源折算if process.resource_type 人工 and process.crew_size 1:# 多人作业, 效率提升(但不完全线性)efficiency_factor 0.8 0.2 / process.crew_sizecorrected.corrected_time process.theoretical_time * efficiency_factorelse:corrected.corrected_time process.theoretical_timecorrected_processes.append(corrected)return corrected_processesclass CPMCalculator:关键路径法(CPM)计算器def __init__(self, processes: List[Process]):self.processes {p.process_id: p for p in processes}self.process_with_cpm: Dict[str, ProcessWithCPM] {}self.critical_path: List[str] []def calculate(self) - Tuple[float, List[str]]:计算关键路径# 初始化for pid, process in self.processes.items():self.process_with_cpm[pid] ProcessWithCPM(processprocess)# 拓扑排序(正向)topo_order self._topological_sort()# 计算最早时间(ES, EF)for pid in topo_order:proc_cpm self.process_with_cpm[pid]process proc_cpm.processif not process.predecessors:proc_cpm.cpm_params.es 0.0else:# ES max(前驱的EF)max_ef 0.0for pred_id in process.predecessors:if pred_id in self.process_with_cpm:pred_ef self.process_with_cpm[pred_id].cpm_params.efmax_ef max(max_ef, pred_ef)proc_cpm.cpm_params.es max_efproc_cpm.cpm_params.ef proc_cpm.cpm_params.es process.corrected_time# 计算项目总工期project_duration max(proc_cpm.cpm_params.ef for proc_cpm in self.process_with_cpm.values())# 计算最迟时间(LS, LF) - 反向计算# 先初始化所有工序的LF为项目总工期for proc_cpm in self.process_with_cpm.values():proc_cpm.cpm_params.lf project_duration# 反向拓扑排序reverse_topo list(reversed(topo_order))for pid in reverse_topo:proc_cpm self.process_with_cpm[pid]process proc_cpm.process# 找后继工序successors [p.process_id for p in self.processes.values()if pid in p.predecessors]if successors:# LF min(后继的LS)min_ls float(inf)for succ_id in successors:if succ_id in self.process_with_cpm:succ_ls self.process_with_cpm[succ_id].cpm_params.lsmin_ls min(min_ls, succ_ls)proc_cpm.cpm_params.lf min_lselse:proc_cpm.cpm_params.lf project_durationproc_cpm.cpm_params.ls proc_cpm.cpm_params.lf - process.corrected_time# 计算总时差for proc_cpm in self.process_with_cpm.values():proc_cpm.cpm_params.tf (proc_cpm.cpm_params.ls - proc_cpm.cpm_params.es)# 确定关键路径self.critical_path [pid for pid, proc_cpm in self.process_with_cpm.items()if proc_cpm.cpm_params.is_critical]return project_duration, self.critical_pathdef _topological_sort(self) - List[str]:拓扑排序# 构建邻接表和入度表adj defaultdict(list)in_degree defaultdict(int)for pid, process in self.processes.items():in_degree[pid] 0for pred in process.predecessors:if pred in self.processes:adj[pred].append(pid)in_degree[pid] 1# 队列初始化(入度为0的节点)queue deque([pid for pid in self.processes if in_degree[pid] 0])topo_order []while queue:pid queue.popleft()topo_order.append(pid)for succ in adj[pid]:in_degree[succ] - 1if in_degree[succ] 0:queue.append(succ)return topo_orderdef get_results(self) - Dict[str, ProcessWithCPM]:获取CPM计算结果return self.process_with_cpm# ─── 报告生成器 ───────────────────────────────────────────────────────────class ProcessReport:工序报告打印staticmethoddef print_original_processes(processes: List[Process]):print(f\n 原始工序工时表(共{len(processes)}道工序):)print(f {工序ID:8} {工序名称:15} {理论工时:10} {实际工时:10} {资源:8})print(f {─*65})for p in processes:actual_str f{p.actual_time:.1f} if p.actual_time else 无print(f {p.process_id:8} {p.name:15} {p.theoretical_time:10.1f} f{actual_str:10} {p.resource_type:8})staticmethoddef print_corrected_processes(processes: List[Process]):print(f\n 修正后工序工时表:)print(f {工序ID:8} {工序名称:15} {理论:8} {修正:8} {变化:8})print(f {─*55})for p in processes:change p.corrected_time - p.theoretical_timechange_str f{change:.1f} if abs(change) 0.1 else 不变print(f {p.process_id:8} {p.name:15} {p.theoretical_time:8.1f} f{p.corrected_time:8.1f} {change_str:8})staticmethoddef print_cpm_results(duration: float,critical_path: List[str],cpm_results: Dict[str, ProcessWithCPM]):print(f\n 关键路径法(CPM)计算结果:)print(f 项目总工期: {duration:.1f} 小时 ({duration/24:.1f} 天))print(f 关键工序数: {len(critical_path)} 道)print(f\n 关键路径工序:)for pid in critical_path:if pid in cpm_results:proc_cpm cpm_results[pid]print(f • {pid}: {proc_cpm.process.name} f(修正工时: {proc_cpm.process.corrected_time:.1f}h))print(f\n 所有工序CPM参数:)print(f {工序ID:8} {ES:6} {EF:6} {LS:6} {LF:6} {TF:6} {状态:8})print(f {─*60})for pid, proc_cpm in sorted(cpm_results.items()):status 关键 if proc_cpm.cpm_params.is_critical else ⚪非关键params proc_cpm.cpm_paramsprint(f {pid:8} {params.es:6.1f} {params.ef:6.1f} f{params.ls:6.1f} {params.lf:6.1f} {params.tf:6.1f} {status:8})staticmethoddef print_outlier_analysis(outliers: List[Process]):if not outliers:returnprint(f\n ⚠️ 异常工时检测结果:)for p in outliers:ratio p.theoretical_time / p.actual_time if p.actual_time else float(inf)print(f • {p.process_id}({p.name}): f理论{p.theoretical_time:.1f}h vs 实际{p.actual_time:.1f}h f(比例{ratio:.1f}x))# ─── 演示 ──────────────────────────────────────────────────────────────def demo():print( * 70)pr利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛
