594 lines
22 KiB
Python
594 lines
22 KiB
Python
from __future__ import annotations
|
||
|
||
import logging
|
||
import sys
|
||
import uuid
|
||
from dataclasses import dataclass
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import matplotlib
|
||
|
||
matplotlib.use("Agg")
|
||
import matplotlib.pyplot as plt
|
||
import pandas as pd
|
||
from matplotlib import font_manager, rcParams
|
||
from werkzeug.datastructures import FileStorage
|
||
from werkzeug.utils import secure_filename
|
||
|
||
from .config import IMAGE_DIR, UPLOAD_DIR
|
||
from .time_utils import utc_now
|
||
|
||
CHINESE_FONT_CANDIDATES = [
|
||
"Noto Sans CJK SC",
|
||
"Noto Sans SC",
|
||
"Noto Sans CJK JP",
|
||
"Noto Sans CJK TC",
|
||
"Source Han Sans SC",
|
||
"WenQuanYi Micro Hei",
|
||
"SimHei",
|
||
"Microsoft YaHei",
|
||
"Arial Unicode MS",
|
||
]
|
||
|
||
CHINESE_FONT_FILES = [
|
||
"/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc",
|
||
"/usr/share/fonts/truetype/noto/NotoSansSC-Regular.ttf",
|
||
"/usr/share/fonts/truetype/noto/NotoSansCJK-Regular.ttc",
|
||
"/usr/local/share/fonts/NotoSansCJK-Regular.ttc",
|
||
"/mnt/c/Windows/Fonts/NotoSansSC-VF.ttf",
|
||
"/mnt/c/Windows/Fonts/msyh.ttc",
|
||
"/mnt/c/Windows/Fonts/simhei.ttf",
|
||
"/mnt/c/Windows/Fonts/simsun.ttc",
|
||
]
|
||
|
||
FEATURES = [
|
||
"管材",
|
||
"管径",
|
||
"流速",
|
||
"压力",
|
||
"温度",
|
||
"降雨量",
|
||
"位置",
|
||
"结构缺陷",
|
||
"功能缺陷",
|
||
]
|
||
|
||
ID_COLUMN = "管道编号"
|
||
PIPE_AGE_COLUMN = "管龄（年）"
|
||
REMAINING_LIFE_COLUMN = "预计剩余寿命（年）（管道健康状态 ≥0.5 的剩余年数）"
|
||
LEGACY_PIPE_AGE_COLUMN = "管龄"
|
||
COLUMN_ALIASES = {
|
||
"ID": ID_COLUMN,
|
||
"管段ID": ID_COLUMN,
|
||
"管段编号": ID_COLUMN,
|
||
"当前管龄": PIPE_AGE_COLUMN,
|
||
"当前管龄（年）": PIPE_AGE_COLUMN,
|
||
"管径（mm）": "管径",
|
||
"管径（mm)": "管径",
|
||
"流速（m/s）": "流速",
|
||
"压力（MPa)": "压力",
|
||
"压力（MPa）": "压力",
|
||
"温度（℃）": "温度",
|
||
"年均降雨量（mm）": "降雨量",
|
||
"降雨量（mm）": "降雨量",
|
||
"结构性缺陷": "结构缺陷",
|
||
"功能性缺陷": "功能缺陷",
|
||
}
|
||
INPUT_COLUMNS = [
|
||
ID_COLUMN,
|
||
PIPE_AGE_COLUMN,
|
||
*FEATURES,
|
||
]
|
||
REQUIRED_INPUT_COLUMNS = [ID_COLUMN, PIPE_AGE_COLUMN, "管材", "管径"]
|
||
CHART_DISPLAY_LIMIT = 10
|
||
MATERIAL_COLUMN = "管材"
|
||
MATERIAL_CODE_TO_NAME = {
|
||
1: "镀锌",
|
||
2: "钢塑",
|
||
3: "铝塑",
|
||
4: "PPR",
|
||
5: "PE",
|
||
6: "UPVC",
|
||
7: "铸铁",
|
||
11: "钢管",
|
||
}
|
||
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}
|
||
|
||
|
||
class PredictionError(Exception):
|
||
def __init__(self, message: str, status_code: int = 400) -> None:
|
||
super().__init__(message)
|
||
self.message = message
|
||
self.status_code = status_code
|
||
|
||
|
||
@dataclass
|
||
class PredictionArtifacts:
|
||
original_filename: str
|
||
saved_path: Path
|
||
excel_path: Path
|
||
image_path: Path
|
||
image_filename: str
|
||
sample_count: int
|
||
summary_rows: list[dict[str, Any]]
|
||
analysis_text: str
|
||
model_version: str
|
||
|
||
|
||
def configure_matplotlib_fonts():
|
||
for font_path in CHINESE_FONT_FILES:
|
||
path = Path(font_path)
|
||
if path.exists():
|
||
font_manager.fontManager.addfont(str(path))
|
||
prop = font_manager.FontProperties(fname=str(path))
|
||
rcParams["font.family"] = [prop.get_name(), "sans-serif"]
|
||
rcParams["font.sans-serif"] = [prop.get_name(), *CHINESE_FONT_CANDIDATES, "DejaVu Sans"]
|
||
rcParams["axes.unicode_minus"] = False
|
||
return prop
|
||
|
||
available_fonts = {font.name for font in font_manager.fontManager.ttflist}
|
||
selected_fonts = [font for font in CHINESE_FONT_CANDIDATES if font in available_fonts]
|
||
if selected_fonts:
|
||
rcParams["font.family"] = [selected_fonts[0], "sans-serif"]
|
||
rcParams["font.sans-serif"] = selected_fonts + ["DejaVu Sans"]
|
||
rcParams["axes.unicode_minus"] = False
|
||
if selected_fonts:
|
||
return font_manager.FontProperties(family=selected_fonts[0])
|
||
logging.warning("未找到中文字体，生成的图表中文可能无法显示。")
|
||
return None
|
||
|
||
|
||
CHINESE_FONT_PROP = configure_matplotlib_fonts()
|
||
|
||
|
||
def chinese_font_kwargs() -> dict[str, Any]:
|
||
return {"fontproperties": CHINESE_FONT_PROP} if CHINESE_FONT_PROP else {}
|
||
|
||
|
||
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:
|
||
try:
|
||
path.unlink(missing_ok=True)
|
||
except OSError as exc:
|
||
logging.warning("删除文件失败 %s: %s", path, exc)
|
||
|
||
|
||
def secure_upload_name(original_filename: str, run_id: str) -> tuple[str, str]:
|
||
suffix = Path(original_filename).suffix.lower()
|
||
if suffix not in SUPPORTED_EXTENSIONS:
|
||
raise PredictionError("不支持的格式，仅支持逗号分隔值文件或电子表格文件")
|
||
|
||
safe_full_name = secure_filename(original_filename)
|
||
safe_stem = Path(safe_full_name).stem if safe_full_name else ""
|
||
if not safe_stem:
|
||
safe_stem = "upload"
|
||
return f"{safe_stem}_{run_id}{suffix}", suffix
|
||
|
||
|
||
def read_input_file(path: Path, suffix: str) -> pd.DataFrame:
|
||
try:
|
||
if suffix == ".csv":
|
||
df = pd.read_csv(path, dtype={ID_COLUMN: "string"})
|
||
else:
|
||
try:
|
||
df = pd.read_excel(path, sheet_name="Template", dtype={ID_COLUMN: "string"})
|
||
except ValueError:
|
||
df = pd.read_excel(path, dtype={ID_COLUMN: "string"})
|
||
df["_excel_row"] = df.index + 2
|
||
except Exception as exc:
|
||
logging.exception("文件解析失败: %s", exc)
|
||
raise PredictionError("文件解析失败，请检查编码或表格格式。")
|
||
df = normalize_input_columns(df)
|
||
if suffix in {".xls", ".xlsx"}:
|
||
df = fill_defects_from_detail_sheet(path, df)
|
||
return df
|
||
|
||
|
||
def normalize_input_columns(df: pd.DataFrame) -> pd.DataFrame:
|
||
aliases = dict(COLUMN_ALIASES)
|
||
if PIPE_AGE_COLUMN not in df.columns and LEGACY_PIPE_AGE_COLUMN in df.columns:
|
||
aliases[LEGACY_PIPE_AGE_COLUMN] = PIPE_AGE_COLUMN
|
||
return df.rename(columns={name: aliases[name] for name in df.columns if name in aliases})
|
||
|
||
|
||
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 defect_grade_value(value: Any) -> float | None:
|
||
if pd.isna(value):
|
||
return None
|
||
text = str(value).strip()
|
||
if text == "":
|
||
return None
|
||
if text in DEFECT_GRADE_VALUES:
|
||
return DEFECT_GRADE_VALUES[text]
|
||
try:
|
||
return float(text)
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
def weighted_defect_score(values: list[Any], weights: list[float]) -> float | None:
|
||
scores = [defect_grade_value(value) for value in values]
|
||
if any(score is None for score in scores):
|
||
return None
|
||
return round(sum(score * weight for score, weight in zip(scores, weights)), 3)
|
||
|
||
|
||
def numeric_or_none(value: Any) -> float | None:
|
||
try:
|
||
if pd.isna(value):
|
||
return None
|
||
return float(value)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
|
||
def fill_defects_from_detail_sheet(path: Path, df: pd.DataFrame) -> pd.DataFrame:
|
||
if "结构缺陷" not in df.columns or "功能缺陷" not in df.columns or "_excel_row" not in df.columns:
|
||
return df
|
||
|
||
structure = pd.to_numeric(df["结构缺陷"], errors="coerce")
|
||
function = pd.to_numeric(df["功能缺陷"], errors="coerce")
|
||
needs_fill = structure.isna() | function.isna()
|
||
df["结构缺陷"] = structure
|
||
df["功能缺陷"] = function
|
||
if not needs_fill.any():
|
||
return df
|
||
|
||
try:
|
||
from openpyxl import load_workbook
|
||
|
||
workbook = load_workbook(path, data_only=False, read_only=True)
|
||
defect_sheet = workbook["缺陷计算"]
|
||
except Exception as exc:
|
||
logging.info("未能读取缺陷计算工作表，跳过缺陷自动计算: %s", exc)
|
||
return df
|
||
|
||
for index in df.index[needs_fill]:
|
||
excel_row = int(df.at[index, "_excel_row"])
|
||
if pd.isna(df.at[index, "结构缺陷"]):
|
||
value = numeric_or_none(defect_sheet.cell(excel_row, 5).value)
|
||
if value is None:
|
||
value = weighted_defect_score(
|
||
[defect_sheet.cell(excel_row, column).value for column in (2, 3, 4)],
|
||
[0.5, 0.4, 0.1],
|
||
)
|
||
df.at[index, "结构缺陷"] = value
|
||
if pd.isna(df.at[index, "功能缺陷"]):
|
||
value = numeric_or_none(defect_sheet.cell(excel_row, 10).value)
|
||
if value is None:
|
||
value = weighted_defect_score(
|
||
[defect_sheet.cell(excel_row, column).value for column in (6, 7, 8, 9)],
|
||
[0.5, 0.2, 0.2, 0.1],
|
||
)
|
||
df.at[index, "功能缺陷"] = value
|
||
|
||
return df
|
||
|
||
|
||
def validate_input_frame(df: pd.DataFrame) -> None:
|
||
if df.empty:
|
||
raise PredictionError("上传文件没有可预测的数据。")
|
||
missing = [col for col in REQUIRED_INPUT_COLUMNS if col not in df.columns]
|
||
if missing:
|
||
raise PredictionError(f"缺少必要字段: {', '.join(missing)}")
|
||
|
||
|
||
def normalize_category_value(
|
||
value: Any,
|
||
aliases: dict[str, str],
|
||
label: str,
|
||
*,
|
||
required: bool = False,
|
||
) -> str | pd.NA:
|
||
if pd.isna(value):
|
||
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 ValueError
|
||
key = str(int(numeric_value))
|
||
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:
|
||
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]:
|
||
if probability <= 0.2:
|
||
return ("I级", "管道安全风险十分严重，需立刻进行抢修或更新改造", "bg-dangerSoft text-dangerText")
|
||
if probability <= 0.4:
|
||
return ("II级", "管道安全风险较为严重，需尽快安排检修及加频巡检", "bg-orange-50 text-orange-600")
|
||
if probability <= 0.6:
|
||
return ("III级", "管道安全风险较低，需安排定期巡检", "bg-amber-50 text-amber-600")
|
||
if probability <= 0.8:
|
||
return ("IV级", "管道安全风险较小，维持常规巡视", "bg-blue-50 text-blue-600")
|
||
return ("V级", "管道安全，维持常规巡视", "bg-blueSoft text-primary")
|
||
|
||
|
||
def interpolate_probability(times: list[float], probs: list[float], target: float) -> float:
|
||
if not times:
|
||
return 0.0
|
||
if target <= times[0]:
|
||
return float(probs[0])
|
||
for idx in range(1, len(times)):
|
||
if times[idx] >= target:
|
||
return float(probs[idx])
|
||
return float(probs[-1])
|
||
|
||
|
||
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_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_state']:.1%}）。"
|
||
)
|
||
|
||
|
||
def run_prediction(uploaded: FileStorage, user_id: int, model) -> PredictionArtifacts:
|
||
original_filename = uploaded.filename or ""
|
||
if not original_filename:
|
||
raise PredictionError("未选择文件")
|
||
|
||
timestamp = utc_now().strftime("%Y%m%d%H%M%S")
|
||
run_id = f"{timestamp}_{uuid.uuid4().hex[:8]}"
|
||
saved_filename, suffix = secure_upload_name(original_filename, run_id)
|
||
|
||
user_dir = UPLOAD_DIR / f"user_{user_id}"
|
||
user_dir.mkdir(parents=True, exist_ok=True)
|
||
original_path = user_dir / saved_filename
|
||
uploaded.save(original_path)
|
||
|
||
try:
|
||
df = read_input_file(original_path, suffix)
|
||
validate_input_frame(df)
|
||
x_test = prepare_model_features(df)
|
||
try:
|
||
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(predictions, image_path)
|
||
|
||
safe_stem = Path(saved_filename).stem
|
||
excel_path = user_dir / f"{safe_stem}_pre.xlsx"
|
||
write_prediction_workbook(excel_path, predictions, summary_rows, summary_sheet_rows)
|
||
return PredictionArtifacts(
|
||
original_filename=original_filename,
|
||
saved_path=original_path,
|
||
excel_path=excel_path,
|
||
image_path=image_path,
|
||
image_filename=image_filename,
|
||
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 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, 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_state": current_health_state,
|
||
"remaining_life": remaining_life,
|
||
"grade_label": grade_label,
|
||
"grade_desc": grade_desc,
|
||
"grade_class": grade_class,
|
||
}
|
||
)
|
||
summary_sheet_rows.append(
|
||
{
|
||
ID_COLUMN: pipe_id,
|
||
PIPE_AGE_COLUMN: pipe_age,
|
||
"当前健康状态": current_health_state,
|
||
"健康等级": grade_label,
|
||
REMAINING_LIFE_COLUMN: format_years(remaining_life, current_age, max_age),
|
||
}
|
||
)
|
||
if i < CHART_DISPLAY_LIMIT:
|
||
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.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:
|
||
plt.legend(loc="best", fontsize=8, prop=CHINESE_FONT_PROP)
|
||
ax = plt.gca()
|
||
if CHINESE_FONT_PROP:
|
||
for label in [*ax.get_xticklabels(), *ax.get_yticklabels()]:
|
||
label.set_fontproperties(CHINESE_FONT_PROP)
|
||
plt.tight_layout(rect=(0, 0.07, 1, 1))
|
||
plt.savefig(image_path, dpi=160, bbox_inches="tight")
|
||
plt.close()
|
||
return summary_rows, summary_sheet_rows
|
||
|
||
|
||
def write_prediction_workbook(
|
||
excel_path: Path,
|
||
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, 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, health_states))))
|
||
|
||
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:
|
||
summary_df = pd.DataFrame(
|
||
summary_sheet_rows,
|
||
columns=[ID_COLUMN, PIPE_AGE_COLUMN, "当前健康状态", "健康等级", REMAINING_LIFE_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["样本数据"]
|
||
summary_worksheet = writer.book["结果摘要"]
|
||
for cell in summary_worksheet["C"][1:]:
|
||
cell.number_format = "0.0%"
|
||
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 = "@"
|