diff --git a/.env.example b/.env.example index 9b191c1..ee3c4fd 100644 --- a/.env.example +++ b/.env.example @@ -15,8 +15,8 @@ APP_TIMEZONE=Asia/Shanghai # Default: 16 MiB MAX_UPLOAD_BYTES=16777216 -# Default model path inside the Docker image. -MODEL_PATH=/app/my_survival_forest_model_quxi-10-0331.joblib +# Default RC1 fusion model core directory inside the Docker image. +FUSION_MODEL_CORE_DIR=/app/model_core # Keep public registration closed by default. ALLOW_REGISTRATION=false diff --git a/.gitignore b/.gitignore index adafec4..6dc4a93 100644 --- a/.gitignore +++ b/.gitignore @@ -28,6 +28,8 @@ static/images/ pipe_survival_0331.db # Large generated or local model artifacts +model_core/ +wheelhouse/ *.joblib *.pkl *.pickle diff --git a/Dockerfile b/Dockerfile index b2e09b2..bd872de 100644 --- a/Dockerfile +++ b/Dockerfile @@ -13,7 +13,7 @@ RUN apt-get update \ COPY requirements.txt . -RUN conda create -n demo python=3.12 -y \ +RUN mamba create -n demo python=3.12 -y \ && conda run -n demo python -m pip install --no-cache-dir -r requirements.txt \ && conda clean -afy @@ -23,7 +23,7 @@ COPY static ./static COPY main.py . COPY example.xlsx . COPY 20260630标准文本——供水管道健康状态与剩余寿命评估技术导则.pdf . -COPY my_survival_forest_model_quxi-10-0331.joblib . +COPY model_core ./model_core RUN mkdir -p data static/images uploads diff --git a/app/__init__.py b/app/__init__.py index 92027f2..9dbe372 100644 --- a/app/__init__.py +++ b/app/__init__.py @@ -39,7 +39,7 @@ def create_app(config_object: type[Config] = Config, *, load_model_on_start: boo if load_model_on_start: try: - app.config["RSF_MODEL"] = load_model(app.config["MODEL_PATH"]) + app.config["RSF_MODEL"] = load_model(app.config["FUSION_MODEL_CORE_DIR"]) logging.info("模型加载成功") except Exception as exc: app.config["RSF_MODEL"] = None diff --git a/app/config.py b/app/config.py index f675d33..f6c813b 100644 --- a/app/config.py +++ b/app/config.py @@ -37,9 +37,9 @@ class Config: MAX_CONTENT_LENGTH = env_int("MAX_UPLOAD_BYTES", 16 * 1024 * 1024) SESSION_COOKIE_HTTPONLY = True SESSION_COOKIE_SAMESITE = "Lax" - MODEL_PATH = os.environ.get( - "MODEL_PATH", - str(BASE_DIR / "my_survival_forest_model_quxi-10-0331.joblib"), + FUSION_MODEL_CORE_DIR = os.environ.get( + "FUSION_MODEL_CORE_DIR", + str(BASE_DIR / "model_core"), ) ALLOW_REGISTRATION = env_bool("ALLOW_REGISTRATION", False) ADMIN_USERNAME = os.environ.get("ADMIN_USERNAME", "admin").strip() or "admin" diff --git a/app/prediction.py b/app/prediction.py index 127c026..bda2fe4 100644 --- a/app/prediction.py +++ b/app/prediction.py @@ -1,18 +1,16 @@ from __future__ import annotations import logging -import os +import sys import uuid from dataclasses import dataclass from pathlib import Path from typing import Any -import joblib import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt -import numpy as np import pandas as pd from matplotlib import font_manager, rcParams from werkzeug.datastructures import FileStorage @@ -59,9 +57,12 @@ FEATURES = [ ID_COLUMN = "管道编号" PIPE_AGE_COLUMN = "管龄（年）" LEGACY_PIPE_AGE_COLUMN = "管龄" -STATUS_COLUMN = "状态" -EVENT_AGE_COLUMN = "事件/观察管龄（年）" COLUMN_ALIASES = { + "ID": ID_COLUMN, + "管段ID": ID_COLUMN, + "管段编号": ID_COLUMN, + "当前管龄": PIPE_AGE_COLUMN, + "当前管龄（年）": PIPE_AGE_COLUMN, "管径（mm）": "管径", "管径（mm)": "管径", "流速（m/s）": "流速", @@ -70,48 +71,42 @@ COLUMN_ALIASES = { "温度（℃）": "温度", "年均降雨量（mm）": "降雨量", "降雨量（mm）": "降雨量", + "结构性缺陷": "结构缺陷", + "功能性缺陷": "功能缺陷", } INPUT_COLUMNS = [ ID_COLUMN, PIPE_AGE_COLUMN, - STATUS_COLUMN, - EVENT_AGE_COLUMN, *FEATURES, ] +REQUIRED_INPUT_COLUMNS = [ID_COLUMN, PIPE_AGE_COLUMN, "管材", "管径"] CHART_DISPLAY_LIMIT = 10 MATERIAL_COLUMN = "管材" -MATERIAL_CODE_OPTIONS = [ - (1, "镀锌"), - (2, "钢塑"), - (3, "铝塑"), - (4, "PPR"), - (5, "PE"), - (6, "UPVC"), - (7, "铸铁"), - (8, "预应力"), - (9, "自应力"), - (10, "玻璃钢夹砂"), - (11, "钢管"), - (12, "钢套混凝土管"), - (13, "球墨铸铁"), - (14, "其他"), -] -MATERIAL_ALIAS_TO_CODE = { - str(code): code - for code, _ in MATERIAL_CODE_OPTIONS +MATERIAL_CODE_TO_NAME = { + 1: "镀锌", + 2: "钢塑", + 3: "铝塑", + 4: "PPR", + 5: "PE", + 6: "UPVC", + 7: "铸铁", + 11: "钢管", } -MATERIAL_ALIAS_TO_CODE.update( - { - name.casefold(): code - for code, name in MATERIAL_CODE_OPTIONS - } -) -MATERIAL_ALIAS_TO_CODE.update( - { - f"{code}-{name}".casefold(): code - for code, name in MATERIAL_CODE_OPTIONS - } -) +LOCATION_CODE_TO_NAME = { + 1: "绿化带", + 2: "行人道", + 3: "非机动车道", + 4: "小区内", + 6: "市政道路", + 7: "桥管", +} +MATERIAL_ALIASES = {str(code): name for code, name in MATERIAL_CODE_TO_NAME.items()} +MATERIAL_ALIASES.update({name.casefold(): name for name in MATERIAL_CODE_TO_NAME.values()}) +MATERIAL_ALIASES.update({f"{code}-{name}".casefold(): name for code, name in MATERIAL_CODE_TO_NAME.items()}) +LOCATION_ALIASES = {str(code): name for code, name in LOCATION_CODE_TO_NAME.items()} +LOCATION_ALIASES.update({name.casefold(): name for name in LOCATION_CODE_TO_NAME.values()}) +LOCATION_ALIASES.update({f"{code}-{name}".casefold(): name for code, name in LOCATION_CODE_TO_NAME.items()}) +LOCATION_ALIASES["人行道"] = "行人道" SUPPORTED_EXTENSIONS = {".csv", ".xls", ".xlsx"} CHINESE_FONT_PROP = None DEFECT_GRADE_VALUES = {"无": 0.0, "轻度": 1.0, "中度": 3.0, "严重": 5.0} @@ -134,30 +129,7 @@ class PredictionArtifacts: sample_count: int summary_rows: list[dict[str, Any]] analysis_text: str - - -class ModelBundleAdapter: - """Expose a bundled preprocessor and survival model as one predictor.""" - - def __init__(self, preprocessor, model) -> None: - self.preprocessor = preprocessor - self.model = model - - def _transform(self, frame: pd.DataFrame): - values = frame[FEATURES].copy() - values[MATERIAL_COLUMN] = values[MATERIAL_COLUMN].map( - lambda value: f"M{int(value)}" if pd.notna(value) else value - ) - values["位置"] = values["位置"].map( - lambda value: f"L{int(value)}" if pd.notna(value) else value - ) - return self.preprocessor.transform(values.to_numpy(dtype=object)).astype(np.float32) - - def predict_survival_function(self, frame: pd.DataFrame): - return self.model.predict_survival_function(self._transform(frame)) - - def predict(self, frame: pd.DataFrame): - return self.model.predict(self._transform(frame)) + model_version: str def configure_matplotlib_fonts(): @@ -190,24 +162,18 @@ def chinese_font_kwargs() -> dict[str, Any]: return {"fontproperties": CHINESE_FONT_PROP} if CHINESE_FONT_PROP else {} -def load_model(model_path: str): - if not os.path.exists(model_path): - raise FileNotFoundError(f"未找到模型文件: {model_path}") - model = joblib.load(model_path) - if isinstance(model, dict): - if not {"preprocessor", "model"}.issubset(model): - raise ValueError("模型包缺少 preprocessor 或 model。") - model = ModelBundleAdapter(model["preprocessor"], model["model"]) - feature_names = getattr(model, "feature_names_in_", None) - if feature_names is not None and list(feature_names) != FEATURES: - raise ValueError( - "模型输入特征与系统配置不一致: " - f"model={list(feature_names)}, app={FEATURES}" - ) - n_features = getattr(model, "n_features_in_", None) - if n_features is not None and int(n_features) != len(FEATURES): - raise ValueError(f"模型特征数量不一致: model={n_features}, app={len(FEATURES)}") - return model +def load_model(core_dir: str): + core_path = Path(core_dir) + if not core_path.exists(): + raise FileNotFoundError(f"未找到模型核心目录: {core_path}") + if not (core_path / "frozen_fusion_inference.py").exists(): + raise FileNotFoundError(f"模型核心目录缺少 frozen_fusion_inference.py: {core_path}") + if str(core_path) not in sys.path: + sys.path.insert(0, str(core_path)) + + from frozen_fusion_inference import FrozenFusionPredictor + + return FrozenFusionPredictor(core_path) def safe_unlink(path: Path) -> None: @@ -339,31 +305,62 @@ def fill_defects_from_detail_sheet(path: Path, df: pd.DataFrame) -> pd.DataFrame def validate_input_frame(df: pd.DataFrame) -> None: if df.empty: raise PredictionError("上传文件没有可预测的数据。") - missing = [col for col in INPUT_COLUMNS if col not in df.columns] + missing = [col for col in REQUIRED_INPUT_COLUMNS if col not in df.columns] if missing: raise PredictionError(f"缺少必要字段: {', '.join(missing)}") -def normalize_material_code(value: Any) -> int: +def normalize_category_value( + value: Any, + aliases: dict[str, str], + label: str, + *, + required: bool = False, +) -> str | pd.NA: if pd.isna(value): - raise PredictionError("管材不能为空。") - if isinstance(value, str): - key = value.strip().casefold() - else: + if required: + raise PredictionError(f"{label}不能为空。") + return pd.NA + text = str(value).strip() + if text == "": + if required: + raise PredictionError(f"{label}不能为空。") + return pd.NA + key = text.casefold() + try: numeric_value = float(value) if not numeric_value.is_integer(): - raise PredictionError(f"管材编码无效: {value}") + raise ValueError key = str(int(numeric_value)) - code = MATERIAL_ALIAS_TO_CODE.get(key) - if code is None: - raise PredictionError(f"管材编码无效: {value}") - return code + except (TypeError, ValueError): + pass + normalized = aliases.get(key) + if normalized is None: + raise PredictionError(f"{label}超出RC1支持范围: {value}") + return normalized def prepare_model_features(df: pd.DataFrame) -> pd.DataFrame: - x_test = df[FEATURES].copy() - x_test[MATERIAL_COLUMN] = x_test[MATERIAL_COLUMN].map(normalize_material_code) - return x_test + result = pd.DataFrame(index=df.index) + result["ID"] = df[ID_COLUMN] + result["Current Age"] = df[PIPE_AGE_COLUMN] + result["Material"] = df[MATERIAL_COLUMN].map( + lambda value: normalize_category_value(value, MATERIAL_ALIASES, "管材", required=True) + ) + result["Diameter"] = df["管径"] + result["Flow Velocity"] = df["流速"] if "流速" in df.columns else pd.NA + result["Pressure"] = df["压力"] if "压力" in df.columns else pd.NA + result["Temperature"] = df["温度"] if "温度" in df.columns else pd.NA + result["Precipitation"] = df["降雨量"] if "降雨量" in df.columns else pd.NA + if "位置" in df.columns: + result["Location"] = df["位置"].map( + lambda value: normalize_category_value(value, LOCATION_ALIASES, "位置") + ) + else: + result["Location"] = pd.NA + result["Structural Defects"] = df["结构缺陷"] if "结构缺陷" in df.columns else pd.NA + result["Functional Defects"] = df["功能缺陷"] if "功能缺陷" in df.columns else pd.NA + return result def grade_info(probability: float) -> tuple[str, str, str]: @@ -389,22 +386,22 @@ def interpolate_probability(times: list[float], probs: list[float], target: floa return float(probs[-1]) -def estimate_remaining_life(times: list[float], probs: list[float]) -> float: - for t, p in zip(times, probs): - if p <= 0.5: - return float(t) - return float(times[-1]) if times else 0.0 +def estimate_remaining_life(times: list[float], health_states: list[float], current_age: float = 0.0) -> float | None: + for age, health_state in zip(times, health_states): + if age >= current_age and health_state <= 0.5: + return max(float(age) - current_age, 0.0) + return None def make_analysis_text(summary_rows: list[dict[str, Any]]) -> str: if not summary_rows: return "当前结果为空，暂无可供解释的样本。" - worst = min(summary_rows, key=lambda x: x["health_probability"]) - best = max(summary_rows, key=lambda x: x["health_probability"]) + worst = min(summary_rows, key=lambda x: x["health_state"]) + best = max(summary_rows, key=lambda x: x["health_state"]) return ( - "阶梯状曲线表示模型对不同管道随时间推移维持在安全健康状态概率的动态预测。" + "曲线表示模型对不同管道随管龄变化的健康状态动态预测。" f"当前样本中风险最高管道为 {worst['pipe_id']}（{worst['grade_label']}），" - f"健康概率最高管道为 {best['pipe_id']}（{best['health_probability']:.1%}）。" + f"当前健康状态最高管道为 {best['pipe_id']}（{best['health_state']:.1%}）。" ) @@ -427,18 +424,21 @@ def run_prediction(uploaded: FileStorage, user_id: int, model) -> PredictionArti validate_input_frame(df) x_test = prepare_model_features(df) try: - curves = model.predict_survival_function(x_test) + predictions = model.predict(x_test, variant="defect_sensitive") + except ValueError as exc: + logging.exception("预测输入校验失败: %s", exc) + raise PredictionError(str(exc)) except Exception as exc: logging.exception("预测失败: %s", exc) raise PredictionError("模型预测失败，请检查输入字段类型是否正确。", 500) image_filename = f"plot_{user_id}_{run_id}.png" image_path = IMAGE_DIR / image_filename - summary_rows, summary_sheet_rows = render_survival_chart(df, curves, image_path) + summary_rows, summary_sheet_rows = render_survival_chart(predictions, image_path) safe_stem = Path(saved_filename).stem excel_path = user_dir / f"{safe_stem}_pre.xlsx" - write_prediction_workbook(excel_path, curves, summary_rows, summary_sheet_rows) + write_prediction_workbook(excel_path, predictions, summary_rows, summary_sheet_rows) return PredictionArtifacts( original_filename=original_filename, saved_path=original_path, @@ -448,36 +448,45 @@ def run_prediction(uploaded: FileStorage, user_id: int, model) -> PredictionArti sample_count=len(summary_rows), summary_rows=summary_rows, analysis_text=make_analysis_text(summary_rows), + model_version=str(getattr(model, "config", {}).get("version", "")), ) except PredictionError: safe_unlink(original_path) raise -def render_survival_chart(df: pd.DataFrame, curves, image_path: Path) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: +def float_list(values: Any) -> list[float]: + return [float(value) for value in list(values)] + + +def format_years(value: float | None, current_age: float, max_age: float) -> str | float: + if value is None: + return f">{max(max_age - current_age, 0.0):g}" + return round(float(value), 3) + + +def render_survival_chart(predictions: list[dict[str, Any]], image_path: Path) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: plt.figure(figsize=(10, 5.6)) summary_rows: list[dict[str, Any]] = [] summary_sheet_rows: list[dict[str, Any]] = [] display_note = f"说明：图中仅展示前{CHART_DISPLAY_LIMIT}条管道的示例数据，不足{CHART_DISPLAY_LIMIT}条则全部展示；完整结果请下载电子表格。" - for i, curve in enumerate(curves): - times = [float(x) for x in list(curve.x)] - probs = [float(y) for y in list(curve.y)] - pipe_id = normalize_id_value(df.iloc[i][ID_COLUMN], f"Pipe_{i+1:03d}") - raw_pipe_age = df.iloc[i][PIPE_AGE_COLUMN] if PIPE_AGE_COLUMN in df.columns else None - pipe_age_value = float(raw_pipe_age) if pd.notna(raw_pipe_age) else 0.0 - pipe_age = f"{raw_pipe_age} 年" if pd.notna(raw_pipe_age) else "-" - health_probability = interpolate_probability(times, probs, pipe_age_value) - health_risk = 1.0 - health_probability - remaining_life = estimate_remaining_life(times, probs) - grade_label, grade_desc, grade_class = grade_info(health_probability) + for i, prediction in enumerate(predictions): + times = float_list(prediction["pipe_age"]) + health_states = float_list(prediction["health_state"]) + pipe_id = normalize_id_value(prediction.get("ID"), f"Pipe_{i+1:03d}") + current_age = float(prediction["current_age"]) + pipe_age = f"{current_age:g} 年" + current_health_state = float(prediction["current_health_state"]) + remaining_life = estimate_remaining_life(times, health_states, current_age) + grade_label, grade_desc, grade_class = grade_info(current_health_state) + max_age = max(times) if times else current_age summary_rows.append( { "pipe_id": pipe_id, "pipe_age": pipe_age, - "health_probability": health_probability, - "health_risk": health_risk, + "health_state": current_health_state, "remaining_life": remaining_life, "grade_label": grade_label, "grade_desc": grade_desc, @@ -488,16 +497,18 @@ def render_survival_chart(df: pd.DataFrame, curves, image_path: Path) -> tuple[l { ID_COLUMN: pipe_id, PIPE_AGE_COLUMN: pipe_age, - "健康风险值": health_risk, + "当前健康状态": current_health_state, + "健康等级": grade_label, + "预计剩余寿命（年）": format_years(remaining_life, current_age, max_age), } ) if i < CHART_DISPLAY_LIMIT: - plt.step(times, probs, where="post", linewidth=2, label=pipe_id) + plt.step(times, health_states, where="post", linewidth=2, label=pipe_id) font_kwargs = chinese_font_kwargs() plt.xlabel("管龄（年）", **font_kwargs) - plt.ylabel("健康风险", **font_kwargs) - plt.title("管道的剩余寿命分析图", **font_kwargs) + plt.ylabel("管道健康状态", **font_kwargs) + plt.title("供水管道健康状态曲线", **font_kwargs) plt.figtext(0.5, 0.02, display_note, ha="center", fontsize=9, color="#475569", **font_kwargs) plt.grid(alpha=0.18) if min(len(summary_rows), CHART_DISPLAY_LIMIT) <= 12: @@ -514,18 +525,18 @@ def render_survival_chart(df: pd.DataFrame, curves, image_path: Path) -> tuple[l def write_prediction_workbook( excel_path: Path, - curves, + predictions: list[dict[str, Any]], summary_rows: list[dict[str, Any]], summary_sheet_rows: list[dict[str, Any]], ) -> None: chart_times: set[float] = set() chart_series: list[tuple[str, dict[float, float]]] = [] - for i, curve in enumerate(curves): - times = [float(x) for x in list(curve.x)] - probs = [float(y) for y in list(curve.y)] + for i, prediction in enumerate(predictions): + times = float_list(prediction["pipe_age"]) + health_states = float_list(prediction["health_state"]) pipe_id = summary_rows[i]["pipe_id"] chart_times.update(times) - chart_series.append((pipe_id, dict(zip(times, probs)))) + chart_series.append((pipe_id, dict(zip(times, health_states)))) sorted_times = sorted(chart_times) sample_columns = ["管龄（年）", *[pipe_id for pipe_id, _ in chart_series]] @@ -535,7 +546,10 @@ def write_prediction_workbook( ] with pd.ExcelWriter(excel_path, engine="openpyxl") as writer: - summary_df = pd.DataFrame(summary_sheet_rows, columns=[ID_COLUMN, PIPE_AGE_COLUMN, "健康风险值"]) + summary_df = pd.DataFrame( + summary_sheet_rows, + columns=[ID_COLUMN, PIPE_AGE_COLUMN, "当前健康状态", "健康等级", "预计剩余寿命（年）"], + ) sample_df = pd.DataFrame(sample_data_rows, columns=sample_columns) if ID_COLUMN in summary_df.columns: summary_df[ID_COLUMN] = summary_df[ID_COLUMN].astype("string") diff --git a/app/routes.py b/app/routes.py index 54004dd..d1edb31 100644 --- a/app/routes.py +++ b/app/routes.py @@ -139,6 +139,7 @@ def prediction_result_payload(artifacts, record: UploadRecord) -> dict: "sample_count": int(artifacts.sample_count), "summary_rows": artifacts.summary_rows[:6], "analysis_text": artifacts.analysis_text, + "model_version": artifacts.model_version, } @@ -437,5 +438,6 @@ def predict(): "result_url": last_result["result_url"], "sample_count": last_result["sample_count"], "original_filename": artifacts.original_filename, + "model_version": last_result["model_version"], } ) diff --git a/docker-compose.yml b/docker-compose.yml index 77f1262..a27ef44 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -13,7 +13,7 @@ services: DEBUG: ${DEBUG:-false} APP_TIMEZONE: ${APP_TIMEZONE:-Asia/Shanghai} MAX_UPLOAD_BYTES: ${MAX_UPLOAD_BYTES:-16777216} - MODEL_PATH: ${MODEL_PATH:-/app/my_survival_forest_model_quxi-10-0331.joblib} + FUSION_MODEL_CORE_DIR: ${FUSION_MODEL_CORE_DIR:-/app/model_core} ALLOW_REGISTRATION: ${ALLOW_REGISTRATION:-false} PASSWORD_RESET_TOKEN_MINUTES: ${PASSWORD_RESET_TOKEN_MINUTES:-30} ports: diff --git a/example.xlsx b/example.xlsx index 6566ace..a370edf 100644 Binary files a/example.xlsx and b/example.xlsx differ diff --git a/requirements.txt b/requirements.txt index 0dfab2c..eccf471 100644 --- a/requirements.txt +++ b/requirements.txt @@ -14,5 +14,5 @@ openpyxl==3.1.5 XlsxWriter==3.2.9 xlrd==2.0.2 -scikit-learn==1.9.0 -scikit-survival==0.28.0 +scikit-learn==1.8.0 +scikit-survival==0.27.0 diff --git a/templates/home.html b/templates/home.html index 15727cf..e1be3ea 100644 --- a/templates/home.html +++ b/templates/home.html @@ -101,7 +101,7 @@
必填基础信息
-
管道编号、管龄、状态、管材、管径
+
管道编号、管龄、管材、管径
选填历史信息
diff --git a/tests/test_prediction.py b/tests/test_prediction.py index 0156937..0c2c445 100644 --- a/tests/test_prediction.py +++ b/tests/test_prediction.py @@ -28,10 +28,21 @@ from app.prediction import ( ) -class DummyCurve: - def __init__(self, x: list[float], y: list[float]) -> None: - self.x = x - self.y = y +def dummy_prediction( + pipe_id: str = "P001", + pipe_age: list[float] | None = None, + health_state: list[float] | None = None, + current_age: float = 12.0, + current_health_state: float = 0.4, +) -> dict: + return { + "ID": pipe_id, + "current_age": current_age, + "current_health_state": current_health_state, + "current_health_grade": "II级", + "pipe_age": pipe_age or [1, 10, 12], + "health_state": health_state or [0.9, 0.7, 0.4], + } class PredictionHelpersTest(unittest.TestCase): @@ -91,7 +102,7 @@ class PredictionHelpersTest(unittest.TestCase): template = workbook.active template.title = "Template" template.append(INPUT_COLUMNS) - template.append(["001", 5, 0, 5, 1, 100, 1.2, 0.4, 20, 800, 1, "=缺陷计算!E2", "=缺陷计算!J2"]) + template.append(["001", 5, 1, 100, 1, 1.2, 0.4, 20, 800, "=缺陷计算!E2", "=缺陷计算!J2"]) detail = workbook.create_sheet("缺陷计算") detail.append([ @@ -114,48 +125,79 @@ class PredictionHelpersTest(unittest.TestCase): self.assertAlmostEqual(float(df.loc[0, "结构缺陷"]), 3.8) self.assertAlmostEqual(float(df.loc[0, "功能缺陷"]), 1.6) - def test_prepare_model_features_maps_material_aliases_to_codes(self) -> None: + def test_prepare_model_features_maps_material_aliases_to_rc1_names(self) -> None: rows = [] - for value in ["镀锌", "2-钢塑", 13]: + for value in ["镀锌", "2-钢塑", 11]: row = {feature: 1 for feature in FEATURES} row["管材"] = value + row[ID_COLUMN] = "P001" + row[PIPE_AGE_COLUMN] = 12 rows.append(row) df = pd.DataFrame(rows) x_test = prepare_model_features(df) - self.assertEqual(x_test["管材"].tolist(), [1, 2, 13]) + self.assertEqual(x_test["Material"].tolist(), ["镀锌", "钢塑", "钢管"]) def test_prepare_model_features_rejects_invalid_material_alias(self) -> None: row = {feature: 1 for feature in FEATURES} row["管材"] = "未知管材" + row[ID_COLUMN] = "P001" + row[PIPE_AGE_COLUMN] = 12 df = pd.DataFrame([row]) with self.assertRaises(PredictionError) as ctx: prepare_model_features(df) - self.assertIn("管材编码无效", ctx.exception.message) + self.assertIn("管材超出RC1支持范围", ctx.exception.message) + + def test_prepare_model_features_rejects_unsupported_legacy_material_code(self) -> None: + row = {feature: 1 for feature in FEATURES} + row["管材"] = 13 + row[ID_COLUMN] = "P001" + row[PIPE_AGE_COLUMN] = 12 + df = pd.DataFrame([row]) + + with self.assertRaises(PredictionError) as ctx: + prepare_model_features(df) + + self.assertIn("管材超出RC1支持范围", ctx.exception.message) + + def test_prepare_model_features_maps_location_aliases_to_rc1_names(self) -> None: + row = {feature: 1 for feature in FEATURES} + row["管材"] = 5 + row["位置"] = "2-行人道" + row[ID_COLUMN] = "P001" + row[PIPE_AGE_COLUMN] = 12 + df = pd.DataFrame([row]) + + x_test = prepare_model_features(df) + + self.assertEqual(x_test["Location"].tolist(), ["行人道"]) def test_prediction_workbook_keeps_sample_data_in_one_sheet(self) -> None: - curves = [ - DummyCurve([1, 2], [0.9, 0.7]), - DummyCurve([1, 2], [0.8, 0.6]), + predictions = [ + dummy_prediction("P001", [1, 2], [0.9, 0.7], 10, 0.7), + dummy_prediction("P002", [1, 2], [0.8, 0.6], 12, 0.6), ] summary_rows = [{"pipe_id": "P001"}, {"pipe_id": "P002"}] summary_sheet_rows = [ - {ID_COLUMN: "P001", PIPE_AGE_COLUMN: "10 年", "健康风险值": 0.3}, - {ID_COLUMN: "P002", PIPE_AGE_COLUMN: "12 年", "健康风险值": 0.4}, + {ID_COLUMN: "P001", PIPE_AGE_COLUMN: "10 年", "当前健康状态": 0.7, "健康等级": "IV级", "预计剩余寿命（年）": ">63"}, + {ID_COLUMN: "P002", PIPE_AGE_COLUMN: "12 年", "当前健康状态": 0.6, "健康等级": "III级", "预计剩余寿命（年）": ">61"}, ] with TemporaryDirectory() as temp_dir: output_path = Path(temp_dir) / "prediction.xlsx" - write_prediction_workbook(output_path, curves, summary_rows, summary_sheet_rows) + write_prediction_workbook(output_path, predictions, summary_rows, summary_sheet_rows) workbook = pd.ExcelFile(output_path) self.assertEqual(workbook.sheet_names, ["结果摘要", "样本数据"]) summary_data = pd.read_excel(output_path, sheet_name="结果摘要") - self.assertEqual(summary_data.columns.tolist(), [ID_COLUMN, PIPE_AGE_COLUMN, "健康风险值"]) + self.assertEqual( + summary_data.columns.tolist(), + [ID_COLUMN, PIPE_AGE_COLUMN, "当前健康状态", "健康等级", "预计剩余寿命（年）"], + ) sample_data = pd.read_excel(output_path, sheet_name="样本数据") self.assertEqual(sample_data.columns.tolist(), ["管龄（年）", "P001", "P002"]) @@ -171,13 +213,15 @@ class PredictionHelpersTest(unittest.TestCase): self.assertEqual(sample_worksheet["B2"].number_format, "0.0%") def test_prediction_workbook_writes_pipe_ids_as_excel_text(self) -> None: - curves = [DummyCurve([1], [0.9])] + predictions = [dummy_prediction("00123", [1], [0.9], 10, 0.9)] summary_rows = [{"pipe_id": "00123"}] - summary_sheet_rows = [{ID_COLUMN: "00123", PIPE_AGE_COLUMN: "10 年", "健康风险值": 0.1}] + summary_sheet_rows = [ + {ID_COLUMN: "00123", PIPE_AGE_COLUMN: "10 年", "当前健康状态": 0.9, "健康等级": "V级", "预计剩余寿命（年）": ">63"} + ] with TemporaryDirectory() as temp_dir: output_path = Path(temp_dir) / "prediction.xlsx" - write_prediction_workbook(output_path, curves, summary_rows, summary_sheet_rows) + write_prediction_workbook(output_path, predictions, summary_rows, summary_sheet_rows) workbook = load_workbook(output_path) summary_worksheet = workbook["结果摘要"] @@ -200,18 +244,17 @@ class PredictionHelpersTest(unittest.TestCase): patch("app.prediction.plt.figtext") as figtext, patch("app.prediction.plt.title") as title, ): - summary_rows, _ = render_survival_chart(df, [DummyCurve([1, 10, 12], [0.9, 0.7, 0.4])], output_path) + summary_rows, _ = render_survival_chart([dummy_prediction()], output_path) - self.assertAlmostEqual(summary_rows[0]["health_probability"], 0.4) - self.assertAlmostEqual(summary_rows[0]["health_risk"], 0.6) + self.assertAlmostEqual(summary_rows[0]["health_state"], 0.4) self.assertEqual(summary_rows[0]["grade_label"], "II级") xlabel.assert_called_once() self.assertEqual(xlabel.call_args.args[0], "管龄（年）") ylabel.assert_called_once() - self.assertEqual(ylabel.call_args.args[0], "健康风险") + self.assertEqual(ylabel.call_args.args[0], "管道健康状态") title.assert_called_once() - self.assertEqual(title.call_args.args[0], "管道的剩余寿命分析图") + self.assertEqual(title.call_args.args[0], "供水管道健康状态曲线") figtext.assert_called_once() self.assertIn(f"前{CHART_DISPLAY_LIMIT}条管道的示例数据", figtext.call_args.args[2]) @@ -223,12 +266,15 @@ class PredictionHelpersTest(unittest.TestCase): PIPE_AGE_COLUMN: [12] * sample_count, } ) - curves = [DummyCurve([1, 10, 12], [0.9, 0.7, 0.4]) for _ in range(sample_count)] + predictions = [ + dummy_prediction(f"P{i:03d}", [1, 10, 12], [0.9, 0.7, 0.4]) + for i in range(sample_count) + ] with TemporaryDirectory() as temp_dir: output_path = Path(temp_dir) / "chart.png" with patch("app.prediction.plt.step") as step, patch("app.prediction.plt.legend"): - summary_rows, summary_sheet_rows = render_survival_chart(df, curves, output_path) + summary_rows, summary_sheet_rows = render_survival_chart(predictions, output_path) self.assertEqual(step.call_count, CHART_DISPLAY_LIMIT) self.assertEqual(len(summary_rows), sample_count)