From 2eb7b9b5c1c92d31b5420f43a73c7733cf2a7b97 Mon Sep 17 00:00:00 2001 From: whm <973418690@qq.com> Date: Thu, 29 Jan 2026 17:45:08 +0800 Subject: [PATCH] =?UTF-8?q?1.=E4=BF=AE=E6=94=B9=E5=B7=A5=E5=86=B5=E6=8E=A8?= =?UTF-8?q?=E7=90=86=E6=8E=A5=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/api/comprehensive_data.py | 12 +- app/utils/get_operating_mode.py | 302 ++++++++++++++++++++++++++++++++ 2 files changed, 310 insertions(+), 4 deletions(-) create mode 100644 app/utils/get_operating_mode.py diff --git a/app/api/comprehensive_data.py b/app/api/comprehensive_data.py index 020a2d8..c431eeb 100644 --- a/app/api/comprehensive_data.py +++ b/app/api/comprehensive_data.py @@ -30,6 +30,7 @@ from ..services.settlement_data import SettlementDataService from ..services.level_data import LevelDataService from ..services.original_data import OriginalDataService from ..services.comprehensive import ComprehensiveDataService +from ..utils.get_operating_mode import OperatingModePredictor import logging router = APIRouter(prefix="/comprehensive_data", tags=["综合数据管理"]) logger = logging.getLogger(__name__) @@ -470,12 +471,14 @@ def get_settlement_by_linecode( settlement_service = SettlementDataService() result = settlement_service.get_settlement_by_linecode(db, linecode) - + settlement_data = result['settlement_data'] + predictor = OperatingModePredictor() + result_1d = predictor.predict(settlement_data) return DataResponse( code=ResponseCode.SUCCESS, message=f"查询成功，共获取{len(result['settlement_data'])}条沉降数据", - total=len(result['settlement_data']), - data=result['settlement_data'] + total=len(result_1d), + data=result_1d ) except Exception as e: @@ -723,4 +726,5 @@ def get_checkpoint_by_point(request: LevelDataQueryRequest, db: Session = Depend message=f"{ResponseMessage.QUERY_FAILED}: {str(e)}", total=0, data=[] - ) \ No newline at end of file + ) + diff --git a/app/utils/get_operating_mode.py b/app/utils/get_operating_mode.py new file mode 100644 index 0000000..3c17142 --- /dev/null +++ b/app/utils/get_operating_mode.py @@ -0,0 +1,302 @@ +from datetime import datetime, date + +class OperatingModePredictor: + """ + 工况预测类（处理二维倒序数据，返回一维列表，仅保留各内嵌列表最新记录） + 功能：根据输入的带时间序列的工况数据，推导每个监测点（point_id）的下一阶段工况 + 特性： + 1. 输入为二维列表，每个内嵌列表对应一个point_id，且为倒序排列（最新记录在索引0） + 2. 输出为一维列表，仅保留每个内嵌列表的最新记录，新增工况推导结果字段 + 3. 无需切换工况时，next_workinfo返回当前工况名称；需要切换时返回目标工况名称 + 4. 时间计算仅按日期（天）维度，忽略时分秒 + """ + + 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() + + 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 + } + + def _load_condition_group(self): + """加载工况分组规则（私有方法，内部使用）""" + return { + "仰拱（底板）施工完成后，第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" + } + + 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 _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对象, 首次到末次持续天数, 今日与首次测量天数差) + """ + # 筛选当前工况的有效记录 + reversed_records = [d for d in data if d.get("workinfoname") == workinfo] + if not reversed_records: + return None, 0, 0 + + # 解析所有有效记录的日期 + reversed_dates = [] + for item in reversed_records: + d = self._parse_to_date(item.get("MTIME_W")) + if d: + reversed_dates.append(d) + if not reversed_dates: + return None, 0, 0 + + # 提取倒序数据的最新、最旧日期 + last_date = reversed_dates[0] + first_date = reversed_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 _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 = [] + + # 遍历二维列表，逐个处理每个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", + "current_workinfo": None, + "first_measure_date": None, + "days_from_first_to_today": None, + "next_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({ + "status": "fail", + "current_workinfo": None, + "first_measure_date": None, + "days_from_first_to_today": None, + "next_workinfo": None, + "error_msg": "工况无效或缺失" + }) + final_result_1d.append(latest_record) + continue + + # 4. 提取时间统计信息 + first_dt, cumulative_days, days_to_today = self._get_time_statistics_from_reversed( + inner_data_list, current_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") + 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) + else: + # 不满足切换条件：返回当前工况 + next_workname = current_workinfo + + # 6. 组装结果字段，更新最新记录 + latest_record.update({ + "status": "success", + "current_workinfo": current_workinfo, + "first_measure_date": first_measure_date, + "days_from_first_to_today": days_to_today, + "next_workinfo": next_workname, + "point_id": point_id + }) + + # 7. 加入最终结果列表 + final_result_1d.append(latest_record) + + return final_result_1d \ No newline at end of file