220 lines
7.1 KiB
Python
220 lines
7.1 KiB
Python
"""Transparent engine health, ranking and spectral explanations."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
from .config import (
|
|
FAULT_LABELS,
|
|
FREQ_COLS,
|
|
LABEL_DISPLAY,
|
|
NOT_APPLICABLE,
|
|
RECOMMENDATIONS,
|
|
SEVERITY_DISPLAY,
|
|
)
|
|
from .service import DiagnosisResult
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class EngineAnalysis:
|
|
engine_id: str
|
|
measurements: pd.DataFrame
|
|
diagnostics: pd.DataFrame
|
|
spectra: np.ndarray
|
|
reference: np.ndarray
|
|
deviation: np.ndarray
|
|
absolute_deviation: np.ndarray
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class EngineSummary:
|
|
engine_id: str
|
|
status: str
|
|
status_tone: str
|
|
highest_severity: str
|
|
highest_severity_display: str
|
|
cylinders: int
|
|
faults: int
|
|
unknown: int
|
|
attention: int
|
|
mean_confidence: float
|
|
top_cylinder: int
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CylinderExplanation:
|
|
cylinder: int
|
|
label: str
|
|
label_display: str
|
|
severity: str
|
|
severity_display: str
|
|
confidence: float
|
|
anomaly_score: float
|
|
top_frequencies: tuple[int, ...]
|
|
reason: str
|
|
recommendation: str
|
|
decision_source: str
|
|
|
|
|
|
def _clean_spectra(frame: pd.DataFrame) -> np.ndarray:
|
|
spectra = frame[FREQ_COLS].apply(pd.to_numeric, errors="coerce")
|
|
spectra = spectra.interpolate(axis=1, limit_direction="both")
|
|
fallback = spectra.median(axis=0)
|
|
spectra = spectra.fillna(fallback).fillna(0.0)
|
|
return spectra.to_numpy(dtype=float)
|
|
|
|
|
|
def _leave_one_out_median(spectra: np.ndarray) -> np.ndarray:
|
|
if len(spectra) < 2:
|
|
raise ValueError("At least two cylinders are required for an engine reference.")
|
|
reference = np.empty_like(spectra)
|
|
positions = np.arange(len(spectra))
|
|
for position in positions:
|
|
reference[position] = np.median(spectra[positions != position], axis=0)
|
|
return reference
|
|
|
|
|
|
def analyze_engine(result: DiagnosisResult, engine_id: str) -> EngineAnalysis:
|
|
measurements = result.engine_measurements(engine_id).sort_values("cylinder").reset_index(drop=True)
|
|
diagnostics = result.engine_diagnostics(engine_id).sort_values("cylinder").reset_index(drop=True)
|
|
if not measurements["cylinder"].equals(diagnostics["cylinder"]):
|
|
raise ValueError("Measurements and diagnostics are not aligned by cylinder.")
|
|
spectra = _clean_spectra(measurements)
|
|
reference = _leave_one_out_median(spectra)
|
|
deviation = spectra - reference
|
|
return EngineAnalysis(
|
|
engine_id=str(engine_id),
|
|
measurements=measurements,
|
|
diagnostics=diagnostics,
|
|
spectra=spectra,
|
|
reference=reference,
|
|
deviation=deviation,
|
|
absolute_deviation=np.abs(deviation),
|
|
)
|
|
|
|
|
|
def _severity_priority(label: str, severity: str) -> int:
|
|
"""Return a transparent ordinal triage level, not a probability."""
|
|
|
|
if label == "unknown":
|
|
return 1
|
|
if label not in FAULT_LABELS:
|
|
return 0
|
|
return {"male": 2, "srednie": 3, "duze": 4}.get(severity, 1)
|
|
|
|
|
|
def summarize_engine(analysis: EngineAnalysis) -> EngineSummary:
|
|
diagnostics = analysis.diagnostics
|
|
attention_mask = ~diagnostics["label"].eq("ok").to_numpy()
|
|
attention = int(attention_mask.sum())
|
|
named_fault = diagnostics["label"].isin(FAULT_LABELS)
|
|
has_unknown = diagnostics["label"].eq("unknown").any()
|
|
|
|
if (diagnostics.loc[named_fault, "severity"] == "duze").any():
|
|
status, tone = "KRYTYCZNY", "critical"
|
|
elif named_fault.any():
|
|
status, tone = "WYMAGA SERWISU", "warning"
|
|
elif has_unknown:
|
|
status, tone = "WYMAGA WERYFIKACJI", "unknown"
|
|
else:
|
|
status, tone = "SPRAWNY", "healthy"
|
|
|
|
observed = diagnostics.loc[named_fault, "severity"]
|
|
severity_order = ["duze", "srednie", "male"]
|
|
highest_severity = next(
|
|
(severity for severity in severity_order if observed.eq(severity).any()),
|
|
NOT_APPLICABLE,
|
|
)
|
|
|
|
ranking = rank_cylinders(analysis)
|
|
return EngineSummary(
|
|
engine_id=analysis.engine_id,
|
|
status=status,
|
|
status_tone=tone,
|
|
highest_severity=highest_severity,
|
|
highest_severity_display=SEVERITY_DISPLAY[highest_severity],
|
|
cylinders=len(diagnostics),
|
|
faults=int(diagnostics["label"].isin(FAULT_LABELS).sum()),
|
|
unknown=int(diagnostics["label"].eq("unknown").sum()),
|
|
attention=attention,
|
|
mean_confidence=float(diagnostics["label_confidence"].mean()),
|
|
top_cylinder=int(ranking.iloc[0]["cylinder"]),
|
|
)
|
|
|
|
|
|
def rank_cylinders(analysis: EngineAnalysis) -> pd.DataFrame:
|
|
diagnostics = analysis.diagnostics.copy()
|
|
diagnostics["triage_level"] = np.asarray(
|
|
[
|
|
_severity_priority(str(row.label), str(row.severity))
|
|
for row in diagnostics.itertuples()
|
|
],
|
|
dtype=int,
|
|
)
|
|
diagnostics["priority_display"] = diagnostics["triage_level"].map(
|
|
{
|
|
4: "Natychmiastowy",
|
|
3: "Wysoki",
|
|
2: "Planowy",
|
|
1: "Weryfikacja",
|
|
0: "Rutynowy",
|
|
}
|
|
)
|
|
diagnostics["label_display"] = diagnostics["label"].map(LABEL_DISPLAY)
|
|
diagnostics["severity_display"] = diagnostics["severity"].map(SEVERITY_DISPLAY)
|
|
return diagnostics.sort_values(
|
|
["triage_level", "anomaly_score_mean_abs_mv", "label_confidence", "cylinder"],
|
|
ascending=[False, False, True, True],
|
|
na_position="first",
|
|
).reset_index(drop=True)
|
|
|
|
|
|
def explain_cylinder(analysis: EngineAnalysis, cylinder: int) -> CylinderExplanation:
|
|
rows = analysis.diagnostics[analysis.diagnostics["cylinder"].eq(cylinder)]
|
|
if rows.empty:
|
|
raise KeyError(f"Unknown cylinder={cylinder}")
|
|
row = rows.iloc[0]
|
|
label = str(row["label"])
|
|
severity = str(row["severity"])
|
|
top_frequencies = tuple(
|
|
int(value)
|
|
for value in str(row["top_anomalous_frequencies_khz"]).split("|")
|
|
if value != ""
|
|
)
|
|
bands = ", ".join(f"{frequency} kHz" for frequency in top_frequencies)
|
|
anomaly = float(row["anomaly_score_mean_abs_mv"])
|
|
confidence = float(row["label_confidence"])
|
|
|
|
if label == "ok":
|
|
reason = (
|
|
f"Widmo pozostaje zgodne z profilem pozostałych cylindrów. "
|
|
f"Największe, nadal akceptowalne odchylenia występują przy {bands}."
|
|
)
|
|
elif label == "unknown":
|
|
reason = (
|
|
f"Cylinder wyraźnie odbiega od profilu silnika (średnio {anomaly:.1f} mV), "
|
|
f"szczególnie przy {bands}, ale wzorzec nie pasuje stabilnie do znanych usterek."
|
|
)
|
|
else:
|
|
reason = (
|
|
f"Charakter odchylenia widma przy {bands} jest najbardziej zgodny z klasą "
|
|
f"„{LABEL_DISPLAY[label]}”. Średnia różnica względem pozostałych cylindrów "
|
|
f"wynosi {anomaly:.1f} mV."
|
|
)
|
|
return CylinderExplanation(
|
|
cylinder=int(cylinder),
|
|
label=label,
|
|
label_display=LABEL_DISPLAY[label],
|
|
severity=severity,
|
|
severity_display=SEVERITY_DISPLAY.get(severity, severity),
|
|
confidence=confidence,
|
|
anomaly_score=anomaly,
|
|
top_frequencies=top_frequencies,
|
|
reason=reason,
|
|
recommendation=RECOMMENDATIONS[label],
|
|
decision_source=str(row.get("decision_source", "classifier")),
|
|
)
|