若已存在关联的措施推荐任务，则下表展示当前风险、建议后风险以及推荐动作；若尚未生成措施任务，则报告保留风险原因与通用治理建议。
{_render_table(["杆塔号", "当前风险", "建议后风险", "改造动作", "措施结论"], mitigation_table_rows or [["-", "-", "-", "当前未关联成功的措施推荐任务", "-"]])} +若已生成与措施任务关联的加装避雷器复算任务，则下表按 {escape(scenario_base_job_type)} 口径展示补装避雷器后的耐雷水平、跳闸率与风险等级；若未生成复算任务，则此表保留为空。
+ {_render_table(["杆塔号", "反击耐雷水平(kA)", "反击跳闸率", "绕击耐雷水平(kA)", "绕击跳闸率", "总雷击跳闸率", "风险等级"], scenario_table_rows or [["-", "-", "-", "-", "-", "-", "当前未关联成功的加装避雷器复算任务"]])} +说明：本报告为可直接下载的 Word 兼容文档，内容按风险评估结果和已关联的措施推荐结果动态生成。
@@ -267,6 +297,14 @@ def _format_job_type(value: str | None) -> str: return "风险评估任务" if value == "mitigation": return "措施推荐任务" + if value == "normal": + return "普通计算任务" + if value == "tongtiao": + return "同跳计算任务" + if value == "scenario": + return "加装避雷器复算任务" + if value == "report": + return "报告任务" return value or "-" @@ -351,6 +389,21 @@ def _display(value: Any) -> str: return text or "-" +def _format_number(value: Any) -> str: + parsed = _as_optional_float(value) + if parsed is None: + return "-" + return f"{parsed:.4f}".rstrip("0").rstrip(".") + + +def _format_total_trip_rate(result_json: Mapping[str, Any]) -> str: + counterstrike_value = _as_optional_float(result_json.get("counterstrike_trip_rate")) + shielding_value = _as_optional_float(result_json.get("shielding_trip_rate")) + if counterstrike_value is None and shielding_value is None: + return "-" + return _format_number((counterstrike_value or 0.0) + (shielding_value or 0.0)) + + def _as_optional_str(value: Any) -> str | None: if value is None: return None @@ -358,6 +411,13 @@ def _as_optional_str(value: Any) -> str | None: return text or None +def _as_optional_float(value: Any) -> float | None: + try: + return float(value) + except (TypeError, ValueError): + return None + + def _as_int(value: Any) -> int | None: if isinstance(value, bool): return int(value) diff --git a/api/app/services/fl_analysis_service.py b/api/app/services/fl_analysis_service.py index 859499f..10ef3f8 100644 --- a/api/app/services/fl_analysis_service.py +++ b/api/app/services/fl_analysis_service.py @@ -186,10 +186,18 @@ def create_job( ) execution_options = _normalize_execution_options(payload.job_type, payload.execution_options_json or {}) + external_adapter = payload.external_adapter + adapter_config_json = payload.adapter_config_json or {} if payload.job_type == "mitigation": total_tower_count = _validate_mitigation_options(db, line_id=line.id, execution_options=execution_options) elif payload.job_type == "report": total_tower_count = _validate_report_options(db, line_id=line.id, execution_options=execution_options) + elif payload.job_type == "scenario": + scenario_config = _validate_scenario_options(db, line_id=line.id, execution_options=execution_options) + total_tower_count = int(scenario_config["total_tower_count"]) + execution_options = dict(scenario_config["execution_options"]) + external_adapter = str(scenario_config["external_adapter"]) + adapter_config_json = dict(scenario_config["adapter_config_json"]) else: total_tower_count = int( db.scalar( @@ -198,16 +206,16 @@ def create_job( or 0 ) if payload.job_type in {"normal", "tongtiao"}: - if payload.external_adapter in {"atp", "wine"}: + if external_adapter in {"atp", "wine"}: try: resolve_external_waveform_job( db, - external_adapter=payload.external_adapter, - adapter_config_json=payload.adapter_config_json or {}, + external_adapter=external_adapter, + adapter_config_json=adapter_config_json, ) except RuntimeError as exc: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc - elif payload.external_adapter != "placeholder": + elif external_adapter != "placeholder": raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="普通计算和同跳计算仅支持 placeholder/atp/wine 适配器", @@ -223,8 +231,8 @@ def create_job( source_kind="line", status="pending", total_tower_count=total_tower_count, - external_adapter=payload.external_adapter, - adapter_config_json=payload.adapter_config_json or {}, + external_adapter=external_adapter, + adapter_config_json=adapter_config_json, execution_options_json=execution_options, result_summary_json={}, create_date=now, @@ -379,6 +387,7 @@ def execute_job(job_id: str) -> None: ) return + waveform_job_type = _resolve_waveform_job_type(job_type=job.job_type, execution_options=execution_options) towers = _load_job_towers(db, job=job, execution_options=execution_options) if not towers: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="当前线路没有可分析的杆塔数据") @@ -426,7 +435,7 @@ def execute_job(job_id: str) -> None: external_job = None stdout_chunks: list[str] = [] stderr_chunks: list[str] = [] - if job.job_type in {"normal", "tongtiao"} and job.external_adapter in {"atp", "wine"}: + if waveform_job_type in {"normal", "tongtiao"} and job.external_adapter in {"atp", "wine"}: external_job = resolve_external_waveform_job( db, external_adapter=job.external_adapter, @@ -442,6 +451,8 @@ def execute_job(job_id: str) -> None: "base_tower_json": snapshot.base_tower_json or {}, "profile_json": snapshot.profile_json or {}, } + if job.job_type == "scenario": + payload["profile_json"] = _apply_scenario_profile_json_overrides(snapshot.profile_json or {}) source_result = source_result_map.get(snapshot.tower_id) if source_result: payload["source_result_json"] = source_result @@ -450,7 +461,7 @@ def execute_job(job_id: str) -> None: payload, non_construction=bool(execution_options.get("non_construction")), ) - elif job.job_type == "normal": + elif waveform_job_type == "normal": baseline_result = grade_normal_snapshot_payload(payload, execution_options=execution_options) if external_job is not None: execution = execute_external_waveform_tower_analysis( @@ -469,7 +480,7 @@ def execute_job(job_id: str) -> None: stderr_chunks.append(f"[{snapshot.tower_no}] {execution.stderr_text}") else: graded = baseline_result - elif job.job_type == "tongtiao": + elif waveform_job_type == "tongtiao": baseline_result = grade_tongtiao_snapshot_payload(payload, execution_options=execution_options) if external_job is not None: execution = execute_external_waveform_tower_analysis( @@ -507,7 +518,7 @@ def execute_job(job_id: str) -> None: _accumulate_result_summary(summary, graded) tower = tower_map.get(snapshot.tower_id) - if tower is not None and job.job_type != "mitigation": + if tower is not None and job.job_type not in {"mitigation", "scenario"}: tower.risk_level = graded["risk_level"] tower.update_date = utcnow() tower.update_user = job.update_user or job.create_user @@ -522,7 +533,22 @@ def execute_job(job_id: str) -> None: summary["source_job_id"] = execution_options.get("source_job_id") summary["source_run_id"] = execution_options.get("source_run_id") summary["non_construction"] = bool(execution_options.get("non_construction")) - elif job.job_type in {"normal", "tongtiao"}: + elif job.job_type == "scenario": + summary["selected_tower_count"] = len(towers) + summary["source_job_id"] = execution_options.get("source_job_id") + summary["source_run_id"] = execution_options.get("source_run_id") + summary["source_job_type"] = execution_options.get("source_job_type") + summary["mitigation_job_id"] = execution_options.get("mitigation_job_id") + summary["mitigation_job_name"] = execution_options.get("mitigation_job_name") + summary["risk_job_id"] = execution_options.get("risk_job_id") + summary["risk_job_name"] = execution_options.get("risk_job_name") + summary["base_job_id"] = execution_options.get("base_job_id") + summary["base_job_name"] = execution_options.get("base_job_name") + summary["base_job_type"] = waveform_job_type + summary["workflow"] = _workflow_summary_from_execution_options(execution_options) + if external_job is not None: + summary["external_engine_adapter"] = job.external_adapter + elif waveform_job_type in {"normal", "tongtiao"}: summary["workflow"] = _workflow_summary_from_execution_options(execution_options) if external_job is not None: summary["external_engine_adapter"] = job.external_adapter @@ -780,6 +806,17 @@ def _prepare_report_payload( ] selected_mitigation_rows = _filter_rows_by_tower_ids(mitigation_rows, selected_tower_ids) + scenario_job_id = str(execution_options.get("scenario_job_id") or "") + scenario_run_id = str(execution_options.get("scenario_run_id") or "") + scenario_job = get_job_by_id(db, scenario_job_id) if scenario_job_id else None + selected_scenario_rows: list[dict[str, Any]] = [] + if scenario_job is not None and scenario_run_id: + scenario_rows = [ + _serialize_report_row(item) + for item in _load_result_rows(db, job_id=scenario_job_id, run_id=scenario_run_id) + ] + selected_scenario_rows = _filter_rows_by_tower_ids(scenario_rows, selected_tower_ids) + source_job = get_job_by_id(db, str(execution_options.get("source_job_id") or "")) line = risk_job.line or job.line @@ -799,10 +836,21 @@ def _prepare_report_payload( "risk_job_name": risk_job.job_name, "mitigation_job_id": mitigation_job.id if mitigation_job else None, "mitigation_job_name": mitigation_job.job_name if mitigation_job else None, + "scenario_job_id": scenario_job.id if scenario_job else None, + "scenario_job_name": scenario_job.job_name if scenario_job else None, + "scenario_base_job_type": ( + _resolve_waveform_job_type( + job_type=scenario_job.job_type, + execution_options=scenario_job.execution_options_json or {}, + ) + if scenario_job is not None + else None + ), }, "risk_rows": risk_rows, "selected_risk_rows": selected_risk_rows, "selected_mitigation_rows": selected_mitigation_rows, + "selected_scenario_rows": selected_scenario_rows, } @@ -1217,6 +1265,37 @@ def _normalize_execution_options(job_type: str, execution_options: dict[str, Any normalized["risk_run_id"] = risk_run_id else: normalized.pop("risk_run_id", None) + scenario_job_id = str(normalized.get("scenario_job_id") or "").strip() + if scenario_job_id: + normalized["scenario_job_id"] = scenario_job_id + else: + normalized.pop("scenario_job_id", None) + scenario_run_id = str(normalized.get("scenario_run_id") or "").strip() + if scenario_run_id: + normalized["scenario_run_id"] = scenario_run_id + else: + normalized.pop("scenario_run_id", None) + base_job_id = str(normalized.get("base_job_id") or "").strip() + if base_job_id: + normalized["base_job_id"] = base_job_id + else: + normalized.pop("base_job_id", None) + base_run_id = str(normalized.get("base_run_id") or "").strip() + if base_run_id: + normalized["base_run_id"] = base_run_id + else: + normalized.pop("base_run_id", None) + base_job_type = str(normalized.get("base_job_type") or "").strip() + if base_job_type: + normalized["base_job_type"] = base_job_type + else: + normalized.pop("base_job_type", None) + for text_key in ("base_job_name", "mitigation_job_name", "risk_job_name"): + text_value = str(normalized.get(text_key) or "").strip() + if text_value: + normalized[text_key] = text_value + else: + normalized.pop(text_key, None) source_job_type = str(normalized.get("source_job_type") or "").strip() if source_job_type: normalized["source_job_type"] = source_job_type @@ -1241,6 +1320,14 @@ def _normalize_execution_options(job_type: str, execution_options: dict[str, Any normalized.pop("mitigation_run_id", None) normalized.pop("risk_job_id", None) normalized.pop("risk_run_id", None) + normalized.pop("scenario_job_id", None) + normalized.pop("scenario_run_id", None) + normalized.pop("base_job_id", None) + normalized.pop("base_run_id", None) + normalized.pop("base_job_type", None) + normalized.pop("base_job_name", None) + normalized.pop("mitigation_job_name", None) + normalized.pop("risk_job_name", None) normalized.pop("source_job_type", None) elif job_type == "report": normalized.pop("current_waveform", None) @@ -1254,6 +1341,61 @@ def _normalize_execution_options(job_type: str, execution_options: dict[str, Any normalized.pop("tail_time_max_us", None) normalized.pop("tail_time_step_us", None) normalized.pop("non_construction", None) + normalized.pop("base_job_id", None) + normalized.pop("base_run_id", None) + normalized.pop("base_job_type", None) + normalized.pop("base_job_name", None) + normalized.pop("mitigation_job_name", None) + normalized.pop("risk_job_name", None) + elif job_type == "scenario": + normalized["current_waveform"] = _normalize_choice( + normalized.get("current_waveform") or normalized.get("current_type"), + allowed={"heidler", "double_slope", "double_exponential"}, + aliases={ + "Heidler": "heidler", + "双斜角": "double_slope", + "双指数": "double_exponential", + }, + default="heidler", + ) + normalized["flashover_method"] = _normalize_choice( + normalized.get("flashover_method"), + allowed={"guideline", "intersection", "leader_development"}, + aliases={ + "规程法": "guideline", + "相交法": "intersection", + "先导发展法": "leader_development", + }, + default="intersection", + ) + normalized["altitude_correction"] = _normalize_choice( + normalized.get("altitude_correction"), + allowed={"none", "formula1", "formula2"}, + aliases={ + "无": "none", + "推荐公式1": "formula1", + "推荐公式2": "formula2", + }, + default="none", + ) + normalized["induced_voltage_formula"] = _normalize_choice( + normalized.get("induced_voltage_formula"), + allowed={"formula1", "formula2"}, + aliases={ + "公式1": "formula1", + "公式2": "formula2", + }, + default="formula1", + ) + normalized["head_time_min_us"] = _normalize_positive_number(normalized.get("head_time_min_us"), default=2.6) + normalized["head_time_max_us"] = _normalize_positive_number(normalized.get("head_time_max_us"), default=2.6) + normalized["head_time_step_us"] = _normalize_positive_number(normalized.get("head_time_step_us"), default=0.1) + normalized["tail_time_min_us"] = _normalize_positive_number(normalized.get("tail_time_min_us"), default=50.0) + normalized["tail_time_max_us"] = _normalize_positive_number(normalized.get("tail_time_max_us"), default=50.0) + normalized["tail_time_step_us"] = _normalize_positive_number(normalized.get("tail_time_step_us"), default=1.0) + normalized.pop("scenario_job_id", None) + normalized.pop("scenario_run_id", None) + normalized.pop("non_construction", None) elif job_type in {"normal", "tongtiao"}: normalized["current_waveform"] = _normalize_choice( normalized.get("current_waveform") or normalized.get("current_type"), @@ -1307,6 +1449,14 @@ def _normalize_execution_options(job_type: str, execution_options: dict[str, Any normalized.pop("mitigation_run_id", None) normalized.pop("risk_job_id", None) normalized.pop("risk_run_id", None) + normalized.pop("scenario_job_id", None) + normalized.pop("scenario_run_id", None) + normalized.pop("base_job_id", None) + normalized.pop("base_run_id", None) + normalized.pop("base_job_type", None) + normalized.pop("base_job_name", None) + normalized.pop("mitigation_job_name", None) + normalized.pop("risk_job_name", None) normalized.pop("source_job_type", None) normalized.pop("non_construction", None) else: @@ -1327,6 +1477,14 @@ def _normalize_execution_options(job_type: str, execution_options: dict[str, Any normalized.pop("mitigation_run_id", None) normalized.pop("risk_job_id", None) normalized.pop("risk_run_id", None) + normalized.pop("scenario_job_id", None) + normalized.pop("scenario_run_id", None) + normalized.pop("base_job_id", None) + normalized.pop("base_run_id", None) + normalized.pop("base_job_type", None) + normalized.pop("base_job_name", None) + normalized.pop("mitigation_job_name", None) + normalized.pop("risk_job_name", None) normalized.pop("source_job_type", None) normalized.pop("non_construction", None) return normalized @@ -1420,6 +1578,11 @@ def _validate_report_options(db: Session, *, line_id: str, execution_options: di else: execution_options.pop("mitigation_job_id", None) execution_options.pop("mitigation_run_id", None) + scenario_job = ( + _find_latest_success_scenario_job(db, line_id=line_id, mitigation_job_id=mitigation_job.id) + if mitigation_job is not None + else None + ) else: mitigation_job = source_job risk_job_id = str((mitigation_job.execution_options_json or {}).get("source_job_id") or "") @@ -1447,10 +1610,21 @@ def _validate_report_options(db: Session, *, line_id: str, execution_options: di run_id=source_run_id, exclude_low_risk=False, ) + scenario_job = _find_latest_success_scenario_job(db, line_id=line_id, mitigation_job_id=mitigation_job.id) if not allowed_tower_ids: raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="报告来源任务暂无可复用杆塔结果") + if scenario_job is not None: + scenario_run_id = _resolve_source_run_id(scenario_job) + if not scenario_run_id: + raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="关联复算任务缺少可复用结果") + execution_options["scenario_job_id"] = scenario_job.id + execution_options["scenario_run_id"] = scenario_run_id + else: + execution_options.pop("scenario_job_id", None) + execution_options.pop("scenario_run_id", None) + invalid_ids = [tower_id for tower_id in selected_tower_ids if tower_id not in allowed_tower_ids] if invalid_ids: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="报告任务包含无效的杆塔选择") @@ -1495,11 +1669,27 @@ def _find_latest_success_mitigation_job(db: Session, *, line_id: str, source_job return None +def _find_latest_success_scenario_job(db: Session, *, line_id: str, mitigation_job_id: str) -> FlAnalysisJob | None: + rows = db.execute( + select(FlAnalysisJob) + .where( + FlAnalysisJob.line_id == line_id, + FlAnalysisJob.job_type == "scenario", + FlAnalysisJob.status == "success", + ) + .order_by(FlAnalysisJob.update_date.desc(), FlAnalysisJob.id.desc()) + ).scalars().all() + for item in rows: + if str((item.execution_options_json or {}).get("mitigation_job_id") or "") == mitigation_job_id: + return item + return None + + def _load_job_towers(db: Session, *, job: FlAnalysisJob, execution_options: dict[str, Any]) -> list[LineTower]: towers = db.execute( select(LineTower).where(LineTower.line_id == job.line_id).order_by(LineTower.seq_no.asc()) ).scalars().all() - if job.job_type not in {"mitigation", "report"}: + if job.job_type not in {"mitigation", "report", "scenario"}: return towers selected_ids = set(execution_options.get("selected_tower_ids") or []) scoped_towers = [tower for tower in towers if tower.id in selected_ids] @@ -1507,10 +1697,105 @@ def _load_job_towers(db: Session, *, job: FlAnalysisJob, execution_options: dict detail = "措施推荐任务的杆塔范围已失效，请重新生成任务" if job.job_type == "report": detail = "报告任务的杆塔范围已失效，请重新生成任务" + elif job.job_type == "scenario": + detail = "加装避雷器复算任务的杆塔范围已失效，请重新生成任务" raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=detail) return scoped_towers +def _validate_scenario_options(db: Session, *, line_id: str, execution_options: dict[str, Any]) -> dict[str, Any]: + mitigation_job_id = str(execution_options.get("source_job_id") or "") + if not mitigation_job_id: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="加装避雷器复算任务缺少前驱措施任务") + selected_tower_ids = execution_options.get("selected_tower_ids") or [] + if not selected_tower_ids: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="加装避雷器复算任务至少需要选择一座杆塔") + base_job_id = str(execution_options.get("base_job_id") or "") + if not base_job_id: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="加装避雷器复算任务缺少复用的计算任务") + + mitigation_job = get_job_by_id(db, mitigation_job_id) + if not mitigation_job: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="前驱措施任务不存在") + if mitigation_job.line_id != line_id: + raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="前驱措施任务与当前线路不匹配") + if mitigation_job.job_type != "mitigation": + raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="前驱任务必须为措施推荐任务") + if mitigation_job.status != "success": + raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="前驱措施任务尚未成功完成") + mitigation_run_id = _resolve_source_run_id(mitigation_job) + if not mitigation_run_id: + raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="前驱措施任务缺少可复用结果") + allowed_tower_ids = _load_report_candidate_tower_ids( + db, + job_id=mitigation_job.id, + run_id=mitigation_run_id, + exclude_low_risk=True, + ) + if not allowed_tower_ids: + raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="前驱措施任务没有可复算的中高风险杆塔") + invalid_ids = [tower_id for tower_id in selected_tower_ids if tower_id not in allowed_tower_ids] + if invalid_ids: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="加装避雷器复算任务包含无效的杆塔选择") + + risk_job_id = str((mitigation_job.execution_options_json or {}).get("source_job_id") or "") + if not risk_job_id: + raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="前驱措施任务缺少关联风险任务") + risk_job = get_job_by_id(db, risk_job_id) + if not risk_job: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="关联风险任务不存在") + if risk_job.line_id != line_id: + raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="关联风险任务与当前线路不匹配") + if risk_job.job_type != "risk" or risk_job.status != "success": + raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="关联风险任务尚未成功完成") + risk_run_id = _resolve_source_run_id(risk_job) + if not risk_run_id: + raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="关联风险任务缺少可复用结果") + + base_job = get_job_by_id(db, base_job_id) + if not base_job: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="复用计算任务不存在") + if base_job.line_id != line_id: + raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="复用计算任务与当前线路不匹配") + if base_job.job_type not in {"normal", "tongtiao"}: + raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="仅普通计算或同跳计算任务可作为复算基线") + if base_job.status != "success": + raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="复用计算任务尚未成功完成") + base_run_id = _resolve_source_run_id(base_job) + if not base_run_id: + raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="复用计算任务缺少可复用结果") + base_tower_ids = _load_result_tower_ids(db, job_id=base_job.id, run_id=base_run_id) + missing_from_base = [tower_id for tower_id in selected_tower_ids if tower_id not in base_tower_ids] + if missing_from_base: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="复算杆塔不在所选基线计算结果中") + + scenario_options = _normalize_execution_options(base_job.job_type, base_job.execution_options_json or {}) + scenario_options.update( + { + "selected_tower_ids": list(selected_tower_ids), + "source_job_id": mitigation_job.id, + "source_run_id": mitigation_run_id, + "source_job_type": mitigation_job.job_type, + "mitigation_job_id": mitigation_job.id, + "mitigation_job_name": mitigation_job.job_name, + "mitigation_run_id": mitigation_run_id, + "risk_job_id": risk_job.id, + "risk_job_name": risk_job.job_name, + "risk_run_id": risk_run_id, + "base_job_id": base_job.id, + "base_job_name": base_job.job_name, + "base_job_type": base_job.job_type, + "base_run_id": base_run_id, + } + ) + return { + "total_tower_count": len(selected_tower_ids), + "execution_options": scenario_options, + "external_adapter": base_job.external_adapter or "placeholder", + "adapter_config_json": base_job.adapter_config_json or {}, + } + + def _load_source_result_map(db: Session, *, execution_options: dict[str, Any]) -> dict[str, dict[str, Any]]: source_job_id = str(execution_options.get("source_job_id") or "") source_run_id = str(execution_options.get("source_run_id") or "") @@ -1555,6 +1840,22 @@ def _load_report_candidate_tower_ids( return tower_ids +def _resolve_waveform_job_type(*, job_type: str, execution_options: dict[str, Any]) -> str: + if job_type == "scenario": + base_job_type = str(execution_options.get("base_job_type") or "").strip() + if base_job_type in {"normal", "tongtiao"}: + return base_job_type + return job_type + + +def _apply_scenario_profile_json_overrides(profile_json: dict[str, Any]) -> dict[str, Any]: + profile = dict(profile_json or {}) + profile["arrester_a"] = "是" + profile["arrester_b"] = "是" + profile["arrester_c"] = "是" + return profile + + def _resolve_source_run_id(job: FlAnalysisJob) -> str | None: if job.latest_run_id: return job.latest_run_id diff --git a/api/pyproject.toml b/api/pyproject.toml index 95ae26d..d1821ea 100644 --- a/api/pyproject.toml +++ b/api/pyproject.toml @@ -18,3 +18,6 @@ dependencies = [ "flower>=2.0.0,<3.0.0", "rasterio>=1.4.0,<2.0.0", ] + +[tool.pytest.ini_options] +pythonpath = ["."] diff --git a/api/tests/test_fl_analysis_report.py b/api/tests/test_fl_analysis_report.py index 1ac7b87..703b2dc 100644 --- a/api/tests/test_fl_analysis_report.py +++ b/api/tests/test_fl_analysis_report.py @@ -22,6 +22,9 @@ def _sample_report_data() -> dict[str, object]: "risk_job_name": "示例线路-风险评估", "mitigation_job_id": "mit-job-1", "mitigation_job_name": "示例线路-措施推荐", + "scenario_job_id": "scenario-job-1", + "scenario_job_name": "示例线路-加装避雷器复算", + "scenario_base_job_type": "normal", }, "risk_rows": [ { @@ -108,6 +111,19 @@ def _sample_report_data() -> dict[str, object]: }, } ], + "selected_scenario_rows": [ + { + "tower_id": "tower-1", + "tower_no": "001", + "risk_level": "low", + "result_json": { + "counterstrike_withstand_ka": 112.45, + "counterstrike_trip_rate": 0.0215, + "shielding_withstand_ka": 136.2, + "shielding_trip_rate": 0.0085, + }, + } + ], } @@ -124,6 +140,9 @@ def test_build_report_summary_payload_counts_risk_and_actions() -> None: assert summary["mitigation_action_counts"]["接地治理"] == 1 assert summary["mitigation_action_counts"]["安装避雷器"] == 1 assert summary["has_mitigation_data"] is True + assert summary["scenario_row_count"] == 1 + assert summary["post_recalc_risk_counts"] == {"high": 0, "medium": 0, "low": 1} + assert summary["has_scenario_data"] is True def test_build_report_document_renders_word_compatible_html() -> None: @@ -136,4 +155,6 @@ def test_build_report_document_renders_word_compatible_html() -> None: assert "雷害风险评估结果" in html assert "差异化防雷措施与预期效果" in html assert "安装避雷器" in html + assert "表14 采取措施后的计算结果表" in html + assert "反击耐雷水平(kA)" in html assert "001" in html diff --git a/api/tests/test_fl_analysis_service.py b/api/tests/test_fl_analysis_service.py new file mode 100644 index 0000000..0df94e2 --- /dev/null +++ b/api/tests/test_fl_analysis_service.py @@ -0,0 +1,422 @@ +from __future__ import annotations + +from types import SimpleNamespace + +from sqlalchemy import create_engine, select +from sqlalchemy.orm import Session, sessionmaker + +from app.core.database import Base +from app.models.fl_analysis import FlAnalysisJob, FlAnalysisRun, FlAnalysisTowerResult, FlAnalysisTowerSnapshot +from app.models.line import Line +from app.models.line_tower import LineTower +from app.models.tower_profile import TowerProfile +from app.schemas.fl_analysis import FlAnalysisJobCreateRequest +from app.services import fl_analysis_service + + +def _build_sessionmaker(): + engine = create_engine("sqlite+pysqlite:///:memory:") + Base.metadata.create_all( + bind=engine, + tables=[ + Line.__table__, + LineTower.__table__, + TowerProfile.__table__, + FlAnalysisJob.__table__, + FlAnalysisRun.__table__, + FlAnalysisTowerSnapshot.__table__, + FlAnalysisTowerResult.__table__, + ], + ) + return sessionmaker(bind=engine, autocommit=False, autoflush=False, expire_on_commit=False) + + +def _create_completed_job( + session: Session, + *, + line: Line, + tower: LineTower, + profile: TowerProfile, + job_type: str, + job_name: str, + risk_level: str, + result_json: dict[str, object], + execution_options_json: dict[str, object] | None = None, + external_adapter: str = "placeholder", +) -> tuple[FlAnalysisJob, FlAnalysisRun]: + job = FlAnalysisJob( + line_id=line.id, + job_name=job_name, + job_type=job_type, + status="success", + source_kind="line", + total_tower_count=1, + snapshotted_tower_count=1, + result_tower_count=1, + external_adapter=external_adapter, + execution_options_json=execution_options_json or {}, + result_summary_json={}, + create_user="tester", + update_user="tester", + ) + session.add(job) + session.flush() + + run = FlAnalysisRun( + job_id=job.id, + status="success", + runner_kind=external_adapter, + create_user="tester", + update_user="tester", + ) + session.add(run) + session.flush() + + job.latest_run_id = run.id + snapshot = FlAnalysisTowerSnapshot( + job_id=job.id, + run_id=run.id, + tower_id=tower.id, + seq_no=tower.seq_no, + tower_no=tower.tower_no, + tower_model=tower.tower_model, + tower_type=tower.tower_type, + longitude=tower.longitude, + latitude=tower.latitude, + altitude_m=tower.altitude_m, + terrain=tower.terrain, + base_tower_json=fl_analysis_service._build_base_tower_json(tower, line), + profile_json=fl_analysis_service._build_profile_json(profile), + ) + session.add(snapshot) + session.flush() + + session.add( + FlAnalysisTowerResult( + job_id=job.id, + run_id=run.id, + snapshot_id=snapshot.id, + status="success", + risk_level=risk_level, + summary_text=str(result_json.get("summary_text") or ""), + result_json=result_json, + ) + ) + session.flush() + return job, run + + +def _seed_job_chain(session: Session) -> dict[str, object]: + line = Line( + code="L-100", + name="迁移验证线路", + voltage_kv=220, + lightning_param_json={"雷电流幅值a": 31.0, "雷电流幅值b": 2.6}, + ) + session.add(line) + session.flush() + + tower = LineTower( + line_id=line.id, + seq_no=1, + tower_no="T001", + tower_model="220-TEST-ZX", + tower_type="直线", + longitude=120.0, + latitude=30.0, + altitude_m=128.0, + terrain="山地", + ground_resistance_ohm=16.0, + lightning_density=4.2, + span_large_m=320.0, + span_small_m=280.0, + slope_1=5.0, + slope_2=3.0, + circuit_geometry_json={ + "I": { + "phase_spacing_m": {"upper": 9.0, "middle": 4.5, "lower": 8.5}, + "phase_height_m": {"upper": 29.0, "middle": 31.0, "lower": 25.0}, + }, + "lightning_wire": { + "left_mid_distance_m": 8.0, + "right_mid_distance_m": 8.0, + "height_m": 38.0, + }, + "insulator_length_mm": 4200.0, + }, + lightning_result_json={}, + ) + session.add(tower) + session.flush() + + profile = TowerProfile( + tower_id=tower.id, + structure_kind="直线", + stroke_mode="反击", + arrester_a="否", + arrester_b="否", + arrester_c="否", + insulator_length_m=4200.0, + current_a=31.0, + current_b=2.6, + current_type="Heidler", + current_head_time_us=2.6, + current_tail_time_us=50.0, + ) + session.add(profile) + session.flush() + + base_job, base_run = _create_completed_job( + session, + line=line, + tower=tower, + profile=profile, + job_type="normal", + job_name="基线普通计算", + risk_level="high", + result_json={ + "risk_level": "high", + "summary_text": "基线普通计算完成", + "score": 88, + }, + execution_options_json={ + "current_waveform": "double_slope", + "flashover_method": "intersection", + "altitude_correction": "formula1", + "induced_voltage_formula": "formula2", + "head_time_min_us": 2.4, + "head_time_max_us": 2.8, + "head_time_step_us": 0.2, + "tail_time_min_us": 45.0, + "tail_time_max_us": 55.0, + "tail_time_step_us": 5.0, + }, + ) + risk_job, risk_run = _create_completed_job( + session, + line=line, + tower=tower, + profile=profile, + job_type="risk", + job_name="风险评估", + risk_level="high", + result_json={ + "risk_level": "high", + "summary_text": "高风险", + "score": 91, + "cause_analysis": "接地电阻偏高", + "mitigation_recommendation": "建议安装避雷器", + "reason_details": [{"code": "ground_resistance", "label": "接地电阻", "triggered": True}], + }, + ) + mitigation_job, mitigation_run = _create_completed_job( + session, + line=line, + tower=tower, + profile=profile, + job_type="mitigation", + job_name="措施推荐", + risk_level="medium", + result_json={ + "risk_level": "medium", + "summary_text": "仍需补强", + "current_risk_level": "high", + "expected_risk_level": "medium", + "cause_analysis": "接地电阻偏高", + "recommendation_result": "需要安装避雷器", + "mitigation_recommendation": "安装避雷器并优化接地", + "mitigation_actions": [ + {"code": "arrester_install", "label": "安装避雷器", "summary": "关键相增设避雷器"}, + ], + }, + execution_options_json={ + "source_job_id": risk_job.id, + "source_run_id": risk_run.id, + "selected_tower_ids": [tower.id], + }, + ) + + session.commit() + return { + "line": line, + "tower": tower, + "profile": profile, + "base_job": base_job, + "base_run": base_run, + "risk_job": risk_job, + "risk_run": risk_run, + "mitigation_job": mitigation_job, + "mitigation_run": mitigation_run, + } + + +def test_execute_scenario_job_reuses_base_waveform_and_forces_arresters(monkeypatch) -> None: + testing_session = _build_sessionmaker() + session = testing_session() + try: + seeded = _seed_job_chain(session) + monkeypatch.setattr(fl_analysis_service, "_publish_change", lambda *args, **kwargs: None) + + created = fl_analysis_service.create_job( + session, + FlAnalysisJobCreateRequest( + line_id=seeded["line"].id, + job_name="加装避雷器复算", + job_type="scenario", + execution_options_json={ + "source_job_id": seeded["mitigation_job"].id, + "base_job_id": seeded["base_job"].id, + "selected_tower_ids": [seeded["tower"].id], + }, + ), + actor=SimpleNamespace(id="tester"), + ) + + assert created.job.external_adapter == "placeholder" + assert created.job.execution_options_json["base_job_type"] == "normal" + assert created.job.execution_options_json["current_waveform"] == "double_slope" + assert created.job.execution_options_json["mitigation_job_id"] == seeded["mitigation_job"].id + + captured_payloads: list[dict[str, object]] = [] + + def fake_grade_normal_snapshot_payload(payload: dict[str, object], *, execution_options: dict[str, object]) -> dict[str, object]: + captured_payloads.append( + { + "profile_json": dict(payload["profile_json"]), + "execution_options": dict(execution_options), + } + ) + return { + "risk_level": "low", + "summary_text": "复算完成", + "score": 12, + "workflow": {"scan_point_count": 1}, + "counterstrike_withstand_ka": 120.5, + "counterstrike_trip_rate": 0.02, + "shielding_withstand_ka": 140.0, + "shielding_trip_rate": 0.005, + } + + monkeypatch.setattr(fl_analysis_service, "SessionLocal", testing_session) + monkeypatch.setattr(fl_analysis_service, "grade_normal_snapshot_payload", fake_grade_normal_snapshot_payload) + monkeypatch.setattr( + fl_analysis_service, + "grade_tongtiao_snapshot_payload", + lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("scenario should reuse normal path")), + ) + monkeypatch.setattr( + fl_analysis_service, + "grade_snapshot_payload", + lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("scenario should not fall back to generic risk grading")), + ) + + session.close() + fl_analysis_service.execute_job(created.job.id) + + verify_session = testing_session() + try: + saved_job = fl_analysis_service.get_job_by_id(verify_session, created.job.id) + assert saved_job is not None + assert saved_job.status == "success" + assert saved_job.result_summary_json["base_job_type"] == "normal" + assert saved_job.result_summary_json["selected_tower_count"] == 1 + assert saved_job.result_summary_json["workflow"]["current_waveform"] == "double_slope" + + saved_row = verify_session.execute( + select(FlAnalysisTowerResult).where(FlAnalysisTowerResult.job_id == created.job.id) + ).scalar_one() + assert saved_row.risk_level == "low" + assert saved_row.result_json["counterstrike_withstand_ka"] == 120.5 + finally: + verify_session.close() + + assert len(captured_payloads) == 1 + assert captured_payloads[0]["profile_json"]["arrester_a"] == "是" + assert captured_payloads[0]["profile_json"]["arrester_b"] == "是" + assert captured_payloads[0]["profile_json"]["arrester_c"] == "是" + assert captured_payloads[0]["execution_options"]["current_waveform"] == "double_slope" + finally: + session.close() + + +def test_report_job_auto_links_latest_scenario_rows(monkeypatch) -> None: + testing_session = _build_sessionmaker() + session = testing_session() + try: + seeded = _seed_job_chain(session) + monkeypatch.setattr(fl_analysis_service, "_publish_change", lambda *args, **kwargs: None) + + scenario_job, scenario_run = _create_completed_job( + session, + line=seeded["line"], + tower=seeded["tower"], + profile=seeded["profile"], + job_type="scenario", + job_name="加装避雷器复算", + risk_level="low", + result_json={ + "risk_level": "low", + "summary_text": "复算完成", + "score": 14, + "counterstrike_withstand_ka": 112.45, + "counterstrike_trip_rate": 0.0215, + "shielding_withstand_ka": 136.2, + "shielding_trip_rate": 0.0085, + }, + execution_options_json={ + "source_job_id": seeded["mitigation_job"].id, + "source_run_id": seeded["mitigation_run"].id, + "source_job_type": "mitigation", + "mitigation_job_id": seeded["mitigation_job"].id, + "mitigation_run_id": seeded["mitigation_run"].id, + "risk_job_id": seeded["risk_job"].id, + "risk_run_id": seeded["risk_run"].id, + "base_job_id": seeded["base_job"].id, + "base_run_id": seeded["base_run"].id, + "base_job_type": "normal", + "base_job_name": seeded["base_job"].job_name, + }, + ) + session.commit() + + created = fl_analysis_service.create_job( + session, + FlAnalysisJobCreateRequest( + line_id=seeded["line"].id, + job_name="效果报告", + job_type="report", + execution_options_json={ + "source_job_id": seeded["mitigation_job"].id, + "selected_tower_ids": [seeded["tower"].id], + }, + ), + actor=SimpleNamespace(id="tester"), + ) + + assert created.job.execution_options_json["scenario_job_id"] == scenario_job.id + assert created.job.execution_options_json["scenario_run_id"] == scenario_run.id + + monkeypatch.setattr(fl_analysis_service, "SessionLocal", testing_session) + session.close() + fl_analysis_service.execute_job(created.job.id) + + verify_session = testing_session() + try: + report_job = fl_analysis_service.get_job_by_id(verify_session, created.job.id) + assert report_job is not None + assert report_job.status == "success" + assert report_job.result_summary_json["scenario_row_count"] == 1 + assert report_job.result_summary_json["has_scenario_data"] is True + assert report_job.result_summary_json["post_recalc_risk_counts"] == {"high": 0, "medium": 0, "low": 1} + + filename, content = fl_analysis_service.download_report_document(verify_session, job_id=created.job.id) + finally: + verify_session.close() + + html = content.decode("utf-8") + assert filename.endswith(".doc") + assert "表14 采取措施后的计算结果表" in html + assert "加装避雷器复算" in html + assert "112.45" in html + finally: + session.close() diff --git a/web/src/app/admin/fl-analysis/page.tsx b/web/src/app/admin/fl-analysis/page.tsx index 8205399..25a211c 100644 --- a/web/src/app/admin/fl-analysis/page.tsx +++ b/web/src/app/admin/fl-analysis/page.tsx @@ -65,6 +65,11 @@ type MitigationFormValues = { non_construction: boolean; }; +type ScenarioFormValues = { + job_name: string; + base_job_id: string; +}; + type ReportFormValues = { job_name: string; }; @@ -175,7 +180,7 @@ function formatJobType(jobType: string, nonConstruction = false): string { if (jobType === "normal") return "普通计算"; if (jobType === "tongtiao") return "同跳计算"; if (jobType === "report") return "报告"; - if (jobType === "scenario") return "场景分析"; + if (jobType === "scenario") return "加装避雷器复算"; return jobType || "-"; } @@ -354,6 +359,20 @@ function mitigationMode(job: FlAnalysisJobDetail | FlAnalysisJobSummary | null): return Boolean(options.non_construction); } +function waveformJobType(job: FlAnalysisJobDetail | FlAnalysisJobSummary | null): "normal" | "tongtiao" | null { + if (!job) { + return null; + } + if (job.job_type === "normal" || job.job_type === "tongtiao") { + return job.job_type; + } + if (job.job_type === "scenario") { + const baseJobType = readOptionalString(readObject(job.execution_options_json), "base_job_type"); + return baseJobType === "tongtiao" ? "tongtiao" : "normal"; + } + return null; +} + function selectedTowerCount(job: FlAnalysisJobDetail | null): number { if (!job) { return 0; @@ -381,12 +400,15 @@ export default function AdminFlAnalysisPage() { const [selectedJobId, setSelectedJobId] = useState