diff --git a/app/utils/get_operating_mode.py b/app/utils/get_operating_mode.py index 1ab050a..aad5a5c 100644 --- a/app/utils/get_operating_mode.py +++ b/app/utils/get_operating_mode.py @@ -9,6 +9,7 @@ class OperatingModePredictor: 2. 输出为一维列表，仅保留每个内嵌列表的最新记录，新增工况推导结果字段 3. 无需切换工况时，next_workinfo返回当前工况名称；需要切换时返回目标工况名称 4. 时间计算仅按日期（天）维度，忽略时分秒 + 5. 支持冬休场景：当前工况为冬休时，以冬休前上一个有效工况作为切换判断依据 """ def __init__(self): @@ -19,6 +20,8 @@ class OperatingModePredictor: 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): """加载基础工况配置（私有方法，内部使用）""" @@ -62,12 +65,13 @@ class OperatingModePredictor: "架桥机(运梁车) 首次通过后": 7, "轨道板(道床)铺设后，第1个月": 14, "轨道板(道床)铺设后，第2至3个月": 30, - "轨道板(道床)铺设后，3个月以后": 90 + "轨道板(道床)铺设后，3个月以后": 90, + "冬休": 0 # 仅保留冬休 } def _load_condition_group(self): """加载工况分组规则（私有方法，内部使用）""" - return { + condition_group = { "仰拱（底板）施工完成后，第1个月": "YG_DIBAN_1", "仰拱(底板)施工完成后，第1个月": "YG_DIBAN_1", "仰拱（底板）施工完成后，第2至3个月": "YG_DIBAN_2_3", @@ -109,6 +113,12 @@ class OperatingModePredictor: "填筑或堆载，两次填筑间隔时间较长情况": "STATIC", "铺路或堆载，两次铺路间隔时间较长情况": "STATIC" } + # 冬休分组（仅冬休） + winter_break_groups = { + "冬休": "STATIC" + } + condition_group.update(winter_break_groups) + return condition_group def _load_group_transition_rules(self): """加载工况切换触发规则（私有方法，内部使用）""" @@ -140,6 +150,10 @@ class OperatingModePredictor: "STATIC": {"trigger_days": None, "next_candidates": None} } + def _load_winter_break_labels(self): + """冬休标识（仅冬休）""" + return {"冬休"} + def _parse_to_date(self, time_str): """ 私有辅助方法：将时间字符串解析为日期对象，仅保留年月日，忽略时分秒 @@ -158,54 +172,68 @@ class OperatingModePredictor: """ 私有辅助方法：从倒序数据中提取时间统计信息 :param data: 单个point_id的倒序数据列表 - :param workinfo: 当前工况名称 + :param workinfo: 目标工况名称（冬休时为冬休前上一工况） :return: 元组 (首次测量日期date对象, 首次到末次持续天数, 今日与首次测量天数差) """ - # 筛选当前工况的有效记录 - reversed_records = [d for d in data if d.get("workinfoname") == workinfo] - if not reversed_records: + # 筛选目标工况的有效记录（忽略冬休） + 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 - # 解析所有有效记录的日期 - reversed_dates = [] - for item in reversed_records: + target_dates = [] + for item in target_records: d = self._parse_to_date(item.get("MTIME_W")) if d: - reversed_dates.append(d) - if not reversed_dates: + target_dates.append(d) + if not target_dates: return None, 0, 0 - # 提取倒序数据的最新、最旧日期 - last_date = reversed_dates[0] - first_date = reversed_dates[-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): + """ + 私有辅助方法：从倒序数据中提取冬休前的上一个有效工况 + :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 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): @@ -230,17 +258,13 @@ class OperatingModePredictor: """ final_result_1d = [] - # 遍历二维列表，逐个处理每个point_id的内嵌数据 for inner_data_list in data_2d_list: - # 跳过空列表 if not isinstance(inner_data_list, list) or len(inner_data_list) == 0: continue - # 1. 提取当前内嵌列表的最新记录（倒序数据索引0为最新） latest_record = inner_data_list[0].copy() - - # 2. 校验point_id一致性 point_id = self._validate_point_id(inner_data_list) + if not point_id: latest_record.update({ "status": "fail", @@ -248,12 +272,12 @@ class OperatingModePredictor: "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 - # 3. 提取并校验当前工况 current_workinfo = latest_record.get("workinfoname") if not current_workinfo or current_workinfo not in self.base_periods: latest_record.update({ @@ -262,41 +286,59 @@ class OperatingModePredictor: "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 - # 4. 提取时间统计信息 + # 冬休逻辑：当前是冬休 → 取冬休前工况作为判断基准 + 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, current_workinfo + inner_data_list, judge_based_workinfo ) first_measure_date = first_dt.strftime("%Y-%m-%d") if first_dt else None - # 5. 判断工况切换条件，推导下一工况 - group_id = self.condition_group.get(current_workinfo, "STATIC") + # 推导下一工况 + 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(current_workinfo, next_candidates) + next_workname = self._match_next_condition(judge_based_workinfo, next_candidates) else: - # 不满足切换条件：返回当前工况 - next_workname = current_workinfo + next_workname = judge_based_workinfo - # 6. 组装结果字段，更新最新记录 + # 组装结果 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 + "point_id": point_id, + "error_msg": "" }) - # 7. 加入最终结果列表 final_result_1d.append(latest_record) return final_result_1d \ No newline at end of file