feat: align prediction report with guidance

app/prediction.py

@@ -56,7 +56,10 @@ FEATURES = [
 
 ID_COLUMN = "管道编号"
 PIPE_AGE_COLUMN = "管龄（年）"
-REMAINING_LIFE_COLUMN = "预计剩余寿命（年）（管道健康状态 ≥0.5 的剩余年数）"
+REMAINING_LIFE_COLUMN = (
+    "预计剩余寿命（年）（管道健康状态 ≥0.5 的剩余年数；"
+    "如管道健康状态在73年内均高于0.5，则为73减去当前管龄）"
+)
 LEGACY_PIPE_AGE_COLUMN = "管龄"
 COLUMN_ALIASES = {
     "ID": ID_COLUMN,
@@ -402,7 +405,7 @@ def make_analysis_text(summary_rows: list[dict[str, Any]]) -> str:
     return (
         "曲线表示模型对不同管道随管龄变化的健康状态动态预测。"
         f"当前样本中风险最高管道为 {worst['pipe_id']}（{worst['grade_label']}），"
-        f"当前健康状态最高管道为 {best['pipe_id']}（{best['health_state']:.1%}）。"
+        f"当前健康状态最高管道为 {best['pipe_id']}（{best['health_state']:.2f}）。"
     )
 
 
@@ -460,9 +463,9 @@ 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:
+def format_years(value: float | None, current_age: float, max_age: float) -> float:
     if value is None:
-        return f">{max(max_age - current_age, 0.0):g}"
+        return round(max(max_age - current_age, 0.0), 3)
     return round(float(value), 3)
 
 
@@ -497,7 +500,7 @@ def render_survival_chart(predictions: list[dict[str, Any]], image_path: Path) -
         summary_sheet_rows.append(
             {
                 ID_COLUMN: pipe_id,
-                PIPE_AGE_COLUMN: pipe_age,
+                PIPE_AGE_COLUMN: current_age,
                 "当前健康状态": current_health_state,
                 "健康等级": grade_label,
                 REMAINING_LIFE_COLUMN: format_years(remaining_life, current_age, max_age),
@@ -509,7 +512,6 @@ def render_survival_chart(predictions: list[dict[str, Any]], image_path: Path) -
     font_kwargs = chinese_font_kwargs()
     plt.xlabel("管龄（年）", **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:
@@ -561,7 +563,7 @@ def write_prediction_workbook(
         sample_worksheet = writer.book["样本数据"]
         summary_worksheet = writer.book["结果摘要"]
         for cell in summary_worksheet["C"][1:]:
-            cell.number_format = "0.0%"
+            cell.number_format = "0.00"
         sample_worksheet.freeze_panes = "A2"
         sample_worksheet.auto_filter.ref = sample_worksheet.dimensions
         sample_worksheet.column_dimensions["A"].width = 12
@@ -569,7 +571,7 @@ def write_prediction_workbook(
             header_cell = column_cells[0]
             sample_worksheet.column_dimensions[header_cell.column_letter].width = max(12, len(str(header_cell.value)) + 2)
             for cell in column_cells[1:]:
-                cell.number_format = "0.0%"
+                cell.number_format = "0.00"
 
         for sheet_name in ("结果摘要", "样本数据"):
             worksheet = writer.book[sheet_name]

docker-compose.yml

@@ -22,3 +22,4 @@ services:
       - ./data:/app/data
       - ./data/uploads:/app/uploads
       - ./data/images:/app/static/images
+      - ./example.xlsx:/app/example.xlsx:ro

tests/test_prediction.py

@@ -18,8 +18,10 @@ from app.prediction import (
     REMAINING_LIFE_COLUMN,
     PredictionError,
     estimate_remaining_life,
+    format_years,
     grade_info,
     interpolate_probability,
+    make_analysis_text,
     prepare_model_features,
     read_input_file,
     render_survival_chart,
@@ -60,6 +62,18 @@ class PredictionHelpersTest(unittest.TestCase):
     def test_probability_helpers(self) -> None:
         self.assertEqual(interpolate_probability([1, 5, 10], [0.9, 0.8, 0.6], 6), 0.6)
         self.assertEqual(estimate_remaining_life([1, 5, 10], [0.9, 0.4, 0.2]), 5.0)
+        self.assertEqual(format_years(None, 10, 73), 63.0)
+
+    def test_analysis_text_displays_health_state_as_decimal(self) -> None:
+        text = make_analysis_text(
+            [
+                {"pipe_id": "P001", "health_state": 0.734, "grade_label": "IV级"},
+                {"pipe_id": "P002", "health_state": 0.2, "grade_label": "I级"},
+            ]
+        )
+
+        self.assertIn("0.73", text)
+        self.assertNotIn("%", text)
 
     def test_grade_boundaries(self) -> None:
         self.assertEqual(grade_info(0.2)[0], "I级")
@@ -79,13 +93,12 @@ class PredictionHelpersTest(unittest.TestCase):
 
         validate_input_frame(df)
 
-    def test_example_workbook_columns_match_backend_contract(self) -> None:
+    def test_example_workbook_contains_backend_contract_columns(self) -> None:
         workbook = load_workbook(Path(__file__).resolve().parents[1] / "example.xlsx", read_only=True)
         worksheet = workbook["Template"]
         headers = [cell.value for cell in next(worksheet.iter_rows(min_row=1, max_row=1))]
 
-        self.assertEqual(len(headers), len(INPUT_COLUMNS))
+        self.assertTrue(set(INPUT_COLUMNS).issubset(headers))
-        self.assertEqual(set(headers), set(INPUT_COLUMNS))
 
     def test_read_input_file_preserves_text_pipe_ids(self) -> None:
         with TemporaryDirectory() as temp_dir:
@@ -183,8 +196,8 @@ class PredictionHelpersTest(unittest.TestCase):
         ]
         summary_rows = [{"pipe_id": "P001"}, {"pipe_id": "P002"}]
         summary_sheet_rows = [
-            {ID_COLUMN: "P001", PIPE_AGE_COLUMN: "10 年", "当前健康状态": 0.7, "健康等级": "IV级", REMAINING_LIFE_COLUMN: ">63"},
+            {ID_COLUMN: "P001", PIPE_AGE_COLUMN: 10, "当前健康状态": 0.7, "健康等级": "IV级", REMAINING_LIFE_COLUMN: 63},
-            {ID_COLUMN: "P002", PIPE_AGE_COLUMN: "12 年", "当前健康状态": 0.6, "健康等级": "III级", REMAINING_LIFE_COLUMN: ">61"},
+            {ID_COLUMN: "P002", PIPE_AGE_COLUMN: 12, "当前健康状态": 0.6, "健康等级": "III级", REMAINING_LIFE_COLUMN: 61},
         ]
 
         with TemporaryDirectory() as temp_dir:
@@ -199,6 +212,8 @@ class PredictionHelpersTest(unittest.TestCase):
                 summary_data.columns.tolist(),
                 [ID_COLUMN, PIPE_AGE_COLUMN, "当前健康状态", "健康等级", REMAINING_LIFE_COLUMN],
             )
+            self.assertEqual(summary_data[PIPE_AGE_COLUMN].tolist(), [10, 12])
+            self.assertEqual(summary_data[REMAINING_LIFE_COLUMN].tolist(), [63, 61])
 
             sample_data = pd.read_excel(output_path, sheet_name="样本数据")
             self.assertEqual(sample_data.columns.tolist(), ["管龄（年）", "P001", "P002"])
@@ -209,15 +224,19 @@ class PredictionHelpersTest(unittest.TestCase):
 
             openpyxl_workbook = load_workbook(output_path)
             sample_worksheet = openpyxl_workbook["样本数据"]
+            summary_worksheet = openpyxl_workbook["结果摘要"]
             self.assertEqual(sample_worksheet.freeze_panes, "A2")
             self.assertEqual(sample_worksheet.auto_filter.ref, "A1:C3")
-            self.assertEqual(sample_worksheet["B2"].number_format, "0.0%")
+            self.assertEqual(summary_worksheet["C2"].number_format, "0.00")
+            self.assertEqual(sample_worksheet["B2"].number_format, "0.00")
+            self.assertIn("73减去当前管龄", summary_worksheet["E1"].value)
+            self.assertNotIn("73 - 当前管龄", summary_worksheet["E1"].value)
 
     def test_prediction_workbook_writes_pipe_ids_as_excel_text(self) -> None:
         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.9, "健康等级": "V级", REMAINING_LIFE_COLUMN: ">63"}
+            {ID_COLUMN: "00123", PIPE_AGE_COLUMN: 10, "当前健康状态": 0.9, "健康等级": "V级", REMAINING_LIFE_COLUMN: 63}
         ]
 
         with TemporaryDirectory() as temp_dir:
@@ -243,19 +262,17 @@ class PredictionHelpersTest(unittest.TestCase):
                 patch("app.prediction.plt.xlabel") as xlabel,
                 patch("app.prediction.plt.ylabel") as ylabel,
                 patch("app.prediction.plt.figtext") as figtext,
-                patch("app.prediction.plt.title") as title,
             ):
-                summary_rows, _ = render_survival_chart([dummy_prediction()], output_path)
+                summary_rows, summary_sheet_rows = render_survival_chart([dummy_prediction()], output_path)
 
         self.assertAlmostEqual(summary_rows[0]["health_state"], 0.4)
         self.assertEqual(summary_rows[0]["grade_label"], "II级")
+        self.assertEqual(summary_sheet_rows[0][PIPE_AGE_COLUMN], 12.0)
 
         xlabel.assert_called_once()
         self.assertEqual(xlabel.call_args.args[0], "管龄（年）")
         ylabel.assert_called_once()
         self.assertEqual(ylabel.call_args.args[0], "管道健康状态")
-        title.assert_called_once()
-        self.assertEqual(title.call_args.args[0], "供水管道健康状态曲线")
         figtext.assert_called_once()
         self.assertIn(f"前{CHART_DISPLAY_LIMIT}条管道的示例数据", figtext.call_args.args[2])