vent_agent_2.0/algorithms/main_fan/analysis.py

"""确定性筛查：命名只生成候选，计算只用明确确认且同步的测点。"""
from collections import Counter
from datetime import datetime
import re
import unicodedata
from zoneinfo import ZoneInfo

from algorithms.main_fan.models import Association, Calculation, Finding, Screening, ScreeningInput

LOCAL_ZONE = ZoneInfo("Asia/Shanghai")
UNITS = {
    "velocity": {"m/s": 1}, "flow": {"m3/s": 1, "m³/s": 1, "m3/min": 1/60, "m³/min": 1/60},
    "static_pressure": {"Pa": 1, "kPa": 1000}, "pressure": {"Pa": 1, "kPa": 1000},
    "shaft_power": {"kW": 1, "W": .001}, "input_power": {"kW": 1, "W": .001},
    "temperature": {"℃": 1, "°C": 1}, "current": {"A": 1}, "voltage": {"V": 1},
    "speed": {"r/min": 1, "rpm": 1}, "frequency": {"Hz": 1},
}


def normalized(value):
    return re.sub(r"\s+", "", unicodedata.normalize("NFKC", value)).lower()


def fan_numbers(value):
    """数字必须带风机语义或号/#；不把 PLC1、一级和测点编码当风机号。"""
    value = normalized(value)
    if not any(word in value for word in ("主通", "主扇", "风机", "通风机")):
        return set()
    for cn, number in {"一": "1", "二": "2", "三": "3", "四": "4", "五": "5",
                       "六": "6", "七": "7", "八": "8", "九": "9", "十": "10"}.items():
        value = value.replace(cn + "号", number + "号")
    matches = re.findall(r"(?:主通|主扇|风机|通风机)[-_]*([0-9]+)(?:号|#)?|([0-9]+)(?:号|#)", value)
    return {int(a or b) for a, b in matches}


def local_time(value):
    return value.replace(tzinfo=LOCAL_ZONE) if value.tzinfo is None else value.astimezone(LOCAL_ZONE)


def same_mine(item, data):
    # 任一已提供的身份冲突都拒绝；不靠去除公司后缀/模糊匹配合并矿井。
    if item.mine_code and item.mine_code != data.mine_code:
        return False
    if item.mine_name and normalized(item.mine_name) != normalized(data.mine_name):
        return False
    return bool(item.mine_code or item.mine_name)


