diff --git a/app/prediction.py b/app/prediction.py index 2215b07..410815e 100644 --- a/app/prediction.py +++ b/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] diff --git a/docker-compose.yml b/docker-compose.yml index a27ef44..ad50df9 100644 --- a/docker-compose.yml +++ b/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 diff --git a/example.xlsx b/example.xlsx index a370edf..500aadc 100644 Binary files a/example.xlsx and b/example.xlsx differ diff --git a/tests/test_prediction.py b/tests/test_prediction.py index 1ccb059..2482d8c 100644 --- a/tests/test_prediction.py +++ b/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.assertEqual(set(headers), set(INPUT_COLUMNS)) + self.assertTrue(set(INPUT_COLUMNS).issubset(headers)) 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: "P002", PIPE_AGE_COLUMN: "12 年", "当前健康状态": 0.6, "健康等级": "III级", REMAINING_LIFE_COLUMN: ">61"}, + {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}, ] 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])