admin/railway_cloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

1.新工况替换

Browse Source
This commit is contained in:
whm
2026-02-10 18:08:11 +08:00
parent 6ccda8f8a7
commit c26bd4e8e1
2 changed files with 293 additions and 190 deletions
Download Patch File Download Diff File Expand all files Collapse all files

441
app/utils/get_operating_mode.py
View File

@@ -7,159 +7,246 @@ class OperatingModePredictor:
特性：
1. 输入为二维列表，每个内嵌列表对应一个point_id，且为倒序排列（最新记录在索引0）
2. 输出为一维列表，仅保留每个内嵌列表的最新记录，新增工况推导结果字段
3. 无需切换工况时，next_workinfo返回当前工况名称；需要切换时返回目标工况名称
3. 推导规则：有等效映射返回新工况，无等效保留旧工况；切换下一工况以base_periods为准
4. 时间计算仅按日期（天）维度，忽略时分秒
5. 支持冬休场景：当前工况为冬休时，以冬休前上一个有效工况作为切换判断依据
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()
# 工况切换触发规则（键：分组ID，值：触发天数+目标工况候选）
self.group_transition_rules = self._load_group_transition_rules()
# 冬休工况标识（仅保留“冬休”）
self.winter_break_labels = self._load_winter_break_labels()
# 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,
"仰拱（底板）施工完成后，第2至3个月": 14,
"仰拱（底板）施工完成后，3个月以后": 30,
# 路基工况（新工况优先）
"路基或预压土填筑，连续填筑": 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,
"墩台施工到一定高度": 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 # 仅保留冬休
# 特殊工况
"冬休": 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):
"""加载工况分组规则（私有方法，内部使用）"""
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",
"""加载工况分组（同义工况归为同一分组）"""
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",
"桥梁主体工程完工后,第1至3个月": "QL_ZHUTI_1_3",
"现浇梁，浇筑前": "STATIC",
"现浇梁上部结构施工中": "STATIC",
"架桥机(运梁车)通过": "STATIC",
"桥梁主体工程完工后，第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",
"桥梁主体工程完工后，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",
"墩台混凝土施工": "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
return group_map
def _load_group_transition_rules(self):
"""加载工况切换触发规则（私有方法，内部使用）"""
def _load_transition_rules(self):
"""加载切换触发规则（沿用原逻辑，以base_periods为准）"""
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}
# 路基切换规则
"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 _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（解析失败时）
"""
"""解析时间字符串为date对象（仅保留年月日）"""
if not time_str:
return None
try:
@@ -168,22 +255,18 @@ class OperatingModePredictor:
except ValueError:
return None
def _get_time_statistics_from_reversed(self, data, workinfo):
"""
私有辅助方法：从倒序数据中提取时间统计信息
:param data: 单个point_id的倒序数据列表
:param workinfo: 目标工况名称（冬休时为冬休前上一工况）
:return: 元组 (首次测量日期date对象, 首次到末次持续天数, 今日与首次测量天数差)
"""
# 筛选目标工况的有效记录（忽略冬休）
def _get_time_statistics(self, data, target_workinfo):
"""从倒序数据中提取目标工况的时间统计信息"""
# 筛选目标工况的有效记录
target_records = [
d for d in data
if d.get("workinfoname") == workinfo
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"))
@@ -192,9 +275,9 @@ class OperatingModePredictor:
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
@@ -202,46 +285,19 @@ class OperatingModePredictor:
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
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 _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（校验失败或列表为空时）
"""
"""校验内嵌列表内point_id是否一致"""
if not inner_list:
return None
base_point_id = inner_list[0].get("point_id")
@@ -252,19 +308,21 @@ class OperatingModePredictor:
def predict(self, data_2d_list):
"""
公有核心方法：执行工况预测，处理二维输入数据，返回一维结果列表
公有核心方法：执行工况预测
:param data_2d_list: 二维倒序数据列表，格式 [[{},{},{}], [{},{},{}]]
:return: 一维结果列表，格式 [{}, {}, {}]，每个元素为对应内嵌列表的最新记录+推导字段
:return: 一维结果列表，格式 [{}, {}, {}]
"""
final_result_1d = []
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",
@@ -275,11 +333,12 @@ class OperatingModePredictor:
"judge_based_workinfo": None,
"error_msg": "point_id不一致或缺失"
})
final_result_1d.append(latest_record)
final_result.append(latest_record)
continue
# 获取当前工况并标准化
current_workinfo = latest_record.get("workinfoname")
if not current_workinfo or current_workinfo not in self.base_periods:
if not current_workinfo or self._standardize_name(current_workinfo) not in self.std_to_canonical:
latest_record.update({
"status": "fail",
"current_workinfo": None,
@@ -289,10 +348,10 @@ class OperatingModePredictor:
"judge_based_workinfo": None,
"error_msg": "工况无效或缺失"
})
final_result_1d.append(latest_record)
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:
@@ -305,40 +364,46 @@ class OperatingModePredictor:
"judge_based_workinfo": None,
"error_msg": "冬休前未找到有效工况"
})
final_result_1d.append(latest_record)
final_result.append(latest_record)
continue
else:
judge_based_workinfo = current_workinfo
# 非冬休：优先使用等效新工况作为判断基准
judge_based_workinfo = self.old_to_new_map.get(current_workinfo, current_workinfo)
# 用判断基准工况计算时间
first_dt, cumulative_days, days_to_today = self._get_time_statistics_from_reversed(
# 计算时间统计信息
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
# 推导下一工况
group_id = self.condition_group.get(judge_based_workinfo, "STATIC")
rule = self.group_transition_rules.get(group_id, {})
# 推导下一工况（以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_candidates", [])
next_candidates = rule.get("next", [])
if trigger_days is not None and cumulative_days >= trigger_days:
next_workname = self._match_next_condition(judge_based_workinfo, next_candidates)
if trigger_days is not None and cumulative_days >= trigger_days and next_candidates:
# 切换工况：取base_periods中的规范名称
next_workinfo = next_candidates[0]
else:
next_workname = judge_based_workinfo
# 不切换：有等效则返回新工况，无则返回旧工况
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,
"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,
"next_workinfo": next_workinfo, # 推导结果：有等效返回新工况，无则返回旧工况
"point_id": point_id,
"error_msg": ""
})
final_result_1d.append(latest_record)
final_result.append(latest_record)
return final_result
return final_result_1d