vent_ai_platform/app/services/diagnosis_service.py

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",
        },
    )