from datetime import datetime, date class OperatingModePredictor: """ 工况预测类（处理二维倒序数据，返回一维列表，仅保留各内嵌列表最新记录） 功能：根据输入的带时间序列的工况数据，推导每个监测点（point_id）的下一阶段工况 特性： 1. 输入为二维列表，每个内嵌列表对应一个point_id，且为倒序排列（最新记录在索引0） 2. 输出为一维列表，仅保留每个内嵌列表的最新记录，新增工况推导结果字段 3. 推导规则：有等效映射返回新工况，无等效保留旧工况；切换下一工况以base_periods为准 4. 时间计算仅按日期（天）维度，忽略时分秒 5. 冬休场景：仅以冬休前上一有效工况为判断基准，切换规则与非冬休完全一致 6. 适配中英文括号、逗号、空格，内部标准化匹配，外部返回规范名称 """ def __init__(self): """初始化类，加载核心配置""" # 1. 基础工况配置（最终返回的规范名称，含所有新旧工况） self.base_periods = self._load_base_periods() # 2. 旧→新等效映射（优先返回新工况） self.old_to_new_map = self._load_old_to_new_map() # 3. 工况分组（同义工况归为同一分组，复用切换规则） self.condition_group = self._load_condition_group() # 4. 切换触发规则（沿用原逻辑，触发天数+目标工况） self.transition_rules = self._load_transition_rules() # 5. 冬休标识 self.winter_break_labels = {"冬休"} # 辅助映射：标准化名称→base_periods中的规范名称（用于最终返回） self.std_to_canonical = { self._standardize_name(name): name for name in self.base_periods.keys() } # 辅助映射：标准化名称→分组ID self.std_to_group = { self._standardize_name(name): group_id for name, group_id in self.condition_group.items() } def _standardize_name(self, name): """ 标准化工况名称（内部匹配用）：去空格、统一中英文符号 :param name: 原始工况名称 :return: 标准化后的名称 """ if not name: return "" # 去所有空格 std_name = name.replace(" ", "").strip() # 统一中英文符号 replace_map = { "（": "(", "）": ")", "，": ",", "。": ",", "~": "至", "；": ";", "：": ":", "＇": "'", "＂": "\"" } for old, new in replace_map.items(): std_name = std_name.replace(old, new) return std_name def _load_base_periods(self): """加载基础工况配置（最终返回的规范名称，无多余数字）""" return { # 路基工况（新工况优先） "路基或预压土填筑，连续填筑": 1, "路基或预压土填筑，两次填筑间隔时间较长": 7, "预压土或路基填筑完成，第1~3个月": 7, "预压土或路基填筑完成，第4~6个月": 14, "预压土或路基填筑完成，6个月以后": 30, "架桥机(运梁车)首次通过前": 1, "架桥机(运梁车)首次通过后，前3天": 1, "架桥机(运梁车)首次通过后": 7, "轨道板(道床)铺设后，第1至3个月": 14, "轨道板(道床)铺设后，第4至6个月": 30, "轨道板(道床)铺设后，6个月以后": 90, # 路基旧工况（保留，无等效则返回） "填筑或堆载，一般情况": 1, "填筑或堆载，两次填筑间隔时间较长情况": 7, "堆载预压或路基填筑完成，6个月以后": 30, "轨道板(道床)铺设后，第1个月": 14, "轨道板(道床)铺设后，第2至3个月": 30, "轨道板(道床)铺设后，3个月以后": 90, # 桥梁工况（新工况优先） "桥墩(台)地面处拆模后": 30, "敦身混凝土施工": 30, "预制梁桥，预制梁架设前": 1, "预制梁桥，预制梁架设后": 7, "现浇梁，浇筑前": 30, "现浇梁上部结构施工中": 1, "架桥机(运梁车)通过": 2, "桥梁主体工程完工后，第1至3个月": 7, "桥梁主体工程完工后，第4至6个月": 14, "桥梁主体工程完工后，6个月以后": 30, "轨道铺设，前": 30, "轨道铺设，后": 14, "轨道铺设完成后，第1个月": 14, "轨道铺设完成后，2至3个月": 30, "轨道铺设完成后，4至12个月": 90, "轨道铺设完成后，12个月以后": 180, # 桥梁旧工况（保留，无等效则返回） "墩台施工到一定高度": 30, "墩台混凝土施工": 30, "预制梁桥，架梁前": 30, "桥位施工桥梁，制梁前": 30, "桥位施工桥梁，上部结构施工中": 1, "桥梁主体工程完工后,第1至3个月": 7, "轨道铺设期间，前": 30, "轨道铺设期间，后": 14, # 隧道工况 "仰拱(底板)施工完成后，第1个月": 7, "仰拱(底板)施工完成后，第2至3个月": 14, "仰拱(底板)施工完成后，3个月以后": 30, "无砟轨道铺设后，第1至3个月": 30, "无砟轨道铺设后，4至12个月": 90, "无砟轨道铺设后，12个月以后": 180, # 特殊工况 "冬休": 0 } def _load_old_to_new_map(self): """加载旧→新等效映射（优先返回新工况）""" return { # 路基等效 "填筑或堆载，一般情况": "路基或预压土填筑，连续填筑", "填筑或堆载，两次填筑间隔时间较长情况": "路基或预压土填筑，两次填筑间隔时间较长", "预压土或路基填筑完成。第1~3个月": "预压土或路基填筑完成，第1~3个月", "堆载预压或路基填筑完成，6个月以后": "预压土或路基填筑完成，6个月以后", "轨道板(道床)铺设后，第1个月": "轨道板(道床)铺设后，第1至3个月", "轨道板(道床)铺设后，第2至3个月": "轨道板(道床)铺设后，第4至6个月", "轨道板(道床)铺设后，3个月以后": "轨道板(道床)铺设后，6个月以后", # 桥梁等效 "墩台施工到一定高度": "桥墩(台)地面处拆模后", "墩台混凝土施工": "敦身混凝土施工", "桥位施工桥梁，制梁前": "现浇梁，浇筑前", "桥位施工桥梁，上部结构施工中": "现浇梁上部结构施工中", "桥梁主体工程完工后,第1至3个月": "桥梁主体工程完工后，第1至3个月", "轨道铺设期间，前": "轨道铺设，前", "轨道铺设期间，后": "轨道铺设，后" } def _load_condition_group(self): """加载工况分组（同义工况归为同一分组）""" group_map = { # 路基分组 "路基或预压土填筑，连续填筑": "DZ_CONTINUE", "路基或预压土填筑，两次填筑间隔时间较长": "DZ_INTERVAL", "预压土或路基填筑完成，第1~3个月": "DZ_FINISH_1_3", "预压土或路基填筑完成，第4~6个月": "DZ_FINISH_4_6", "预压土或路基填筑完成，6个月以后": "DZ_FINISH_AFTER_6", "架桥机(运梁车)首次通过前": "JQJ_FIRST_BEFORE", "架桥机(运梁车)首次通过后，前3天": "JQJ_FIRST_AFTER_3D", "架桥机(运梁车)首次通过后": "JQJ_FIRST_AFTER", "轨道板(道床)铺设后，第1至3个月": "GDB_FINISH_1_3", "轨道板(道床)铺设后，第4至6个月": "GDB_FINISH_4_6", "轨道板(道床)铺设后，6个月以后": "GDB_FINISH_AFTER_6", # 路基旧工况分组（复用新工况分组） "填筑或堆载，一般情况": "DZ_CONTINUE", "填筑或堆载，两次填筑间隔时间较长情况": "DZ_INTERVAL", "堆载预压或路基填筑完成，6个月以后": "DZ_FINISH_AFTER_6", "轨道板(道床)铺设后，第1个月": "GDB_FINISH_1_3", "轨道板(道床)铺设后，第2至3个月": "GDB_FINISH_4_6", "轨道板(道床)铺设后，3个月以后": "GDB_FINISH_AFTER_6", # 桥梁分组 "桥墩(台)地面处拆模后": "STATIC", "敦身混凝土施工": "STATIC", "预制梁桥，架梁前": "STATIC", "预制梁桥，预制梁架设前": "YZLQ_BEFORE_JS", "预制梁桥，预制梁架设后": "YZLQ_AFTER_JS", "现浇梁，浇筑前": "STATIC", "现浇梁上部结构施工中": "STATIC", "架桥机(运梁车)通过": "STATIC", "桥梁主体工程完工后，第1至3个月": "QL_ZHUTI_1_3", "桥梁主体工程完工后，第4至6个月": "QL_ZHUTI_4_6", "桥梁主体工程完工后，6个月以后": "QL_ZHUTI_AFTER_6", "轨道铺设，前": "STATIC", "轨道铺设，后": "STATIC", "轨道铺设完成后，第1个月": "GD_FINISH_1", "轨道铺设完成后，2至3个月": "GD_FINISH_2_3", "轨道铺设完成后，4至12个月": "GD_FINISH_4_12", "轨道铺设完成后，12个月以后": "GD_FINISH_AFTER_12", # 桥梁旧工况分组（复用新工况分组） "墩台施工到一定高度": "STATIC", "墩台混凝土施工": "STATIC", "桥位施工桥梁，制梁前": "STATIC", "桥位施工桥梁，上部结构施工中": "STATIC", "桥梁主体工程完工后,第1至3个月": "QL_ZHUTI_1_3", "轨道铺设期间，前": "STATIC", "轨道铺设期间，后": "STATIC", # 隧道分组 "仰拱(底板)施工完成后，第1个月": "YG_DIBAN_1", "仰拱(底板)施工完成后，第2至3个月": "YG_DIBAN_2_3", "仰拱(底板)施工完成后，3个月以后": "YG_DIBAN_AFTER_3", "无砟轨道铺设后，第1至3个月": "WZGD_1_3", "无砟轨道铺设后，4至12个月": "WZGD_4_12", "无砟轨道铺设后，12个月以后": "WZGD_AFTER_12", # 特殊工况 "冬休": "STATIC" } return group_map def _load_transition_rules(self): """加载切换触发规则（沿用原逻辑，以base_periods为准）""" return { # 路基切换规则 "DZ_FINISH_1_3": {"trigger_days": 90, "next": ["预压土或路基填筑完成，第4~6个月"]}, "DZ_FINISH_4_6": {"trigger_days": 90, "next": ["预压土或路基填筑完成，6个月以后"]}, "DZ_FINISH_AFTER_6": {"trigger_days": None, "next": None}, "JQJ_FIRST_BEFORE": {"trigger_days": 1, "next": ["架桥机(运梁车)首次通过后，前3天"]}, "JQJ_FIRST_AFTER_3D": {"trigger_days": 3, "next": ["架桥机(运梁车)首次通过后"]}, "JQJ_FIRST_AFTER": {"trigger_days": None, "next": None}, "GDB_FINISH_1_3": {"trigger_days": 30, "next": ["轨道板(道床)铺设后，第4至6个月"]}, "GDB_FINISH_4_6": {"trigger_days": 30, "next": ["轨道板(道床)铺设后，6个月以后"]}, "GDB_FINISH_AFTER_6": {"trigger_days": None, "next": None}, # 桥梁切换规则 "YZLQ_BEFORE_JS": {"trigger_days": 1, "next": ["架桥机(运梁车)通过"]}, "YZLQ_AFTER_JS": {"trigger_days": 7, "next": ["桥梁主体工程完工后，第1至3个月"]}, "QL_ZHUTI_1_3": {"trigger_days": 90, "next": ["桥梁主体工程完工后，第4至6个月"]}, "QL_ZHUTI_4_6": {"trigger_days": 90, "next": ["桥梁主体工程完工后，6个月以后"]}, "QL_ZHUTI_AFTER_6": {"trigger_days": None, "next": None}, "GD_FINISH_1": {"trigger_days": 30, "next": ["轨道铺设完成后，2至3个月"]}, "GD_FINISH_2_3": {"trigger_days": 60, "next": ["轨道铺设完成后，4至12个月"]}, "GD_FINISH_4_12": {"trigger_days": 240, "next": ["轨道铺设完成后，12个月以后"]}, "GD_FINISH_AFTER_12": {"trigger_days": None, "next": None}, # 隧道切换规则 "YG_DIBAN_1": {"trigger_days": 30, "next": ["仰拱(底板)施工完成后，第2至3个月"]}, "YG_DIBAN_2_3": {"trigger_days": 60, "next": ["仰拱(底板)施工完成后，3个月以后"]}, "YG_DIBAN_AFTER_3": {"trigger_days": None, "next": None}, "WZGD_1_3": {"trigger_days": 90, "next": ["无砟轨道铺设后，4至12个月"]}, "WZGD_4_12": {"trigger_days": 240, "next": ["无砟轨道铺设后，12个月以后"]}, "WZGD_AFTER_12": {"trigger_days": None, "next": None}, # 静态分组（无切换） "DZ_CONTINUE": {"trigger_days": None, "next": None}, "DZ_INTERVAL": {"trigger_days": None, "next": None}, "STATIC": {"trigger_days": None, "next": None} } def _parse_to_date(self, time_str): """解析时间字符串为date对象（仅保留年月日）""" if not time_str: return None try: dt = datetime.strptime(str(time_str).strip(), "%Y-%m-%d %H:%M:%S") return dt.date() except ValueError: return None def _get_time_statistics(self, data, target_workinfo): """从倒序数据中提取目标工况的时间统计信息""" # 筛选目标工况的有效记录 target_records = [ d for d in data if self._standardize_name(d.get("workinfoname")) == self._standardize_name(target_workinfo) and d.get("workinfoname") not in self.winter_break_labels ] if not target_records: return None, 0, 0 # 解析日期 target_dates = [] for item in target_records: d = self._parse_to_date(item.get("MTIME_W")) if d: target_dates.append(d) if not target_dates: return None, 0, 0 # 计算时间差（倒序数据：最新在0，最早在-1） last_date = target_dates[0] first_date = target_dates[-1] cumulative_days = (last_date - first_date).days today = date.today() days_to_today = (today - first_date).days if first_date else 0 return first_date, cumulative_days, days_to_today def _get_pre_winter_break_workinfo(self, inner_data_list): """提取冬休前的上一个有效工况""" if not inner_data_list: return None # 倒序遍历，跳过冬休，找第一个有效工况 for record in inner_data_list: current_work = record.get("workinfoname") if current_work and current_work not in self.winter_break_labels: # 优先返回等效新工况 return self.old_to_new_map.get(current_work, current_work) return None def _validate_point_id(self, inner_list): """校验内嵌列表内point_id是否一致""" if not inner_list: return None base_point_id = inner_list[0].get("point_id") for item in inner_list: if item.get("point_id") != base_point_id: return None return base_point_id def predict(self, data_2d_list): """ 公有核心方法：执行工况预测 :param data_2d_list: 二维倒序数据列表，格式 [[{},{},{}], [{},{},{}]] :return: 一维结果列表，格式 [{}, {}, {}] """ final_result = [] for inner_data_list in data_2d_list: if not isinstance(inner_data_list, list) or len(inner_data_list) == 0: continue # 取最新记录并复制（避免修改原数据） latest_record = inner_data_list[0].copy() point_id = self._validate_point_id(inner_data_list) # 校验point_id if not point_id: latest_record.update({ "status": "fail", "current_workinfo": None, "first_measure_date": None, "days_from_first_to_today": None, "next_workinfo": None, "judge_based_workinfo": None, "error_msg": "point_id不一致或缺失" }) final_result.append(latest_record) continue # 获取当前工况并标准化 current_workinfo = latest_record.get("workinfoname") if not current_workinfo or self._standardize_name(current_workinfo) not in self.std_to_canonical: latest_record.update({ "status": "fail", "current_workinfo": None, "first_measure_date": None, "days_from_first_to_today": None, "next_workinfo": None, "judge_based_workinfo": None, "error_msg": "工况无效或缺失" }) final_result.append(latest_record) continue # 冬休逻辑：取冬休前工况作为判断基准 if current_workinfo in self.winter_break_labels: judge_based_workinfo = self._get_pre_winter_break_workinfo(inner_data_list) if not judge_based_workinfo: latest_record.update({ "status": "fail", "current_workinfo": current_workinfo, "first_measure_date": None, "days_from_first_to_today": None, "next_workinfo": None, "judge_based_workinfo": None, "error_msg": "冬休前未找到有效工况" }) final_result.append(latest_record) continue else: # 非冬休：优先使用等效新工况作为判断基准 judge_based_workinfo = self.old_to_new_map.get(current_workinfo, current_workinfo) # 计算时间统计信息 first_dt, cumulative_days, days_to_today = self._get_time_statistics( inner_data_list, judge_based_workinfo ) first_measure_date = first_dt.strftime("%Y-%m-%d") if first_dt else None # 推导下一工况（以base_periods中的名称为准） std_judge_work = self._standardize_name(judge_based_workinfo) group_id = self.std_to_group.get(std_judge_work, "STATIC") rule = self.transition_rules.get(group_id, {}) trigger_days = rule.get("trigger_days") next_candidates = rule.get("next", []) if trigger_days is not None and cumulative_days >= trigger_days and next_candidates: # 切换工况：取base_periods中的规范名称 next_workinfo = next_candidates[0] else: # 不切换：有等效则返回新工况，无则返回旧工况 next_workinfo = self.old_to_new_map.get(judge_based_workinfo, judge_based_workinfo) # 组装最终结果 latest_record.update({ "status": "success", "current_workinfo": current_workinfo, # 保留原始输入工况 "judge_based_workinfo": judge_based_workinfo, # 实际判断用的工况（新工况优先） "first_measure_date": first_measure_date, "days_from_first_to_today": days_to_today, "next_workinfo": next_workinfo, # 推导结果：有等效返回新工况，无则返回旧工况 "point_id": point_id, "error_msg": "" }) final_result.append(latest_record) return final_result