from __future__ import annotations from datetime import datetime, timezone from sqlalchemy.orm import Session from app.db.base import new_id from app.db.models.diagnosis import DiagnosisTask from app.db.repositories import diagnosis as diagnosis_repo from app.db.repositories import mine as mine_repo from app.db.repositories import monitoring as monitoring_repo from app.schemas.diagnosis import DiagnosisRunRequest from app.services.mine_service import require_mine SENSOR_SCENES = { "gas": "灾害防治", "dust": "灾害防治", "fire": "灾害防治", "co": "灾害防治", "airflow": "风量分配", } def _task_no() -> str: stamp = datetime.now(timezone.utc).strftime("%Y%m%d%H%M%S") return f"DX-{stamp}-{new_id()[:6].upper()}" def _severity_for(sensor, value: float) -> str | None: if sensor.alarm_threshold is not None and value >= sensor.alarm_threshold: return "high" if sensor.warning_threshold is not None and value >= sensor.warning_threshold: return "medium" return None def run_diagnosis( db: Session, payload: DiagnosisRunRequest, ) -> DiagnosisTask: mine = require_mine(db, payload.mine_id) task = diagnosis_repo.create_task( db, { "mine_id": mine.id, "task_no": _task_no(), "scene": payload.scene, "trigger_type": "manual", "status": "running", "started_at": datetime.now(timezone.utc), "requested_by": payload.requested_by, "model_name": "rule-engine", "input_snapshot": payload.context, }, ) sensors = monitoring_repo.list_sensors(db, mine.id) for sensor in sensors: scene = SENSOR_SCENES.get(sensor.sensor_type) if scene is None: continue if payload.scene != "all" and payload.scene != scene: continue if sensor.last_value is None: continue severity = _severity_for(sensor, float(sensor.last_value)) if severity is None: continue result = diagnosis_repo.create_result( db, { "task_id": task.id, "defect_code": f"SENSOR-{sensor.sensor_type.upper()}-LIMIT", "defect_name": f"{sensor.name}监测值超过阈值", "scene": scene, "severity": severity, "confidence": 0.96 if severity == "high" else 0.88, "status": "open", "root_cause": "", "conclusion": ( f"当前值 {sensor.last_value}{sensor.unit}," f"报警阈值 {sensor.alarm_threshold or '-'}{sensor.unit}。" ), "recommendation": "核实现场工况,完成参数复测并记录处置过程。", "related_objects": [{"type": "sensor", "id": sensor.id, "code": sensor.code}], "raw_output": { "value": sensor.last_value, "warning_threshold": sensor.warning_threshold, "alarm_threshold": sensor.alarm_threshold, }, }, ) diagnosis_repo.create_evidence( db, { "result_id": result.id, "evidence_type": "monitoring", "source_id": sensor.id, "source_name": sensor.name, "content": f"最近监测值 {sensor.last_value}{sensor.unit}", "confidence": result.confidence, "occurred_at": sensor.last_seen_at, "metadata_json": {"sensor_code": sensor.code}, }, ) facilities = monitoring_repo.list_facilities(db, mine.id) for facility in facilities: if payload.scene not in {"all", "通风设施"} or facility.status == "normal": continue result = diagnosis_repo.create_result( db, { "task_id": task.id, "defect_code": "FACILITY-STATUS-ABNORMAL", "defect_name": f"{facility.name}状态异常", "scene": "通风设施", "severity": "medium", "confidence": 0.9, "status": "open", "conclusion": f"设施当前状态为 {facility.status}。", "recommendation": "安排现场复核设施开闭状态、传感器状态和视频画面。", "related_objects": [ {"type": "ventilation_facility", "id": facility.id, "code": facility.code} ], "raw_output": {"status": facility.status, "open_state": facility.open_state}, }, ) diagnosis_repo.create_evidence( db, { "result_id": result.id, "evidence_type": "facility", "source_id": facility.id, "source_name": facility.name, "content": f"设施状态 {facility.status},开闭状态 {facility.open_state or '-'}", "confidence": 0.9, "occurred_at": facility.last_checked_at, "metadata_json": {}, }, ) work_faces = mine_repo.list_work_faces(db, mine.id) for work_face in work_faces: if payload.scene not in {"all", "风量分配"}: continue planned = work_face.planned_air_volume actual = work_face.actual_air_volume if not planned or actual is None or actual >= planned * 0.9: continue severity = "high" if actual < planned * 0.8 else "medium" result = diagnosis_repo.create_result( db, { "task_id": task.id, "defect_code": "AIRFLOW-INSUFFICIENT", "defect_name": f"{work_face.name}风量不足", "scene": "风量分配", "severity": severity, "confidence": 0.92, "status": "open", "conclusion": f"实际风量 {actual},计划风量 {planned}。", "recommendation": "核查风窗开度、漏风点和通风阻力,完成风量调整后复测。", "related_objects": [ {"type": "work_face", "id": work_face.id, "code": work_face.code} ], "raw_output": {"actual_air_volume": actual, "planned_air_volume": planned}, }, ) diagnosis_repo.create_evidence( db, { "result_id": result.id, "evidence_type": "business_metric", "source_id": work_face.id, "source_name": work_face.name, "content": f"实际风量 {actual},计划风量 {planned}", "confidence": 0.92, "occurred_at": None, "metadata_json": {}, }, ) return diagnosis_repo.update_task( db, task, { "status": "completed", "finished_at": datetime.now(timezone.utc), "model_name": "rule-engine", }, )