feat: update prediction workbook output

This commit is contained in:
2026-07-06 18:33:52 +08:00
parent c7ee2adb82
commit 0a12eb7f3b
8 changed files with 219 additions and 78 deletions
+117 -20
View File
@@ -58,6 +58,39 @@ FEATURES = [
ID_COLUMN = "管道编号"
PIPE_AGE_COLUMN = "管龄"
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_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
}
)
SUPPORTED_EXTENSIONS = {".csv", ".xls", ".xlsx"}
CHINESE_FONT_PROP = None
@@ -150,13 +183,21 @@ def secure_upload_name(original_filename: str, run_id: str) -> tuple[str, str]:
def read_input_file(path: Path, suffix: str) -> pd.DataFrame:
try:
if suffix == ".csv":
return pd.read_csv(path)
return pd.read_excel(path)
return pd.read_csv(path, dtype={ID_COLUMN: "string"})
return pd.read_excel(path, dtype={ID_COLUMN: "string"})
except Exception as exc:
logging.exception("文件解析失败: %s", exc)
raise PredictionError("文件解析失败,请检查编码或表格格式。")
def normalize_id_value(value: Any, fallback: str) -> str:
if pd.isna(value):
return fallback
if isinstance(value, str):
return value.strip()
return str(value)
def validate_input_frame(df: pd.DataFrame) -> None:
if df.empty:
raise PredictionError("上传文件没有可预测的数据。")
@@ -166,6 +207,28 @@ def validate_input_frame(df: pd.DataFrame) -> None:
raise PredictionError(f"缺少必要字段: {', '.join(missing)}")
def normalize_material_code(value: Any) -> int:
if pd.isna(value):
raise PredictionError("管材不能为空。")
if isinstance(value, str):
key = value.strip().casefold()
else:
numeric_value = float(value)
if not numeric_value.is_integer():
raise PredictionError(f"管材编码无效: {value}")
key = str(int(numeric_value))
code = MATERIAL_ALIAS_TO_CODE.get(key)
if code is None:
raise PredictionError(f"管材编码无效: {value}")
return code
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
def grade_info(probability: float) -> tuple[str, str, str]:
if probability <= 0.2:
return ("I级", "管道安全风险十分严重,需立刻进行抢修或更新改造", "bg-dangerSoft text-dangerText")
@@ -318,7 +381,7 @@ def run_prediction(uploaded: FileStorage, user_id: int, model) -> PredictionArti
try:
df = read_input_file(original_path, suffix)
validate_input_frame(df)
x_test = df[FEATURES].copy()
x_test = prepare_model_features(df)
try:
curves = model.predict_survival_function(x_test)
except Exception as exc:
@@ -365,7 +428,7 @@ def render_survival_chart(df: pd.DataFrame, curves, image_path: Path) -> tuple[l
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 = str(df.iloc[i][ID_COLUMN]) if pd.notna(df.iloc[i][ID_COLUMN]) else f"Pipe_{i+1:03d}"
pipe_id = normalize_id_value(df.iloc[i][ID_COLUMN], f"Pipe_{i+1:03d}")
pipe_age = f"{df.iloc[i][PIPE_AGE_COLUMN]}" if PIPE_AGE_COLUMN in df.columns and pd.notna(df.iloc[i][PIPE_AGE_COLUMN]) else "-"
health_probability = interpolate_probability(times, probs, 10)
remaining_life = estimate_remaining_life(times, probs)
@@ -386,15 +449,13 @@ def render_survival_chart(df: pd.DataFrame, curves, image_path: Path) -> tuple[l
{
"管道编号": pipe_id,
"管龄": pipe_age,
"健康概率": health_probability,
"预计剩余寿命": remaining_life,
"健康等级": grade_label,
}
)
plt.step(times, probs, where="post", linewidth=2, label=pipe_id)
font_kwargs = chinese_font_kwargs()
plt.xlabel("预测时间轴(年)", **font_kwargs)
plt.xlabel("管龄(年)", **font_kwargs)
plt.ylabel("生存概率", **font_kwargs)
plt.title("预测分析图", **font_kwargs)
plt.grid(alpha=0.18)
@@ -416,22 +477,58 @@ def write_prediction_workbook(
summary_rows: list[dict[str, Any]],
summary_sheet_rows: list[dict[str, Any]],
) -> None:
sample_data_rows: list[dict[str, Any]] = []
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)]
pipe_id = summary_rows[i]["pipe_id"]
for time, probability in zip(times, probs):
sample_data_rows.append(
{
"管道编号": pipe_id,
"样本序号": i + 1,
"时间(年)": time,
"生存概率": probability,
"风险概率": 1 - probability,
}
)
chart_times.update(times)
chart_series.append((pipe_id, dict(zip(times, probs))))
sorted_times = sorted(chart_times)
sample_columns = ["管龄(年)", *[pipe_id for pipe_id, _ in chart_series]]
sample_data_rows = [
[time, *[series.get(time) for _, series in chart_series]]
for time in sorted_times
]
with pd.ExcelWriter(excel_path, engine="openpyxl") as writer:
pd.DataFrame(summary_sheet_rows).to_excel(writer, sheet_name="结果摘要", index=False)
pd.DataFrame(sample_data_rows).to_excel(writer, sheet_name="样本数据", index=False)
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")
summary_df.to_excel(writer, sheet_name="结果摘要", index=False)
sample_df.to_excel(writer, sheet_name="样本数据", index=False)
sample_worksheet = writer.book["样本数据"]
sample_worksheet.freeze_panes = "A2"
sample_worksheet.auto_filter.ref = sample_worksheet.dimensions
sample_worksheet.column_dimensions["A"].width = 12
for column_cells in sample_worksheet.iter_cols(min_col=2, max_col=sample_worksheet.max_column):
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%"
for sheet_name in ("结果摘要", "样本数据"):
worksheet = writer.book[sheet_name]
header_cells = next(worksheet.iter_rows(min_row=1, max_row=1), [])
id_column_index = None
for cell in header_cells:
if cell.value == ID_COLUMN:
id_column_index = cell.column
break
if id_column_index is None:
continue
for cell in worksheet.iter_cols(
min_col=id_column_index,
max_col=id_column_index,
min_row=2,
max_row=worksheet.max_row,
):
for item in cell:
if item.value is not None:
item.value = str(item.value)
item.number_format = "@"
+2 -1
View File
@@ -30,6 +30,7 @@ from .time_utils import format_datetime_for_timezone
bp = Blueprint("main", __name__)
REFERENCE_PDF_NAME = "20260630标准文本——供水管道健康状态与剩余寿命评估技术导则.pdf"
TEMPLATE_EXCEL_NAME = "管道预测数据模板.xlsx"
REGISTRATION_SETTING_KEY = "allow_registration"
RECORDS_PER_PAGE = 10
@@ -375,7 +376,7 @@ def download_template():
template_path = BASE_DIR / "example.xlsx"
if not template_path.exists():
abort(404)
return send_file(template_path, as_attachment=True, download_name="example.xlsx")
return send_file(template_path, as_attachment=True, download_name=TEMPLATE_EXCEL_NAME)
@bp.route("/reference_pdf")
BIN
View File
Binary file not shown.
-41
View File
@@ -19,7 +19,6 @@ const resultImportanceWrap = document.getElementById('resultImportanceWrap');
const resultImportanceImage = document.getElementById('resultImportanceImage');
const excelBtn = document.getElementById('excelBtn');
const resultPageBtn = document.getElementById('resultPageBtn');
const homeStateKey = 'pipelineLifetime.homeState';
let selectedFile = null;
let alertTimer = null;
let alertHideTimer = null;
@@ -66,24 +65,6 @@ function updateSelectedFile(file) {
selectedFile = file;
selectedFileName.textContent = '已选择文件:' + file.name;
selectedFileName.classList.remove('hidden');
saveHomeState({ selectedFilename: file.name });
}
function readHomeState() {
try {
return JSON.parse(sessionStorage.getItem(homeStateKey)) || {};
} catch (err) {
return {};
}
}
function saveHomeState(nextState) {
try {
const currentState = readHomeState();
sessionStorage.setItem(homeStateKey, JSON.stringify({ ...currentState, ...nextState }));
} catch (err) {
// Ignore storage failures; prediction still works without client-side restore.
}
}
function renderResult(data) {
@@ -102,17 +83,6 @@ function renderResult(data) {
resultContent.classList.add('flex');
}
function restoreHomeState() {
const savedState = readHomeState();
if (savedState.selectedFilename) {
selectedFileName.textContent = '上次选择文件:' + savedState.selectedFilename + '(需重新选择)';
selectedFileName.classList.remove('hidden');
}
if (savedState.result) {
renderResult(savedState.result);
}
}
fileInput.addEventListener('change', () => {
const file = fileInput.files[0];
if (!file) return;
@@ -176,15 +146,6 @@ form.addEventListener('submit', async (e) => {
return;
}
renderResult(data);
saveHomeState({
selectedFilename: data.original_filename || selectedFile.name,
result: {
image_url: data.image_url,
importance_url: data.importance_url || '',
excel_url: data.excel_url,
result_url: data.result_url
}
});
showAlert('预测完成,已生成图表与电子表格报告。', 'success');
inlineResult.scrollIntoView({ behavior: 'smooth', block: 'nearest' });
} catch (err) {
@@ -197,5 +158,3 @@ form.addEventListener('submit', async (e) => {
submitIcon = restoredIcon;
}
});
restoreHomeState();
-9
View File
@@ -142,15 +142,6 @@
showAppNotification(message, category === 'error' ? 'error' : 'info');
}
document.querySelectorAll('[data-clear-home-state]').forEach((form) => {
form.addEventListener('submit', () => {
try {
sessionStorage.removeItem('pipelineLifetime.homeState');
} catch (err) {
// Ignore storage failures during logout.
}
});
});
})();
</script>
{% block scripts %}{% endblock %}
+4 -2
View File
@@ -51,12 +51,13 @@
</section>
<section class="flex flex-1 flex-col rounded-lg border border-line bg-white p-5 shadow-panel">
<h2 class="text-lg font-extrabold">样本预览</h2>
<h2 class="text-lg font-extrabold">结果展示</h2>
<div class="mt-4 overflow-hidden rounded-md border border-line">
<table class="w-full text-sm">
<thead class="bg-slate-50 text-xs font-bold text-textSub">
<tr>
<th class="px-3 py-3 text-left">管道编号</th>
<th class="px-3 py-3 text-left">编号</th>
<th class="px-3 py-3 text-left">管龄</th>
<th class="px-3 py-3 text-left">等级</th>
</tr>
</thead>
@@ -64,6 +65,7 @@
{% for item in result.summary_rows %}
<tr class="border-t border-line">
<td class="px-3 py-3 font-semibold">{{ item.pipe_id }}</td>
<td class="px-3 py-3 text-textSub">{{ item.pipe_age }}</td>
<td class="px-3 py-3"><span class="inline-flex rounded-full px-2.5 py-1 text-xs font-bold {{ item.grade_class }}">{{ item.grade_label }}</span></td>
</tr>
{% endfor %}
+13
View File
@@ -130,6 +130,19 @@ class RegistrationRoutesTest(unittest.TestCase):
self.assertIn('name="captcha"', html)
self.assertIn(captcha, html)
def test_template_download_uses_chinese_filename(self) -> None:
with TemporaryDirectory() as temp_dir:
app = self.create_test_app(temp_dir, allow_registration=True)
response = app.test_client().get("/download_template")
self.assertEqual(response.status_code, 200)
self.assertIn("attachment", response.headers["Content-Disposition"])
self.assertIn(
"filename*=UTF-8''%E7%AE%A1%E9%81%93%E9%A2%84%E6%B5%8B%E6%95%B0%E6%8D%AE%E6%A8%A1%E6%9D%BF.xlsx",
response.headers["Content-Disposition"],
)
def test_register_post_does_not_create_user_when_registration_is_closed(self) -> None:
with TemporaryDirectory() as temp_dir:
app = self.create_test_app(temp_dir, allow_registration=False)
+83 -5
View File
@@ -3,16 +3,22 @@ from __future__ import annotations
import unittest
from pathlib import Path
from tempfile import TemporaryDirectory
from unittest.mock import patch
import pandas as pd
from openpyxl import load_workbook
from app.prediction import (
FEATURES,
ID_COLUMN,
PIPE_AGE_COLUMN,
PredictionError,
estimate_remaining_life,
grade_info,
interpolate_probability,
prepare_model_features,
read_input_file,
render_survival_chart,
secure_upload_name,
validate_input_frame,
write_prediction_workbook,
@@ -53,6 +59,37 @@ class PredictionHelpersTest(unittest.TestCase):
self.assertIn("缺少必要字段", ctx.exception.message)
def test_read_input_file_preserves_text_pipe_ids(self) -> None:
with TemporaryDirectory() as temp_dir:
output_path = Path(temp_dir) / "input.xlsx"
pd.DataFrame({ID_COLUMN: ["00123"], FEATURES[0]: [1]}).to_excel(output_path, index=False)
df = read_input_file(output_path, ".xlsx")
self.assertEqual(df[ID_COLUMN].tolist(), ["00123"])
def test_prepare_model_features_maps_material_aliases_to_codes(self) -> None:
rows = []
for value in ["镀锌", "2-钢塑", 13]:
row = {feature: 1 for feature in FEATURES}
row["管材"] = value
rows.append(row)
df = pd.DataFrame(rows)
x_test = prepare_model_features(df)
self.assertEqual(x_test["管材"].tolist(), [1, 2, 13])
def test_prepare_model_features_rejects_invalid_material_alias(self) -> None:
row = {feature: 1 for feature in FEATURES}
row["管材"] = "未知管材"
df = pd.DataFrame([row])
with self.assertRaises(PredictionError) as ctx:
prepare_model_features(df)
self.assertIn("管材编码无效", ctx.exception.message)
def test_prediction_workbook_keeps_sample_data_in_one_sheet(self) -> None:
curves = [
DummyCurve([1, 2], [0.9, 0.7]),
@@ -60,8 +97,8 @@ class PredictionHelpersTest(unittest.TestCase):
]
summary_rows = [{"pipe_id": "P001"}, {"pipe_id": "P002"}]
summary_sheet_rows = [
{"管道编号": "P001", "健康概率": 0.7},
{"管道编号": "P002", "健康概率": 0.6},
{"管道编号": "P001", "管龄": "10 年", "健康概率": 0.7, "预计剩余寿命": 2.0, "健康等级": "IV级"},
{"管道编号": "P002", "管龄": "12 年", "健康概率": 0.6, "预计剩余寿命": 1.5, "健康等级": "III级"},
]
with TemporaryDirectory() as temp_dir:
@@ -71,10 +108,51 @@ class PredictionHelpersTest(unittest.TestCase):
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, "健康等级"])
sample_data = pd.read_excel(output_path, sheet_name="样本数据")
self.assertEqual(len(sample_data), 4)
self.assertEqual(sample_data["管道编号"].tolist(), ["P001", "P001", "P002", "P002"])
self.assertIn("风险概率", sample_data.columns)
self.assertEqual(sample_data.columns.tolist(), ["管龄(年)", "P001", "P002"])
self.assertEqual(len(sample_data), 2)
self.assertEqual(sample_data["管龄(年)"].tolist(), [1, 2])
self.assertEqual(sample_data["P001"].tolist(), [0.9, 0.7])
self.assertEqual(sample_data["P002"].tolist(), [0.8, 0.6])
openpyxl_workbook = load_workbook(output_path)
sample_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%")
def test_prediction_workbook_writes_pipe_ids_as_excel_text(self) -> None:
curves = [DummyCurve([1], [0.9])]
summary_rows = [{"pipe_id": "00123"}]
summary_sheet_rows = [{ID_COLUMN: "00123", PIPE_AGE_COLUMN: "10 年", "健康等级": "V级"}]
with TemporaryDirectory() as temp_dir:
output_path = Path(temp_dir) / "prediction.xlsx"
write_prediction_workbook(output_path, curves, summary_rows, summary_sheet_rows)
workbook = load_workbook(output_path)
summary_worksheet = workbook["结果摘要"]
self.assertEqual(summary_worksheet["A2"].value, "00123")
self.assertEqual(summary_worksheet["A2"].data_type, "s")
self.assertEqual(summary_worksheet["A2"].number_format, "@")
sample_worksheet = workbook["样本数据"]
self.assertEqual(sample_worksheet["B1"].value, "00123")
self.assertEqual(sample_worksheet["B1"].data_type, "s")
def test_survival_chart_uses_pipe_age_x_axis_label(self) -> None:
df = pd.DataFrame({ID_COLUMN: ["P001"], PIPE_AGE_COLUMN: [12]})
with TemporaryDirectory() as temp_dir:
output_path = Path(temp_dir) / "chart.png"
with patch("app.prediction.plt.xlabel") as xlabel:
render_survival_chart(df, [DummyCurve([1, 2], [0.9, 0.7])], output_path)
xlabel.assert_called_once()
self.assertEqual(xlabel.call_args.args[0], "管龄(年)")
if __name__ == "__main__":