from datetime import datetime, date class OperatingModePredictor: """ 工况预测类（处理二维倒序数据，返回一维列表，仅保留各内嵌列表最新记录） 功能：根据输入的带时间序列的工况数据，推导每个监测点（point_id）的下一阶段工况 特性： 1. 输入为二维列表，每个内嵌列表对应一个point_id，且为倒序排列（最新记录在索引0） 2. 输出为一维列表，仅保留每个内嵌列表的最新记录，新增工况推导结果字段 3. 无需切换工况时，next_workinfo返回当前工况名称；需要切换时返回目标工况名称 4. 时间计算仅按日期（天）维度，忽略时分秒 5. 支持冬休场景：当前工况为冬休时，以冬休前上一个有效工况作为切换判断依据 """ def __init__(self): """初始化类，加载内置的工况配置、分组规则和切换触发规则""" # 基础工况配置（更新为最新的工况列表，重复key保留后面的值） self.base_periods = self._load_base_periods() # 工况分组（同步更新分组规则，匹配最新工况名称） self.condition_group = self._load_condition_group() # 工况切换触发规则（键：分组ID，值：触发天数+目标工况候选） self.group_transition_rules = self._load_group_transition_rules() # 冬休工况标识（仅保留“冬休”） self.winter_break_labels = self._load_winter_break_labels() def _load_base_periods(self): """加载基础工况配置（更新为最新工况列表，重复key自动保留后面的值）""" return { "路基或预压土填筑，连续填筑":1, "路基或预压土填筑，两次填筑间隔时间较长":7, "预压土或路基填筑完成，第1~3个月":7, "预压土或路基填筑完成，第4~6个月":14, "仰拱（底板）施工完成后，第1个月": 7, "预压土或路基填筑完成，6个月以后":30, "仰拱（底板）施工完成后，第2至3个月": 14, "仰拱（底板）施工完成后，3个月以后": 30, "仰拱(底板)施工完成后，第1个月": 7, # 重复key，保留此值 "仰拱(底板)施工完成后，第2至3个月": 14, # 重复key，保留此值 "仰拱(底板)施工完成后，3个月以后": 30, # 重复key，保留此值 "无砟轨道铺设后，第1至3个月": 30, # 重复key，保留此值 "无砟轨道铺设后，4至12个月": 90, # 重复key，保留此值 "无砟轨道铺设后，12个月以后": 180, # 重复key，保留此值 "墩台施工到一定高度": 30, "墩台混凝土施工": 30, "预制梁桥，架梁前": 30, # 重复key，保留此值 "预制梁桥，预制梁架设前": 1, # 重复key，保留此值 "预制梁桥，预制梁架设后": 7, "桥位施工桥梁，制梁前": 30, "桥位施工桥梁，上部结构施工中": 1, "架桥机(运梁车)通过": 2, # 重复key，保留此值（原7→2） "桥梁主体工程完工后,第1至3个月": 7, "桥梁主体工程完工后，第4至6个月": 14, # 重复key，保留此值 "桥梁主体工程完工后,6个月以后": 30, # 重复key，保留此值 "轨道铺设期间，前": 30, "轨道铺设期间，后": 14, "轨道铺设完成后，第1个月": 14, # 重复key，保留此值 "轨道铺设完成后，2至3个月": 30, # 重复key，保留此值 "轨道铺设完成后，4至12个月": 90, # 重复key，保留此值 "轨道铺设完成后，12个月以后": 180, # 重复key，保留此值 "铺路或堆载，一般情况": 1, "填筑或堆载，一般情况": 1, "铺路或堆载，沉降量突变情况": 1, "填筑或堆载，两次填筑间隔时间较长情况":3, "铺路或堆载，两次铺路间隔时间较长情况": 3, "堆载预压或路基填筑完成，第1至3个月":7, "堆载预压或路基填筑完成，第4至6个月": 14, "堆载预压或路基填筑完成，6个月以后": 30, "架桥机(运梁车) 首次通过前": 1, "架桥机(运梁车) 首次通过后，前3天": 1, "架桥机(运梁车) 首次通过后": 7, "轨道板(道床)铺设后，第1个月": 14, "轨道板(道床)铺设后，第2至3个月": 30, "轨道板(道床)铺设后，3个月以后": 90, "架桥机(运梁车)首次通过前": 1, "架桥机(运梁车)首次通过后，前3天": 1, "架桥机(运梁车)首次通过后": 7, "轨道板铺设前": 14, # 新增工况 "轨道板(道床)铺设后，第1至3个月": 14, # 新增工况 "轨道板(道床)铺设后，第4至6个月": 30, # 新增工况 "轨道板(道床)铺设后，6个月以后": 90, # 新增工况 "站场填方路基段填筑完成至静态验收": 14, # 新增工况 "桥墩(台)地面处拆模后": 30, # 新增工况 "敦身混凝土施工": 30, # 新增工况 "预制梁桥预制梁架设后": 7, # 新增工况 "现浇梁，浇筑前": 30, # 新增工况 "现浇梁上部结构施工中": 1, # 新增工况 "侨梁主体工程完工后，6个月以后": 30, # 新增工况（注意错别字） "轨道铺设，前": 30, # 新增工况 "轨道铺设，后": 14, # 新增工况 "仰拱(底板)施工完成后第，2至3个月": 14, # 新增工况（标点位置异常） "特殊地段隧道施工完成后至静态验收": 14, # 新增工况 "冬休": 0 # 保留冬休工况 } def _load_condition_group(self): """加载工况分组规则（同步更新，匹配最新工况名称）""" condition_group = { # 仰拱相关 "仰拱（底板）施工完成后，第1个月": "YG_DIBAN_1", "仰拱(底板)施工完成后，第1个月": "YG_DIBAN_1", "仰拱（底板）施工完成后，第2至3个月": "YG_DIBAN_2_3", "仰拱(底板)施工完成后，第2至3个月": "YG_DIBAN_2_3", "仰拱（底板）施工完成后，3个月以后": "YG_DIBAN_AFTER_3", "仰拱(底板)施工完成后，3个月以后": "YG_DIBAN_AFTER_3", "仰拱(底板)施工完成后第，2至3个月": "YG_DIBAN_2_3", # 新增异常格式工况分组 # 架桥机相关 "架桥机(运梁车) 首次通过前": "JQJ_FIRST_BEFORE", "架桥机(运梁车)首次通过前": "JQJ_FIRST_BEFORE", # 新增无空格版本 "架桥机(运梁车) 首次通过后，前3天": "JQJ_FIRST_AFTER_3D", "架桥机(运梁车)首次通过后，前3天": "JQJ_FIRST_AFTER_3D", # 新增无空格版本 "架桥机(运梁车) 首次通过后": "JQJ_FIRST_AFTER", "架桥机(运梁车)首次通过后": "JQJ_FIRST_AFTER", # 新增无空格版本 "架桥机(运梁车)通过": "STATIC", # 堆载预压/路基填筑相关 "堆载预压或路基填筑完成，第1至3个月": "DZYY_1_3", "堆载预压或路基填筑完成，第4至6个月": "DZYY_4_6", "堆载预压或路基填筑完成，6个月以后": "DZYY_AFTER_6", "路基或预压土填筑，连续填筑": "STATIC", "路基或预压土填筑，两次填筑间隔时间较长": "STATIC", "预压土或路基填筑完成，第1~3个月": "DZYY_1_3", "预压土或路基填筑完成，第4~6个月": "DZYY_4_6", "预压土或路基填筑完成，6个月以后": "DZYY_AFTER_6", # 轨道板相关 "轨道板(道床)铺设后，第1个月": "GDB_1", "轨道板(道床)铺设后，第2至3个月": "GDB_2_3", "轨道板(道床)铺设后，3个月以后": "GDB_AFTER_3", "轨道板铺设前": "STATIC", "轨道板(道床)铺设后，第1至3个月": "GDB_1", "轨道板(道床)铺设后，第4至6个月": "GDB_2_3", "轨道板(道床)铺设后，6个月以后": "GDB_AFTER_3", # 预制梁桥相关 "预制梁桥，架梁前": "STATIC", "预制梁桥，预制梁架设前": "YZLQ_BEFORE_JS", "预制梁桥，预制梁架设后": "YZLQ_AFTER_JS", "预制梁桥预制梁架设后": "YZLQ_AFTER_JS", # 新增工况分组 # 桥梁主体相关 "桥梁主体工程完工后,第1至3个月": "QL_ZHUTI_1_3", "桥梁主体工程完工后，第4至6个月": "QL_ZHUTI_4_6", "桥梁主体工程完工后,6个月以后": "QL_ZHUTI_AFTER_6", "侨梁主体工程完工后，6个月以后": "QL_ZHUTI_AFTER_6", # 错别字工况分组 # 轨道铺设相关 "轨道铺设完成后，第1个月": "GD_1", "轨道铺设完成后，2至3个月": "GD_2_3", "轨道铺设完成后，4至12个月": "GD_4_12", "轨道铺设完成后，12个月以后": "GD_AFTER_12", "轨道铺设期间，前": "STATIC", "轨道铺设期间，后": "STATIC", "轨道铺设，前": "STATIC", "轨道铺设，后": "STATIC", # 无砟轨道相关 "无砟轨道铺设后，第1至3个月": "WZGD_1_3", "无砟轨道铺设后，4至12个月": "WZGD_4_12", "无砟轨道铺设后，12个月以后": "WZGD_AFTER_12", # 静态工况（无切换规则） "墩台施工到一定高度": "STATIC", "墩台混凝土施工": "STATIC", "桥位施工桥梁，制梁前": "STATIC", "桥位施工桥梁，上部结构施工中": "STATIC", "铺路或堆载，一般情况": "STATIC", "填筑或堆载，一般情况": "STATIC", "铺路或堆载，沉降量突变情况": "STATIC", "填筑或堆载，两次填筑间隔时间较长情况": "STATIC", "铺路或堆载，两次铺路间隔时间较长情况": "STATIC", "站场填方路基段填筑完成至静态验收": "STATIC", "桥墩(台)地面处拆模后": "STATIC", "敦身混凝土施工": "STATIC", "现浇梁，浇筑前": "STATIC", "现浇梁上部结构施工中": "STATIC", "特殊地段隧道施工完成后至静态验收": "STATIC", "冬休": "STATIC" } return condition_group def _load_group_transition_rules(self): """加载工况切换触发规则（保持原有规则，适配新增工况）""" return { "YG_DIBAN_1": {"trigger_days": 30, "next_candidates": ["仰拱（底板）施工完成后，第2至3个月", "仰拱(底板)施工完成后，第2至3个月"]}, "YG_DIBAN_2_3": {"trigger_days": 60, "next_candidates": ["仰拱（底板）施工完成后，3个月以后", "仰拱(底板)施工完成后，3个月以后"]}, "YG_DIBAN_AFTER_3": {"trigger_days": None, "next_candidates": None}, "JQJ_FIRST_BEFORE": {"trigger_days": 1, "next_candidates": ["架桥机(运梁车) 首次通过后，前3天"]}, "JQJ_FIRST_AFTER_3D": {"trigger_days": 3, "next_candidates": ["架桥机(运梁车) 首次通过后"]}, "JQJ_FIRST_AFTER": {"trigger_days": None, "next_candidates": None}, "DZYY_1_3": {"trigger_days": 90, "next_candidates": ["堆载预压或路基填筑完成，第4至6个月"]}, "DZYY_4_6": {"trigger_days": 90, "next_candidates": ["堆载预压或路基填筑完成，6个月以后"]}, "DZYY_AFTER_6": {"trigger_days": None, "next_candidates": None}, "GDB_1": {"trigger_days": 30, "next_candidates": ["轨道板(道床)铺设后，第2至3个月"]}, "GDB_2_3": {"trigger_days": 30, "next_candidates": ["轨道板(道床)铺设后，3个月以后"]}, "GDB_AFTER_3": {"trigger_days": None, "next_candidates": None}, "YZLQ_BEFORE_JS": {"trigger_days": 1, "next_candidates": ["架桥机(运梁车)通过"]}, "YZLQ_AFTER_JS": {"trigger_days": 7, "next_candidates": ["桥梁主体工程完工后,第1至3个月"]}, "QL_ZHUTI_1_3": {"trigger_days": 90, "next_candidates": ["桥梁主体工程完工后，第4至6个月"]}, "QL_ZHUTI_4_6": {"trigger_days": 90, "next_candidates": ["桥梁主体工程完工后，6个月以后"]}, "QL_ZHUTI_AFTER_6": {"trigger_days": None, "next_candidates": None}, "GD_1": {"trigger_days": 30, "next_candidates": ["轨道铺设完成后，2至3个月"]}, "GD_2_3": {"trigger_days": 60, "next_candidates": ["轨道铺设完成后，4至12个月"]}, "GD_4_12": {"trigger_days": 240, "next_candidates": ["轨道铺设完成后，12个月以后"]}, "GD_AFTER_12": {"trigger_days": None, "next_candidates": None}, "WZGD_1_3": {"trigger_days": 90, "next_candidates": ["无砟轨道铺设后，4至12个月"]}, "WZGD_4_12": {"trigger_days": 240, "next_candidates": ["无砟轨道铺设后，12个月以后"]}, "WZGD_AFTER_12": {"trigger_days": None, "next_candidates": None}, "STATIC": {"trigger_days": None, "next_candidates": None} } def _load_winter_break_labels(self): """冬休标识（仅冬休）""" return {"冬休"} def _parse_to_date(self, time_str): """ 私有辅助方法：将时间字符串解析为日期对象，仅保留年月日，忽略时分秒 :param time_str: 时间字符串，格式为 "YYYY-MM-DD HH:MM:SS" :return: date对象 / None（解析失败时） """ 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_from_reversed(self, data, workinfo): """ 私有辅助方法：从倒序数据中提取时间统计信息 :param data: 单个point_id的倒序数据列表 :param workinfo: 目标工况名称（冬休时为冬休前上一工况） :return: 元组 (首次测量日期date对象, 首次到末次持续天数, 今日与首次测量天数差) """ # 筛选目标工况的有效记录（忽略冬休） target_records = [ d for d in data if d.get("workinfoname") == 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 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): """ 私有辅助方法：从倒序数据中提取冬休前的上一个有效工况 :param inner_data_list: 单个point_id的倒序数据列表 :return: 冬休前工况名称 / None（未找到时） """ 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 and current_work in self.base_periods: return current_work return None def _match_next_condition(self, current_name, candidates): """ 私有辅助方法：按符号风格匹配下一个工况名称（全角/半角括号对应） :param current_name: 当前工况名称（冬休时为冬休前上一工况） :param candidates: 目标工况候选列表 :return: 匹配的工况名称 / 当前工况名称（无候选时） """ if not candidates: return current_name if "（" in current_name: for cand in candidates: if "（" in cand: return cand if "(" in current_name: for cand in candidates: if "(" in cand: return cand return candidates[0] def _validate_point_id(self, inner_list): """ 私有辅助方法：校验内嵌列表内所有元素的point_id是否一致 :param inner_list: 单个point_id的倒序数据列表 :return: point_id / None（校验失败或列表为空时） """ 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_1d = [] 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) 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_1d.append(latest_record) continue current_workinfo = latest_record.get("workinfoname") if not current_workinfo or current_workinfo not in self.base_periods: 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_1d.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_1d.append(latest_record) continue else: judge_based_workinfo = current_workinfo # 用判断基准工况计算时间 first_dt, cumulative_days, days_to_today = self._get_time_statistics_from_reversed( inner_data_list, judge_based_workinfo ) first_measure_date = first_dt.strftime("%Y-%m-%d") if first_dt else None # 推导下一工况 group_id = self.condition_group.get(judge_based_workinfo, "STATIC") rule = self.group_transition_rules.get(group_id, {}) trigger_days = rule.get("trigger_days") next_candidates = rule.get("next_candidates", []) if trigger_days is not None and cumulative_days >= trigger_days: next_workname = self._match_next_condition(judge_based_workinfo, next_candidates) else: next_workname = 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_workname, "point_id": point_id, "error_msg": "" }) final_result_1d.append(latest_record) return final_result_1d