vent_agent_2.0/algorithms/main_fan/models.py

"""统一业务事实。单位、来源和人工确认均显式记录。"""
from datetime import date, datetime
from typing import Literal

from pydantic import AliasChoices, BaseModel, ConfigDict, Field, field_validator


class FactModel(BaseModel):
    model_config = ConfigDict(extra="forbid", populate_by_name=True, allow_inf_nan=False,
                             str_strip_whitespace=True)


class Evidence(FactModel):
    source: str = ""
    page: int | None = Field(default=None, ge=1)
    label: str = ""
    excerpt: str = ""
    sha256: str = ""
    method: Literal["api", "manual_transcription", "model_extraction", "text_extraction", "synthetic"] = "api"


class Device(FactModel):
    device_code: str = Field(min_length=1, validation_alias=AliasChoices("device_code", "deviceCode"))
    name: str = Field(min_length=1, validation_alias=AliasChoices("name", "deviceName"))
    mine_code: str = Field(default="", validation_alias=AliasChoices("mine_code", "mineCode"))
    mine_name: str = Field(default="", validation_alias=AliasChoices("mine_name", "mineName"))
    location: str = ""
    state: Literal["running", "stopped", "unknown"] = "unknown"
    fan_role: Literal["in_service", "standby", "upgrading", "replacing", "unknown"] = "unknown"
    availability_status: Literal["available", "unavailable", "unknown"] = "unknown"
    observed_at: datetime | None = None
    evidence: Evidence = Field(default_factory=Evidence)

    @field_validator("state", mode="before")
    @classmethod
    def state_label(cls, value):
        return {"开": "running", "停": "stopped"}.get(value, value)


Metric = Literal["velocity", "flow", "static_pressure", "pressure", "shaft_power",
                 "input_power", "temperature", "current", "voltage", "speed", "frequency", "unknown"]


class MonitorPoint(FactModel):
    point_code: str = Field(min_length=1, validation_alias=AliasChoices("point_code", "pointCode"))
    name: str = Field(min_length=1, validation_alias=AliasChoices("name", "pointName"))
    mine_code: str = Field(default="", validation_alias=AliasChoices("mine_code", "mineCode"))
    mine_name: str = Field(default="", validation_alias=AliasChoices("mine_name", "mineName"))
    device_name: str = ""
    location: str = ""
    metric: Metric = "unknown"
    value: float | None = None
    unit: str = ""
    status: str = "unknown"
    observed_at: datetime | None = None
    evidence: Evidence = Field(default_factory=Evidence)


class ConfirmedBinding(FactModel):
    point_code: str = Field(min_length=1)
    fan_id: str = Field(min_length=1)
    confirmed_by: str = Field(min_length=1)
    reason: str = Field(min_length=1)


class PerformancePoint(FactModel):
    label: str = ""
    frequency_hz: float | None = Field(default=None, gt=0)
    blade_angle_deg: float | None = None
    # 现场实测与折算值不可混用；每条记录都是同一行的量。
    condition: Literal["measured", "converted"] = "measured"
    flow_m3_s: float = Field(ge=0)
    static_pressure_pa: float = Field(ge=0)
    shaft_power_kw: float = Field(gt=0)
    input_power_kw: float | None = Field(default=None, gt=0)
    reported_efficiency_pct: float | None = Field(default=None, ge=0, le=100)
    evidence: Evidence = Field(default_factory=Evidence)


class PerformanceReport(FactModel):
    report_id: str = Field(min_length=1)
    mine_name: str = Field(min_length=1)
    mine_code: str = ""
    fan_name: str = Field(min_length=1)
    serial_number: str = ""
    model: str = ""
    motor_model: str = ""
    rated_power_raw: str = Field(default="", description="铭牌额定功率原表达式，如2×185 kW；不以运行功率代替")
    rated_power_kw: float | None = Field(default=None, gt=0, description="仅直接载明的总额定功率；组合表达式由工具计算")
    rated_speed_rpm: float | None = Field(default=None, gt=0)
    rated_voltage_raw: str = ""
    manufacturer: str = ""
    impeller_diameter_m: float | None = Field(default=None, gt=0)
    ventilation_mode: str = ""
    structure_type: str = ""
    rated_flow_range_m3_min: list[float] = Field(default_factory=list, max_length=2)
    rated_pressure_range_pa: list[float] = Field(default_factory=list, max_length=2)
    blade_angle_deg: float | None = Field(default=None, description="测试叶片安装角；原表/或空白表示未载明，不填0")
    inspected_on: date | None = None
    recommended_next_on: date | None = None
    original_conclusion: str = ""
    basis: list[str] = Field(default_factory=list)
    points: list[PerformancePoint] = Field(default_factory=list, max_length=250)
    curves: list[PerformancePoint] = Field(default_factory=list, max_length=30,
        description="只录曲线图注印出的工况数字，不从曲线像素猜读，和数据表分开保存")
    metadata_evidence: dict[str, Evidence] = Field(default_factory=dict,
        description="每个身份/日期字段的原文与物理页码，日期须含检测/建议下次标签")
    evidence: Evidence = Field(default_factory=Evidence)


