Jakub Famulski 2
dab5264ed7
All checks were successful
ENGIN CI / Build, test and smoke (push) Successful in 36s
430 lines
15 KiB
Python
430 lines
15 KiB
Python
"""ENGIN industrial diagnostic console built on the tested application core."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import html
|
|
import logging
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
import streamlit as st
|
|
|
|
from engin.charts import cylinder_spectrum, deviation_chart, engine_heatmap
|
|
from engin.config import (
|
|
LABEL_COLORS,
|
|
LABEL_DISPLAY,
|
|
SEVERITY_DISPLAY,
|
|
AppConfig,
|
|
)
|
|
from engin.errors import UserFacingError
|
|
from engin.explainability import (
|
|
analyze_engine,
|
|
explain_cylinder,
|
|
rank_cylinders,
|
|
summarize_engine,
|
|
)
|
|
from engin.io import PandasCsvReader
|
|
from engin.model import PrecomputedPredictionModel, SklearnPredictionModel
|
|
from engin.service import DiagnosisResult, DiagnosticService
|
|
from engin.validation import SpectrumFrameValidator
|
|
|
|
LOGGER = logging.getLogger("engin.app")
|
|
BASE_DIR = Path(__file__).resolve().parent
|
|
MODEL_VERSION = "engin-2026.08.25-1"
|
|
MODEL_ARTIFACT_DIR = BASE_DIR / "artifacts" / MODEL_VERSION
|
|
VIEW_OVERVIEW = "Przegląd"
|
|
VIEW_DETAIL = "Szczegóły cylindra"
|
|
PLOTLY_CONFIG = {"displayModeBar": False, "displaylogo": False}
|
|
DIAGNOSTIC_COLUMN_DISPLAY = {
|
|
"engine_id": "Silnik",
|
|
"cylinder": "Cylinder",
|
|
"label": "Diagnoza",
|
|
"severity": "Nasilenie",
|
|
"raw_model_label": "Surowa diagnoza modelu",
|
|
"decision_source": "Źródło decyzji",
|
|
"label_confidence": "Wynik diagnozy",
|
|
"raw_model_confidence": "Surowy wynik diagnozy",
|
|
"label_margin": "Margines decyzji",
|
|
"severity_confidence": "Wynik oceny nasilenia",
|
|
"anomaly_score_mean_abs_mv": "Średnie odchylenie bezwzględne [mV]",
|
|
"anomaly_ratio_to_engine_median": "Odchylenie względem mediany silnika",
|
|
"ood_absolute_threshold_mv": "Bezwzględny próg anomalii [mV]",
|
|
"ood_ratio_threshold": "Względny próg anomalii",
|
|
"top_anomalous_frequencies_khz": "Anomalne częstotliwości [kHz]",
|
|
"missing_spectral_cells": "Brakujące pomiary widma",
|
|
"model_version": "Wersja modelu",
|
|
"model_artifact_sha256": "Suma SHA-256 artefaktu",
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AppDependencies:
|
|
service: DiagnosticService
|
|
demo_payload: bytes
|
|
live_inference: bool = True
|
|
startup_warning: str | None = None
|
|
model_version: str = "unknown"
|
|
|
|
|
|
def _reader() -> PandasCsvReader:
|
|
return PandasCsvReader(AppConfig().max_upload_bytes)
|
|
|
|
|
|
@st.cache_resource(show_spinner="Ładowanie modelu diagnostycznego…")
|
|
def build_dependencies() -> AppDependencies:
|
|
reader = _reader()
|
|
validator = SpectrumFrameValidator()
|
|
demo_path = BASE_DIR / "test.csv"
|
|
demo_payload = demo_path.read_bytes()
|
|
try:
|
|
model = SklearnPredictionModel.from_artifact(
|
|
MODEL_ARTIFACT_DIR,
|
|
expected_model_version=MODEL_VERSION,
|
|
)
|
|
return AppDependencies(
|
|
service=DiagnosticService(reader=reader, validator=validator, model=model),
|
|
demo_payload=demo_payload,
|
|
model_version=MODEL_VERSION,
|
|
)
|
|
except Exception:
|
|
LOGGER.exception("Live model initialization failed; enabling demo fallback")
|
|
submission = pd.read_csv(BASE_DIR / "predictions.csv")
|
|
diagnostics = pd.read_csv(BASE_DIR / "prediction_diagnostics.csv")
|
|
fallback = PrecomputedPredictionModel(submission, diagnostics)
|
|
return AppDependencies(
|
|
service=DiagnosticService(reader=reader, validator=validator, model=fallback),
|
|
demo_payload=demo_payload,
|
|
live_inference=False,
|
|
startup_warning=(
|
|
"Artefakt modelu nie został załadowany. Aplikacja działa w bezpiecznym "
|
|
"trybie demonstracyjnym na zapisanych predykcjach."
|
|
),
|
|
model_version="tryb-demonstracyjny",
|
|
)
|
|
|
|
|
|
@st.cache_data(show_spinner="Analiza widm i klasyfikacja cylindrów…")
|
|
def diagnose_payload(_service: DiagnosticService, payload: bytes) -> DiagnosisResult:
|
|
return _service.diagnose_bytes(payload)
|
|
|
|
|
|
def _load_css() -> None:
|
|
css_path = BASE_DIR / "assets" / "app.css"
|
|
st.markdown(f"<style>{css_path.read_text(encoding='utf-8')}</style>", unsafe_allow_html=True)
|
|
|
|
|
|
def _render_error(error: UserFacingError) -> None:
|
|
st.error(f"**{error.title}**\n\n{error.message}")
|
|
if error.hint:
|
|
st.info(error.hint)
|
|
st.caption(f"Kod błędu: `{error.code}`")
|
|
|
|
|
|
def _format_confidence(value: float | int | None) -> str:
|
|
if value is None or pd.isna(value):
|
|
return "—"
|
|
return f"{100 * float(value):.0f}%"
|
|
|
|
|
|
def _diagnostics_for_display(frame: pd.DataFrame) -> pd.DataFrame:
|
|
display = frame.copy()
|
|
for column in ("label", "raw_model_label"):
|
|
if column in display:
|
|
display[column] = display[column].map(LABEL_DISPLAY).fillna(display[column])
|
|
if "severity" in display:
|
|
display["severity"] = (
|
|
display["severity"].map(SEVERITY_DISPLAY).fillna(display["severity"])
|
|
)
|
|
if "decision_source" in display:
|
|
display["decision_source"] = display["decision_source"].replace(
|
|
{
|
|
"classifier": "Klasyfikator spektralny",
|
|
"ood_override": "Reguła anomalii",
|
|
}
|
|
)
|
|
if "model_version" in display:
|
|
display["model_version"] = display["model_version"].replace(
|
|
{"demo-precomputed": "tryb demonstracyjny"}
|
|
)
|
|
return display.rename(columns=DIAGNOSTIC_COLUMN_DISPLAY)
|
|
|
|
|
|
def _render_header(live_inference: bool) -> None:
|
|
if live_inference:
|
|
mode = "MODEL GOTOWY"
|
|
verification = "ARTEFAKT ZWERYFIKOWANY"
|
|
tone = "verified"
|
|
else:
|
|
mode = "TRYB DEMONSTRACYJNY"
|
|
verification = "ZAPISANE WYNIKI"
|
|
tone = "fallback"
|
|
st.markdown(
|
|
f"""
|
|
<div class="product-header">
|
|
<div>
|
|
<div class="eyebrow">AESTEEL · DIAGNOSTYKA WTRYSKU DIESEL</div>
|
|
<h1>Konsola diagnostyczna ENGIN</h1>
|
|
<p>Diagnoza cylindra, nasilenie i następny krok.</p>
|
|
</div>
|
|
<div class="runtime-badge runtime-{tone}">● {mode}<br><span>{verification}</span></div>
|
|
</div>
|
|
""",
|
|
unsafe_allow_html=True,
|
|
)
|
|
|
|
|
|
def _render_summary(summary) -> None:
|
|
st.markdown(
|
|
f"""
|
|
<div class="status-strip status-{html.escape(summary.status_tone)}">
|
|
<div><span>STATUS SILNIKA</span><strong>{html.escape(summary.status)}</strong></div>
|
|
<div><span>NAJWYŻSZE NASILENIE</span><strong>{html.escape(summary.highest_severity_display)}</strong></div>
|
|
<div><span>WYMAGA UWAGI</span><strong>{summary.attention} / {summary.cylinders}</strong></div>
|
|
<div><span>ŚR. WYNIK MODELU</span><strong>{_format_confidence(summary.mean_confidence)}</strong></div>
|
|
</div>
|
|
""",
|
|
unsafe_allow_html=True,
|
|
)
|
|
|
|
|
|
def _render_engine_overview(analysis) -> None:
|
|
left, right = st.columns(2, gap="large")
|
|
with left:
|
|
st.markdown("### Mapa odchyleń")
|
|
st.plotly_chart(
|
|
engine_heatmap(analysis),
|
|
width="stretch",
|
|
key=f"heatmap_{analysis.engine_id}",
|
|
config=PLOTLY_CONFIG,
|
|
)
|
|
with right:
|
|
st.markdown("### Priorytet kontroli")
|
|
ranking = rank_cylinders(analysis).head(6).copy()
|
|
ranking["Cylinder"] = ranking["cylinder"].map(lambda value: f"C{int(value):02d}")
|
|
ranking["Diagnoza"] = ranking["label_display"]
|
|
ranking["Nasilenie"] = ranking["severity_display"]
|
|
ranking["Wynik modelu"] = ranking["label_confidence"].map(_format_confidence)
|
|
ranking["Priorytet"] = ranking["priority_display"]
|
|
st.dataframe(
|
|
ranking[["Cylinder", "Diagnoza", "Nasilenie", "Wynik modelu", "Priorytet"]],
|
|
hide_index=True,
|
|
width="stretch",
|
|
height=360,
|
|
)
|
|
|
|
|
|
def _render_cylinder_detail(
|
|
analysis,
|
|
cylinder: int,
|
|
comparison_cylinders: list[int],
|
|
) -> None:
|
|
explanation = explain_cylinder(analysis, cylinder)
|
|
row = analysis.diagnostics[analysis.diagnostics["cylinder"].eq(cylinder)].iloc[0]
|
|
color = LABEL_COLORS[explanation.label]
|
|
severity_confidence = row.get("severity_confidence", np.nan)
|
|
st.markdown(
|
|
f"""
|
|
<div class="diagnosis-card" style="--diagnosis-color:{color}">
|
|
<div>
|
|
<span class="diagnosis-kicker">CYLINDER {explanation.cylinder:02d}</span>
|
|
<h2>{html.escape(explanation.label_display)}</h2>
|
|
<p>{html.escape(explanation.severity_display)}</p>
|
|
</div>
|
|
<div class="diagnosis-metrics">
|
|
<div><span>Wynik diagnozy</span><strong>{_format_confidence(explanation.confidence)}</strong></div>
|
|
<div><span>Wynik oceny nasilenia</span><strong>{_format_confidence(severity_confidence)}</strong></div>
|
|
<div><span>Średnie odchylenie</span><strong>{explanation.anomaly_score:.1f} mV</strong></div>
|
|
</div>
|
|
</div>
|
|
""",
|
|
unsafe_allow_html=True,
|
|
)
|
|
st.markdown("#### Widma porównawcze")
|
|
st.plotly_chart(
|
|
cylinder_spectrum(analysis, cylinder, comparison_cylinders),
|
|
width="stretch",
|
|
key=(
|
|
f"spectrum_{analysis.engine_id}_"
|
|
+ "_".join(str(value) for value in comparison_cylinders)
|
|
),
|
|
config=PLOTLY_CONFIG,
|
|
)
|
|
|
|
st.markdown("#### Odchylenie cylindra głównego")
|
|
st.plotly_chart(
|
|
deviation_chart(analysis, cylinder),
|
|
width="stretch",
|
|
key=f"deviation_{analysis.engine_id}_{cylinder}",
|
|
config=PLOTLY_CONFIG,
|
|
)
|
|
|
|
why, next_step = st.columns(2, gap="large")
|
|
with why:
|
|
st.markdown("#### Uzasadnienie")
|
|
st.write(explanation.reason)
|
|
with next_step:
|
|
st.markdown("#### Następny krok")
|
|
st.write(explanation.recommendation)
|
|
source_label = (
|
|
"Reguła anomalii"
|
|
if explanation.decision_source == "ood_override"
|
|
else "Klasyfikator spektralny"
|
|
)
|
|
st.caption(
|
|
f"Źródło: {source_label} · wynik modelu nie jest prawdopodobieństwem."
|
|
)
|
|
|
|
|
|
def _render_technical(result: DiagnosisResult) -> None:
|
|
st.markdown("### Walidacja i model")
|
|
c1, c2, c3, c4 = st.columns(4)
|
|
c1.metric("Makro F1 (grupowe)", "0.981")
|
|
c2.metric("Trafność nasilenia", "0.930")
|
|
c3.metric("Punkty walidacyjne", "33.65 / 40")
|
|
c4.metric("Testy", "38 / 38")
|
|
st.caption(
|
|
"Walidacja grupowa według silników · średnia z 5 uruchomień · "
|
|
"makro F1 przy 5% braków: 0.983 · wnioskowanie lokalne na CPU."
|
|
)
|
|
st.markdown("#### Dane diagnostyczne")
|
|
st.dataframe(
|
|
_diagnostics_for_display(result.diagnostics),
|
|
width="stretch",
|
|
hide_index=True,
|
|
)
|
|
|
|
|
|
def render_app(dependencies: AppDependencies | None = None) -> None:
|
|
st.set_page_config(
|
|
page_title="Konsola diagnostyczna ENGIN",
|
|
page_icon="⚙️",
|
|
layout="wide",
|
|
initial_sidebar_state="expanded",
|
|
)
|
|
_load_css()
|
|
deps = dependencies or build_dependencies()
|
|
_render_header(deps.live_inference)
|
|
|
|
with st.sidebar:
|
|
st.markdown("## Centrum diagnostyczne")
|
|
source = st.radio(
|
|
"Źródło danych",
|
|
["Dane demonstracyjne", "Wgraj plik CSV"],
|
|
captions=["50 silników testowych", "Własne kompletne silniki 8/12/16"],
|
|
)
|
|
payload: bytes | None
|
|
if source == "Dane demonstracyjne":
|
|
payload = deps.demo_payload
|
|
st.success("Dane demonstracyjne załadowane")
|
|
else:
|
|
uploaded = st.file_uploader("Plik pomiarowy CSV", type=["csv"])
|
|
payload = uploaded.getvalue() if uploaded is not None else None
|
|
if payload is None:
|
|
st.info("Wgraj CSV, aby rozpocząć diagnozę.")
|
|
st.caption("Wymagane: kompletne silniki oraz mV_0...mV_20.")
|
|
|
|
if deps.startup_warning:
|
|
st.warning(deps.startup_warning)
|
|
if payload is None:
|
|
st.markdown("### Oczekiwanie na dane")
|
|
st.write("Po wgraniu pliku aplikacja zweryfikuje strukturę przed uruchomieniem modelu.")
|
|
return
|
|
|
|
try:
|
|
result = diagnose_payload(deps.service, payload)
|
|
except UserFacingError as exc:
|
|
_render_error(exc)
|
|
return
|
|
except Exception:
|
|
LOGGER.exception("Unexpected application failure")
|
|
st.error("**Nieoczekiwany błąd aplikacji**\n\nDiagnoza została bezpiecznie przerwana; dane nie zostały zmodyfikowane.")
|
|
st.info("Uruchom aplikację ponownie lub użyj zestawu demonstracyjnego.")
|
|
return
|
|
|
|
for warning in result.warnings:
|
|
st.warning(warning)
|
|
|
|
with st.sidebar:
|
|
engine_id = st.selectbox("Aktywny silnik", result.engine_ids)
|
|
st.download_button(
|
|
"Pobierz wyniki (CSV)",
|
|
data=result.submission.to_csv(index=False).encode("utf-8"),
|
|
file_name="predictions.csv",
|
|
mime="text/csv",
|
|
width="stretch",
|
|
)
|
|
st.download_button(
|
|
"Pobierz dane diagnostyczne (CSV)",
|
|
data=result.diagnostics.to_csv(index=False).encode("utf-8"),
|
|
file_name="prediction_diagnostics.csv",
|
|
mime="text/csv",
|
|
width="stretch",
|
|
)
|
|
st.divider()
|
|
st.caption(f"Wersja modelu: {deps.model_version}")
|
|
|
|
analysis = analyze_engine(result, str(engine_id))
|
|
summary = summarize_engine(analysis)
|
|
_render_summary(summary)
|
|
|
|
session_key = f"selected_cylinder_{analysis.engine_id}"
|
|
available = analysis.measurements["cylinder"].astype(int).tolist()
|
|
if session_key not in st.session_state or st.session_state[session_key] not in available:
|
|
st.session_state[session_key] = summary.top_cylinder
|
|
|
|
view_key = f"active_view_{analysis.engine_id}"
|
|
if view_key not in st.session_state:
|
|
st.session_state[view_key] = VIEW_OVERVIEW
|
|
view = st.segmented_control(
|
|
"Widok",
|
|
[VIEW_OVERVIEW, VIEW_DETAIL],
|
|
required=True,
|
|
key=view_key,
|
|
label_visibility="collapsed",
|
|
width="stretch",
|
|
)
|
|
|
|
if view == VIEW_OVERVIEW:
|
|
_render_engine_overview(analysis)
|
|
elif view == VIEW_DETAIL:
|
|
selector_col, comparison_col = st.columns([1.0, 2.0], gap="large")
|
|
with selector_col:
|
|
selected_from_box = st.selectbox(
|
|
"Cylinder główny",
|
|
available,
|
|
format_func=lambda value: f"Cylinder {value:02d}",
|
|
key=session_key,
|
|
)
|
|
comparison_options = [
|
|
cylinder for cylinder in available if cylinder != selected_from_box
|
|
]
|
|
with comparison_col:
|
|
additional_cylinders = st.multiselect(
|
|
"Porównaj z cylindrami",
|
|
comparison_options,
|
|
max_selections=3,
|
|
format_func=lambda value: f"Cylinder {value:02d}",
|
|
key=(
|
|
f"comparison_{analysis.engine_id}_{int(selected_from_box)}"
|
|
),
|
|
help="Cylinder główny jest zawsze pokazany; można dodać trzy kolejne.",
|
|
)
|
|
comparison_cylinders = [
|
|
int(selected_from_box),
|
|
*(int(value) for value in additional_cylinders),
|
|
]
|
|
_render_cylinder_detail(
|
|
analysis,
|
|
int(selected_from_box),
|
|
comparison_cylinders,
|
|
)
|
|
|
|
with st.expander("Informacje techniczne", expanded=False):
|
|
_render_technical(result)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
render_app()
|