def screen(data: ScreeningInput, *, now=None, freshness_seconds=600, sync_seconds=120,
           efficiency_tolerance=.5) -> Screening:
    now = local_time(data.evaluation_time or now or datetime.now(LOCAL_ZONE))
    findings, associations, calculations, missing = [], [], [], []

    def note(code, severity, message, *evidence):
        findings.append(Finding(code=code, severity=severity, message=message, evidence=list(evidence)))

    scoped_devices = [d for d in data.devices if same_mine(d, data)]
    selected = [d for d in scoped_devices if d.device_code == data.fan_id]
    target_name = data.fan_name or (selected[0].name if len(selected) == 1 else data.fan_id)
    target_numbers = fan_numbers(target_name)
    if len(target_numbers) != 1:
        missing.append("明确的目标风机名称/编号；PLC 编号及一级、二级不是风机编号")
    bindings = {}
    for binding in data.confirmed_bindings:
        if binding.point_code in bindings:
            raise ValueError("同一测点存在重复人工确认，请消除冲突")
        bindings[binding.point_code] = binding
    duplicates = {code for code, n in Counter(p.point_code for p in data.points).items() if n > 1}
    usable, in_scope = {}, bool(scoped_devices)
    for point in data.points:
        state, confidence, basis = "unmatched", "none", []
        if not same_mine(point, data):
            state = "excluded"
            basis.append("矿井身份冲突或缺失，未参与关联和计算")
            note("point_identity", "blocking", f"测点 {point.point_code} 矿井身份不匹配或缺失。", point.evidence)
        elif point.point_code in duplicates:
            state = "ambiguous"
            basis.append("测点编码重复，未选择任意一条覆盖其他记录")
            note("duplicate_point", "blocking", f"测点 {point.point_code} 存在重复记录。", point.evidence)
        else:
            in_scope = True
            binding = bindings.get(point.point_code)
            numbers = fan_numbers(point.name + " " + point.device_name)
            if binding and binding.fan_id == data.fan_id and data.fan_id:
                if len(target_numbers) == 1 and numbers and numbers != target_numbers:
                    state = "ambiguous"
                    basis.append("人工绑定与名称中的风机编号冲突，需复核确认记录")
                    note("binding_conflict", "blocking", f"测点 {point.point_code} 人工绑定与名称编号冲突。", point.evidence)
                else:
                    state, confidence = "confirmed", "confirmed"
                    basis.append(f"人工确认：{binding.confirmed_by}；依据：{binding.reason}")
            elif len(numbers) > 1:
                state = "ambiguous"
                basis.append("名称含多个风机编号")
            elif len(target_numbers) == 1 and numbers == target_numbers:
                state, confidence = "candidate", "medium"
                basis.append("同矿井且名称风机编号一致；不能证明安装拓扑")
            else:
                basis.append("未获得可唯一指向目标风机的命名证据")
            if state in {"candidate", "ambiguous", "unmatched"}:
                missing.append(f"测点 {point.point_code} 与目标风机的人工确认")
            if point.value is None:
                note("missing_value", "review", f"测点 {point.point_code} 缺少数值。", point.evidence)
            unit_valid = point.unit in UNITS.get(point.metric, {})
            if not unit_valid:
                note("unknown_unit", "review", f"测点 {point.point_code} 的类型/单位尚不能用于计算：{point.metric}/{point.unit}。", point.evidence)
            valid_time = point.observed_at is not None
            if not valid_time:
                note("missing_time", "review", f"测点 {point.point_code} 缺少采集时间。", point.evidence)
            else:
                age = (now - local_time(point.observed_at)).total_seconds()
                valid_time = -sync_seconds <= age <= freshness_seconds
                if not valid_time:
                    note("stale_point" if age > freshness_seconds else "future_point", "review",
                         f"测点 {point.point_code} 采集时间不在当前筛查窗口内。", point.evidence)
            valid_status = point.status in {"normal", "正常", "online", "在线"}
            if not valid_status:
                note("point_status", "review", f"测点 {point.point_code} 采集状态为 {point.status}，暂不用于计算。", point.evidence)
            nonnegative = point.value is not None and (point.value >= 0 or point.metric == "temperature")
            if point.value is not None and not nonnegative:
                note("negative_value", "review", f"测点 {point.point_code} 负值不能直接作为正向风量、风机静压增量或功率。", point.evidence)
            eligible = state == "confirmed" and unit_valid and valid_time and valid_status and nonnegative
            if eligible:
                usable.setdefault(point.metric, []).append(point)
            if point.metric == "velocity" and point.value == 0:
                note("zero_velocity", "review", f"测点 {point.point_code} 记录风速为 0；仅此记录不能判定风机停转或矿井停风。", point.evidence)
                if eligible:
                    running = [d for d in scoped_devices if d.state == "running" and d.observed_at and
                        (d.device_code == data.fan_id or fan_numbers(d.name) == target_numbers) and
                        -sync_seconds <= (now - local_time(d.observed_at)).total_seconds() <= freshness_seconds and
                        abs((local_time(d.observed_at) - local_time(point.observed_at)).total_seconds()) <= sync_seconds]
                    if running:
                        note("running_zero_velocity", "review", "已确认关联的零风速与同步开机记录不一致，需核查采集、测点或设备状态；不能仅据此断定停风。",
                             point.evidence, *(d.evidence for d in running))
            if point.metric == "velocity":
                missing.append("直接风量测点，或经确认的断面面积和断面平均风速；单点风速不能直接换算风量")
            if point.metric == "pressure":
                missing.append("风压测点的物理含义与测压位置；未标定的风压不能替代风机静压增量")
        associations.append(Association(point_code=point.point_code, candidate_fan=target_name,
            state=state, confidence=confidence, basis=basis,
            eligible_for_calculation=bool(state == "confirmed" and point in usable.get(point.metric, []))))

    # 阶段开关只描述该阶段，缺少时间及主备配置时不推断整套系统停风。
    for device in scoped_devices:
        if device.observed_at is None:
            note("device_missing_time", "review", f"设备 {device.name} 开停记录缺少时间，不能与测点作同步冲突判断。", device.evidence)
        elif not -sync_seconds <= (now - local_time(device.observed_at)).total_seconds() <= freshness_seconds:
            note("stale_device" if local_time(device.observed_at) < now else "future_device", "review",
                 f"设备 {device.name} 开停记录不在当前筛查窗口内。", device.evidence)
    for metric in ("flow", "static_pressure", "shaft_power"):
        if len(usable.get(metric, [])) != 1:
            missing.append({"flow": "唯一且已确认关联的当前风量", "static_pressure": "唯一且已确认关联的当前风机静压增量",
                            "shaft_power": "唯一且已确认关联的当前轴功率（电输入功率不可替代）"}[metric])
    if all(len(usable.get(metric, [])) == 1 for metric in ("flow", "static_pressure", "shaft_power")):
        chosen = [usable[metric][0] for metric in ("flow", "static_pressure", "shaft_power")]
        times = [local_time(p.observed_at) for p in chosen]
        if (max(times) - min(times)).total_seconds() > sync_seconds:
            note("unsynchronized", "blocking", "风量、静压和轴功率不同步，未计算当前效率。", *(p.evidence for p in chosen))
        else:
            q, pressure, power = [p.value * UNITS[p.metric][p.unit] for p in chosen]
            if power <= 0:
                note("zero_shaft_power", "review", "轴功率为零，未计算效率。", chosen[2].evidence)
            else:
                eta = q * pressure / (1000 * power) * 100
                calculations.append(Calculation(scope="current_confirmed", label=target_name,
                    values={"flow_m3_s": q, "static_pressure_pa": pressure, "shaft_power_kw": power},
                    efficiency_pct=round(eta, 4), evidence=[p.evidence for p in chosen]))
                if eta > 100:
                    note("impossible_efficiency", "review", "计算效率超过 100%，应核查关联、单位、压力和功率口径。", *(p.evidence for p in chosen))

    for report in data.reports:
        if not same_mine(report, data):
            note("report_mine_mismatch", "blocking", f"报告 {report.report_id} 属于 {report.mine_name}，不能用于 {data.mine_name} 的风机研判。", report.evidence)
            continue
        if len(target_numbers) != 1 or fan_numbers(report.fan_name) != target_numbers:
            note("report_fan_mismatch", "review", f"报告 {report.report_id} 对应 {report.fan_name}，未唯一匹配目标风机。", report.evidence)
            continue
        in_scope = True
        if report.recommended_next_on and now.date() > report.recommended_next_on:
            note("report_recommended_date_passed", "review",
                 f"报告 {report.report_id} 所载建议下次检验日期 {report.recommended_next_on} 已过；需核查更新报告，不据此直接认定违法或当前不合格。", report.evidence)
        if report.inspected_on and report.inspected_on > now.date():
            note("report_future_date", "blocking", f"报告 {report.report_id} 检验日期晚于研判日期，未用于数值复核。", report.evidence)
            continue
        note("historical_report", "info", f"报告 {report.report_id} 的原结论“{report.original_conclusion or '未提取'}”仅为历史检测结论，不能证明当前状态。", report.evidence)
        for row in report.points:
            eta = row.flow_m3_s * row.static_pressure_pa / (1000 * row.shaft_power_kw) * 100
            difference = None if row.reported_efficiency_pct is None else abs(eta - row.reported_efficiency_pct)
            calculations.append(Calculation(scope="historical_report", label=f"{report.report_id}/{row.label}/{row.condition}",
                values={"flow_m3_s": row.flow_m3_s, "static_pressure_pa": row.static_pressure_pa,
                        "shaft_power_kw": row.shaft_power_kw}, efficiency_pct=round(eta, 4),
                reported_efficiency_pct=row.reported_efficiency_pct,
                difference_percentage_points=None if difference is None else round(difference, 4), evidence=[row.evidence]))
            if eta > 100 or difference is not None and difference > efficiency_tolerance:
                note("report_formula_inconsistent", "review", f"报告 {report.report_id} 的 {row.label} 效率复核存在差异，应核验原表、识别结果及单位口径。", row.evidence)
        if not report.points:
            missing.append(f"报告 {report.report_id} 可定位到原页的性能测定数据行")
        for caption in report.curves:
            caption_eta = caption.flow_m3_s * caption.static_pressure_pa / (10 * caption.shaft_power_kw)
            if caption_eta > 100 or (caption.reported_efficiency_pct is not None and
                    abs(caption_eta-caption.reported_efficiency_pct) > efficiency_tolerance):
                note("curve_formula_inconsistent", "review", f"报告 {report.report_id} 的 {caption.frequency_hz or 0:g}Hz 图注数字与其效率不一致，需核验图注原件及识别结果。", caption.evidence)
            comparable = [r for r in report.points if r.frequency_hz == caption.frequency_hz and
                          r.condition == caption.condition and "运行工况" in r.label and
                          abs(r.flow_m3_s-caption.flow_m3_s)<=max(.001,caption.flow_m3_s*.005) and
                          abs(r.shaft_power_kw-caption.shaft_power_kw)<=max(.001,caption.shaft_power_kw*.005)]
            if len(comparable) == 1:
                row = comparable[0]
                if abs(caption.static_pressure_pa-row.static_pressure_pa) > max(1, row.static_pressure_pa*.005) or (
                        caption.reported_efficiency_pct is not None and row.reported_efficiency_pct is not None and
                        abs(caption.reported_efficiency_pct-row.reported_efficiency_pct) > efficiency_tolerance):
                    condition_label = "现场" if caption.condition == "measured" else "折算"
                    note("curve_table_disagreement", "review", f"报告 {report.report_id} 的 {caption.frequency_hz:g}Hz 图注与同频率{condition_label}运行工况表不一致，保留各自原页，需复核。",
                         caption.evidence, row.evidence)
    if not any(r for r in data.reports if same_mine(r, data) and fan_numbers(r.fan_name) == target_numbers):
        missing.append("本矿井目标风机对应的性能测定报告")
    missing.append("主备风机配置、切换记录与停风处置资料；现有开停快照不足以核验连续通风能力")
    return Screening(engineering_status="review_required" if in_scope else "insufficient_data", has_in_scope_data=in_scope,
                     findings=findings, associations=associations, calculations=calculations,
                     missing_materials=list(dict.fromkeys(missing)))