344 lines
18 KiB
Python
344 lines
18 KiB
Python
from datetime import datetime, date
|
||
|
||
class OperatingModePredictor:
|
||
"""
|
||
工况预测类（处理二维倒序数据，返回一维列表，仅保留各内嵌列表最新记录）
|
||
功能：根据输入的带时间序列的工况数据，推导每个监测点（point_id）的下一阶段工况
|
||
特性：
|
||
1. 输入为二维列表，每个内嵌列表对应一个point_id，且为倒序排列（最新记录在索引0）
|
||
2. 输出为一维列表，仅保留每个内嵌列表的最新记录，新增工况推导结果字段
|
||
3. 无需切换工况时，next_workinfo返回当前工况名称；需要切换时返回目标工况名称
|
||
4. 时间计算仅按日期（天）维度，忽略时分秒
|
||
5. 支持冬休场景：当前工况为冬休时，以冬休前上一个有效工况作为切换判断依据
|
||
"""
|
||
|
||
def __init__(self):
|
||
"""初始化类，加载内置的工况配置、分组规则和切换触发规则"""
|
||
# 基础工况配置（键：工况名称，值：基础监测周期参考）
|
||
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):
|
||
"""加载基础工况配置（私有方法，内部使用）"""
|
||
return {
|
||
"仰拱（底板）施工完成后，第1个月": 7,
|
||
"仰拱（底板）施工完成后，第2至3个月": 14,
|
||
"仰拱（底板）施工完成后，3个月以后": 30,
|
||
"仰拱(底板)施工完成后，第1个月": 7,
|
||
"仰拱(底板)施工完成后，第2至3个月": 14,
|
||
"仰拱(底板)施工完成后，3个月以后": 30,
|
||
"无砟轨道铺设后，第1至3个月": 30,
|
||
"无砟轨道铺设后，4至12个月": 90,
|
||
"无砟轨道铺设后，12个月以后": 180,
|
||
"墩台施工到一定高度": 30,
|
||
"墩台混凝土施工": 30,
|
||
"预制梁桥，架梁前": 30,
|
||
"预制梁桥，预制梁架设前": 1,
|
||
"预制梁桥，预制梁架设后": 7,
|
||
"桥位施工桥梁，制梁前": 30,
|
||
"桥位施工桥梁，上部结构施工中": 1,
|
||
"架桥机(运梁车)通过": 7,
|
||
"桥梁主体工程完工后，第1至3个月": 7,
|
||
"桥梁主体工程完工后，第4至6个月": 14,
|
||
"桥梁主体工程完工后，6个月以后": 30,
|
||
"轨道铺设期间，前": 30,
|
||
"轨道铺设期间，后": 14,
|
||
"轨道铺设完成后，第1个月": 14,
|
||
"轨道铺设完成后，2至3个月": 30,
|
||
"轨道铺设完成后，4至12个月": 90,
|
||
"轨道铺设完成后，12个月以后": 180,
|
||
"铺路或堆载，一般情况": 1,
|
||
"填筑或堆载，一般情况": 1,
|
||
"铺路或堆载，沉降量突变情况": 1,
|
||
"填筑或堆载，两次填筑间隔时间较长情况": 3,
|
||
"铺路或堆载，两次铺路间隔时间较长情况": 3,
|
||
"堆载预压或路基填筑完成，第1至3个月": 7,
|
||
"堆载预压或路基填筑完成，第4至6个月": 14,
|
||
"堆载预压或路基填筑完成，6个月以后": 30,
|
||
"架桥机(运梁车) 首次通过前": 1,
|
||
"架桥机(运梁车) 首次通过后，前3天": 1,
|
||
"架桥机(运梁车) 首次通过后": 7,
|
||
"轨道板(道床)铺设后，第1个月": 14,
|
||
"轨道板(道床)铺设后，第2至3个月": 30,
|
||
"轨道板(道床)铺设后，3个月以后": 90,
|
||
"冬休": 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",
|
||
"架桥机(运梁车) 首次通过前": "JQJ_FIRST_BEFORE",
|
||
"架桥机(运梁车) 首次通过后，前3天": "JQJ_FIRST_AFTER_3D",
|
||
"架桥机(运梁车) 首次通过后": "JQJ_FIRST_AFTER",
|
||
"堆载预压或路基填筑完成，第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",
|
||
"预制梁桥，预制梁架设前": "YZLQ_BEFORE_JS",
|
||
"预制梁桥，预制梁架设后": "YZLQ_AFTER_JS",
|
||
"桥梁主体工程完工后,第1至3个月": "QL_ZHUTI_1_3",
|
||
"桥梁主体工程完工后，第4至6个月": "QL_ZHUTI_4_6",
|
||
"桥梁主体工程完工后,6个月以后": "QL_ZHUTI_AFTER_6",
|
||
"轨道铺设完成后，第1个月": "GD_1",
|
||
"轨道铺设完成后，2至3个月": "GD_2_3",
|
||
"轨道铺设完成后，4至12个月": "GD_4_12",
|
||
"轨道铺设完成后，12个月以后": "GD_AFTER_12",
|
||
"无砟轨道铺设后，第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"
|
||
}
|
||
# 冬休分组（仅冬休）
|
||
winter_break_groups = {
|
||
"冬休": "STATIC"
|
||
}
|
||
condition_group.update(winter_break_groups)
|
||
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 |