class DocumentFacts(FactModel):
    reports: list[PerformanceReport] = Field(default_factory=list)
    devices: list[Device] = Field(default_factory=list)
    points: list[MonitorPoint] = Field(default_factory=list)
    warnings: list[str] = Field(default_factory=list)
    extraction_audit: dict = Field(default_factory=dict, description="仅由服务端填写，记录路由、缓存和模型调用")


class SeriesSample(FactModel):
    observed_at: datetime
    value: float
    status: str = "normal"


class OperatingSeries(FactModel):
    mine_code: str = ""
    mine_name: str = ""
    fan_id: str = Field(min_length=1)
    metric: Metric
    unit: str
    sensor_id: str = Field(min_length=1)
    mapping_confirmed: bool = False
    samples: list[SeriesSample] = Field(default_factory=list, max_length=200000)
    evidence: Evidence = Field(default_factory=Evidence)


class SwitchRecord(FactModel):
    mine_code: str = ""
    mine_name: str = ""
    event_id: str = Field(min_length=1)
    switched_at: datetime
    from_fan_id: str = Field(min_length=1)
    to_fan_id: str = Field(min_length=1)
    from_stopped_at: datetime | None = None
    to_started_at: datetime | None = None
    reason: str = "unknown"
    evidence: Evidence = Field(default_factory=Evidence)


class FanStateRecord(FactModel):
    mine_code: str = ""
    mine_name: str = ""
    fan_id: str = Field(min_length=1)
    observed_at: datetime
    state: Literal["running", "stopped", "unknown"]
    complete_fan_state: bool = False
    frequency_hz: float | None = Field(default=None, gt=0)
    blade_angle_deg: float | None = None
    event_type: Literal["state", "start_stop", "frequency_change", "angle_change", "fault"] = "state"
    evidence: Evidence = Field(default_factory=Evidence)


class ScreeningInput(FactModel):
    mine_code: str = Field(min_length=1, validation_alias=AliasChoices("mine_code", "mineCode"))
    mine_name: str = Field(min_length=1, validation_alias=AliasChoices("mine_name", "mineName"))
    fan_id: str = Field(default="", validation_alias=AliasChoices("fan_id", "fanId"))
    fan_name: str = ""
    fan_pair: list[str] = Field(default_factory=list, max_length=2, description="两台风机的明确标识；不表示谁主用谁备用")
    operating_series: list[OperatingSeries] = Field(default_factory=list, max_length=200)
    switch_records: list[SwitchRecord] = Field(default_factory=list, max_length=1000)
    fan_state_records: list[FanStateRecord] = Field(default_factory=list, max_length=200000)
    evaluation_time: datetime | None = None
    devices: list[Device] = Field(default_factory=list, max_length=500)
    points: list[MonitorPoint] = Field(default_factory=list, max_length=3000)
    reports: list[PerformanceReport] = Field(default_factory=list, max_length=50)
    confirmed_bindings: list[ConfirmedBinding] = Field(default_factory=list)


class Finding(FactModel):
    code: str
    severity: Literal["info", "review", "blocking"]
    message: str
    evidence: list[Evidence] = Field(default_factory=list)


class Association(FactModel):
    point_code: str
    candidate_fan: str
    state: Literal["candidate", "confirmed", "unmatched", "ambiguous", "excluded"]
    confidence: Literal["none", "low", "medium", "confirmed"]
    basis: list[str]
    eligible_for_calculation: bool = False


class Calculation(FactModel):
    scope: Literal["historical_report", "current_confirmed"]
    label: str
    formula: str = "eta_static = Q[m3/s] * p_static[Pa] / (1000 * P_shaft[kW]) * 100%"
    values: dict[str, float]
    efficiency_pct: float
    reported_efficiency_pct: float | None = None
    difference_percentage_points: float | None = None
    evidence: list[Evidence] = Field(default_factory=list)


class Screening(FactModel):
    scope: Literal["preliminary_screening"] = "preliminary_screening"
    engineering_status: Literal["insufficient_data", "review_required"]
    has_in_scope_data: bool
    findings: list[Finding]
    associations: list[Association]
    calculations: list[Calculation]
    missing_materials: list[str]
    document_audits: list[dict] = Field(default_factory=list)