"""底座 ``ChartService``：真实置信区间（CI）带与 QbD 风险色块渲染（任务 8）。 对应 design.md「10. ChartService」与需求 4.1 / 4.2 / 4.4 / 11.5 / 14.3： - **真实 CI 带，零硬编码系数**（需求 4.1 / 4.2）：``prediction_band()`` 接收 ``compute`` 阶段在连续时间网格上算出的**真实** CI 数据（``times / point / lower / upper``）并**直接渲染**喇叭形阴影。本模块**绝不**自行从点预测推导带宽 （不存在 ``se_scale=0.02`` 之类的硬编码比例系数）——上下界完全来自传入数据， 渲染层仅作几何绘制。 - **一级动力学非对称带**（需求 4.4）：带的非对称性由 ``compute`` 传入的非对称 上下界（``upper-point ≠ point-lower``）自然呈现，渲染层原样保留，不强制对称化。 - **多 CQA 叠加图并标注短板 CQA**（需求 11.5）：``multi_cqa_overlay()`` 将多个 CQA 以对齐的时间轴面板（或单轴叠加）呈现，并以风险色高亮标注决定货架期的**短板 （木桶）CQA**。 - **统一 QbD 风险色板**（需求 14.3）：红 / 黄 / 绿（高 / 中 / 低危）色板集中定义， 供报告与图表复用；``risk_color()`` 同时识别中英文与高/中/低三档别名。 - **内嵌中文字体与降级**（需求 14.1 关联）：加载 ``fonts/NotoSansSC-Regular.otf`` 用于中文标签；**字体缺失则降级为英文标签**，不报错、不中断。 可降级性（呼应 design Property 7）：matplotlib / numpy 等重依赖一律 **try-import**； 缺失时各渲染方法返回带提示的 :class:`ChartResult`（``ok=False`` + ``warning``）， **绝不抛出导致应用崩溃**。 输出约定：渲染成功时 ``ChartResult.image_base64`` 为 ``"data:image/png;base64,..."`` 形式的内联 PNG（与既有 ``utils/chart_executor.py`` 约定一致），可直接嵌入 HTML 报告。 """ from __future__ import annotations import base64 import io import logging import os from dataclasses import dataclass, field from pathlib import Path from typing import Optional, Sequence logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # QbD 风险色板（红 / 黄 / 绿）— 需求 14.3 # --------------------------------------------------------------------------- #: 统一 QbD 风险矩阵色板：高危(红) / 中危(黄) / 低危(绿)。 #: 与 ``config/chart_styles.py`` 的 compliant/marginal/non_compliant 取色保持一致。 QBD_RISK_COLORS: dict[str, str] = { "high": "#dc3545", # 红 —— 高危 / 不合规 / 超规格 "medium": "#ffc107", # 黄 —— 中危 / 临界 / 接近规格 "low": "#28a745", # 绿 —— 低危 / 合规 / 满足规格 } #: 风险等级别名 → 标准档位（识别中英文与多种表述）。 _RISK_ALIASES: dict[str, str] = { # 高危 "high": "high", "高": "high", "高危": "high", "高风险": "high", "danger": "high", "fail": "high", "failed": "high", "critical": "high", "non_compliant": "high", "noncompliant": "high", "不合规": "high", "不合格": "high", "超规格": "high", "超标": "high", "red": "high", # 中危 "medium": "medium", "mid": "medium", "中": "medium", "中危": "medium", "中风险": "medium", "warning": "medium", "warn": "medium", "marginal": "medium", "临界": "medium", "注意": "medium", "yellow": "medium", "amber": "medium", # 低危 "low": "low", "低": "low", "低危": "low", "低风险": "low", "safe": "low", "pass": "low", "passed": "low", "ok": "low", "compliant": "low", "合规": "low", "合格": "low", "满足": "low", "green": "low", } #: 默认档位（无法识别风险等级时使用中危黄，保守提示）。 _DEFAULT_RISK_LEVEL = "medium" # --------------------------------------------------------------------------- # 双语标签目录（中文字体可用→中文；缺失→英文）— 需求 14.1 关联 # --------------------------------------------------------------------------- _LABELS: dict[str, dict[str, str]] = { "time_axis": {"zh": "时间 (月)", "en": "Time (months)"}, "value_axis": {"zh": "数值", "en": "Value"}, "point_pred": {"zh": "点预测", "en": "Point prediction"}, "ci_band": {"zh": "置信区间", "en": "Confidence interval"}, "observed": {"zh": "实测值", "en": "Observed"}, "spec_limit": {"zh": "规格限", "en": "Spec limit"}, "shelf_life": {"zh": "货架期", "en": "Shelf life"}, "limiting_cqa": {"zh": "短板 CQA（木桶原理）", "en": "Limiting CQA (weakest-link)"}, "prediction_title": {"zh": "预测趋势与置信带", "en": "Prediction trend with CI band"}, "multi_cqa_title": {"zh": "多 CQA 联合评估", "en": "Multi-CQA joint assessment"}, "observed_title": {"zh": "实测数据趋势", "en": "Observed data trends"}, "target_point": {"zh": "预测时间点", "en": "Prediction timepoint"}, "months": {"zh": "个月", "en": "months"}, } #: 仓库内内嵌中文字体的默认路径（相对本文件定位 ``/fonts``）。 #: ``platform/services/chart_service.py`` → 上溯三级到仓库根。 _DEFAULT_FONT_PATH = ( Path(__file__).resolve().parents[2] / "fonts" / "NotoSansSC-Regular.otf" ) # --------------------------------------------------------------------------- # 数据载体 # --------------------------------------------------------------------------- @dataclass class CIBand: """连续时间网格上的**真实** CI 带数据（由 ``compute`` 阶段产出）。 渲染层据此直接绘制，不做任何带宽推导： - ``times``：时间网格（与 point/lower/upper 等长）。 - ``point``：点预测序列。 - ``lower`` / ``upper``：CI 下界 / 上界序列（一级动力学下天然非对称）。 - ``label``：CQA 名称（用于图例 / 子图标题）。 - ``spec_type``：``"upper"``（上限型，如杂质）或 ``"lower"``（下限型，如含量）。 - ``spec_limit``：规格限（可选，绘制为参考线）。 - ``shelf_life``：该 CQA 满足规格的最长时间（可选，绘制为竖线标记）。 - ``risk_level``：该 CQA 的风险档位（可选，影响标注色，见 :data:`_RISK_ALIASES`）。 - ``observed_t`` / ``observed_y``：实测散点（可选，叠加显示）。 """ times: Sequence[float] point: Sequence[float] lower: Sequence[float] upper: Sequence[float] label: str = "" spec_type: str = "upper" spec_limit: Optional[float] = None shelf_life: Optional[float] = None risk_level: Optional[str] = None observed_t: Optional[Sequence[float]] = None observed_y: Optional[Sequence[float]] = None target_timepoints: Optional[Sequence[float]] = None @dataclass class ObservedTrace: """一条**实测**时间序列（用于数据梳理趋势图，不含任何模型外推）。 - ``times`` / ``values``：实测时间点（月）与对应测定值，等长且非空。 - ``label``：序列图例名（通常为「批次@条件」）。 - ``spec_limit``：可选规格限，仅当**确为用户/数据提供**时绘制参考线（避免 在无规格时凭空画线，呼应「无规格不得判定合规」的科学约束）。 """ times: Sequence[float] values: Sequence[float] label: str = "" spec_limit: Optional[float] = None @dataclass class ChartResult: """统一的图表渲染结果。 - ``ok``：是否成功生成图像。 - ``image_base64``：成功时为 ``"data:image/png;base64,..."`` 内联 PNG。 - ``warning``：非致命提示（如重依赖缺失已降级、未生成图像）。 - ``error``：失败原因（数据不合法、渲染异常等）。 - ``used_chinese_font``：本次渲染是否使用了中文字体（否则为英文标签降级）。 """ ok: bool image_base64: str = "" warning: str = "" error: str = "" used_chinese_font: bool = False def __bool__(self) -> bool: # 便于 ``if result:`` 直接判定成功与否 return self.ok # --------------------------------------------------------------------------- # ChartService # --------------------------------------------------------------------------- class ChartService: """真实 CI 带与 QbD 风险色块渲染服务。 构造参数： - ``font_path``：内嵌中文字体路径。默认指向 ``/fonts/NotoSansSC-Regular.otf``。 - ``enable_chinese``：是否启用中文标签（即便字体存在也可显式关闭以测试英文降级）。 - ``dpi``：导出 PNG 的分辨率。 """ def __init__( self, font_path: Optional[str] = None, *, enable_chinese: bool = True, dpi: int = 150, ) -> None: self.dpi = dpi self._font_path = ( Path(font_path) if font_path is not None else _DEFAULT_FONT_PATH ) # 是否真正具备中文渲染能力：启用开关 + 字体文件存在。 self._enable_chinese = bool(enable_chinese) self._font_available = self._enable_chinese and self._font_path.is_file() self._font_prop = None # 延迟到首次渲染时构造（依赖 matplotlib） # ------------------------------------------------------------------ # 语言 / 字体 # ------------------------------------------------------------------ @property def chinese_available(self) -> bool: """当前是否以中文标签渲染（字体可用且未被显式关闭）。""" return self._font_available @property def lang(self) -> str: """当前标签语言：``"zh"``（中文字体可用）或 ``"en"``（降级英文）。""" return "zh" if self._font_available else "en" def label(self, key: str) -> str: """按当前语言取标签文案；缺 key 时回退 key 本身。""" entry = _LABELS.get(key) if not entry: return key return entry.get(self.lang) or entry.get("en") or key # ------------------------------------------------------------------ # QbD 风险色板（需求 14.3） # ------------------------------------------------------------------ @staticmethod def risk_palette() -> dict[str, str]: """返回统一 QbD 风险色板（high/medium/low → 红/黄/绿）的副本。""" return dict(QBD_RISK_COLORS) @staticmethod def normalize_risk_level(level: Optional[str]) -> str: """把任意风险表述归一化为 ``"high"`` / ``"medium"`` / ``"low"`` 三档。 识别中英文及多种别名（见 :data:`_RISK_ALIASES`）；无法识别时返回 默认档位（中危）。 """ if level is None: return _DEFAULT_RISK_LEVEL key = str(level).strip().lower() if key in _RISK_ALIASES: return _RISK_ALIASES[key] # 容错：子串匹配（如「中等风险」「high risk」）。 for alias, normalized in _RISK_ALIASES.items(): if alias and alias in key: return normalized return _DEFAULT_RISK_LEVEL @classmethod def risk_color(cls, level: Optional[str]) -> str: """按风险等级返回 QbD 红 / 黄 / 绿色值（识别中英文与高/中/低）。""" return QBD_RISK_COLORS[cls.normalize_risk_level(level)] # ------------------------------------------------------------------ # 带数据校验（纯函数，无副作用，便于测试） # ------------------------------------------------------------------ @staticmethod def validate_band(band: "CIBand"): """校验并归一化一条 CI 带为等长 numpy 数组，返回 ``(times, point, lower, upper)``。 - 四个序列必须等长且非空。 - **不**对带宽做任何缩放 / 推导——上下界原样保留（需求 4.2 的硬保证）。 - 任一界越界（``lower > upper``）将抛出 ``ValueError``，以便尽早暴露上游错误。 依赖 numpy；numpy 不可用时抛出 ``RuntimeError``（由调用方降级处理）。 """ try: import numpy as np except ImportError as exc: # pragma: no cover - 环境缺 numpy 时降级 raise RuntimeError("numpy 不可用，无法处理 CI 带数据。") from exc times = np.asarray(band.times, dtype=float) point = np.asarray(band.point, dtype=float) lower = np.asarray(band.lower, dtype=float) upper = np.asarray(band.upper, dtype=float) n = times.size if n == 0: raise ValueError("CI 带时间网格为空。") if not (point.size == lower.size == upper.size == n): raise ValueError( "CI 带各序列长度不一致：" f"times={times.size}, point={point.size}, " f"lower={lower.size}, upper={upper.size}。" ) if np.any(lower > upper + 1e-9): raise ValueError("CI 带存在 lower > upper 的非法区间。") return times, point, lower, upper @classmethod def band_halfwidths(cls, band: "CIBand"): """返回 ``(upper-point, point-lower)`` 两个半宽序列（numpy 数组）。 用于**证明**渲染层直接消费真实上下界：当 ``upper-point != point-lower`` 时， 带为非对称（一级动力学，需求 4.4）。此方法不做任何缩放，仅作差。 """ _, point, lower, upper = cls.validate_band(band) return upper - point, point - lower # ------------------------------------------------------------------ # 重依赖准备（matplotlib / numpy）— try-import 优雅降级 # ------------------------------------------------------------------ def _ensure_backend(self): """准备渲染后端，返回 ``(plt, np)``；任一缺失返回 ``None``（调用方降级）。""" try: import matplotlib matplotlib.use("Agg") # 非交互后端，适合服务端 / 测试 import matplotlib.pyplot as plt import numpy as np except ImportError as exc: logger.warning("matplotlib / numpy 不可用，图表渲染降级：%s", exc) return None # 字体配置：中文字体可用则注册并设为首选；缺失则保持默认（英文标签）。 font_name = None if self._font_available and self._font_prop is None: try: from matplotlib import font_manager font_manager.fontManager.addfont(str(self._font_path)) self._font_prop = font_manager.FontProperties(fname=str(self._font_path)) except Exception as exc: # noqa: BLE001 - 字体加载失败即降级英文 logger.warning("加载中文字体失败，降级英文标签：%s", exc) self._font_available = False self._font_prop = None if self._font_prop is not None: try: font_name = self._font_prop.get_name() except Exception: # noqa: BLE001 font_name = None # 出版级（SCI）排版基线：统一字体、去顶/右边框、细线宽、外向刻度、 # 一致字号与高分辨率导出。所有图表共用，无需逐方法重复设置。 self._apply_publication_style(plt, font_name) return plt, np def _apply_publication_style(self, plt, font_name: Optional[str]) -> None: """设置 matplotlib 出版级 rcParams（SCI 论文取向）。 - **简洁坐标轴**：去除上 / 右边框、外向短刻度、细线宽——符合期刊制图规范。 - **一致字号与高 DPI**：正文 9.5、标题略大、加粗标题；高分辨率导出。 说明：**不**全局改写 ``font.family`` / ``font.sans-serif``——既有渲染通过 逐文本 ``fontproperties`` 应用内嵌字体（见 ``_font_kwargs``），全局改写字体族会 与字体缓存 / ``tight_layout`` 交互而引入跨渲染不确定性。这里仅设置确定性、 与文本字体解析无关的外观参数。 """ rc = { "axes.titleweight": "bold", "figure.titleweight": "bold", # 简洁坐标轴（去顶/右框、细线宽、外向刻度） "axes.spines.top": False, "axes.spines.right": False, "axes.linewidth": 0.8, "axes.edgecolor": "#333333", "axes.labelcolor": "#222222", "axes.titlecolor": "#1a1a1a", "text.color": "#222222", "xtick.color": "#333333", "ytick.color": "#333333", "xtick.direction": "out", "ytick.direction": "out", "xtick.major.width": 0.8, "ytick.major.width": 0.8, "xtick.major.size": 3.5, "ytick.major.size": 3.5, # 网格（默认关闭，由各图按需开启；统一为浅虚线风格） "grid.color": "#dfe3e6", "grid.linewidth": 0.6, "grid.linestyle": "--", "legend.frameon": False, # 负号正常显示（自定义字体下尤为重要） "axes.unicode_minus": False, # 高质量导出（注意：不设 savefig.bbox="tight" —— 它会按文本范围裁剪， # 使画布尺寸随标签文字长度变化、并引入渲染不确定性，破坏可复现比较。 # 统一裁剪由各图的 tight_layout + 固定 figsize 保证）。 "figure.dpi": self.dpi, "savefig.dpi": self.dpi, } try: plt.rcParams.update(rc) except Exception as exc: # noqa: BLE001 - 个别键不被某版本支持时不致命 logger.info("应用出版级图表样式时部分参数被忽略：%s", exc) plt.rcParams["axes.unicode_minus"] = False def _font_kwargs(self) -> dict: """返回供 matplotlib 文本接口使用的字体参数（中文时附 FontProperties）。""" if self._font_available and self._font_prop is not None: return {"fontproperties": self._font_prop} return {} def _fig_to_result(self, plt, fig) -> ChartResult: """把图形导出为内联 PNG 的 :class:`ChartResult`，并释放资源。""" try: buf = io.BytesIO() fig.savefig( buf, format="png", dpi=self.dpi, bbox_inches="tight", facecolor="white", edgecolor="none", ) buf.seek(0) encoded = base64.b64encode(buf.getvalue()).decode("ascii") return ChartResult( ok=True, image_base64=f"data:image/png;base64,{encoded}", used_chinese_font=self._font_available, ) finally: plt.close(fig) # ------------------------------------------------------------------ # 单 CQA 预测带（需求 4.1 / 4.2 / 4.4） # ------------------------------------------------------------------ def prediction_band( self, band: "CIBand", *, title: Optional[str] = None, value_axis_label: Optional[str] = None, ) -> ChartResult: """渲染单个 CQA 的真实预测带（喇叭形 / 非对称阴影直接来自传入上下界）。 **不含任何硬编码比例系数**：阴影由 ``band.lower`` / ``band.upper`` 直接绘制； 一级动力学的非对称形态原样保留（需求 4.2 / 4.4）。 """ backend = self._ensure_backend() if backend is None: return ChartResult( ok=False, warning="matplotlib / numpy 不可用，已跳过图表渲染（核心分析不受影响）。", ) plt, np = backend try: times, point, lower, upper = self.validate_band(band) except (ValueError, RuntimeError) as exc: return ChartResult(ok=False, error=f"CI 带数据不合法：{exc}") try: fig, ax = plt.subplots(figsize=(7.2, 4.5)) self._draw_band(ax, plt, np, times, point, lower, upper, band) ax.set_xlabel(self.label("time_axis"), **self._font_kwargs()) ax.set_ylabel( value_axis_label or band.label or self.label("value_axis"), **self._font_kwargs(), ) ax.set_title( title or self.label("prediction_title"), **self._font_kwargs(), fontsize=13, ) self._apply_legend(ax) ax.grid(True, color="#e0e0e0", linestyle="--", linewidth=0.6) fig.tight_layout() return self._fig_to_result(plt, fig) except Exception as exc: # noqa: BLE001 - 渲染异常不得中断主流程 logger.warning("渲染预测带失败：%s", exc, exc_info=True) plt.close("all") return ChartResult(ok=False, error=f"渲染预测带失败：{exc}") # ------------------------------------------------------------------ # 实测数据趋势图（数据梳理可视化）—— 仅画实测点，不含外推 # ------------------------------------------------------------------ def observed_trends( self, traces: Sequence["ObservedTrace"], *, title: Optional[str] = None, value_axis_label: Optional[str] = None, spec_limit: Optional[float] = None, ) -> ChartResult: """把多条**实测**序列绘成折线散点趋势图（数据梳理可视化）。 与 :meth:`prediction_band` 严格区分：本图**只呈现实测数据本身**（按批次×条件 分系列的折线 + 散点），不绘制任何模型外推或置信带，用于「先把原始数据梳理 清楚、再谈建模」。规格限仅在 ``spec_limit`` 显式提供（即确有规格）时绘制参考 线——无规格时不画线，避免误导性的「合规」暗示。 可降级：matplotlib/numpy 缺失或无有效序列时返回 ``ok=False`` 的 :class:`ChartResult`，绝不抛出。 """ backend = self._ensure_backend() if backend is None: return ChartResult( ok=False, warning="matplotlib / numpy 不可用，已跳过图表渲染（核心分析不受影响）。", ) plt, np = backend # 过滤出至少含 1 个点的有效序列。 valid: list[tuple[ObservedTrace, Any, Any]] = [] for tr in traces or []: try: t = np.asarray(tr.times, dtype=float) y = np.asarray(tr.values, dtype=float) except Exception: # noqa: BLE001 - 单序列异常跳过 continue if t.size and t.size == y.size: valid.append((tr, t, y)) if not valid: return ChartResult(ok=False, error="无有效实测序列可绘制。") try: palette = ["#1f77b4", "#ff7f0e", "#2ca02c", "#9467bd", "#8c564b", "#e377c2", "#17becf", "#bcbd22"] fig, ax = plt.subplots(figsize=(7.6, 4.6)) for idx, (tr, t, y) in enumerate(valid): color = palette[idx % len(palette)] order = np.argsort(t) ax.plot( t[order], y[order], color=color, linewidth=1.6, marker="o", markersize=5, markeredgecolor="white", markeredgewidth=0.6, label=(tr.label or f"{self.label('observed')} {idx + 1}"), ) # 规格限参考线：仅在显式提供时绘制（无规格不画，避免误导）。 eff_spec = spec_limit if eff_spec is None: for tr, _, _ in valid: if tr.spec_limit is not None: eff_spec = tr.spec_limit break if eff_spec is not None: ax.axhline( eff_spec, color=QBD_RISK_COLORS["high"], linewidth=1.2, linestyle="--", label=f"{self.label('spec_limit')}={eff_spec:g}", ) ax.set_xlabel(self.label("time_axis"), **self._font_kwargs()) ax.set_ylabel( value_axis_label or self.label("value_axis"), **self._font_kwargs(), ) ax.set_title( title or self.label("observed_title"), **self._font_kwargs(), fontsize=13, ) ax.grid(True, color="#e0e0e0", linestyle="--", linewidth=0.6) self._apply_legend(ax) fig.tight_layout() return self._fig_to_result(plt, fig) except Exception as exc: # noqa: BLE001 - 渲染异常不得中断主流程 logger.warning("渲染实测趋势图失败：%s", exc, exc_info=True) plt.close("all") return ChartResult(ok=False, error=f"渲染实测趋势图失败：{exc}") # ------------------------------------------------------------------ # 多 CQA 叠加 / 分图，并标注短板 CQA（需求 11.5） # ------------------------------------------------------------------ def multi_cqa_overlay( self, bands: Sequence["CIBand"], *, limiting_cqa: Optional[str] = None, title: Optional[str] = None, mode: str = "panels", ) -> ChartResult: """渲染多个 CQA 并标注决定货架期的短板（木桶）CQA（需求 11.5）。 - ``mode="panels"``（默认）：每个 CQA 一个对齐时间轴的子图面板，避免不同 量纲（含量 ~100% vs 杂质 ~0.3%）混轴失真；短板 CQA 的标题以高危红高亮。 - ``mode="overlay"``：所有 CQA 叠加于同一坐标轴（适用于同量纲对比）。 ``limiting_cqa`` 指定短板 CQA 的 ``label``；缺省时若各 band 提供 ``shelf_life``，自动取最短者为短板（木桶原理）。 """ backend = self._ensure_backend() if backend is None: return ChartResult( ok=False, warning="matplotlib / numpy 不可用，已跳过图表渲染（核心分析不受影响）。", ) plt, np = backend if not bands: return ChartResult(ok=False, error="未提供任何 CQA 带数据。") # 预校验全部 band，任一不合法即整体报错（尽早暴露上游问题）。 validated = [] try: for band in bands: validated.append((band, *self.validate_band(band))) except (ValueError, RuntimeError) as exc: return ChartResult(ok=False, error=f"CI 带数据不合法：{exc}") limiting = self._resolve_limiting(bands, limiting_cqa) try: if mode == "overlay": fig = self._render_overlay(plt, np, validated, limiting, title) else: fig = self._render_panels(plt, np, validated, limiting, title) return self._fig_to_result(plt, fig) except Exception as exc: # noqa: BLE001 - 渲染异常不得中断主流程 logger.warning("渲染多 CQA 图失败：%s", exc, exc_info=True) plt.close("all") return ChartResult(ok=False, error=f"渲染多 CQA 图失败：{exc}") # ------------------------------------------------------------------ # 短板 CQA 判定（木桶原理） # ------------------------------------------------------------------ @staticmethod def _resolve_limiting( bands: Sequence["CIBand"], limiting_cqa: Optional[str] ) -> Optional[str]: """确定短板 CQA 的 label：优先显式指定，否则取 shelf_life 最短者。""" if limiting_cqa: return limiting_cqa candidates = [ b for b in bands if b.shelf_life is not None and b.label ] if not candidates: return None shortest = min(candidates, key=lambda b: b.shelf_life) return shortest.label # ------------------------------------------------------------------ # 内部绘制原语 # ------------------------------------------------------------------ def _draw_band(self, ax, plt, np, times, point, lower, upper, band, *, color=None): """在给定坐标轴上绘制一条带（阴影 + 点预测 + 规格线 + 货架期 + 实测点）。 阴影直接由 ``lower`` / ``upper`` 填充——这是「真实 CI、零硬编码系数」的核心： 渲染层不重算带宽，非对称形态原样呈现（需求 4.2 / 4.4）。 """ line_color = color or "#003366" # 真实 CI 阴影（喇叭形 / 非对称均由数据决定）。 ax.fill_between( times, lower, upper, color=line_color, alpha=0.18, linewidth=0, label=self.label("ci_band"), ) # 上下包络线（虚线），强调带边界形态。 ax.plot(times, upper, color=line_color, alpha=0.5, linewidth=0.8, linestyle=":") ax.plot(times, lower, color=line_color, alpha=0.5, linewidth=0.8, linestyle=":") # 点预测线。 ax.plot(times, point, color=line_color, linewidth=1.8, label=self.label("point_pred")) # 规格限参考线（按上限/下限型着色）。 if band.spec_limit is not None: ax.axhline( band.spec_limit, color=QBD_RISK_COLORS["high"], linewidth=1.2, linestyle="--", label=f"{self.label('spec_limit')}={band.spec_limit:g}", ) # 货架期竖线标记。 if band.shelf_life is not None: ax.axvline( band.shelf_life, color=QBD_RISK_COLORS["medium"], linewidth=1.2, linestyle="-.", label=f"{self.label('shelf_life')}={band.shelf_life:g}", ) # 实测散点。 if band.observed_t is not None and band.observed_y is not None: ot = np.asarray(band.observed_t, dtype=float) oy = np.asarray(band.observed_y, dtype=float) if ot.size and ot.size == oy.size: ax.scatter( ot, oy, color="#0066cc", s=28, zorder=5, edgecolors="white", linewidths=0.6, label=self.label("observed"), ) # 目标预测时间点高亮标记（不同颜色 + 竖线 + 数值标注），便于用户直观感知。 self._mark_target_timepoints(ax, np, times, point, band) def _mark_target_timepoints(self, ax, np, times, point, band) -> None: """在预测曲线上以醒目颜色标注用户的目标预测时间点（竖线 + 星形点 + 数值）。 预测值由真实点预测序列在该时间点插值得到（不另算）；时间点超出网格则跳过。 """ tps = getattr(band, "target_timepoints", None) if not tps: return try: t_arr = np.asarray(times, dtype=float) y_arr = np.asarray(point, dtype=float) except Exception: # noqa: BLE001 return if t_arr.size == 0: return marker_color = "#d6336c" # 醒目品红，与 QbD 红/黄/绿区分，专指"预测时间点" labeled = False for tp in tps: try: tp_f = float(tp) except (TypeError, ValueError): continue if tp_f < float(t_arr.min()) or tp_f > float(t_arr.max()): continue y_at = float(np.interp(tp_f, t_arr, y_arr)) ax.axvline( tp_f, color=marker_color, linewidth=1.0, linestyle="--", alpha=0.7, ) ax.scatter( [tp_f], [y_at], color=marker_color, s=90, marker="*", zorder=6, edgecolors="white", linewidths=0.8, label=(self.label("target_point") if not labeled else None), ) labeled = True ax.annotate( f"{tp_f:g}{self.label('months')}: {y_at:.3g}", xy=(tp_f, y_at), xytext=(4, 8), textcoords="offset points", fontsize=8, color=marker_color, **self._font_kwargs(), ) def _render_panels(self, plt, np, validated, limiting, title): """每个 CQA 一个子图面板，短板 CQA 以高危红高亮标题。""" n = len(validated) fig, axes = plt.subplots( n, 1, figsize=(7.2, 2.8 * n + 0.4), squeeze=False, sharex=True ) for idx, (band, times, point, lower, upper) in enumerate(validated): ax = axes[idx][0] self._draw_band(ax, plt, np, times, point, lower, upper, band) is_limiting = bool(limiting) and band.label == limiting label = band.label or f"CQA {idx + 1}" if is_limiting: marker = self.label("limiting_cqa") ax.set_title( f"★ {label} — {marker}", color=QBD_RISK_COLORS["high"], **self._font_kwargs(), fontsize=12, ) # 用高危红描边强调短板面板。 for spine in ax.spines.values(): spine.set_edgecolor(QBD_RISK_COLORS["high"]) spine.set_linewidth(1.6) else: ax.set_title(label, **self._font_kwargs(), fontsize=12) ax.set_ylabel(label, **self._font_kwargs()) ax.grid(True, color="#e0e0e0", linestyle="--", linewidth=0.6) self._apply_legend(ax, fontsize=8) axes[-1][0].set_xlabel(self.label("time_axis"), **self._font_kwargs()) fig.suptitle( title or self.label("multi_cqa_title"), **self._font_kwargs(), fontsize=14, ) fig.tight_layout(rect=(0, 0, 1, 0.97)) return fig def _render_overlay(self, plt, np, validated, limiting, title): """所有 CQA 叠加于同一坐标轴；短板 CQA 加粗并以高危红高亮。""" palette = ["#1f77b4", "#ff7f0e", "#2ca02c", "#9467bd", "#8c564b", "#e377c2"] fig, ax = plt.subplots(figsize=(8.0, 5.0)) for idx, (band, times, point, lower, upper) in enumerate(validated): is_limiting = bool(limiting) and band.label == limiting color = QBD_RISK_COLORS["high"] if is_limiting else palette[idx % len(palette)] label = band.label or f"CQA {idx + 1}" ax.fill_between(times, lower, upper, color=color, alpha=0.12, linewidth=0) ax.plot( times, point, color=color, linewidth=2.6 if is_limiting else 1.6, label=(f"★ {label}" if is_limiting else label), ) if band.spec_limit is not None: ax.axhline(band.spec_limit, color=color, linewidth=0.9, linestyle="--", alpha=0.7) if band.observed_t is not None and band.observed_y is not None: ot = np.asarray(band.observed_t, dtype=float) oy = np.asarray(band.observed_y, dtype=float) if ot.size and ot.size == oy.size: ax.scatter(ot, oy, color=color, s=24, zorder=5, edgecolors="white", linewidths=0.5) ax.set_xlabel(self.label("time_axis"), **self._font_kwargs()) ax.set_ylabel(self.label("value_axis"), **self._font_kwargs()) subtitle = title or self.label("multi_cqa_title") if limiting: subtitle = f"{subtitle}（{self.label('limiting_cqa')}: {limiting}）" \ if self._font_available else f"{subtitle} ({self.label('limiting_cqa')}: {limiting})" ax.set_title(subtitle, **self._font_kwargs(), fontsize=13) ax.grid(True, color="#e0e0e0", linestyle="--", linewidth=0.6) self._apply_legend(ax) fig.tight_layout() return fig def _apply_legend(self, ax, *, fontsize: int = 9) -> None: """添加图例；中文字体可用时为图例文本设定字体属性。""" handles, labels = ax.get_legend_handles_labels() if not handles: return legend = ax.legend(loc="best", fontsize=fontsize, framealpha=0.9) if self._font_available and self._font_prop is not None: for text in legend.get_texts(): text.set_fontproperties(self._font_prop) # ------------------------------------------------------------------ # 自适应可视化新增图型（adaptive-report-visualization 任务 9） # 全部：matplotlib 缺失→ok=False；中文字体缺失→英文标签仍出图；不抛异常。 # ------------------------------------------------------------------ def _unavailable(self) -> ChartResult: return ChartResult(ok=False, warning="matplotlib/numpy 不可用，图表已降级。") def grouped_bar(self, labels, values, *, title="", value_label="", reference_lines=None, dot=False) -> ChartResult: """分组条形图（``dot=True`` 时为点图）：每个分组一个数值。 ``reference_lines``：``[{value,label,kind}]``，仅绘制由调用方（来自 compute） 提供的限度参考线，不推导任何值。 """ backend = self._ensure_backend() if backend is None: return self._unavailable() plt, np = backend try: fig, ax = plt.subplots(figsize=(7.2, 4.0)) x = list(range(len(labels))) fk = self._font_kwargs() if dot: ax.scatter(x, values, s=70, color="#1f6f78", zorder=3) else: ax.bar(x, values, color="#1f6f78", width=0.6, zorder=3) for xi, v in zip(x, values): ax.annotate(f"{v:g}", (xi, v), textcoords="offset points", xytext=(0, 5), ha="center", fontsize=8, **fk) for rl in (reference_lines or []): if not isinstance(rl.get("value"), (int, float)): continue ax.axhline(rl["value"], color="#d08a1d", linestyle="--", linewidth=1.2) ax.annotate(str(rl.get("label", "")), (x[-1] if x else 0, rl["value"]), fontsize=8, color="#8a5a12", **fk) ax.set_xticks(x) ax.set_xticklabels([str(l) for l in labels], **fk) if value_label: ax.set_ylabel(value_label, **fk) if title: ax.set_title(title, **fk) ax.grid(axis="y", linestyle=":", alpha=0.5) return self._fig_to_result(plt, fig) except Exception as exc: # noqa: BLE001 logger.warning("grouped_bar 渲染失败：%s", exc) return ChartResult(ok=False, error=str(exc)) def distribution_dot(self, units, *, title="", value_label="", mean=None, reference_lines=None, box=False) -> ChartResult: """重复单位分布点图（``box=True`` 叠加箱线）：逐单位散点 + 均值线 + 参考线。""" backend = self._ensure_backend() if backend is None: return self._unavailable() plt, np = backend try: nums = [float(v) for v in units if isinstance(v, (int, float))] fig, ax = plt.subplots(figsize=(6.4, 4.0)) fk = self._font_kwargs() if box and len(nums) >= 2: ax.boxplot(nums, vert=True, widths=0.4, positions=[1], patch_artist=True, boxprops=dict(facecolor="#eef6f7", color="#1f6f78")) jitter = (np.random.default_rng(0).uniform(-0.06, 0.06, size=len(nums)) if nums else []) ax.scatter([1 + j for j in jitter], nums, s=60, color="#1f6f78", zorder=3, alpha=0.85) if mean is None and nums: mean = float(np.mean(nums)) if isinstance(mean, (int, float)): ax.axhline(mean, color="#1f6f78", linewidth=1.5, label=(self.label("mean") if self.label("mean") != "mean" else "mean")) ax.annotate(f"{mean:g}", (1.15, mean), fontsize=8, color="#155057", **fk) for rl in (reference_lines or []): if not isinstance(rl.get("value"), (int, float)): continue ax.axhline(rl["value"], color="#d08a1d", linestyle="--", linewidth=1.2) ax.annotate(str(rl.get("label", "")), (0.6, rl["value"]), fontsize=8, color="#8a5a12", **fk) ax.set_xticks([]) if value_label: ax.set_ylabel(value_label, **fk) if title: ax.set_title(title, **fk) ax.grid(axis="y", linestyle=":", alpha=0.5) return self._fig_to_result(plt, fig) except Exception as exc: # noqa: BLE001 logger.warning("distribution_dot 渲染失败：%s", exc) return ChartResult(ok=False, error=str(exc)) def status_matrix(self, rows, *, title="") -> ChartResult: """跨属性合规状态矩阵：颜色 + 符号/文字双通道（无障碍，需求 15）。 ``rows``：``[{label, status in {pass,fail,na}}]``。 """ backend = self._ensure_backend() if backend is None: return self._unavailable() plt, np = backend try: fk = self._font_kwargs() color_map = {"pass": QBD_RISK_COLORS["low"], "fail": QBD_RISK_COLORS["high"], "na": "#c7ccd1"} mark_map = {"pass": "√", "fail": "×", "na": "—"} n = len(rows) fig, ax = plt.subplots(figsize=(6.6, max(1.2, 0.42 * n + 0.6))) for i, r in enumerate(rows): y = n - 1 - i status = str(r.get("status", "na")) ax.add_patch(plt.Rectangle((0, y), 1, 0.9, facecolor=color_map.get(status, "#c7ccd1"), edgecolor="white")) # 冗余编码：色块内放符号；右侧放文字标签。 ax.text(0.5, y + 0.45, mark_map.get(status, "—"), ha="center", va="center", fontsize=13, color="white", **fk) ax.text(1.1, y + 0.45, str(r.get("label", "")), ha="left", va="center", fontsize=9, **fk) ax.set_xlim(0, 4) ax.set_ylim(0, n) ax.axis("off") if title: ax.set_title(title, **fk) return self._fig_to_result(plt, fig) except Exception as exc: # noqa: BLE001 logger.warning("status_matrix 渲染失败：%s", exc) return ChartResult(ok=False, error=str(exc)) __all__ = [ "ChartService", "CIBand", "ObservedTrace", "ChartResult", "QBD_RISK_COLORS", ]